前两天有个朋友问我你手头已经有翻译Agent、写代码Agent、查资料的Agent了能不能让翻译Agent把结果直接丢给写代码Agent让它们自己商量着把活干完我说能但要加一个东西就是标题里这个a2a-python。a2a-python是A2A协议Agent-to-Agent的Python官方SDK解决的核心问题很简单不同框架、不同机器、不同厂商跑着的AI Agent怎么互相发现能力、怎么发消息、怎么把任务执行完再回传结果。这篇文章我会把它的语法、参数和一套完整可跑的实践案例拆开讲不废话直接上手。A2A这套协议刚出来的时候很多人的第一反应是又要学一个轮子。但等我把自己的几个Agent按A2A串起来之后才发现它解决的不是怎么调一个API而是Agent之间怎么协作的标准问题。尤其当你需要把Agent开放给外部系统、其他团队甚至跨语言服务调用时a2a-python提供了一整套现成的服务封装和数据类型比你自己拿FastAPI硬写要省心太多。适合正在做多Agent系统、被Agent互联互通折腾过、或者准备把Agent能力服务化的Python开发者。基础要求不高会装包、会写函数就够了。1. 为什么需要a2a-pythonAgent之间也需要普通话1.1 AI Agent协作的痛点能力没法互相发现先说个真实场景。我自己用Python接过大模型、搭过检索增强也写过几个小Agent。每个Agent都暴露一个HTTP接口但每个接口的入参、出参都不一样有的接收表单有的接收JSON有的直接让你把prompt拼在URL里。真想让两个Agent协作时发现能力靠文档传参靠猜调用靠复制粘贴curl结果一个字段对不上就全崩了。A2A协议做的事情就是把这个过程标准化。每个Agent对外提供一份AgentCard能力名片说明自己叫什么、能做什么、支持哪些技能、是否需要流式输出。另一个Agent或者客户端拿到这份名片就知道该怎么调用它。而a2a-python就是把协议里的消息格式、任务状态、服务路由都封装成Python类你只需要填参数不需要自己实现JSON-RPC那套细节。1.2 A2A协议的工作方式名片、工单和消息我用一个生活化的类比帮你理解A2A的运行逻辑。想象每个Agent是一家外包公司A2A是行业统一印制的《合作规范》。规范分三块AgentCard档案和名片挂在外墙HTTP服务上路过的人一看就知道这公司能干什么。Task工单甲方发起一次任务拿到的不是结果而是一个工单号Task ID工单有状态待办、执行中、需要补充材料、完成、失败、取消。Message和Part工单里传递的具体内容消息分角色user/agent内容由多个Part组成Part可以是文本、文件或者结构化数据。这套设计的巧妙之处在于解耦。调用方不需要知道服务方内部用了什么框架、什么模型、什么代码结构只要协议一致翻译Agent的输出可以被关键词Agent直接当作输入就像不同公司之间只认统一格式的合同不关心对方公司内部怎么管考勤。1.3 a2a-python在协议里扮演的角色a2a-python不是协议本身而是协议在Python生态里的落地实现。它负责三件事服务端通过A2AAServer把FastAPI应用包装好挂上AgentCard路由和A2A标准通信端点你只管声明几个技能函数。客户端通过A2AClient或手写HTTP调用按协议把消息发过去轮询任务状态最终拿回artifacts产物。数据类型AgentCard、AgentSkill、Task、Message、TextPart这些协议对象都有了Python类IDE自动补全和类型检查都能用写起来比拼字典舒服得多。有一点我特别想提醒在PyPI上这个包的名字是a2a-sdk不是a2a-python。标题里的a2a-python指的是GitHub仓库名你pip安装时会发现官网推荐的是pip install a2a-sdk。我第一次装的时候照着a2a-python去搜吃了不少亏这个坑下面还会讲。2. 环境准备与最小服务先让一个Agent在网上挂牌2.1 环境要求和安装命令先交代环境。我本机是Python 3.10建议用虚拟环境别直接往系统Python里装尤其如果你的开发机还要跑别的项目依赖冲突会让人崩溃。python -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install a2a-sdk安装完可以验证一下版本和导入路径import a2a print(a2a.__version__)如果你是第一次在VS Code里跑这种项目记得让VS Code选中你刚创建的虚拟环境解释器否则代码能导入但命令行跑了另一个Python也会出现装了包但找不到模块的诡异问题。还有一个小技巧在launch.json里把justMyCode: false打开调试时能看到SDK内部是怎么组装请求的这对于理解协议的调用链非常有帮助。2.2 最小Agent服务代码下面是一段最小但完整的Agent服务。作用很朴素上线一个Agent声明一个叫echo的技能能接收文本并返回文本。# echo_agent.py from a2a import A2AAServer, AgentCard, AgentCapabilities, AgentSkill def make_echo_server(port: int 8000): card AgentCard( nameEchoAgent, description一个只会复读的Agent接收文本并原样返回。, urlfhttp://localhost:{port}, version1.0.0, capabilitiesAgentCapabilities(streamingFalse), skills[ AgentSkill( idecho, nameecho, description把用户输入的文本原样返回。, tags[echo], ) ], ) return A2AAServer( host0.0.0.0, portport, cardcard, ) if __name__ __main__: import uvicorn server make_echo_server(8000) uvicorn.run(server, host0.0.0.0, port8000)这里要注意A2AAServer本身就是FastAPI应用所以启动用uvicorn.run(server, ...)不需要额外创建FastAPI实例。如果SDK版本较新可能还支持server.run()但用uvicorn是最稳妥也最好控制的方式。运行python echo_agent.py看到Uvicorn running on http://0.0.0.0:8000就说明服务起来了。2.3 用curl验证AgentCard和消息发送A2A协议约定AgentCard默认放在/.well-known/agent.json路径下。开个新终端验证curl http://localhost:8000/.well-known/agent.json返回的JSON里应该能看到name、description、skills这些字段。这一步很关键它证明你的Agent已经挂了牌其他Agent只要知道这个地址就能发现你的能力。再试一下发送消息。A2A协议底层走的是JSON-RPC 2.0用message/send方法发消息curl -X POST http://localhost:8000/ \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, parts: [{type: text, text: hello a2a}] } } }如果SDK里没有绑定具体的处理逻辑你会看到一个Task对象状态可能是working或者completedartifacts可能是空数组这正常——因为AgentCard里声明了skills但服务端还没有把这些技能映射到真实函数。接下来要做的就是注册技能函数把回声从协议层面变成业务层面。2.4 为什么我说能跑通和能干活是两码事很多人装好SDK、把示例服务跑起来就觉得完事了。但示例服务只证明了协议通了没有证明业务通了。真正的Agent要干的事是收到消息 - 解析参数 - 调模型/查数据/写代码 - 把结果包装成Task返回。这个中间层就是第3章和第4章要解决的内容。先通协议再接业务这个顺序别反了。3. AgentCard语法和Server参数拆解把能力写成JSON给别的Agent看3.1 AgentCard各字段的语法与写法AgentCard是整个A2A体系里最像产品文档的部分。它决定了别的Agent会不会信任你、会不会调用你、会不会正确调用你。我按字段逐个拆一遍。字段必填含义我的建议name是Agent名称用简短、无空格的英文名比如translate_agentdescription是能力描述写清业务边界、输入、输出、限制条件url是服务地址必须是调用方真正能访问到的地址version是服务版本协议或行为有变就升版本capabilities是能力标记声明是否支持流式、推送通知skills否技能列表每个技能单独写descriptionsecurity否认证信息有鉴权需求时使用provider否服务方信息方便调用方定位联系documentation_url否文档地址复杂Agent建议填description这个字段我特别想多说两句。我自己吃过亏一开始把一个翻译Agent的description写成提供翻译服务结果别的Agent什么文本都往这儿丢你让它翻译、总结、写诗它全收最后返回一堆莫名其妙的输出。后来改成仅支持英译中输入英文文本输出中文翻译一次只处理一段不支持会议纪要生成调用准确率立刻上来了。A2A协议里没有意图识别中间层AgentCard的description就是你的寻路向导。3.2 AgentSkill的写法让技能边界清晰技能是AgentCard里最重要的结构。一个Agent可以声明多个技能每个技能是一个AgentSkill对象。以翻译Agent为例skills[ AgentSkill( idtranslate_en2zh, nametranslate_en2zh, description把英文文本翻译成中文输入纯英文文本输出中文译文。, tags[translation, nlp], examples[Translate hello world to Chinese], input_modes[text], output_modes[text], ) ]每个字段都有用。id是技能的唯一标识name建议和函数名一致description是给别人看的使用说明examples可以给调用方一个few-shot参考input_modes和output_modes则声明了数据形态。比如一个技能接收图片、输出文本就把input_modes设为[file]output_modes设为[text]。这样调用方可以提前判断数据格式省得发了半天文件你只会处理纯文本。3.3 A2AAServer的参数host、port之外你还得知道什么A2AAServer是a2a-python服务端的主入口。不同版本里参数名会有差异但核心参数通常包括参数常见用途备注host绑定地址0.0.0.0表示所有网卡可访问port监听端口注意端口冲突cardAgentCard对象必填handler / agent技能处理函数或Agent实例版本差异较大以官方example为准log_level日志级别排查问题建议INFO或DEBUGcors是否开启跨域浏览器端调用时需要有个反直觉的坑很多人觉得host传0.0.0.0就能被公网访问实际上云服务器还要在安全组/防火墙里放行端口。我第一次部署时在本地curl一切正常换到云服务器上怎么都连不上最后发现是安全组规则没加。还有一点如果服务是放在nginx反代后面的AgentCard.url一定要填外部能访问的地址不能写http://localhost:8000否则别的Agent拿到名片后访问的是它自己的localhost。3.4 Task、Message、Part三个数据结构帮你理解状态流转A2A协议里一次完整的调用是这么流转的调用方发一条Message给服务端服务端创建一个TaskTask状态从submitted变成working处理完变为completed响应里带着artifacts通常是包含结果的Message列表。如果处理中需要用户补材料状态会变成input-required。Message由角色和Part列表组成message Message( roleuser, parts[ TextPart(text今天北京天气如何), DataPart(data{location: Beijing}), ], )Part是真正承载内容的对象最常用的是TextPart还有FilePart传文件、DataPart传结构化JSON。Task里的artifacts也是Message列表只是角色是agent作用是承载最终输出。理解了这个模型你就能明白为什么A2A适合做Agent协作输出的artifacts可以直接作为下一个Agent的输入Message整个流水线在数据结构上天然通畅。4. 参数校验与客户端调用两种参数都得讲清楚4.1 服务端参数host、port、日志与超时标题里的参数其实有两层含义一层是服务配置参数另一层是技能函数的入参。先讲配置参数。服务端常见的坑集中在四个地方。绑定地址本机调试用127.0.0.1对外服务用0.0.0.0。端口8000是常用端口如果被占用就换8010、8020但记得AgentCard.url也要跟着改。日志级别排查慢任务或参数问题时把log_level设为DEBUG你能看到SDK收到的原始请求。超时大模型推理通常要几秒到几十秒如果前置有nginx记得把nginx的proxy_read_timeout调大。我有个任务因为超时被网关切掉返回值在客户端看来是空响应排查了很久才发现根本不是代码问题。4.2 技能函数入参从自由文本到结构化参数校验A2A协议本身只传Message没有强约束说消息里必须是JSON。所以实际项目中技能函数的入参需要你自己约法三章。最推荐的做法是约定调用方把JSON字符串放在TextPart.text里服务端解析后做校验。配合Pydantic非常舒服。from pydantic import BaseModel, ValidationError from a2a.types import Task, TaskStatus, Message, TextPart class WeatherQuery(BaseModel): city: str date: str | None None async def weather_handler(message: Message) - Task: try: query WeatherQuery.model_validate_json(message.parts[0].text) except ValidationError as e: return Task( idtask-error, statusTaskStatus.FAILED, artifacts[ Message( roleagent, parts[TextPart(textf参数错误: {e})], ) ], ) result f{query.city} 在 {query.date or 今天} 的天气是晴25℃。 return Task( idtask-1, statusTaskStatus.COMPLETED, artifacts[Message(roleagent, parts[TextPart(textresult)])], )这段代码有两个重点。一是用model_validate_json直接从文本解析JSON省去手动dict取值二是校验失败时返回TaskStatus.FAILED而不是抛异常这样调用方拿到的是结构化错误信息而不是一串traceback。我在项目中凡是入参复杂的技能全部走这个模式参数解析和业务逻辑解耦测试也好写。4.3 客户端调用用SDK类还是直接用HTTP如果你的环境里只有a2a-sdk直接用它的客户端类最省事。但有个现实问题很多场景下调用方不是Python或者调用方不想引入SDK这时候掌握协议层的调用格式就更重要。A2A协议基于JSON-RPC 2.0我用httpx写过最简单的一个客户端import httpx def send_message(url: str, text: str, timeout: float 30.0) - dict: payload { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, parts: [{type: text, text: text}], } }, } resp httpx.post(url, jsonpayload, timeouttimeout) resp.raise_for_status() return resp.json() def get_task_result(url: str, task_id: str, timeout: float 30.0) - dict: payload { jsonrpc: 2.0, id: 2, method: task/get, params: {id: task_id}, } resp httpx.post(url, jsonpayload, timeouttimeout) resp.raise_for_status() return resp.json()这个封装足够你在99%的场景里用。SDK的A2AClient本质也是这套协议只不过把URL、超时、会话ID管理都封装好了。我个人的习惯是快速验证用curl业务代码里用简洁的httpx封装只有做复杂多Agent编排时才引入SDK客户端。原因无他少依赖一个类的API差异就少一个踩坑点。4.4 一个我调了很久的参数坑超时单位引发的错觉有次我调一个Agent明明服务端处理只要5秒客户端却总是报超时。看服务端日志发现任务都completed了响应却回不去。查了半天发现是nginx配置里proxy_read_timeout默认60秒、按理说够用但客户端请求里我自己又传了timeout5把5秒当成连接超时其实httpx的timeout参数是整个请求读超时大模型一推理就超了。换成timeoutNone后服务端把任务id返回客户端再轮询task/get问题解决。这个坑的教训是长任务不要死等一个HTTP响应应该先用message/send拿到Task ID并返回submitted或working然后再轮询task/get拿最终结果。这也是A2A协议设计Task状态机的初衷。你在实际项目中如果感觉等待响应很别扭多半是没有正确使用异步任务模型。5. 实战翻译Agent和关键词Agent的串联流水线5.1 场景与架构下面这个案例是我自己做过的一个小流水线很适合演示A2A的实际价值输入一段英文学术文本先让翻译Agent翻成中文再把中文交给关键词提取Agent最终输出一个关键词列表。整个过程里两个Agent彼此不知道对方存在是一个客户端脚本按顺序调用它们。架构很简单服务A翻译Agent监听8001端口技能translate_en2zh。服务B关键词Agent监听8002端口技能extract_keywords。客户端普通Python脚本串行调用两个Agent打印最终结果。这里我不绑定特定SDK的handler接口而是用一个make_server模板来统一包装以便你的SDK版本即使接口有差异也能照抄这个模式。5.2 翻译Agent实现# server_a_translate.py from a2a import A2AAServer, AgentCard, AgentCapabilities, AgentSkill from a2a.types import Task, TaskStatus, Message, TextPart async def translate_handler(message: Message) - Task: text .join(part.text for part in message.parts if part.type text) # 实际项目里这里替换成大模型API即可 translated f[中文译文] {text} return Task( idtranslate-1, statusTaskStatus.COMPLETED, artifacts[ Message(roleagent, parts[TextPart(texttranslated)]) ], ) def make_translate_server(port: int 8001): card AgentCard( nametranslate_agent, description把英文文本翻译成中文只处理English到中文的翻译。, urlfhttp://localhost:{port}, version1.0.0, capabilitiesAgentCapabilities(streamingFalse), skills[ AgentSkill( idtranslate_en2zh, nametranslate_en2zh, descriptionTranslate English text to Chinese., input_modes[text], output_modes[text], ) ], ) server A2AAServer(host0.0.0.0, portport, cardcard) # 不同版本SDK绑定handler方式不同请按官方example调整 server.bind_handler(translate_handler) return server if __name__ __main__: import uvicorn uvicorn.run(make_translate_server(8001), host0.0.0.0, port8001)我在代码里留了一个bind_handler的示例如果你的SDK版本里方法名不同去安装目录下的examples里搜一下关键字即可。核心思路是一致的把处理函数挂到Server上协议层收到消息后调用它把返回值包装成Task。5.3 关键词提取Agent实现# server_b_extract.py from a2a import A2AAServer, AgentCard, AgentCapabilities, AgentSkill from a2a.types import Task, TaskStatus, Message, TextPart # 演示用实际可以接 jieba / HanLP / 大模型 def extract_keywords_from_zh(text: str) - list[str]: words [w for w in text.replace(, ).replace(。, ).split() if len(w) 2] return list(dict.fromkeys(words))[:5] async def extract_handler(message: Message) - Task: text .join(part.text for part in message.parts if part.type text) keywords extract_keywords_from_zh(text) result .join(keywords) return Task( idextract-1, statusTaskStatus.COMPLETED, artifacts[ Message(roleagent, parts[TextPart(textresult)]) ], ) def make_extract_server(port: int 8002): card AgentCard( namekeyword_agent, description从中文文本中提取关键词输入中文文本输出逗号分隔的关键词列表。, urlfhttp://localhost:{port}, version1.0.0, capabilitiesAgentCapabilities(streamingFalse), skills[ AgentSkill( idextract_keywords, nameextract_keywords, descriptionExtract keywords from Chinese text., input_modes[text], output_modes[text], ) ], ) server A2AAServer(host0.0.0.0, portport, cardcard) server.bind_handler(extract_handler) return server if __name__ __main__: import uvicorn uvicorn.run(make_extract_server(8002), host0.0.0.0, port8002)这个Agent相当简陋但不影响演示。关键在于它接收的是上游翻译Agent的输出这正好体现了A2A流水线里artifacts可以无缝作为下一个Message的特点。5.4 客户端串联调用与结果验证客户端脚本不需要任何Agent框架纯Python加httpx就能完成串联# client_pipeline.py import json import httpx def send_and_get_result(url: str, text: str) - str: payload { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, parts: [{type: text, text: text}], } }, } resp httpx.post(url, jsonpayload, timeout60.0) resp.raise_for_status() data resp.json() # 提取第一个artifact里的text artifacts data[result][artifacts] return artifacts[0][parts][0][text] if __name__ __main__: en_text Agent to Agent protocol defines how agents discover and communicate with each other. zh_text send_and_get_result(http://localhost:8001/, en_text) print(翻译结果:, zh_text) keywords send_and_get_result(http://localhost:8002/, zh_text) print(关键词:, keywords)运行结果类似翻译结果: [中文译文] Agent to Agent protocol defines how agents discover and communicate with each other. 关键词: ProtocolAgentdefinesagentsdiscover虽然演示里翻译是假的但这个流程是完整的一个客户端两个Agent服务一条数据流水线。把翻译函数换成真实大模型API后就是一套可上线的多Agent协作雏形。5.5 这个模型怎么扩展如果你想把它扩展到真实系统有几个很自然的升级方向。翻译和提取都接大模型只要改handler内部实现协议层不用动增加会话管理在Task或Message里传递sessionId实现多轮对话增加路由Agent把客户端脚本里的顺序调用改成由一个路由Agent自动决策先调谁再调谁接入文件Part让翻译Agent直接接收PDF或Word文件。核心思路始终一致Agent之间不直接依赖对方代码只依赖对方的AgentCard、Task和Message格式。6. 部署到真实环境后的坑绑定、跨域、超时与版本差异6.1 绑定0.0.0.0之后外部还是连不上这个问题几乎每个人都遇到过。本地跑得好好的部署到服务器上外部机器访问不了。99%的情况是云服务商的安全组或服务器防火墙没有放行对应端口。我自己的排查顺序是先在服务器本机curl通了说明服务正常再检查防火墙firewall-cmd --list-ports或ufw status最后查安全组规则。还有一个经常被忽视的点如果AgentCard.url写的是http://localhost:8000外部Agent拿到名片后会去调用它自己的localhost根本到不了你的服务。所以正式环境里url必须改成对外可解析的域名或IP。6.2 浏览器端调用跨域CORS配置如果Agent服务不只是被后端调用还要被浏览器里的Agent管理后台直接调用就会碰到跨域问题。A2AAServer底层是FastAPI处理方式也很直接加CORSMiddleware。from fastapi.middleware.cors import CORSMiddleware server make_translate_server(8001) server.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], )allow_origins[*]只在开发环境用生产环境建议只放行你自己的控制台域名。这个配置本身不难麻烦的是如果你用SDK的封装属性找不到add_middleware容易卡住。记住A2AAServer就是FastAPI应用FastAPI能干的它都能干。6.3 长任务场景从同步等待到流式输出前面说过长任务不要在一个HTTP请求里死等。A2A协议里有两种解法要么先返回Task ID客户端轮询task/get要么用message/stream走SSE流式返回。a2a-sdk里AgentCapabilities有个streaming字段就是这个开关。我建议项目一开始就用轮询模式因为流式会引入连接管理、断线重连等复杂度。等你的调用方明确需要打字机效果时再升级到流式。协议里把这两种机制都定义好了只是实现深度不同。6.4 版本差异SDK更新导致接口对不上a2a-sdk迭代速度很快网上教程里写的代码到了你本地可能就报AttributeError。这不是你笨是版本差异。解决这个问题有两个土办法。一是看官方examples在你虚拟环境的Lib/site-packages/a2a/examples目录下面通常有完整的server和client示例直接对比你的代码。二是用dir(server)查看A2AAServer实例上到底有哪些方法快速定位bind_handler还是register_handler还是别的什么名字。我在升级SDK后经常用这两个方法比查文档快。6.5 几个我养成的调试习惯最后分享几个让我少掉头发的调试习惯。先用curl验证/.well-known/agent.json这一步能确认服务通了、AgentCard挂上了。再用curl发一次message/send观察返回的Task状态如果卡在working就说明handler没绑定成功如果直接failed就去看业务逻辑。日志一定要开DEBUGA2A协议是JSON-RPC格式日志里能看到原始请求和响应参数有没有传对一目了然。还有每次升级SDK版本后把部署环境重跑一遍最小示例确认协议没变化再继续改代码。这些都不是什么高级技巧但在Agent这种多服务协作的场景里能帮你把问题局限在最小的范围内。