今年做多智能体项目时被一个老问题折磨得不轻Agent 用不同框架开发通信方式全靠自己写胶水代码A 系统的输出 B 系统读不懂光是消息格式对齐就来回改了三天。后来社区里讨论 Google 发布的 A2A 协议官方配套的 Python 包a2a-protocol也放了出来。我第一时间在本地搭了个 Demo 跑通实测下来它对“让不同的 Agent 互相打电话”这件事的解决方式是系统性的不止给了协议规范还直接给了现成的服务端、客户端和类型定义。这篇文章不堆概念只讲三件事这个包的语法结构、核心参数含义以及一个能直接复现的实战案例。如果你也在做 Agent 间的互操作这篇应该能帮你少走不少弯路。1. 先搞清楚 a2a-protocol 解决什么问题1.1 Agent 之间为什么需要一套“通话协议”单个 Agent 的能力再强也只是一个孤岛。真正复杂的业务场景里可能需要一个“翻译 Agent”先做语义理解再把任务转给“代码生成 Agent”最后由“质检 Agent”检查结果。这种协作结构一出现就立刻面临一个基础问题Agent 之间用什么格式沟通最早的方案是自己定义 JSON。今天你定义一个{sender, receiver, content}明天对方框架要{from, to, payload}两套系统对接就要写字段映射。更麻烦的是任务状态的跟踪——这个任务到底完成了没有、中间有没有产生中间产物、失败原因是什么这些问题各家实现都不一样。A2A 协议解决的正是这一层用一套标准化的消息格式和任务生命周期让不同 Agent 之间能互相发现、互相调用、追踪进度。a2a-protocol包就是这个协议的 Python 实现。它不是某个具体平台的私有 SDK而是对协议规范的代码落地里面包含通信双方所需的核心组件。你不需要自己实现 JSON-RPC 协议细节也不需要反复发明任务状态机的轮子直接把包装进来定义好 Agent 的能力就能对外提供标准的服务。1.2 协议的工作方式与消息模型理解了它解决什么问题再看它的工作方式就顺了。A2A 底层走的是 JSON-RPC 2.0 这套成熟的远程调用规范传输层用 HTTP同时支持 SSE 做流式推送。消息模型的核心是 Task任务Task 内部挂载 Message消息、Artifact产物和 Part内容片段。我给你打个生活化的比方。把 Agent 当成一家公司的客服专员Task 就是一次工单客户发起工单客服接单中间可能实时回复进展Message最终交付一份文档Artifact。工单有明确的状态进行中、完成、失败、取消。这套模型天然适合 Agent 之间有来有往的协作场景。包的代码结构也跟这个模型对齐。a2a.types里就是这些消息类型定义a2a.server提供服务端框架负责接收 JSON-RPC 请求a2a.client是客户端负责发起调用。后面两个自然是a2a.server和a2a.client用起来非常直观。2. 安装与运行环境准备2.1 安装条件与 pip 安装步骤先说环境。a2a-protocol是基于 Python 3.9 开发的建议至少用 3.10 以上因为包内部的类型标注大量使用|联合类型和Optional语义老版本解析会有问题。我实测在 Python 3.10 和 3.11 下都稳定3.12 也兼容。安装官方版本直接用 pippip install a2a-protocol如果你在某个全新环境里装建议先把 pip 本身更新到较新版本再装避免老 pip 解析依赖失败python -m pip install --upgrade pip pip install a2a-protocol安装过程会自动带上几个关键依赖FastAPI、uvicorn、Pydantic、httpx、sse-starlette。这说明什么服务端不是从零造轮子而是基于 FastAPI 搭建的客户端则基于 httpx。所以实际使用时你完全可以复用之前积累的 FastAPI 和 Pydantic 经验上手成本很低。装完顺手验证一下pip show a2a-protocol python -c import a2a; print(a2a.__version__)如果第二个命令能输出版本号说明安装就绪。有些版本可能没有暴露顶层__version__那就看第一个命令的 Version 字段。2.2 安装后快速验证协议组件装好之后先确认核心模块能不能正常导入。我常用的一条命令是这样python -c from a2a.types import AgentCard, AgentSkill, AgentCapabilities; print(types ok) python -c from a2a.server import A2AServer; print(server ok) python -c from a2a.client import A2AClient; print(client ok)三行全部输出 OK说明核心组件都已就位。这里有个细节不同小版本的导入路径可能有差别。早期版本AgentCard直接放在a2a.types下面后来的版本把它挪到了a2a.server.agent或者别的子模块里。如果导入报ImportError先别慌在 Python 里输入dir(a2a.types)和dir(a2a.server)看一眼实际暴露了哪些名字按实际情况调整即可。用help()查看类签名也是排查这类问题最快的方式。3. 语法与核心 API 拆解3.1 类型定义从 Message 到 AgentCarda2a.types是理解整个包的关键。它定义了几组最重要的类型我按使用频率从高到低给你梳理。第一组是AgentCard相当于 Agent 挂在门口的名片。其他 Agent 要跟你协作先看这张名片判断你有什么能力、怎么联系你。核心字段包括name、description、url、version、capabilities、skills。其中capabilities是AgentCapabilities类型内部用streaming和push_notifications这两个布尔值声明是否支持流式和主动推送skills是AgentSkill列表用来描述你这个 Agent 会哪些“手艺”。第二组是消息体包括Message、TextPart和Artifact。Message是用户或 Agent 发给任务的输入消息TextPart是消息里的文本片段对应最终返回的文本内容Artifact则是任务完成后的产物容器里面装着一组Part。第三组是任务状态包括Task、TaskState和TaskStatus。Task对象挂着一个唯一id整个生命周期里它的状态会从SUBMITTED逐步走向COMPLETED、FAILED或CANCELED。TaskStatus里可以直接存一段人类可读的状态描述。来看一段创建核心对象的代码from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, Part ) agent_card AgentCard( nameqa_agent, description负责代码审查和质量检查的质检代理, urlhttp://127.0.0.1:8001, version1.0.0, capabilitiesAgentCapabilities(streamingTrue, push_notificationsFalse), skills[ AgentSkill(idcode_review, name代码审查, description检查代码规范与潜在缺陷), AgentSkill(idtest_advice, name测试建议, description生成测试用例建议), ], )注意AgentCapabilities(streamingTrue)这个参数。声明了流式支持客户端就知道可以长期挂连接接收流式输出如果不声明客户端会默认你走一次性返回等不到中间秒回信息容易误判超时。任务的构造大致是这样task Task( idtask-001-abc, stateTaskState.COMPLETED, statusTaskStatus(statuscompleted, message任务顺利完成), artifacts[ Artifact(parts[TextPart(text这是最终生成的回复内容)]) ], )在实际调用链里你从请求里收到的是TaskId和Message做完业务处理后返回的则是完整Task对象。这个“请求-响应”闭环只要跑通A2A 基础的协作流程就通了一半。3.2 客户端与服务端核心类服务端核心是a2a.server.A2AServer客户端核心是a2a.client.A2AClient。这两个类是整个 SDK 的门面。服务端的使用方式是一个类组合的过程先定义一个代理类继承 SDK 里的Agent基类重写核心处理函数然后实例化A2AServer把AgentCard丢进去最后用 uvicorn 把服务跑起来。在 FastAPI 的语义下A2AServer可以理解成一个预配置好的 FastAPI 应用加载完毕之后可以直接交给uvicorn.run()。客户端的用法更直接。A2AClient接受一个服务端地址就能初始化from a2a.client import A2AClient client A2AClient(http://127.0.0.1:8001) # 一次性调用 task await client.send_message(请帮我审查这段代码) # 流式调用适合长任务 async for event in client.send_message_stream(请分析这个问题的根因): if event.type TaskStatusUpdateEvent: print(进度:, event.task.status) elif event.type TaskArtifactUpdateEvent: print(产物:, event.artifact.parts)流式调用的返回值是一系列事件对象代码里可以根据事件类型分别处理。在实战里流式接口的价值特别大长任务卡住的时候至少能知道 Agent 正在工作而不是干等到超时。4. 参数体系与配置细节4.1 AgentCard 参数说明参数体系是新手最容易踩坑的地方。AgentCard的字段看着简单但实际上每个字段都影响对方 Agent 对你这个服务的感知。参数类型说明实践建议namestrAgent 唯一名称用机器可读的短名称如qa_agentdescriptionstr能力自然语言描述直接决定别的 Agent 要不要调用你写清楚边界urlstr服务对外地址部署后一定要改成对外可达的地址不要留 127.0.0.1versionstr版本号升级接口时记得递增方便对端识别capabilitiesAgentCapabilities是否流式、是否支持推送不支持的别乱开否则对端挂流式连接会超时skillsList[AgentSkill]技能列表每个技能要有独立id对内路由用AgentSkill的id是一个容易被忽略的细节。如果你的 Agent 要处理多种不同的任务最好在技能 id 上做区分比如code_review、doc_generate之类。服务端拿到任务的意图分类结果后可以直接根据skill_id路由到对应的处理逻辑省去在内部再打一层条件判断。4.2 服务端启动参数A2AServer本身的初始化参数不多通常只需要传agent和port两个server A2AServer(agentagent_card, port8001)但真正跑服务时很多问题出在 uvicorn 层的参数上。我强烈建议不要只写uvicorn.run(server.app)就完事至少把host、port、log_level都显式写出来import uvicorn if __name__ __main__: uvicorn.run(server.app, host0.0.0.0, port8001, log_levelinfo)host0.0.0.0在容器部署或者是多机协作时是必须的。如果写成127.0.0.1别的机器上的 Agent 根本连不进来。log_levelinfo能让你看到每一次请求的处理日志排查问题的时候日志比任何 debugger 都管用。生产环境下建议再加两个 uvicorn 参数workers1和timeout_keep_alive75。A2A 服务属于有状态交互多 worker 会导致任务 ID 归属不一致先保持单 worker 最稳timeout_keep_alive设长一点避免长连接被网关掐断。4.3 客户端连接参数客户端A2AClient同样有值得斟酌的参数。构造时除了服务地址还可以传入timeout和auth_headersclient A2AClient( http://127.0.0.1:8001, timeout30.0, auth_headers{Authorization: Bearer your-token}, )调接口时send_message还有两个核心参数task_id和message。task_id如果不传SDK 大概率会帮你生成一个随机 ID但在真实协作场景里我建议由调用方显式生成并且让任务 ID 带业务前缀比如order-20250601-001。这样日志追踪可以直接从任务 ID 反查业务单据排障效率高一个档次。send_message_stream还有timeout参数它控制的是整个流式会话的总体空闲超时不是单次读取超时。如果 Agent 中间思考了很久没有产出客户端那边就会断开。5. 一个可直接运行的 A2A 实战案例5.1 服务端实现做一个带技能的智能客服 Agent理论讲再多都不如直接跑一个能用的程序。这里我实现一个“订单咨询 Agent”技能有两个查订单状态、估算到货时间。数据直接用内存字典模拟不依赖外部数据库方便你照抄后 5 分钟内跑通。import uvicorn from typing import Optional from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, ) from a2a.server import A2AServer, Agent class OrderAgent(Agent): 订单咨询代理处理订单状态查询和时间估算 def __init__(self): super().__init__() # 模拟订单数据 self._orders { A1001: {status: 已发货, eta: 3天内到货}, A1002: {status: 待支付, eta: 支付后48小时内发货}, A1003: {status: 已签收, eta: 已完成}, } async def handle_message(self, message: Message) - Task: # 从消息里取文本内容 text .join( part.text for part in message.parts if hasattr(part, text) ) # 简单规则路由 if 单号 in text: order_id None for token in text.split(): if token.startswith(A): order_id token.strip(。) break if order_id and order_id in self._orders: order self._orders[order_id] reply f订单 {order_id} 当前状态{order[status]}{order[eta]} else: reply 抱歉没有查询到这个订单号。 elif 预计 in text or 到货 in text: reply 常规订单一般 3 天到 5 天偏远地区以物流信息为准。 else: reply 我可以帮你查单号和估算到货时间请提供订单号例如 A1001。 return Task( idmessage.task_id, stateTaskState.COMPLETED, statusTaskStatus(statuscompleted, message已生成回复), artifacts[Artifact(parts[TextPart(textreply)])], ) agent_card AgentCard( nameorder_agent, description订单咨询代理支持查询订单状态并给出预计到货时间。提问时请包含以A开头的订单号。, urlhttp://127.0.0.1:8002, version1.0.0, capabilitiesAgentCapabilities(streamingFalse, push_notificationsFalse), skills[ AgentSkill(idorder_status, name查单, description根据订单号查询物流状态), AgentSkill(ideta_estimate, name到货时间, description估算预计到货时间), ], ) server A2AServer(agentagent_card, port8002) if __name__ __main__: uvicorn.run(server.app, host0.0.0.0, port8002, log_levelinfo)注意几个实操细节。第一handle_message必须是个异步函数因为底层消息处理是 async 的用同步函数硬扛会导致事件循环阻塞。第二构造Task时id我直接复用了message.task_id这样调用方传什么任务 ID返回结果就对应什么 ID日志关联最容易。第三回复文本是一个完整的TextPart如果要返回多段内容就多放几个TextPart到parts列表里。5.2 客户端实现请求、流式与异常处理服务端跑起来之后客户端就是一个A2AClient的事。这里我写一个包含普通请求、流式请求和异常处理的完整脚本import asyncio from a2a.client import A2AClient async def main(): client A2AClient(http://127.0.0.1:8002, timeout15.0) # 1. 普通请求 task await client.send_message( task_idorder-demo-001, message帮我查一下订单 A1001 的状态, ) print(任务状态:, task.state) for artifact in task.artifacts: for part in artifact.parts: print(回复:, part.text) # 2. 不存在的订单场景 missing await client.send_message( task_idorder-demo-002, message查一下订单 B0001, ) for artifact in missing.artifacts: for part in artifact.parts: print(回复:, part.text) if __name__ __main__: asyncio.run(main())执行之后客户端打印的内容会是这样任务状态: TaskState.COMPLETED 回复: 订单 A1001 当前状态已发货3天内到货 回复: 抱歉没有查询到这个订单号。这个流程看起来简单但你已经完整走了一遍 A2A 的标准链路客户端发消息到服务端服务端解析消息、路由到技能逻辑、生成任务产物、返回完整任务对象。协议层的编码解码、HTTP 传输、JSON-RPC 封装全被a2a-protocol包替你消化了。5.3 流式输出场景与长任务体验上面的客服案例因为是即时返回用不到流式。但真实世界里一定会有长任务让一个 Agent 去检索一堆资料、生成一篇长文可能要几十秒。这时候如果不支持流式客户端只能傻等体验极差。要让服务端支持流式需要改两个地方AgentCapabilities(streamingTrue)以及继承类里实现生成器风格的流式输出方法。不同的 SDK 版本可能在细节上有差异但总体思路一致把一次任务拆成多个状态更新事件每产生一个中间产物就推送一次。我习惯的做法是先立刻回一个TaskState.WORKING的状态事件让客户端知道“已经接单了”中间有中间结果就更新Artifact真正全部结束再发布TaskState.COMPLETED事件。这样客户端即使拿到一个很长的任务也能持续看到进度不会因为 30 秒没动静就判断失败。6. 常见问题与排查经验6.1 网络连接与超时问题问题现象。客户端调用直接抛超时或者服务端收到请求但迟迟没有响应。这类问题的排查顺序是固定的先确认服务端进程还活着再确认地址端口能连通最后才怀疑协议层毛病。我的排查套路是这样的# 1. 确认服务进程端口 lsof -i :8002 # 2. 直接 curl 测健康检查接口如果服务有 / 或 /health curl http://127.0.0.1:8002/ # 3. 看客户端连接的地址是不是 127.0.0.1而不是 0.0.0.0其中最容易踩的坑就是把服务端地址写成了0.0.0.0。0.0.0.0是服务端监听地址不是客户端访问地址。客户端只能写127.0.0.1或者服务器的真实 IP。6.2 类型序列化与字段不匹配问题现象。客户端报ValidationError最常见的就是TaskState和字符串之间互转失败。A2A 的TaskState是枚举类型JSON 传输层是字符串如果你在版本差异之间切换可能老版本传的是小写字符串新版本枚举只认大写后台直接报错。解决办法也简单统一通过 SDK 提供的枚举对象传值不要手动写字符串。比如from a2a.types import TaskState task.state TaskState.COMPLETED # 不要写 task.state completed如果是从外部系统接入对方传了一个大写的字符串那也要在边界代码里先做一次映射转换再交给核心逻辑处理。6.3 SSE 流式连接被断开问题现象。流式调用刚开始能收到消息但几分钟后没有新事件客户端连接被断。这个问题的成因基本都指向空闲超时。中间件、反向代理、负载均衡器都有空闲超时设置比如 Nginx 默认proxy_read_timeout是 60 秒。Agent 思考 60 秒没有产出连接就被网关切断。处理办法有两层。第一层服务端开启流式后要定期发心跳事件哪怕没有正式产出也发一个状态更新占位第二层在部署配置里调大反向代理的超时参数给长任务留够空间。如果你在本地起 Nginx 做代理可以在同一个配置里给 A2A 服务的 location 加上proxy_read_timeout 300s;。6.4 异步函数与事件循环的隐含冲突问题现象。代理处理函数里调用了另一个同步的、耗时的库比如某个官方 SDK 的同步版本结果整个服务卡住。这在 FastAPI 异步环境里非常典型事件循环被同步阻塞以后所有并发请求都要排队。对策是耗时的操作用run_in_executor丢到线程池或者干脆换异步 SDK。如果第三方库没有异步版本就在继承类的内部用await asyncio.to_thread(func, ...)包一下。这个细节在别人看起来不起眼但在并发请求来临时直接决定服务是稳定还是雪崩。7. 部署与上线前的几个关键检查7.1 服务地址与能力声明的对齐上线前最容易被忽视的一个问题AgentCard里的url是声明地址但实际绑定的监听地址和端口可能跟声明不一致。举个例子AgentCard里写的是http://192.168.1.10:8002但 Docker 里映射的端口是38002别的 Agent 拿到名片去请求8002自然吃闭门羹。这块我的习惯是写一个启动自检脚本启动后用AgentClient拉取AgentCard比对声明url和当前服务进程监听的地址端口不一致就打印醒目的警告日志。这套自检逻辑不复杂但对多 Agent 协作的团队很有价值。7.2 安全认证与最小权限A2A 协议天然是网络服务只要是网络服务就得考虑认证。本地 Demo 可以裸奔但到了企业环境至少要在类里加一层 token 校验。a2a-protocol的客户端支持构造时传auth_headers服务端则需要自己在 FastAPI 的中间件里处理。我推荐的入门做法是先加一个最简单的静态 Token。客户端初始化时传{Authorization: Bearer xxxxx}服务端用 FastAPI middleware 检查请求头没有 token 或者 token 不对的直接返回 401不进业务逻辑。等团队规模大了、Agent 数量多了再引入正式的身份管理方案。8. 我的一些个人体会用a2a-protocol的经验里最大的心得不是某个函数怎么写而是“标准化带来的是减负”。以前做 Agent 协作每个人都在定义自己的通信格式字段名、状态机、超时策略全靠口头对齐。现在有了统一协议和官方包哪怕合作方用的是别家语言实现的 SDK大家面向的仍是同一套语义沟通成本直线下降。另外也想提个建议开始千万别追求大而全的框架先把一个单一技能的 Agent 用 A2A 包跑起来再逐步加技能、加流式、加认证。我见过一些项目一上来就想落地多 Agent 编排平台结果连最基本的send_message链路都没调通。走得稳比走得快重要。这个包的 API 还在快速迭代中如果你读到这篇时发现个别导入路径变了不用慌顺着模块列表去看一眼就能找到对应位置——协议本身是稳的变的只是代码组织方式。
a2a-protocol实战：Python多智能体互操作与核心API解析
今年做多智能体项目时被一个老问题折磨得不轻Agent 用不同框架开发通信方式全靠自己写胶水代码A 系统的输出 B 系统读不懂光是消息格式对齐就来回改了三天。后来社区里讨论 Google 发布的 A2A 协议官方配套的 Python 包a2a-protocol也放了出来。我第一时间在本地搭了个 Demo 跑通实测下来它对“让不同的 Agent 互相打电话”这件事的解决方式是系统性的不止给了协议规范还直接给了现成的服务端、客户端和类型定义。这篇文章不堆概念只讲三件事这个包的语法结构、核心参数含义以及一个能直接复现的实战案例。如果你也在做 Agent 间的互操作这篇应该能帮你少走不少弯路。1. 先搞清楚 a2a-protocol 解决什么问题1.1 Agent 之间为什么需要一套“通话协议”单个 Agent 的能力再强也只是一个孤岛。真正复杂的业务场景里可能需要一个“翻译 Agent”先做语义理解再把任务转给“代码生成 Agent”最后由“质检 Agent”检查结果。这种协作结构一出现就立刻面临一个基础问题Agent 之间用什么格式沟通最早的方案是自己定义 JSON。今天你定义一个{sender, receiver, content}明天对方框架要{from, to, payload}两套系统对接就要写字段映射。更麻烦的是任务状态的跟踪——这个任务到底完成了没有、中间有没有产生中间产物、失败原因是什么这些问题各家实现都不一样。A2A 协议解决的正是这一层用一套标准化的消息格式和任务生命周期让不同 Agent 之间能互相发现、互相调用、追踪进度。a2a-protocol包就是这个协议的 Python 实现。它不是某个具体平台的私有 SDK而是对协议规范的代码落地里面包含通信双方所需的核心组件。你不需要自己实现 JSON-RPC 协议细节也不需要反复发明任务状态机的轮子直接把包装进来定义好 Agent 的能力就能对外提供标准的服务。1.2 协议的工作方式与消息模型理解了它解决什么问题再看它的工作方式就顺了。A2A 底层走的是 JSON-RPC 2.0 这套成熟的远程调用规范传输层用 HTTP同时支持 SSE 做流式推送。消息模型的核心是 Task任务Task 内部挂载 Message消息、Artifact产物和 Part内容片段。我给你打个生活化的比方。把 Agent 当成一家公司的客服专员Task 就是一次工单客户发起工单客服接单中间可能实时回复进展Message最终交付一份文档Artifact。工单有明确的状态进行中、完成、失败、取消。这套模型天然适合 Agent 之间有来有往的协作场景。包的代码结构也跟这个模型对齐。a2a.types里就是这些消息类型定义a2a.server提供服务端框架负责接收 JSON-RPC 请求a2a.client是客户端负责发起调用。后面两个自然是a2a.server和a2a.client用起来非常直观。2. 安装与运行环境准备2.1 安装条件与 pip 安装步骤先说环境。a2a-protocol是基于 Python 3.9 开发的建议至少用 3.10 以上因为包内部的类型标注大量使用|联合类型和Optional语义老版本解析会有问题。我实测在 Python 3.10 和 3.11 下都稳定3.12 也兼容。安装官方版本直接用 pippip install a2a-protocol如果你在某个全新环境里装建议先把 pip 本身更新到较新版本再装避免老 pip 解析依赖失败python -m pip install --upgrade pip pip install a2a-protocol安装过程会自动带上几个关键依赖FastAPI、uvicorn、Pydantic、httpx、sse-starlette。这说明什么服务端不是从零造轮子而是基于 FastAPI 搭建的客户端则基于 httpx。所以实际使用时你完全可以复用之前积累的 FastAPI 和 Pydantic 经验上手成本很低。装完顺手验证一下pip show a2a-protocol python -c import a2a; print(a2a.__version__)如果第二个命令能输出版本号说明安装就绪。有些版本可能没有暴露顶层__version__那就看第一个命令的 Version 字段。2.2 安装后快速验证协议组件装好之后先确认核心模块能不能正常导入。我常用的一条命令是这样python -c from a2a.types import AgentCard, AgentSkill, AgentCapabilities; print(types ok) python -c from a2a.server import A2AServer; print(server ok) python -c from a2a.client import A2AClient; print(client ok)三行全部输出 OK说明核心组件都已就位。这里有个细节不同小版本的导入路径可能有差别。早期版本AgentCard直接放在a2a.types下面后来的版本把它挪到了a2a.server.agent或者别的子模块里。如果导入报ImportError先别慌在 Python 里输入dir(a2a.types)和dir(a2a.server)看一眼实际暴露了哪些名字按实际情况调整即可。用help()查看类签名也是排查这类问题最快的方式。3. 语法与核心 API 拆解3.1 类型定义从 Message 到 AgentCarda2a.types是理解整个包的关键。它定义了几组最重要的类型我按使用频率从高到低给你梳理。第一组是AgentCard相当于 Agent 挂在门口的名片。其他 Agent 要跟你协作先看这张名片判断你有什么能力、怎么联系你。核心字段包括name、description、url、version、capabilities、skills。其中capabilities是AgentCapabilities类型内部用streaming和push_notifications这两个布尔值声明是否支持流式和主动推送skills是AgentSkill列表用来描述你这个 Agent 会哪些“手艺”。第二组是消息体包括Message、TextPart和Artifact。Message是用户或 Agent 发给任务的输入消息TextPart是消息里的文本片段对应最终返回的文本内容Artifact则是任务完成后的产物容器里面装着一组Part。第三组是任务状态包括Task、TaskState和TaskStatus。Task对象挂着一个唯一id整个生命周期里它的状态会从SUBMITTED逐步走向COMPLETED、FAILED或CANCELED。TaskStatus里可以直接存一段人类可读的状态描述。来看一段创建核心对象的代码from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, Part ) agent_card AgentCard( nameqa_agent, description负责代码审查和质量检查的质检代理, urlhttp://127.0.0.1:8001, version1.0.0, capabilitiesAgentCapabilities(streamingTrue, push_notificationsFalse), skills[ AgentSkill(idcode_review, name代码审查, description检查代码规范与潜在缺陷), AgentSkill(idtest_advice, name测试建议, description生成测试用例建议), ], )注意AgentCapabilities(streamingTrue)这个参数。声明了流式支持客户端就知道可以长期挂连接接收流式输出如果不声明客户端会默认你走一次性返回等不到中间秒回信息容易误判超时。任务的构造大致是这样task Task( idtask-001-abc, stateTaskState.COMPLETED, statusTaskStatus(statuscompleted, message任务顺利完成), artifacts[ Artifact(parts[TextPart(text这是最终生成的回复内容)]) ], )在实际调用链里你从请求里收到的是TaskId和Message做完业务处理后返回的则是完整Task对象。这个“请求-响应”闭环只要跑通A2A 基础的协作流程就通了一半。3.2 客户端与服务端核心类服务端核心是a2a.server.A2AServer客户端核心是a2a.client.A2AClient。这两个类是整个 SDK 的门面。服务端的使用方式是一个类组合的过程先定义一个代理类继承 SDK 里的Agent基类重写核心处理函数然后实例化A2AServer把AgentCard丢进去最后用 uvicorn 把服务跑起来。在 FastAPI 的语义下A2AServer可以理解成一个预配置好的 FastAPI 应用加载完毕之后可以直接交给uvicorn.run()。客户端的用法更直接。A2AClient接受一个服务端地址就能初始化from a2a.client import A2AClient client A2AClient(http://127.0.0.1:8001) # 一次性调用 task await client.send_message(请帮我审查这段代码) # 流式调用适合长任务 async for event in client.send_message_stream(请分析这个问题的根因): if event.type TaskStatusUpdateEvent: print(进度:, event.task.status) elif event.type TaskArtifactUpdateEvent: print(产物:, event.artifact.parts)流式调用的返回值是一系列事件对象代码里可以根据事件类型分别处理。在实战里流式接口的价值特别大长任务卡住的时候至少能知道 Agent 正在工作而不是干等到超时。4. 参数体系与配置细节4.1 AgentCard 参数说明参数体系是新手最容易踩坑的地方。AgentCard的字段看着简单但实际上每个字段都影响对方 Agent 对你这个服务的感知。参数类型说明实践建议namestrAgent 唯一名称用机器可读的短名称如qa_agentdescriptionstr能力自然语言描述直接决定别的 Agent 要不要调用你写清楚边界urlstr服务对外地址部署后一定要改成对外可达的地址不要留 127.0.0.1versionstr版本号升级接口时记得递增方便对端识别capabilitiesAgentCapabilities是否流式、是否支持推送不支持的别乱开否则对端挂流式连接会超时skillsList[AgentSkill]技能列表每个技能要有独立id对内路由用AgentSkill的id是一个容易被忽略的细节。如果你的 Agent 要处理多种不同的任务最好在技能 id 上做区分比如code_review、doc_generate之类。服务端拿到任务的意图分类结果后可以直接根据skill_id路由到对应的处理逻辑省去在内部再打一层条件判断。4.2 服务端启动参数A2AServer本身的初始化参数不多通常只需要传agent和port两个server A2AServer(agentagent_card, port8001)但真正跑服务时很多问题出在 uvicorn 层的参数上。我强烈建议不要只写uvicorn.run(server.app)就完事至少把host、port、log_level都显式写出来import uvicorn if __name__ __main__: uvicorn.run(server.app, host0.0.0.0, port8001, log_levelinfo)host0.0.0.0在容器部署或者是多机协作时是必须的。如果写成127.0.0.1别的机器上的 Agent 根本连不进来。log_levelinfo能让你看到每一次请求的处理日志排查问题的时候日志比任何 debugger 都管用。生产环境下建议再加两个 uvicorn 参数workers1和timeout_keep_alive75。A2A 服务属于有状态交互多 worker 会导致任务 ID 归属不一致先保持单 worker 最稳timeout_keep_alive设长一点避免长连接被网关掐断。4.3 客户端连接参数客户端A2AClient同样有值得斟酌的参数。构造时除了服务地址还可以传入timeout和auth_headersclient A2AClient( http://127.0.0.1:8001, timeout30.0, auth_headers{Authorization: Bearer your-token}, )调接口时send_message还有两个核心参数task_id和message。task_id如果不传SDK 大概率会帮你生成一个随机 ID但在真实协作场景里我建议由调用方显式生成并且让任务 ID 带业务前缀比如order-20250601-001。这样日志追踪可以直接从任务 ID 反查业务单据排障效率高一个档次。send_message_stream还有timeout参数它控制的是整个流式会话的总体空闲超时不是单次读取超时。如果 Agent 中间思考了很久没有产出客户端那边就会断开。5. 一个可直接运行的 A2A 实战案例5.1 服务端实现做一个带技能的智能客服 Agent理论讲再多都不如直接跑一个能用的程序。这里我实现一个“订单咨询 Agent”技能有两个查订单状态、估算到货时间。数据直接用内存字典模拟不依赖外部数据库方便你照抄后 5 分钟内跑通。import uvicorn from typing import Optional from a2a.types import ( AgentCard, AgentCapabilities, AgentSkill, Task, TaskState, TaskStatus, Message, TextPart, Artifact, ) from a2a.server import A2AServer, Agent class OrderAgent(Agent): 订单咨询代理处理订单状态查询和时间估算 def __init__(self): super().__init__() # 模拟订单数据 self._orders { A1001: {status: 已发货, eta: 3天内到货}, A1002: {status: 待支付, eta: 支付后48小时内发货}, A1003: {status: 已签收, eta: 已完成}, } async def handle_message(self, message: Message) - Task: # 从消息里取文本内容 text .join( part.text for part in message.parts if hasattr(part, text) ) # 简单规则路由 if 单号 in text: order_id None for token in text.split(): if token.startswith(A): order_id token.strip(。) break if order_id and order_id in self._orders: order self._orders[order_id] reply f订单 {order_id} 当前状态{order[status]}{order[eta]} else: reply 抱歉没有查询到这个订单号。 elif 预计 in text or 到货 in text: reply 常规订单一般 3 天到 5 天偏远地区以物流信息为准。 else: reply 我可以帮你查单号和估算到货时间请提供订单号例如 A1001。 return Task( idmessage.task_id, stateTaskState.COMPLETED, statusTaskStatus(statuscompleted, message已生成回复), artifacts[Artifact(parts[TextPart(textreply)])], ) agent_card AgentCard( nameorder_agent, description订单咨询代理支持查询订单状态并给出预计到货时间。提问时请包含以A开头的订单号。, urlhttp://127.0.0.1:8002, version1.0.0, capabilitiesAgentCapabilities(streamingFalse, push_notificationsFalse), skills[ AgentSkill(idorder_status, name查单, description根据订单号查询物流状态), AgentSkill(ideta_estimate, name到货时间, description估算预计到货时间), ], ) server A2AServer(agentagent_card, port8002) if __name__ __main__: uvicorn.run(server.app, host0.0.0.0, port8002, log_levelinfo)注意几个实操细节。第一handle_message必须是个异步函数因为底层消息处理是 async 的用同步函数硬扛会导致事件循环阻塞。第二构造Task时id我直接复用了message.task_id这样调用方传什么任务 ID返回结果就对应什么 ID日志关联最容易。第三回复文本是一个完整的TextPart如果要返回多段内容就多放几个TextPart到parts列表里。5.2 客户端实现请求、流式与异常处理服务端跑起来之后客户端就是一个A2AClient的事。这里我写一个包含普通请求、流式请求和异常处理的完整脚本import asyncio from a2a.client import A2AClient async def main(): client A2AClient(http://127.0.0.1:8002, timeout15.0) # 1. 普通请求 task await client.send_message( task_idorder-demo-001, message帮我查一下订单 A1001 的状态, ) print(任务状态:, task.state) for artifact in task.artifacts: for part in artifact.parts: print(回复:, part.text) # 2. 不存在的订单场景 missing await client.send_message( task_idorder-demo-002, message查一下订单 B0001, ) for artifact in missing.artifacts: for part in artifact.parts: print(回复:, part.text) if __name__ __main__: asyncio.run(main())执行之后客户端打印的内容会是这样任务状态: TaskState.COMPLETED 回复: 订单 A1001 当前状态已发货3天内到货 回复: 抱歉没有查询到这个订单号。这个流程看起来简单但你已经完整走了一遍 A2A 的标准链路客户端发消息到服务端服务端解析消息、路由到技能逻辑、生成任务产物、返回完整任务对象。协议层的编码解码、HTTP 传输、JSON-RPC 封装全被a2a-protocol包替你消化了。5.3 流式输出场景与长任务体验上面的客服案例因为是即时返回用不到流式。但真实世界里一定会有长任务让一个 Agent 去检索一堆资料、生成一篇长文可能要几十秒。这时候如果不支持流式客户端只能傻等体验极差。要让服务端支持流式需要改两个地方AgentCapabilities(streamingTrue)以及继承类里实现生成器风格的流式输出方法。不同的 SDK 版本可能在细节上有差异但总体思路一致把一次任务拆成多个状态更新事件每产生一个中间产物就推送一次。我习惯的做法是先立刻回一个TaskState.WORKING的状态事件让客户端知道“已经接单了”中间有中间结果就更新Artifact真正全部结束再发布TaskState.COMPLETED事件。这样客户端即使拿到一个很长的任务也能持续看到进度不会因为 30 秒没动静就判断失败。6. 常见问题与排查经验6.1 网络连接与超时问题问题现象。客户端调用直接抛超时或者服务端收到请求但迟迟没有响应。这类问题的排查顺序是固定的先确认服务端进程还活着再确认地址端口能连通最后才怀疑协议层毛病。我的排查套路是这样的# 1. 确认服务进程端口 lsof -i :8002 # 2. 直接 curl 测健康检查接口如果服务有 / 或 /health curl http://127.0.0.1:8002/ # 3. 看客户端连接的地址是不是 127.0.0.1而不是 0.0.0.0其中最容易踩的坑就是把服务端地址写成了0.0.0.0。0.0.0.0是服务端监听地址不是客户端访问地址。客户端只能写127.0.0.1或者服务器的真实 IP。6.2 类型序列化与字段不匹配问题现象。客户端报ValidationError最常见的就是TaskState和字符串之间互转失败。A2A 的TaskState是枚举类型JSON 传输层是字符串如果你在版本差异之间切换可能老版本传的是小写字符串新版本枚举只认大写后台直接报错。解决办法也简单统一通过 SDK 提供的枚举对象传值不要手动写字符串。比如from a2a.types import TaskState task.state TaskState.COMPLETED # 不要写 task.state completed如果是从外部系统接入对方传了一个大写的字符串那也要在边界代码里先做一次映射转换再交给核心逻辑处理。6.3 SSE 流式连接被断开问题现象。流式调用刚开始能收到消息但几分钟后没有新事件客户端连接被断。这个问题的成因基本都指向空闲超时。中间件、反向代理、负载均衡器都有空闲超时设置比如 Nginx 默认proxy_read_timeout是 60 秒。Agent 思考 60 秒没有产出连接就被网关切断。处理办法有两层。第一层服务端开启流式后要定期发心跳事件哪怕没有正式产出也发一个状态更新占位第二层在部署配置里调大反向代理的超时参数给长任务留够空间。如果你在本地起 Nginx 做代理可以在同一个配置里给 A2A 服务的 location 加上proxy_read_timeout 300s;。6.4 异步函数与事件循环的隐含冲突问题现象。代理处理函数里调用了另一个同步的、耗时的库比如某个官方 SDK 的同步版本结果整个服务卡住。这在 FastAPI 异步环境里非常典型事件循环被同步阻塞以后所有并发请求都要排队。对策是耗时的操作用run_in_executor丢到线程池或者干脆换异步 SDK。如果第三方库没有异步版本就在继承类的内部用await asyncio.to_thread(func, ...)包一下。这个细节在别人看起来不起眼但在并发请求来临时直接决定服务是稳定还是雪崩。7. 部署与上线前的几个关键检查7.1 服务地址与能力声明的对齐上线前最容易被忽视的一个问题AgentCard里的url是声明地址但实际绑定的监听地址和端口可能跟声明不一致。举个例子AgentCard里写的是http://192.168.1.10:8002但 Docker 里映射的端口是38002别的 Agent 拿到名片去请求8002自然吃闭门羹。这块我的习惯是写一个启动自检脚本启动后用AgentClient拉取AgentCard比对声明url和当前服务进程监听的地址端口不一致就打印醒目的警告日志。这套自检逻辑不复杂但对多 Agent 协作的团队很有价值。7.2 安全认证与最小权限A2A 协议天然是网络服务只要是网络服务就得考虑认证。本地 Demo 可以裸奔但到了企业环境至少要在类里加一层 token 校验。a2a-protocol的客户端支持构造时传auth_headers服务端则需要自己在 FastAPI 的中间件里处理。我推荐的入门做法是先加一个最简单的静态 Token。客户端初始化时传{Authorization: Bearer xxxxx}服务端用 FastAPI middleware 检查请求头没有 token 或者 token 不对的直接返回 401不进业务逻辑。等团队规模大了、Agent 数量多了再引入正式的身份管理方案。8. 我的一些个人体会用a2a-protocol的经验里最大的心得不是某个函数怎么写而是“标准化带来的是减负”。以前做 Agent 协作每个人都在定义自己的通信格式字段名、状态机、超时策略全靠口头对齐。现在有了统一协议和官方包哪怕合作方用的是别家语言实现的 SDK大家面向的仍是同一套语义沟通成本直线下降。另外也想提个建议开始千万别追求大而全的框架先把一个单一技能的 Agent 用 A2A 包跑起来再逐步加技能、加流式、加认证。我见过一些项目一上来就想落地多 Agent 编排平台结果连最基本的send_message链路都没调通。走得稳比走得快重要。这个包的 API 还在快速迭代中如果你读到这篇时发现个别导入路径变了不用慌顺着模块列表去看一眼就能找到对应位置——协议本身是稳的变的只是代码组织方式。
