一、Agent和Agent之间到底怎么“对话”python的a2a-protocol包很多人第一次看到名字会以为是某个加密通信库其实它解决的是近两年AI应用里最挠头的问题之一多个AI Agent智能体之间怎么互相找到对方、发消息、交付结果。A2A全称是Agent-to-Agent由Google在2025年4月联合数十家厂商发起随后捐给了Linux基金会目前已经有一堆头部AI公司参与进来。简单理解MCP解决的是“Agent怎么调用工具”A2A解决的是“Agent怎么调用另一个Agent”。我做AI应用开发这几年最大的体会是单机Agent跑demo很容易跑生产环境就难在协作。项目一复杂你就会有多个Agent——一个负责数据处理一个负责写代码一个负责做质检它们彼此之间怎么协调过去要么自己写HTTP接口要么直接用消息队列硬怼结果就是每家的对接都不一样换个Agent基本等于重写一套集成代码。a2a-protocol就是想把这个“A对A”的过程标准化让大家不用重复造轮子。这篇文章我会从语法层面、参数配置、以及一个完整的实际应用案例来拆解这个包适合已经在用LangChain、CrewAI或自己搭过多Agent系统的开发者。如果你是刚开始接触Agent方向看完也能明白这套协议在工程上到底替我们省了多少事。二、a2a-protocol的核心语法这不只是一个HTTP封装2.1 安装与最小可运行示例老规矩先安装。这个包在PyPI上叫a2a-protocolPython 3.10都支持建议装在独立虚拟环境里避免依赖打架。pip install a2a-protocol安装完成后最常用的入口是a2a模块它同时包含了客户端和服务端两套能力。我们先看一个最小的“发现Agent”动作import asyncio from a2a import Client async def main(): client Client() agent_card await client.get_agent_card(http://localhost:10000/) print(agent_card.name) print(agent_card.description) print(agent_card.skills) asyncio.run(main())这里注意几个关键点Client()不需要传认证参数就能用协议本身的设计是让AgentCard这个“名片”完全公开。get_agent_card()返回的是一个Pydantic模型不是裸字典所以后续你可以用agent_card.name、agent_card.skills这种属性访问方式拿到结构化字段。所有方法都是异步的统一走asyncio这和现代Python生态的取向一致。2.2 五个核心类型把通信语言定下来a2a协议最大的价值不在传输层而在它定义了一套“话语体系”。我把它拆成五个必须理解的核心类型你在读官方文档时遇到的大部分陌生名词都逃不出这五类。AgentCard智能体名片每个Agent必须通过/.well-known/agent.json路径公开自己的AgentCard内容包括名字、简介、技能列表、以及一个默认的endpoint。这就像你微信头像下面那段自我介绍其他Agent过来先看名片再决定要不要和你协作。Message消息通信的基本单位。它有role字段分为user和agent。注意这里的role不一定指的是人类用户在A2A语境里“user”可能是另一个Agent。消息内部由若干个Part组成Part分为TextPart、FilePart、DataPart三种。Task任务这是整个协议里最值得琢磨的类型。A2A认为Agent之间的交互不应该是无状态的“一问一答”而应该围绕一个可追踪的任务展开。每个Task有全局唯一的id有state状态机比如submitted、working、completed、failed还有一个可选的metadata用来带业务上下文。Artifact产物Agent干活之后产出物可能是一份报告、一个数据文件、一张图。Artifact和Message的区别在于Message是“说的话”Artifact是“交付的成果”。协议允许Artifact挂在Task下面这样干完活之后调用方可以直接从Task里把产物取走。Part内容块上面说了Message和Artifact的内容都由Part承载。你不需要自己去定义JSON结构直接用TextPart(text...)、FilePart(file...)或者DataPart(data...)就行。这一套类型设计说白了就是把“Agent协作”翻译成了“任务驱动的异步工作流”。和HTTP随便发个POST不同A2A要求你显式创建任务、推送消息、轮询状态——麻烦是麻烦一点但换来的是整个调用链可追溯、可恢复生产环境里出现“活干了一半没人管”的概率大大降低。2.3 消息编码与JSON-RPC 2.0底层这里我必须提一句很多人第一次看a2a代码会觉得奇怪为什么不是RESTful风格接口因为协议底层是基于JSON-RPC 2.0规范的。这意味着客户端和服务端之间通过jsonrpc、method、params、id这些标准字段通信。比如message/send这个核心方法理论上请求体长这样{ jsonrpc: 2.0, method: message/send, params: { task_id: task-001, message: { role: user, parts: [ { type: text, text: 帮我写一段Python冒泡排序 } ] } }, id: 1 }好消息是a2a-protocol包把这一坨JSON构造细节全部封装好了你直接用Python对象操作即可。坏消息是如果你要自己实现一个非Python的A2A服务端就必须吃透JSON-RPC规范没有捷径可走。三、参数详解每个常用参数到底是什么意思3.1 Client客户端参数从使用者的角度绝大部分参数都集中在Client对象和相关方法上。我把高频参数整理成一张表方便你查阅。参数类型必填说明与建议agent_urlstr是Agent的根地址比如http://localhost:10000。不要在末尾加多余的斜杠容易拼接出错timeoutfloat否单次请求超时时间默认30秒。如果Agent内部要跑大模型推理建议调大到120秒以上auth_providerAuthProvider否认证提供者企业内部环境常用BearerTokenAuthProvidertask_idstr视方法而定任务ID一般无脑用str(uuid.uuid4())生成messageMessage是要发送的消息对象必须用a2a.types构建partslist[Part]是消息内容块列表可包含多个不同类型的PartstateTaskState是任务状态回执多用于服务端推送状态更新artifactslist[Artifact]否结果产物列表完成或失败状态时返回这里我想重点说两个容易被忽视的参数。第一个是timeout。本地联调的时候没事一部署到Kubernetes容器里Agent内部再调一次外部大模型API常见的就是一个请求等60秒以上。我见过不止一个开发者在生产环境收到TimeoutError后一脸懵——不是代码写错是Agent干活真的慢。所以没事别用默认值除非你的Agent只做秒回级别的计算。第二个是auth_provider。公开测试网上的Agent可以不用认证但是一旦接入公司内部服务你就需要类似这样的写法from a2a import Client from a2a.auth import BearerTokenAuthProvider client Client( auth_providerBearerTokenAuthProvider(tokenyour-company-token) )这个设计比在URL里拼token干净得多建议从第一天就养成习惯。3.2 Message与Part的参数细节构造消息时最容易踩坑的地方在于Message和Part的嵌套关系。Message可以包含多个Part每个Part必须指定type。比如from a2a.types import Message, TextPart, FilePart message Message( roleuser, parts[ TextPart(text请统计这个CSV文件的行数), FilePart(file{uri: file:///data/users.csv, mime_type: text/csv}) ] )注意几个细节FilePart的file参数接受的是字典形式里面需要uri和mime_type两个键。如果你只给了uri协议校验会报错。TextPart、FilePart、DataPart分别对应纯文本、文件引用、结构化数据。道理和MIME类型差不多文本用text文件用uri引用数据用data字段直接内嵌JSON。role枚举只有user和agent。这里容易产生误解——协同工作的两个Agent之间调用方就扮演user被调用方扮演agent并没有第三个角色。3.3 AgentCard参数与技能声明每个Agent暴露出来的AgentCard实际上就是一个Pydantic模型。常见字段包括字段说明nameAgent名称要唯一description一句话描述这个Agent能做什么urlAgent的endpoint地址version协议版本号skills技能列表每个技能有id和namecapabilities声明是否支持流式、是否支持事件推送等authentication认证方式声明skills这个字段值得多说一句。我在实际项目里发现与其让调用方硬编码调用某个Agent不如让它先拉取AgentCard、看看skills、做动态路由。比如你有三个Agent分别负责“简历解析”“技能匹配”“面试问题生成”你就通过判断skill.id来决定把任务发给谁。这样后续新Agent上线只要更新自己的AgentCard就行调用方代码完全不用改。四、实际应用案例从零搭一个多Agent招聘助理4.1 场景设定与架构思路光讲语法和参数太干了我直接用最近做的“多Agent招聘助理”当案例带你走一遍完整流程。这个项目解决一个很现实的需求把一份候选人简历丢进去系统自动完成“解析简历 - 匹配岗位 - 生成面试问题”三条流水线。传统的做法是把所有逻辑写在一个Agent里但后续扩展性差。比如你想换一家新的简历解析服务就得把整个Agent推倒重来。用A2A拆分成三个独立Agent之后每个Agent只负责一件事CVAgent接收简历文件输出候选人的结构化信息。MatchAgent接收结构化信息和岗位描述输出匹配分数和建议。InterviewAgent接收匹配结果生成定制化的面试问题列表。三者通过A2A协议互相调用。这里选A2A而不是自己写HTTP原因只有一个我想让每个Agent保持完全独立将来谁都能替换掉它。4.2 用a2a-protocol实现CV解析Agent的服务端先实现被调用的服务端。a2a-protocol提供A2AHandler和A2AHTTPServer两个类你可以把协议接入点全部交给它们自己只写业务逻辑。from datetime import datetime from a2a import A2AHandler, A2AHTTPServer from a2a.types import ( Message, Task, AgentCard, AgentSkill, TextPart, FilePart, DataPart, ) class CVAgentHandler(A2AHandler): async def on_message_request(self, message, task): parts message.parts # 找到第一个文件Part视为简历文件 file_part next(p for p in parts if isinstance(p, FilePart)) # 模拟解析简历实际工程里可以调用OCR 大模型抽取 parsed await self._parse_resume(file_part.file.get(uri)) return Message( roleagent, parts[DataPart(dataparsed)] ) async def _parse_resume(self, uri: str): # 假装返回结构化数据 return { name: 张三, skills: [python, sql], years: 5, } agent_card AgentCard( nameCVAgent, description简历解析Agent支持把简历文件转换为结构化数据, urlhttp://localhost:10001/, version1.0.0, skills[AgentSkill(idresume_parse, name解析简历)], ) handler CVAgentHandler(agent_cardagent_card) server A2AHTTPServer(handlerhandler, port10001) server.serve_forever()这段代码里有几点是平时文档不会细讲但是很重要的on_message_request是核心方法接收message和task两个参数。task里可以拿到task_id方便你做状态追踪或者日志关联。解析返回内容直接用DataPart协议会自动帮你序列化。不需要自己再写一遍json.dumps。A2AHTTPServer默认支持启动健康检查路径/.well-known/agent.json所以别的Agent才能“看到”你。4.3 调用方的实现拼接任务流接下来实现调用方也就是“调度中心”。这个调度中心本身不干活它只负责把任务发给对应的Agent再在结果回来后继续下一步。import asyncio import uuid from a2a import Client from a2a.types import Message, TextPart, FilePart async def run_pipeline(resume_uri: str, job_description: str): cv_client Client() match_client Client() interview_client Client() # 第一步解析简历 cv_msg Message( roleuser, parts[FilePart(file{uri: resume_uri, mime_type: application/pdf})] ) cv_result await cv_client.send_message( agent_urlhttp://localhost:10001/, task_idstr(uuid.uuid4()), messagecv_msg ) cv_data cv_result.parts[0].data # 第二步岗位匹配 match_msg Message( roleuser, parts[ TextPart(textf岗位描述{job_description}), DataPart(datacv_data), ] ) match_result await match_client.send_message( agent_urlhttp://localhost:10002/, task_idstr(uuid.uuid4()), messagematch_msg ) match_data match_result.parts[0].data # 第三步生成面试题 interview_msg Message( roleuser, parts[DataPart(datamatch_data)] ) interview_result await interview_client.send_message( agent_urlhttp://localhost:10003/, task_idstr(uuid.uuid4()), messageinterview_msg ) return interview_result.parts[0].text asyncio.run(run_pipeline(file:///resume.pdf, 高级后端工程师))我用这个例子想表达一个理念A2A让“Agent编排”变成了和“调用函数”一样简单的事。你不是在对接三个服务构造三次HTTP请求你是在调用三个能力边界清晰的Agent方法。当然真实生产环境里轮询式的send_message可能不够用。如果某个Agent任务要跑几分钟你不想让HTTP连接一直挂着那就应该用send_task_start启动任务、再定时调用get_task获取状态。这个模式我已经在后面的避坑章节细说。4.4 让Agent真正“看到”彼此AgentCard路由实战最后一个实战点利用AgentCard实现动态路由。假设调度中心启动时不知道有哪些Agent在线它可以通过协议自带的“发现”能力去问。from a2a import Client async def discover_agents(agent_urls): client Client() agents {} for url in agent_urls: try: card await client.get_agent_card(url.rstrip(/)) for skill in card.skills: agents[skill.id] url except Exception: # 网络抖动时跳过不拖垮整个调度中心 continue return agents async def main(): registry await discover_agents([ http://localhost:10001/, http://localhost:10002/, http://localhost:10003/, ]) print(registry) # 输出示例{resume_parse: http://localhost:10001/, ...}这段代码的价值在于当你有10个Agent甚至50个Agent时你不需要维护一份硬编码的调用表Agent自己会“报名”。新增Agent只要部署起来、配上AgentCard调度中心下一轮发现循环就能自动识别它。五、常见问题与排查技巧实录5.1 问题速查表这几个坑我全都踩过现象可能原因解决方案客户端报404AgentCard路径必须是/.well-known/agent.json且末尾不能有自定义路径检查服务端路由确认协议能访问到agent.jsonFilePart一直校验不过file字段没有同时给uri和mime_type两个字段必须都提供缺一不可大文件任务超时默认timeout30秒大模型处理时间远超于此调大timeout或改用异步任务轮询模式结果乱序/丢失并发发了多个任务但没有维护task_id的映射关系每个任务用独立UUID并在本地存task_id到请求的映射服务端收不到消息AgentCard里配置的url和实际监听端口不一致检查AgentCard的url字段这是其他Agent调你的唯一依据数据全部变成字符串发消息时用了TextPart却给的是JSON字符串结构化数据请用DataPart不要用TextPart包JSON5.2 从阻塞调用迁移到异步任务模式我在案例里用的是send_message它简单但也有局限——如果任务执行时间太长HTTP连接会被挂起中间任何网络抖动都会导致你误判为失败。所以当你确认某个Agent单次要跑几秒以上建议尽早切换到任务的异步模式。核心方法是send_task_start给Agent发一个启动信号立刻返回。get_task带上task_id去轮询轮询间隔建议2~5秒。如果支持可以订阅事件推送让Agent主动通知你任务完成省去轮询。from a2a import Client import asyncio async def run_long_task(agent_url: str, message): client Client() task_id task-optimize-001 await client.send_task_start( agent_urlagent_url, task_idtask_id, messagemessage ) while True: task await client.get_task(agent_urlagent_url, task_idtask_id) if task.state in (completed, failed): return task await asyncio.sleep(2)这个模式的好处是哪怕任务跑到一半容器重启了只要日志里记了task_id重启后还能继续查询状态。数据不因为进程重启而丢。5.3 独门经验先写AgentCard再写业务逻辑我从这个包上学到的最大教训是——不要一上来就写业务代码先定义好AgentCard和Part类型再往里填逻辑。原因很简单A2A的协作边界完全靠公开的类型和文档来保证。你自己内部怎么实现都行但如果你输出的Part结构不稳定下游Agent就会频繁踩解析错误的坑。我现在的习惯是先写下每个Agent的技能ID和描述发给团队评审。再定义输入输出Part的字段类型写成Python dataclass锁定结构。最后才写内部业务逻辑和异常处理。这样做的体验类比一下就是写微服务之前先定好Swagger文档而不是写完代码再补文档。先定协议边界后期联调的痛苦会少一个量级。另外metadata字段别浪费掉。建议把debug_trace、source_agent、request_time这样的上下文信息塞进去排查问题的时候全靠这些关键信息定位链路。a2a本身不会帮你做链路追踪但你可以用标准字段把业务自己需要的可观测性信息带起来。写到最后想说的这套协议目前还在快速演进阶段a2a-protocol包的版本变数也不小。但无论如何它已经替开发者夯实了大量通用层的工作JSON-RPC封装、AgentCard发现机制、任务状态机、Part结构定义这些都不需要你再去操心。真正常写的是自己的Agent业务逻辑。如果后面还想深挖建议从两方面入手一是把A2A和MCP结合起来用让Agent既能调用工具也能调用其他Agent整条链路就完整了二是自己实现一个A2A服务端脱离Python原生包用Go或Java去造一遍轮子吃透协议细节。我个人更推荐第二条路因为从客户端使用者到协议实现者这个视角切换带来的理解深度是完全不一样的。
A2A协议实战：用a2a-protocol实现多Agent协作通信
一、Agent和Agent之间到底怎么“对话”python的a2a-protocol包很多人第一次看到名字会以为是某个加密通信库其实它解决的是近两年AI应用里最挠头的问题之一多个AI Agent智能体之间怎么互相找到对方、发消息、交付结果。A2A全称是Agent-to-Agent由Google在2025年4月联合数十家厂商发起随后捐给了Linux基金会目前已经有一堆头部AI公司参与进来。简单理解MCP解决的是“Agent怎么调用工具”A2A解决的是“Agent怎么调用另一个Agent”。我做AI应用开发这几年最大的体会是单机Agent跑demo很容易跑生产环境就难在协作。项目一复杂你就会有多个Agent——一个负责数据处理一个负责写代码一个负责做质检它们彼此之间怎么协调过去要么自己写HTTP接口要么直接用消息队列硬怼结果就是每家的对接都不一样换个Agent基本等于重写一套集成代码。a2a-protocol就是想把这个“A对A”的过程标准化让大家不用重复造轮子。这篇文章我会从语法层面、参数配置、以及一个完整的实际应用案例来拆解这个包适合已经在用LangChain、CrewAI或自己搭过多Agent系统的开发者。如果你是刚开始接触Agent方向看完也能明白这套协议在工程上到底替我们省了多少事。二、a2a-protocol的核心语法这不只是一个HTTP封装2.1 安装与最小可运行示例老规矩先安装。这个包在PyPI上叫a2a-protocolPython 3.10都支持建议装在独立虚拟环境里避免依赖打架。pip install a2a-protocol安装完成后最常用的入口是a2a模块它同时包含了客户端和服务端两套能力。我们先看一个最小的“发现Agent”动作import asyncio from a2a import Client async def main(): client Client() agent_card await client.get_agent_card(http://localhost:10000/) print(agent_card.name) print(agent_card.description) print(agent_card.skills) asyncio.run(main())这里注意几个关键点Client()不需要传认证参数就能用协议本身的设计是让AgentCard这个“名片”完全公开。get_agent_card()返回的是一个Pydantic模型不是裸字典所以后续你可以用agent_card.name、agent_card.skills这种属性访问方式拿到结构化字段。所有方法都是异步的统一走asyncio这和现代Python生态的取向一致。2.2 五个核心类型把通信语言定下来a2a协议最大的价值不在传输层而在它定义了一套“话语体系”。我把它拆成五个必须理解的核心类型你在读官方文档时遇到的大部分陌生名词都逃不出这五类。AgentCard智能体名片每个Agent必须通过/.well-known/agent.json路径公开自己的AgentCard内容包括名字、简介、技能列表、以及一个默认的endpoint。这就像你微信头像下面那段自我介绍其他Agent过来先看名片再决定要不要和你协作。Message消息通信的基本单位。它有role字段分为user和agent。注意这里的role不一定指的是人类用户在A2A语境里“user”可能是另一个Agent。消息内部由若干个Part组成Part分为TextPart、FilePart、DataPart三种。Task任务这是整个协议里最值得琢磨的类型。A2A认为Agent之间的交互不应该是无状态的“一问一答”而应该围绕一个可追踪的任务展开。每个Task有全局唯一的id有state状态机比如submitted、working、completed、failed还有一个可选的metadata用来带业务上下文。Artifact产物Agent干活之后产出物可能是一份报告、一个数据文件、一张图。Artifact和Message的区别在于Message是“说的话”Artifact是“交付的成果”。协议允许Artifact挂在Task下面这样干完活之后调用方可以直接从Task里把产物取走。Part内容块上面说了Message和Artifact的内容都由Part承载。你不需要自己去定义JSON结构直接用TextPart(text...)、FilePart(file...)或者DataPart(data...)就行。这一套类型设计说白了就是把“Agent协作”翻译成了“任务驱动的异步工作流”。和HTTP随便发个POST不同A2A要求你显式创建任务、推送消息、轮询状态——麻烦是麻烦一点但换来的是整个调用链可追溯、可恢复生产环境里出现“活干了一半没人管”的概率大大降低。2.3 消息编码与JSON-RPC 2.0底层这里我必须提一句很多人第一次看a2a代码会觉得奇怪为什么不是RESTful风格接口因为协议底层是基于JSON-RPC 2.0规范的。这意味着客户端和服务端之间通过jsonrpc、method、params、id这些标准字段通信。比如message/send这个核心方法理论上请求体长这样{ jsonrpc: 2.0, method: message/send, params: { task_id: task-001, message: { role: user, parts: [ { type: text, text: 帮我写一段Python冒泡排序 } ] } }, id: 1 }好消息是a2a-protocol包把这一坨JSON构造细节全部封装好了你直接用Python对象操作即可。坏消息是如果你要自己实现一个非Python的A2A服务端就必须吃透JSON-RPC规范没有捷径可走。三、参数详解每个常用参数到底是什么意思3.1 Client客户端参数从使用者的角度绝大部分参数都集中在Client对象和相关方法上。我把高频参数整理成一张表方便你查阅。参数类型必填说明与建议agent_urlstr是Agent的根地址比如http://localhost:10000。不要在末尾加多余的斜杠容易拼接出错timeoutfloat否单次请求超时时间默认30秒。如果Agent内部要跑大模型推理建议调大到120秒以上auth_providerAuthProvider否认证提供者企业内部环境常用BearerTokenAuthProvidertask_idstr视方法而定任务ID一般无脑用str(uuid.uuid4())生成messageMessage是要发送的消息对象必须用a2a.types构建partslist[Part]是消息内容块列表可包含多个不同类型的PartstateTaskState是任务状态回执多用于服务端推送状态更新artifactslist[Artifact]否结果产物列表完成或失败状态时返回这里我想重点说两个容易被忽视的参数。第一个是timeout。本地联调的时候没事一部署到Kubernetes容器里Agent内部再调一次外部大模型API常见的就是一个请求等60秒以上。我见过不止一个开发者在生产环境收到TimeoutError后一脸懵——不是代码写错是Agent干活真的慢。所以没事别用默认值除非你的Agent只做秒回级别的计算。第二个是auth_provider。公开测试网上的Agent可以不用认证但是一旦接入公司内部服务你就需要类似这样的写法from a2a import Client from a2a.auth import BearerTokenAuthProvider client Client( auth_providerBearerTokenAuthProvider(tokenyour-company-token) )这个设计比在URL里拼token干净得多建议从第一天就养成习惯。3.2 Message与Part的参数细节构造消息时最容易踩坑的地方在于Message和Part的嵌套关系。Message可以包含多个Part每个Part必须指定type。比如from a2a.types import Message, TextPart, FilePart message Message( roleuser, parts[ TextPart(text请统计这个CSV文件的行数), FilePart(file{uri: file:///data/users.csv, mime_type: text/csv}) ] )注意几个细节FilePart的file参数接受的是字典形式里面需要uri和mime_type两个键。如果你只给了uri协议校验会报错。TextPart、FilePart、DataPart分别对应纯文本、文件引用、结构化数据。道理和MIME类型差不多文本用text文件用uri引用数据用data字段直接内嵌JSON。role枚举只有user和agent。这里容易产生误解——协同工作的两个Agent之间调用方就扮演user被调用方扮演agent并没有第三个角色。3.3 AgentCard参数与技能声明每个Agent暴露出来的AgentCard实际上就是一个Pydantic模型。常见字段包括字段说明nameAgent名称要唯一description一句话描述这个Agent能做什么urlAgent的endpoint地址version协议版本号skills技能列表每个技能有id和namecapabilities声明是否支持流式、是否支持事件推送等authentication认证方式声明skills这个字段值得多说一句。我在实际项目里发现与其让调用方硬编码调用某个Agent不如让它先拉取AgentCard、看看skills、做动态路由。比如你有三个Agent分别负责“简历解析”“技能匹配”“面试问题生成”你就通过判断skill.id来决定把任务发给谁。这样后续新Agent上线只要更新自己的AgentCard就行调用方代码完全不用改。四、实际应用案例从零搭一个多Agent招聘助理4.1 场景设定与架构思路光讲语法和参数太干了我直接用最近做的“多Agent招聘助理”当案例带你走一遍完整流程。这个项目解决一个很现实的需求把一份候选人简历丢进去系统自动完成“解析简历 - 匹配岗位 - 生成面试问题”三条流水线。传统的做法是把所有逻辑写在一个Agent里但后续扩展性差。比如你想换一家新的简历解析服务就得把整个Agent推倒重来。用A2A拆分成三个独立Agent之后每个Agent只负责一件事CVAgent接收简历文件输出候选人的结构化信息。MatchAgent接收结构化信息和岗位描述输出匹配分数和建议。InterviewAgent接收匹配结果生成定制化的面试问题列表。三者通过A2A协议互相调用。这里选A2A而不是自己写HTTP原因只有一个我想让每个Agent保持完全独立将来谁都能替换掉它。4.2 用a2a-protocol实现CV解析Agent的服务端先实现被调用的服务端。a2a-protocol提供A2AHandler和A2AHTTPServer两个类你可以把协议接入点全部交给它们自己只写业务逻辑。from datetime import datetime from a2a import A2AHandler, A2AHTTPServer from a2a.types import ( Message, Task, AgentCard, AgentSkill, TextPart, FilePart, DataPart, ) class CVAgentHandler(A2AHandler): async def on_message_request(self, message, task): parts message.parts # 找到第一个文件Part视为简历文件 file_part next(p for p in parts if isinstance(p, FilePart)) # 模拟解析简历实际工程里可以调用OCR 大模型抽取 parsed await self._parse_resume(file_part.file.get(uri)) return Message( roleagent, parts[DataPart(dataparsed)] ) async def _parse_resume(self, uri: str): # 假装返回结构化数据 return { name: 张三, skills: [python, sql], years: 5, } agent_card AgentCard( nameCVAgent, description简历解析Agent支持把简历文件转换为结构化数据, urlhttp://localhost:10001/, version1.0.0, skills[AgentSkill(idresume_parse, name解析简历)], ) handler CVAgentHandler(agent_cardagent_card) server A2AHTTPServer(handlerhandler, port10001) server.serve_forever()这段代码里有几点是平时文档不会细讲但是很重要的on_message_request是核心方法接收message和task两个参数。task里可以拿到task_id方便你做状态追踪或者日志关联。解析返回内容直接用DataPart协议会自动帮你序列化。不需要自己再写一遍json.dumps。A2AHTTPServer默认支持启动健康检查路径/.well-known/agent.json所以别的Agent才能“看到”你。4.3 调用方的实现拼接任务流接下来实现调用方也就是“调度中心”。这个调度中心本身不干活它只负责把任务发给对应的Agent再在结果回来后继续下一步。import asyncio import uuid from a2a import Client from a2a.types import Message, TextPart, FilePart async def run_pipeline(resume_uri: str, job_description: str): cv_client Client() match_client Client() interview_client Client() # 第一步解析简历 cv_msg Message( roleuser, parts[FilePart(file{uri: resume_uri, mime_type: application/pdf})] ) cv_result await cv_client.send_message( agent_urlhttp://localhost:10001/, task_idstr(uuid.uuid4()), messagecv_msg ) cv_data cv_result.parts[0].data # 第二步岗位匹配 match_msg Message( roleuser, parts[ TextPart(textf岗位描述{job_description}), DataPart(datacv_data), ] ) match_result await match_client.send_message( agent_urlhttp://localhost:10002/, task_idstr(uuid.uuid4()), messagematch_msg ) match_data match_result.parts[0].data # 第三步生成面试题 interview_msg Message( roleuser, parts[DataPart(datamatch_data)] ) interview_result await interview_client.send_message( agent_urlhttp://localhost:10003/, task_idstr(uuid.uuid4()), messageinterview_msg ) return interview_result.parts[0].text asyncio.run(run_pipeline(file:///resume.pdf, 高级后端工程师))我用这个例子想表达一个理念A2A让“Agent编排”变成了和“调用函数”一样简单的事。你不是在对接三个服务构造三次HTTP请求你是在调用三个能力边界清晰的Agent方法。当然真实生产环境里轮询式的send_message可能不够用。如果某个Agent任务要跑几分钟你不想让HTTP连接一直挂着那就应该用send_task_start启动任务、再定时调用get_task获取状态。这个模式我已经在后面的避坑章节细说。4.4 让Agent真正“看到”彼此AgentCard路由实战最后一个实战点利用AgentCard实现动态路由。假设调度中心启动时不知道有哪些Agent在线它可以通过协议自带的“发现”能力去问。from a2a import Client async def discover_agents(agent_urls): client Client() agents {} for url in agent_urls: try: card await client.get_agent_card(url.rstrip(/)) for skill in card.skills: agents[skill.id] url except Exception: # 网络抖动时跳过不拖垮整个调度中心 continue return agents async def main(): registry await discover_agents([ http://localhost:10001/, http://localhost:10002/, http://localhost:10003/, ]) print(registry) # 输出示例{resume_parse: http://localhost:10001/, ...}这段代码的价值在于当你有10个Agent甚至50个Agent时你不需要维护一份硬编码的调用表Agent自己会“报名”。新增Agent只要部署起来、配上AgentCard调度中心下一轮发现循环就能自动识别它。五、常见问题与排查技巧实录5.1 问题速查表这几个坑我全都踩过现象可能原因解决方案客户端报404AgentCard路径必须是/.well-known/agent.json且末尾不能有自定义路径检查服务端路由确认协议能访问到agent.jsonFilePart一直校验不过file字段没有同时给uri和mime_type两个字段必须都提供缺一不可大文件任务超时默认timeout30秒大模型处理时间远超于此调大timeout或改用异步任务轮询模式结果乱序/丢失并发发了多个任务但没有维护task_id的映射关系每个任务用独立UUID并在本地存task_id到请求的映射服务端收不到消息AgentCard里配置的url和实际监听端口不一致检查AgentCard的url字段这是其他Agent调你的唯一依据数据全部变成字符串发消息时用了TextPart却给的是JSON字符串结构化数据请用DataPart不要用TextPart包JSON5.2 从阻塞调用迁移到异步任务模式我在案例里用的是send_message它简单但也有局限——如果任务执行时间太长HTTP连接会被挂起中间任何网络抖动都会导致你误判为失败。所以当你确认某个Agent单次要跑几秒以上建议尽早切换到任务的异步模式。核心方法是send_task_start给Agent发一个启动信号立刻返回。get_task带上task_id去轮询轮询间隔建议2~5秒。如果支持可以订阅事件推送让Agent主动通知你任务完成省去轮询。from a2a import Client import asyncio async def run_long_task(agent_url: str, message): client Client() task_id task-optimize-001 await client.send_task_start( agent_urlagent_url, task_idtask_id, messagemessage ) while True: task await client.get_task(agent_urlagent_url, task_idtask_id) if task.state in (completed, failed): return task await asyncio.sleep(2)这个模式的好处是哪怕任务跑到一半容器重启了只要日志里记了task_id重启后还能继续查询状态。数据不因为进程重启而丢。5.3 独门经验先写AgentCard再写业务逻辑我从这个包上学到的最大教训是——不要一上来就写业务代码先定义好AgentCard和Part类型再往里填逻辑。原因很简单A2A的协作边界完全靠公开的类型和文档来保证。你自己内部怎么实现都行但如果你输出的Part结构不稳定下游Agent就会频繁踩解析错误的坑。我现在的习惯是先写下每个Agent的技能ID和描述发给团队评审。再定义输入输出Part的字段类型写成Python dataclass锁定结构。最后才写内部业务逻辑和异常处理。这样做的体验类比一下就是写微服务之前先定好Swagger文档而不是写完代码再补文档。先定协议边界后期联调的痛苦会少一个量级。另外metadata字段别浪费掉。建议把debug_trace、source_agent、request_time这样的上下文信息塞进去排查问题的时候全靠这些关键信息定位链路。a2a本身不会帮你做链路追踪但你可以用标准字段把业务自己需要的可观测性信息带起来。写到最后想说的这套协议目前还在快速演进阶段a2a-protocol包的版本变数也不小。但无论如何它已经替开发者夯实了大量通用层的工作JSON-RPC封装、AgentCard发现机制、任务状态机、Part结构定义这些都不需要你再去操心。真正常写的是自己的Agent业务逻辑。如果后面还想深挖建议从两方面入手一是把A2A和MCP结合起来用让Agent既能调用工具也能调用其他Agent整条链路就完整了二是自己实现一个A2A服务端脱离Python原生包用Go或Java去造一遍轮子吃透协议细节。我个人更推荐第二条路因为从客户端使用者到协议实现者这个视角切换带来的理解深度是完全不一样的。
