上篇我们把 A2A 协议的基本定位、AgentCard 的作用、Task 状态机的雏形都过了一遍。今天这篇换一个角度不再停留在“A2A 是什么、为什么需要它”而是直接聊“怎么把它跑起来”。我会用一个最小可运行的示例把整个通信链路打通再拆解任务状态流转、消息体设计、流式输出这些实操里避不开的细节最后把我在真实项目里遇到过的高频坑整理成一张速查表。无论你是刚看完上篇准备动手还是已经在调研多 Agent 协作方案这篇都应该能帮你省下不少试错时间。1. 先对齐基础A2A 到底在解决什么问题1.1 AgentCardAgent 对外发布的“名片”如果你已经看过 A2A 协议规范一定对 AgentCard 不陌生。它就是每个 Agent 暴露给外界的元数据描述文件通常是一个 JSON 结构里面写着这个 Agent 的能力名称、支持的任务类型、输入输出格式、认证方式、接口地址等。客户端通过访问/.well-known/agent-card.json这类固定路径拿到卡片就知道对面这个 Agent 支持什么、不支持什么、该怎么调用。我第一次接触这个设计时觉得很简单像极了公司前台给访客递的访客卡。但实际写代码时才发现AgentCard 的 schema 版本、字段命名、context 字段的兼容性都是坑。比如有些实现早期版本没有version字段导致客户端无法判断它对端协议版本是 0.1 还是 0.2直接用旧字段解析会静默失败。所以如果你们团队的多 Agent 系统要对接到不同团队开发的 Agent务必让 AgentCard 自己带有明确的版本信息并在客户端做 schema 校验而不是裸解析 JSON。AgentCard 在大多数情况下的获取方式是 HTTP GET这决定了 A2A 天然是“以 HTTP 为传输底座”的。这一点很重要因为它意味着我们现网所有负载均衡、限流、监控、鉴权体系都可以直接用不需要另起一套通信中间件。对比一下如果所有 Agent 都走消息队列或 RPC 框架虽然延迟可能更低但基础设施改造成本马上翻倍。1.2 Task 状态机A2A 真正值钱的部分很多没有深入看过 A2A 的人有个误解以为它就是 “JSON 格式的 Agent API”。其实 A2A 最有价值的部分是 Task 状态机。一个 Task任务产生以后不会因为 HTTP 请求结束就消失它会在服务端维持自己的生命周期客户端可以反复查询状态直到任务进入终态。标准的状态流转大概是这样的request - submitted - working - input-required / completed / failed / canceled / rejected其中submitted表示任务已被接受working表示正在执行input-required表示 Agent 执行到一半发现需要用户更多输入completed、failed、canceled、rejected是四个终态。为什么这个设计值钱因为它把 Agent 之间“你问一句我答一句”的轻量交互升级成了“你委托一个任务我去异步执行然后同步结果”的生产级交互。这对编排复杂流程非常关键。举个例子Agent A 把自己的用户转给 Agent BB 执行过程中发现用户意图其实是“退款”而不是“查询物流”B 不能直接瞎编要把状态置为input-required回传一段明确的问题再等 A 把用户的补充信息拿回来。这比传统的回调 URL 方式清晰得多传统的做法往往是回调触发后对方再发起一个新任务旧任务和新任务之间根本没有关联。Task 状态机的一个直接收益就是一个任务从头到尾是同一个 ID、同一份上下文排障信息完整。1.3 消息的角色与 Part 设计在 A2A 中Message是真正承载语义的对象它有role字段——user和agent。这个设计有一点值得注意多 Agent 协作时消息的发起方不一定是真实用户。比如“用户先问天气Agent A 再找 Agent B 订餐厅”Agent A 是以agent角色向 Agent B 发消息的但消息体里同时通过context字段携带了真实用户身份否则 B 不知道自己在替谁服务。每个 Message 里的parts是一个数组支持TextPart、FilePart、DataPart等不同内容类型。我在项目里干脆把parts当成一个“通用信封”文本是文字文件是字节引用结构数据是 JSON。这样消息可以灵活承载多种形态的内容而不像普通的 REST 接口参数定死了就很难扩展。具体怎么设计结合后面第 3 章实战部分讲。2. 跑通一个最小 A2A 通信示例理论讲太多没有用先把 Demo 跑起来。我假设你用的 Python 3.10当前 A2A 的官方 SDK 更新比较勤直接装最新版本就好。如果你在公司环境不方便直接用 PyPI也可以把协议仓库里的核心代码拷到本地毕竟 A2A 本质是协议按 JSON 实现并不复杂。2.1 环境准备与依赖安装python -m venv venv source venv/bin/activate pip install a2a-sdk fastapi uvicorn如果a2a-sdk装不上fastapi和uvicorn一定没问题因为最终跑服务还得靠 FastAPI。此外建议装一个pydantic的较新版本SDK 里的类型声明依赖它。提示如果你在公司内网环境记得先确认 pip 源是否同步到了最新版本否则可能装到旧版 SDK接口签名不一致导致代码跑不起来。2.2 服务端注册 Agent 并实现任务处理逻辑下面这段代码是一个最小可跑的服务端。功能很简单接收一个message提取其中的文本返回“接收到了”的文本结果。import uuid from fastapi import FastAPI from a2a.types import Task, Message, TextPart, TaskStatus from a2a.server import InMemoryAgentServer app FastAPI() agent_server InMemoryAgentServer( agent_namedemo-agent, agent_urlhttp://127.0.0.1:8000, auth_token, ) agent_server.task_handler async def handle_task(request): # 从请求里取出用户消息 user_message request.message user_text .join( part.text for part in user_message.parts if isinstance(part, TextPart) ) # 构造一个回复消息 reply Message( message_idstr(uuid.uuid4()), roleagent, parts[TextPart(textf已收到你的消息: {user_text})], task_idrequest.task_id, ) return reply app.include_router(agent_server.router)注意task_handler是异步函数。在真实项目中这里的业务逻辑往往是一段耗时的处理。如果处理时间比较长直接这个函数内部完成后才返回也说得通但更好的方式是把任务先存入队列立刻返回Task后续再更新状态。因为 A2A 协议允许我们同步返回一个尚未完成的 Task客户端会持续轮询状态。2.3 客户端发现 Agent、发送任务、轮询结果客户端逻辑更像我们日常调用方先拉 AgentCard 拿到入口信息再发送消息生成任务然后轮询直到终态。import asyncio from a2a.client import A2AClient from a2a.types import Message, TextPart async def main(): client A2AClient(http://127.0.0.1:8000) # 获取 AgentCard card await client.agent_card() print(agent 名称:, card.name) # 构造一条用户消息并发送 msg Message( message_idmsg-001, roleuser, parts[TextPart(text你好请处理这个任务)], task_idtask-001, ) task await client.send_task(msg) print(任务初始状态:, task.status) # 轮询直到完成 while task.status not in [TaskStatus.COMPLETED, TaskStatus.FAILED]: await asyncio.sleep(0.5) task await client.get_task(task.id) # 取结果 if task.status TaskStatus.COMPLETED: for part in task.artifacts[0].parts: if isinstance(part, TextPart): print(最终结果:, part.text) asyncio.run(main())这段代码并不复杂但你亲手跑一次之后会意识到两个关键点第一task.id是 A2A 协议的全局关联键。所有后续查询都必须拿着这个 ID。所以客户端侧必须做好 ID 的保存不能直接把任务结果当 HTTP 响应用完就扔。第二message_id最好用 UUID 或具备唯一性的 ID。这个字段用作幂等防止同一个消息被重复执行。我在一个项目里因为偷懒用了固定字符串结果一个定时重试任务被执行了两次下游业务直接炸了。2.4 运行与验证终端开两个窗口一个跑uvicorn main:app --port 8000另一个跑客户端脚本。能看到agent 名称: demo-agent和最终的最终结果: 已收到你的消息: 你好请处理这个任务就算通了。这个 Demo 大概 60 行代码完整覆盖了“发现 Agent - 发送任务 - 查询状态 - 获取结果”的闭环。别急着往下看建议先在本机把这个跑通。因为后面所有进阶玩法——流式输出、多 Agent 编排、文件传输——都是在这个闭环上做加减法。3. 掌握状态机与消息设计的细节Demo 能跑只代表链路通了。真正决定系统稳不稳的是对状态机和消息体的细致处理。这一章我把实操中容易搞错的地方逐个拆开讲。3.1 状态流转input-required 到底怎么用input-required是我认为 A2A 最容易被忽略但最实用的状态。它解决的是“Agent 执行过程中发现条件不足需要回到用户那边补充信息”的场景。比如你做客服机器人Agent A 负责接待用户问“我要退款”A 发现自己没有退款权限就把这个退款意图提交给 Agent B。B 经过分析发现需要用户提供订单号就把任务状态置为input-required同时在返回的消息里附上一个TextPart“请提供订单号”。这时候 Agent A 作为任务发起方看到状态是input-required就知道要回去问真实用户要订单号。等用户给了订单号A 再向 B 发送一条新的消息B 重新进入working状态继续处理。在代码上服务端处理函数内部通常是这样if not order_id: task.status TaskStatus.INPUT_REQUIRED task.message Message( roleagent, parts[TextPart(text请提供订单号)] ) return task这里有个隐晦的细节input-required不是错误它是正常业务语义的一部分。很多团队把一切非completed状态都当异常处理结果用户明明只差一个字段却被判定成“服务不可用”。排查系统问题时这一类误判往往很隐蔽。3.2 流式输出与增量更新A2A 协议支持流式输出非常适合大模型场景下的逐字回复。流式输出的核心是两个概念lastEvent和event流。每次客户端查询get_task时可以传入lastEventId服务端只返回从该事件之后的新事件减少重复传输。同时服务端可以通过events: text delta这类事件流类型推送增量文本。我最初以为流式输出就是 WebSocket 或者 SSE 推消息但 A2A 的做法更克制它允许你通过查询接口拿到事件流。换句话说即使底层只用了普通 HTTP 轮询协议层也能模拟出流式效果只是频率不够高的话体验不够丝滑。在实现时服务端要注意给事件加上自增 ID否则客户端无法判断增量从哪开始。我曾经遇到一个现象客户端拿到的事件 ID 居然重复了排查了半天发现是服务端把“事件序号”和“消息序号”放在同一个字段里更新。后者的 ID 按 task 维度重置当然会出现重复。3.3 消息体的 Part 设计不止是文本Part 有不同类型最常用的三种TextPart纯文本适合绝大多数对话消息。FilePart文件引用字段里包含文件名、MIME 类型、字节内容或下载链接。DataPart结构化 JSON 数据适合传输业务对象。在设计多 Agent 协作的消息体时我建议把“面向显示的内容”和“面向程序的数据”分开。比如“用户上传了一张身份证照片”这个场景FilePart负责传递图片本身DataPart负责传递识别出来的结构化身份信息。两者放在同一个parts数组里接收方可以选择性地消费它需要的部分而不是强制解析一套包含所有信息的自定义 JSON。3.4 文件附件与用户上下文传递在 Agent 之间传递文件有一点得特别注意A2A 的FilePart里的bytes字段虽然可以内联文件内容但我不建议真的这么干。比如一个动辄几十 MB 的 PDF直接塞进 JSON 请求既浪费带宽又容易触发网关的请求体大小上限。更可靠的做法是文件先落到对象存储或服务器临时目录FilePart里写文件的下载 URL接收方按需拉取。我把这个方案讲给一个团队听时有人问那不是要自己实现文件服务其实不用很多文件服务器、对象存储都自带临时签名 URL 的能力A2A 只需要传递一个带时效的链接就够了。用户上下文方面A2A 的Message.context是“带外信息”的容器。里面可以放user_id、device_id、会话 ID 等。这个字段的作用简单理解就是“委托任务时把用户的户口本复印件带上”让下游 Agent 知道自己在替谁服务。注意如果真实用户的隐私信息敏感建议只传 ID 或经过脱敏的标识而不是全量个人信息。4. 多 Agent 协作的三种编排思路与工程落地单 Agent 通信讲完了现在聊多 Agent 协作。只靠协议你依然可以设计出五花八门的协作方式。我从工程落地的角度整理三种模式并聊聊各自的适用场景。4.1 单播路由让 Agent 收到它该收的任务最常见的方式是由一个“路由 Agent”接收用户请求根据请求内容分发给下游 Agent。这和微服务架构里的 API 网关特别像。路由 Agent 手上有各下游 Agent 的 AgentCard可以根据关键字、意图分类、甚至语义向量把请求转发给合适对象。这种模式的好处是收敛入口用户只面对一个 Agent。坏处是路由 Agent 可能成为性能瓶颈也可能因为路由判断错误把请求发给错误的 Agent导致来回踢皮球。我建议在路由层做一层“意图置信度”判断低于某阈值时不要硬转发而是直接让用户重述需求。4.2 广播与订阅当一个任务需要多个 Agent 并行参与某些场景下一个任务需要多个 Agent 各自处理一部分。例如“为一篇产品稿生成配图和摘要”写作 Agent 负责摘要制图 Agent 负责画图。这种场景可以把一个主任务拆成多个子任务并行发给多个 Agent最后做一个汇聚。A2A 对广播模式没有特殊支持它只提供“一对一任务”的模型。所以广播模式本质上属于编排层逻辑。你在主任务里持有子任务 ID 和对应 Agent 的列表等所有子任务终态后再汇总结果。实践中有两个细节要留意一是子任务的超时时间要分别管理某个 Agent 挂了不能阻塞整个主任务的汇聚二是汇聚阶段要注意产出去重多个子任务可能返回内容上有重叠。4.3 链式传递A2A MCP 的分工与搭配我在前一篇里提到过 MCPModel Context Protocol很多读者问 A2A 和 MCP 是不是竞争关系。我的理解是MCP 解决的是“大模型怎么调用工具和数据”A2A 解决的是“Agent 之间怎么交接任务”。两者天然可以组合使用。举一个组合的例子在一个售后流程里售后 Agent 通过 MCP 调用订单系统查询订单详情当你发现订单来自另一个业务线它不能查询时就把任务整体委托给那个业务线的 Agent。委托过程用 A2A内部查数据用 MCP。A2A 管跨系统协作MCP 管单 Agent 内的能力扩展分工明确。组合使用时建议在 A2A 的context里带上“需要的工具白名单”或“MCP 会话标识”避免下游 Agent 重新初始化一堆连接。4.4 可靠性设计幂等、超时与重试多 Agent 系统一旦上了生产可靠性比功能丰富度重要得多。我总结了三个必须做的设计第一消息幂等。前面提过message_id的唯一性这里再强调一次发送消息前应该检查是否已存在相同message_id的任务如果存在直接返回原有的 Task而不是新建。这样可以挡住网络重试导致的重复执行。第二任务超时。A2A 的get_task查询是同步的但如果服务端长时间不返回终态客户端不能无限等待。建议所有轮询设置一个总超时上限超时后先发cancel请求再标记任务失败。第三重试有退避。轮询频率不要写死可以采用指数退避前几次 0.5 秒然后 1 秒、2 秒、4 秒最大不超过 10 秒。这既保证了短任务反馈及时又不会给服务端造成无谓压力。下面是一个简化的轮询逻辑示意async def wait_task_completed(client, task_id, timeout60): deadline time.time() timeout delay 0.5 task await client.get_task(task_id) while task.status not in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELED]: if time.time() deadline: await client.cancel_task(task_id) raise TimeoutError(任务超时) await asyncio.sleep(delay) delay min(delay * 2, 10) task await client.get_task(task_id) return task5. 线上排查与避坑速查多 Agent 系统的一个麻烦是问题往往跨模块发生。消息明明发出去了对方说没收到对方说处理完了调用方却拿不到结果。这类问题排查起来非常耗时。我把自己的经验整理成一张速查表按症状、可能原因、处理办法排列。5.1 高频问题速查表症状可能原因处理办法客户端读取 AgentCard 失败URL 配置错误AgentCard 路径不符合规范检查/.well-known/agent-card.json是否可访问确认协议版本消息发送后没有任何 Task 返回服务端抛异常但未捕获路由方法未注册查看服务端日志确认task_handler已注册轮询长期停在working服务端处理函数内部阻塞任务未更新状态检查异步函数是否用了同步阻塞调用显式更新 Task 状态状态已completed但artifacts为空服务端返回了 Message 但没有生成 artifact确认是否把回复放到了artifacts而不是messageinput-required后无法恢复客户端不回传补充信息服务端不重新进入working确认补充消息发送后状态是否被置回working消息被重复处理message_id重复缺少幂等校验改用 UUID服务端增加按message_id查重流式输出事件 ID 重复事件序号和消息 ID 混用单独维护事件自增序列这张表看着简单但每一条背后我都踩过雷。最典型的是“状态已 completed 但 artifacts 为空”一开始我以为是自己代码写错了后来发现是 A2A 服务端 SDK 对“消息回复”和“任务产物”是两个概念前者回给用户后者作为任务交付结果。如果你把结果写在message里而不是artifacts里客户端按 artifacts 拿结果自然是空的。5.2 性能调优轮询频率与连接管理很多人一上来就问A2A 任务全靠轮询是不是效率很低其实 A2A 同时支持服务端推送事件流如果对实时性要求高可以让服务端通过events通道把状态变更推送出来客户端订阅事件流而不是高频轮询。但推送通道往往需要额外的前置服务支撑。如果你们团队基础设施以普通 HTTP 为主轮询配合指数退避完全够用。这里有一个可以日常注意的优化点连接复用。A2AClient底层用的是异步 HTTP 客户端如果你在一个循环里反复发任务尽量复用同一个 client 实例而不是每发一条消息就 new 一个。否则每次都要重新握手指纹认证性能开销成倍增长。5.3 安全与权限边界A2A 协议本身允许在 AgentCard 上声明认证方式常见的像 Bearer Token、OAuth 等。但这只是传输层的认证业务权限还需要在 Agent 内部自己控制。我特别想提醒的是不要把所有 Agent 的密钥放在同一个环境的同一个文件里。在一个真实项目里我把三个 Agent 的 Token 都写在一个.env后来一个 Agent 的服务端日志泄露直接导致另外两个 Agent 被扫。正确做法是按 Agent 拆分配置来源并尽快把 Token 换成短时有效的临时凭据。另一个安全细节是多 Agent 之间传递文件链接时一定要设置有效期和访问限制。曾经有内部文档的下载链接因为没有过期时间被下游 Agent 缓存后泄露给无关人员。问题不在 A2A 协议在于实现方的落地姿势。协议能力再强最后还是要人写得对。5.4 调试工具与测试建议多 Agent 调试比单进程调试麻烦很多。我目前的习惯是本地启动多个服务端实例每个实例端口不同AgentCard 里的url直接指向本地端口。用curl直接打 AgentCard 接口验证卡片内容是否符合预期。客户端侧增加“回放”能力把每次任务请求的 request/response 落日志出问题时可以按task_id拉出完整链路记录。如果你用 FastAPI 作为服务端FastAPI 自带的/docs接口非常方便可以直接在浏览器里调试每个 Agent 暴露的 HTTP 接口。但注意这不等于验证 A2A 协议正确只能验证 HTTP 层可用。真正的协议验证还是要走一个真实的 A2A 客户端。6. 我实际操作中踩过的几个坑前面零零散散提了一些最后一章集中讲我在真实项目里踩过的、且不太好在文档里找到答案的几个坑。第一个坑是 AgentCard 缓存。我们有一个多 Agent 平台各 Agent 的卡片由中心化配置管理每次更新配置后需要 Agent 重新拉取卡片。一开始缓存时间是 24 小时结果下游 Agent 一直调用旧接口上线新功能的 Agent 根本等不到流量。后来我把卡片缓存时间改短同时在更新配置时主动往下游服务推一个“卡片变更通知”。A2A 规范并没有规定卡片缓存策略但生产系统一定要把“更新传播”这件事当一等公民来做。第二个坑是任务并发没有加隔离。我们的服务端用InMemoryAgentServer任务处理函数里直接改共享内存字典。上线后发现两个并发任务互相覆盖状态查了半天才发现是同一个 dict 被多个协程同时写。解决办法很简单给共享状态加一个异步锁或者换用支持序列化的存储。这里多说一句真要上生产不要用内存存任务状态进程一重启任务全丢了至少换成 Redis。第三个坑是Task的终结状态判空。有的 SDK 版本里failed状态任务的error字段可能是空的只有状态码。如果客户端只检查error字段来判断失败原因就会拿到一堆“只有状态但没原因”的废日志。正确做法是在客户端侧给所有非completed的终态都设置兜底错误信息并尽早把服务端返回的原始对象保存下来方便定位。第四个坑是我自己对“流式输出”抱有错误预期。我一开始以为只要服务端生成了一点内容客户端马上就能看到。实际测试发现如果没有事件推送通道光靠轮询查询用户感知到的延迟仍然是轮询间隔的一半以上。后来在交互场景里我直接把轮询间隔调短到 200ms效果接近实时但代价是服务端请求量上升。最终才想明白如果产品要求 ChatGPT 那种逐字效果应该走事件通道而不是轮询。回到开头的那个 Demo。你细心一点会发现整个 A2A 链路本质上是在“任务”这个抽象上构建协作。无论你的 Agent 是文本对话、写代码、画图还是查数据库只要它遵守 AgentCard、Message、Task 这三板斧就能被其他 Agent 用统一的方式发现、调用和追踪。我个人在实际项目中体会最深的一点是A2A 最不需要纠结的反而是“协议”真正要花心思的是每个 Agent 的能力边界和任务语义划分。协议给了一个相对标准的协作底座但每个 Agent 在什么情况下该把什么任务交给谁这永远是业务设计问题而不是技术问题。如果你正准备引入多 Agent 协作建议不要一上来就铺太多 Agent。先用两三个 Agent 把最小闭环跑通把状态流转、幂等、超时这些基本功做扎实再逐步扩展。等到你需要在十个以上 Agent 之间编排任务时你会庆幸当初花了时间把消息设计和状态管理打磨得足够干净。后续如果大家感兴趣我还可以专门写一篇基于事件通道的高实时性 Agent 协作方案那个话题比现在的轮询模式要更有意思得多。