这两年AI Agent火得一塌糊涂但你会发现一个尴尬的现象每个Agent单拎出来都挺聪明放到一起却谁也不理谁。翻译Agent干完活不知道把结果丢给谁审校Agent想接手又听不懂对方在说什么。我们做集成的时候最头疼的就是让这些Agent互相通信每家都有自己的私有接口对接起来恨不得写一整套翻译层。A2A协议Agent-to-Agent Protocol就是冲着这个痛点来的它给Agent之间定了一套统一的“对话规矩”让不同团队、不同技术栈开发的Agent能直接协作。这篇文章我会从协议的核心概念讲起拆解任务流转的完整过程再给出一份可以直接跑通的最小协作Demo最后聊聊我在实测中踩过的坑。不管你是刚入门的开发者还是已经在做Agent集成的老兵跟着过一遍应该都能少走不少弯路。1. 先说清楚A2A到底解决什么问题1.1 现在的AI Agent都是“独行侠”单看一个Agent能力边界其实很清晰你给它一个目标它自己规划步骤、调用工具、产出一份结果。问题出在多个Agent需要配合的时候。我见过不少项目团队费了好大劲把Agent A的输出解析成JSON再写一堆胶水代码喂给Agent BA和B之间的状态同步全靠自己造轮子。这种点对点的集成方式每增加一个Agent复杂度就往上涨一截时间全耗在联调上业务本身反而没怎么推进。A2A想做的事情就是把这种“点对点私聊”变成“标准化群聊”。每个Agent只要暴露一个符合A2A规范的HTTP端点别的Agent就能通过统一的JSON-RPC方法给它派活、查状态、拿结果。你不需要关心对方是用什么框架写的也不需要为每个Agent单独定制通信格式。这就像两个人交流一个人说中文一个人说英文A2A就是那个同声传译大家按同一套规则来谁也不用迁就谁。1.2 A2A的前置拼图MCP先把工具层打通聊A2A之前必须提一下MCPModel Context Protocol。很多人会把这两个协议搞混其实它们分工完全不同。MCP解决的是Agent和工具之间的连接Agent要查数据库、发邮件、操作浏览器通过MCP就能统一调用这些外部能力。而A2A解决的是Agent和Agent之间的连接Agent发现另一个Agent能干活就把任务直接派过去还能持续跟踪任务进度。用生活化一点的话说MCP相当于给Agent装上了四肢让它能操控外部工具A2A相当于给Agent办了一张对外名片和一套接线规范让它可以光明正大地找别的Agent合作。两者是互补关系一个管“怎么干活”一个管“怎么协作”。在实际项目里你可能先用MCP让翻译Agent接入词典工具再用A2A让审校Agent调用翻译Agent的结果两个协议各管一段互不冲突。1.3 A2A的地位给Agent发一张“对外名片”A2A协议里有一个叫Agent Card的概念相当于每个Agent的“自我介绍”。这份元数据会写明Agent叫什么、能做什么、有哪些技能、请求端点在哪里、支持什么加密方式。客户端Agent拿到这份名片就能判断“这个Agent是不是我要找的”然后照着名片上的端点发起调用。这个设计特别像一个开放市场的公示牌。你想要找人翻译文档先看各家翻译服务的招牌Agent Card选中之后按照招牌上写的联系方式打电话过去HTTP调用对方接单后不断更新任务进度Task生命周期。整个过程中你不需要知道对方公司内部怎么组织员工干活只要确认它能接单、能交付就行。A2A把Agent之间的交互抽象成了这样一套标准化的“接单-干活-交付”模型这就是它最核心的价值。2. A2A协议的核心概念搞懂这几个词就够了2.1 Agent CardAgent的自我介绍Agent Card是A2A协议的入口也是所有协作发生前的第一站。它本质上是一个JSON文档里面至少包含这几块信息字段作用示例值nameAgent名称translate-agentdescription能力描述便于其他Agent理解中英双语翻译支持语气审校skills支持的技能列表translate、reviewendpoints通信端点协议类型和URIa2a: https://example.com/a2aauth端点认证方式api-key / oauth2客户端Agent拿到这份卡片后先读description判断能力匹不匹配再从endpoints里找到真正的调用地址。我个人习惯把Agent Card放在一个固定的公开路径比如/.well-known/agent.json这样别人只需要知道你服务的域名就能自动发现你的能力。字段的命名和结构在不同版本里略有差异但核心思想一直没有变让Agent的能力对外可见、可发现。2.2 Task与Message一次协作就是一个任务A2A把Agent之间的一次合作称为Task。比如“翻译这篇文章并检查语气”在协议层面就是一个Task。Task有自己独立的ID和状态从提交、工作中、等待补充信息到完成、失败或取消整个生命周期的变化都是可追踪的。这跟我们在项目管理系统里建工单是一个道理一个工单从创建到关闭每一步都有状态记录方便各方确认进度。Task运行过程中Agent之间通过Message来交换信息。Message是真正的“聊天内容”它由一组Part组成可以包含文本、文件或者结构化数据。你可以理解成Task是这场协作的“会话窗口”Message是对话里的一条条消息而Part就是每条消息里的具体内容片段。协议规定Message有一个owner字段标记这条消息是哪个Agent发出的避免了多方协作时“这话到底谁说的”这种低级混乱。2.3 Part与Artifact消息里装的到底是什么Part是Message的组成单元也是A2A协议里最底层的“通信字节”。它支持三种基本类型TextPart纯文本、FilePart文件、DataPart结构化数据。比如翻译Agent完成翻译后可以在一条Message里同时放一个TextPart翻译后的文本和一个DataPart包含原文字数、目标语言等元数据。接收端Agent不需要猜测对方发的是什么只要按Part类型逐个解析就行。Artifact则是任务运行过程中产生的结构化产物。这个概念容易和Message混淆我来区分一下Message是Agent之间交流用的即时消息Artifact是Task积累下来的最终成果。比如审校Agent在翻译结果上做了批注这些批注放在Message里发给翻译Agent最后集成好的审校报告就是这次Task的Artifact。协议允许Artifact里也包含多个Part方便同时交付文本、表格和附件。要记住的核心是Agent之间通过Message沟通但任务的价值最终落在Artifact上。3. 任务从头到尾是怎么流转的状态机与两种调用模式3.1 同步模式小任务一把梭A2A支持两种任务执行模式。第一种是同步模式客户端Agent发送message/send请求后服务端Agent会在同一个HTTP响应里直接返回任务最终状态。这种模式适合耗时短、结果一次性就能出来的场景比如“查一下明天的天气”、“算一道数学题”。它的优点是实现简单、调试方便一个请求响应就完成了整个协作。但同步模式有个天然缺陷如果任务执行时间超过HTTP请求的超时设置连接就会断开客户端拿不到结果。我实测下来凡是涉及外部API调用、模型推理较重的场景同步模式都很容易翻车。所以协议里还提供了第二种模式也是生产环境中更推荐的异步模式。3.2 异步模式长任务的正确姿势异步模式下客户端Agent发送任务请求后服务端立即返回任务ID和当前状态通常是working然后客户端通过两种方式跟进后续进度。第一种是轮询客户端定期调用task/get方法带着任务ID去问服务端“干完了没”。这种方式实现最简单只要支持最基本的方法就能用代价是会占用一些固定的请求量。第二种是推送客户端在发起任务时顺便订阅一个回调URL服务端任务状态一变化就主动往这个URL发送通知。这种方式的实时性最好也不会浪费无效轮询的请求但要求客户端的回调地址能从公网访问到内网开发时经常需要做内网穿透。我在实际项目里一般这么选内部Agent之间网络互通、延迟要求不高的用轮询跨团队、跨公网协作的用推送前提是先把回调的鉴权做好避免别人伪造通知。3.3 状态机一张表看清Task的一生Task的状态流转是A2A协议里最重要的机制调试Agent协作问题的时候几乎都是在看状态卡在哪一步。核心状态有六种状态含义常见场景submitted任务已提交等待处理刚收到请求还没开始干活working正在处理中模型推理中、外部API调用中input-required需要补充信息任务缺少必要参数等待输入completed任务成功完成已产出最终结果failed任务失败服务异常、模型报错canceled任务被主动取消用户中途放弃任务这些状态之间是可以来回切换的最典型的是input-required和working之间的循环。比如翻译Agent发现原文里有专有名词不知道怎么翻译它会先进入input-required状态向客户端Agent请求确认客户端Agent补充信息后它又回到working继续干活。这种设计让Agent之间的协作更接近真实团队的工作方式遇到卡点先沟通而不是一味埋头干活或者直接报错。我在做状态机联调时踩过最深的坑是服务端Agent忘了更新任务状态客户端一直等到超时才发现任务早就应该报错了。所以我自己写服务端的时候会强制在每个耗时操作的前后端都打点记录状态宁可多打几条日志也不能让任务状态“悬空”。4. 动手实现一个最小A2A协作Demo4.1 环境准备与项目结构讲再多理论不如直接跑一个Demo。这一节我做一个“翻译加审校”的小场景客户端Agent把任务派给翻译Agent翻译Agent干完活之后内部再调用审校Agent检查语气是否正式审校通过后把最终结果返回给客户端。为了降低理解门槛我不用重型框架就用Python的FastAPI加两个HTTP端点来模拟A2A协议的核心交互。生产环境你可以直接用现成的A2A SDK但先用最原始的方式实现一遍你会对协议运行机制有更直观的体感。环境准备只需要三样Python 3.10以上、FastAPI、以及一个能发HTTP请求的库我用httpx来写Agent内部的互相调用。项目结构大概是这样的a2a-demo/ agent_a_translate.py # 翻译Agent接收任务调用审校Agent agent_b_review.py # 审校Agent接收翻译结果输出审校意见 client.py # 客户端Agent发起整个协作流程4.2 实现服务端Agent翻译与审校先写审校Agent因为它的逻辑更简单只接收一条文本消息返回一段审校意见。按照A2A协议请求体走JSON-RPC 2.0格式method是message/sendparams里带着任务信息。# agent_b_review.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import uuid app FastAPI() app.post(/a2a) async def handle_review(request: Request): payload await request.json() if payload.get(method) ! message/send: return JSONResponse({jsonrpc: 2.0, error: {code: -32601, message: Method not found}, id: payload.get(id)}) params payload.get(params, {}) message params.get(message, {}) text for part in message.get(parts, []): if part.get(kind) text: text part.get(text, ) # 模拟审校逻辑 if urgent in text.lower(): review 语气偏急促建议增加礼貌用语补充截止时间说明。 else: review 语气总体正式个别地方可精炼。 task_id str(uuid.uuid4()) result { taskId: task_id, status: {state: completed}, artifacts: [ { kind: artifact, parts: [{kind: text, text: review}] } ] } return JSONResponse({jsonrpc: 2.0, result: result, id: payload.get(id)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8001)这段代码的核心点在于解析message.parts并从中提取文本这正好对应前面说的Part概念。服务端Agent只要处理好parts里的内容就能拿到对方传来的任务不需要关心上游是谁。任务完成后我构造了一个artifact结构把审校意见作为文本Part放进artifacts里返回。注意这里为了演示直接同步返回completed状态实际生产中更稳妥的做法是先返回working异步处理完再通过推送或轮询更新状态。4.3 Agent之间互相调用翻译Agent里串联审校Agent翻译Agent是这次Demo里最核心的部分因为它不仅要响应客户端请求还要在内部用A2A的方式调用审校Agent。这才能真正体现“Agent与Agent直接通信”而不是靠客户端做中间人转发。# agent_a_translate.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import httpx, uuid app FastAPI() app.post(/a2a) async def handle_translate(request: Request): payload await request.json() if payload.get(method) ! message/send: return JSONResponse({jsonrpc: 2.0, error: {code: -32601, message: Method not found}, id: payload.get(id)}) params payload.get(params, {}) message params.get(message, {}) text for part in message.get(parts, []): if part.get(kind) text: text part.get(text, ) # 模拟翻译 translated f[EN] {text} # 通过A2A调用审校Agent async with httpx.AsyncClient() as client: review_payload { jsonrpc: 2.0, id: str(uuid.uuid4()), method: message/send, params: { message: { owner: translate-agent, parts: [{kind: text, text: translated}] } } } resp await client.post(http://localhost:8001/a2a, jsonreview_payload) review_data resp.json() review_text result review_data.get(result, {}) artifacts result.get(artifacts, []) for artifact in artifacts: for part in artifact.get(parts, []): if part.get(kind) text: review_text part.get(text, ) task_id str(uuid.uuid4()) final_message ( f原文本{text}\n f翻译结果{translated}\n f审校意见{review_text} ) return JSONResponse({ jsonrpc: 2.0, result: { taskId: task_id, status: {state: completed}, artifacts: [{kind: artifact, parts: [{kind: text, text: final_message}]}] }, id: payload.get(id) }) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8002)这里有个细节值得关注翻译Agent在收到客户端请求之后自己在内部构造了一个新的message/send请求发向审校Agent的端点http://localhost:8001/a2a。这个请求完全符合A2A规范所以审校Agent根本不知道、也不需要知道发起方是谁。这个过程就是最纯粹的Agent间协作链条是“客户端 - 翻译Agent - 审校Agent”而不是客户端分别调用两个Agent再自己拼结果。4.4 跑通端到端协作最后写一个客户端脚本模拟用户把“请把这段话翻译成英文并检查语气是否正式”这个需求发给翻译Agent。# client.py import httpx import uuid async def main(): async with httpx.AsyncClient() as client: payload { jsonrpc: 2.0, id: str(uuid.uuid4()), method: message/send, params: { message: { owner: client-agent, parts: [{kind: text, text: 请把这段话翻译成英文并检查语气是否正式周末之前务必全部完成任务。}] } } } resp await client.post(http://localhost:8002/a2a, jsonpayload) result resp.json().get(result, {}) artifacts result.get(artifacts, []) for artifact in artifacts: for part in artifact.get(parts, []): if part.get(kind) text: print(最终结果) print(part.get(text)) if __name__ __main__: import asyncio asyncio.run(main())跑起来的方式很简单先启动审校Agent端口8001再启动翻译Agent端口8002最后运行客户端脚本。python agent_b_review.py python agent_a_translate.py python client.py整个流程跑通以后你会看到客户端打印出翻译结果和审校意见。这个Demo麻雀虽小五脏俱全客户端发起任务、服务端接收任务、Agent内部链式调用另一个Agent、最终结果以Artifact的形式返回A2A协议的核心交互全过了一遍。5. 常见问题与排查技巧实录5.1 任务卡在working状态不动这是我被问得最多的问题。服务端Agent确实收到了请求但任务状态一直是working客户端等到超时也没有拿到结果。最常见的原因有两个一是服务端在处理任务过程中出现了未捕获的异常导致状态更新逻辑没有执行二是任务里有耗时很长的外部调用但服务端没有实现异步推送客户端轮询周期又不匹配。排查这类问题我一般按三步走先看服务端日志确认任务处理流程是否真的执行完了再curl手动调一次A2A端点检查请求返回的JSON结构是否符合规范最后看客户端拿到的任务ID是否跟服务端处理任务ID一致很多时候是ID拼接错误导致查不到状态。还有一种隐蔽情况就是服务端在返回结果里把status.state拼错了比如completed写成了complete客户端只认规范里的标准枚举值拼错直接当成无效状态。5.2 推送回调收不到通知异步模式下配置了推送URL但服务端状态变化时客户端根本收不到通知。这个问题九成是网络穿透没做好导致的。开发环境里的回调地址经常是localhost或者内网IP服务端在公网根本访问不到。我自己用内网开发的时候会临时开一个隧道工具把本地端口暴露出去把公网地址填到推送配置里回调就能通了。第二类原因是回调签名校验不过。规范里允许在推送通知上附加签名信息客户端收到回调后要校验签名如果密钥配置不一致回调会被静默丢弃。这种情况服务端日志里可能根本看不到痕迹因为请求已经发出去了是客户端主动拒绝处理。排查的时候一定要在客户端回调入口加日志先确认“有请求进来没”再去查“是不是校验失败了”。5.3 消息格式对不上双方都在甩锅A2A规范迭代速度不慢不同版本的方法名和字段结构会有差异。比如早期版本和后续版本对任务状态枚举的命名不完全一致某些方法在迭代过程中被拆分或改名。这就导致一个客户端Agent用新规范发请求老版本的服务端Agent却解析不了。我的建议是做一个集中式的协议适配层避免每个Agent各自处理协议版本问题。方法是在团队里维护一个公共库统一封装message/send、task/get这些核心方法版本升级时只改这个公共库各个Agent业务逻辑完全不用动。实测下来这套做法能显著降低联调成本尤其是Agent数量超过五个之后收益会非常明显。另外强烈建议在Agent Card里声明兼容的协议版本号方便客户端Agent做能力探测。客户端拿到Agent Card后先检查版本不匹配就直接提示不要等到发请求之后才报错。这个“先探测后调用”的习惯能帮你省掉大量无意义的联调时间。5.4 Agent Card能力描述含糊导致匹配失败有些团队写Agent Card特别敷衍description就一句“提供翻译功能”客户端Agent根本判断不了这个Agent到底适不适合当前任务。我见过一个项目客户端Agent因为description太泛把文本摘要任务发给了一个翻译Agent两边在input-required状态来回拉扯了好久。写Agent Card的时候description一定要把输入、输出、场景边界说清楚。比如“根据用户提供的原文和目标语言代码返回翻译文本及译文质量评分适用于正式文档和日常对话”就比“提供翻译功能”好得多。skills字段也要细化不要只写一个“translate”可以拆成“translate-zh-en”、“translate-en-zh”、“tone-review”等多个技能让客户端Agent能精确匹配。最后再分享一个实操心得从我个人的实践经验看A2A协议最适合的切入点不是一上来就把所有Agent接入协议而是先挑一条最简单的链路做试点。像我这篇的翻译加审校Demo十分钟就能跑通跑通之后你才会真正理解状态、消息、产物这几个概念在真实场景里是什么感觉然后再去对接复杂业务心里就有底了。还有一个小提醒Agent之间协作的时候消息内容尽量保持结构化别把一堆文本塞进一个TextPart里完事。比如翻译结果和审校意见分开放在不同的Part中下游Agent解析起来就清爽得多也不会因为你的文本格式一变就跟着崩溃。A2A还在快速演进中现在学会这套通信思路后面版本怎么升级你都能跟得上。