上篇我们把 A2A 协议的基本定位、AgentCard 的作用、Task 状态机的雏形都过了一遍。今天这篇换一个角度，不再停留在“A2A 是什么、为什么需要它”，而是直接聊“怎么把它跑起来”。我会用一个最小可运行的示例把整个通信链路打通，再拆解任务状态流转、消息体设计、流式输出这些实操里避不开的细节，最后把我在真实项目里遇到过的高频坑整理成一张速查表。无论你是刚看完上篇准备动手，还是已经在调研多 Agent 协作方案，这篇都应该能帮你省下不少试错时间。
1. 先对齐基础：A2A 到底在解决什么问题
1.1 AgentCard：Agent 对外发布的“名片”
如果你已经看过 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 B，B 执行过程中发现用户意图其实是“退款”而不是“查询物流”，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_name="demo-agent", agent_url="http://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_id=str(uuid.uuid4()), role="agent", parts=[TextPart(text=f"已收到你的消息: {user_text}")], task_id=request.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_id="msg-001", role="user", parts=[TextPart(text="你好，请处理这个任务")], task_id="task-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( role="agent", 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 的分工与搭配
我在前一篇里提到过 MCP（Model 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, timeout=60): 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而不是message |
input-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 协作方案，那个话题比现在的轮询模式要更有意思得多。
