最近我接手了一个内部AI平台升级的活儿要把散落在不同工程团队手里的几个智能体串成一个自动化处理链路。按理说都是智能体但A团队用LangGraphB团队自研状态机C团队直接依赖某个外部托管服务想让它们互相配合光接口参数对齐就能耗掉一个迭代。就在我准备硬写胶水层的时候接触到了A2A这套用于智能体协作的开放协议。这篇把我从概念到实践的调研笔记沉淀下来当作一次学习记录也希望能帮正在做多智能体编排的你少走几步弯路。1. 先搞明白智能体之间为什么需要一套“普通话”1.1 异构Agent越多互连的困难越现实现在的Agent开发可以说是“诸侯割据”。每个框架都有自己的抽象有的把Agent封装成可调用的函数有的暴露成HTTP服务有的只能在框架内部互相协作离开自家生态就寸步难行。你在项目里想用A框架的规划Agent搭配B框架的执行Agent听起来很合理做起来却非常痛苦。需要自己写一层又一层的适配代码把一个Agent的输出翻译成另一个Agent的输入还要处理任务状态同步、错误重试、多轮对话上下文最后往往变成一堆无法维护的补丁代码。我举个例子假设你有两个Agent一个负责从技术社区抓取用户反馈并总结趋势另一个负责根据趋势生成一份周报。如果没有统一协议你得自己定义请求格式、响应结构、错误码、回调地址甚至还要协调两边的时间轮询逻辑。A2A做的事情就是把这层“胶水”标准化让不同框架、不同语言、不同部署方式的Agent之间可以互相发现、互相调用。它定了一套共同的语言至于Agent内部用的是什么大模型、什么推理方式完全不管也完全不需要管。1.2 A2A到底管了什么又没管什么要理解A2A先得划清边界。它管的是智能体之间的“互操作层”也就是两个智能体之间如何建立连接、如何交换信息、如何跟踪一个协作任务的状态。核心东西有这么几块如何发现远程智能体通过Agent Card智能体能力卡片暴露自己是谁、能干什么、怎么连接。如何发起和跟踪任务通过标准化的Task任务概念定义任务从提交到完成的完整生命周期。如何交换消息和产物消息由Part组成Part可以是文本、文件或结构化数据任务可以产出Artifact最终产物。如何处理长任务和异步支持同步响应也支持通过SSE流推送或客户端轮询获取任务进展。如何做认证和鉴权在Agent Card中声明认证方式调用方按标准流程完成鉴权。它明确不管的是智能体内部的思考过程、模型选择、Prompt编排和具体业务逻辑。你可以用任何框架实现远程Agent只要它暴露出来的接口符合A2A标准就能被任何符合标准的客户端调用。这种“管连接、不管内政”的思路和很多基础通信协议的设计哲学是一样的也是A2A能被生态接受的重要原因。1.3 和MCP的关系一个管“手”一个管“口”聊A2A绕不开MCP很多人会把它们搞混。我自己的理解是这样MCPModel Context Protocol解决的是大模型怎么调用工具和数据相当于给模型接上了“手”A2A解决的是智能体怎么和其他智能体对话协作相当于给智能体配上了“口”。两者方向不同但能形成互补关系。实际的协作链路上你会看到一个Agent可能先通过MCP调用内部工具系统完成数据获取再通过A2A把需要专业能力处理的任务委派给另一个远程Agent最后把多方结果汇总起来输出。MCP是模型通向工具世界的标准A2A是智能体通向其他智能体的标准它们不是替代关系而是分层配合的关系。理解了这个图景再看A2A的核心机制就不会觉得它只是个通信协议那么简单了。2. Agent Card远程Agent发给世界的“能力名片”2.1 卡片里有哪些字段哪些字段会影响调用方的决策Agent Card是A2A协议里最具“工程味道”的设计。它本质上是一份JSON文档描述这个Agent的能力边界。调用方拿到卡片后不需要查内部文档就能决定要不要调用、怎么调用。一个标准的Agent Card通常包含这些核心字段字段作用调用方关注点name / descriptionAgent的名称和自然语言描述用LLM来“阅读理解”Agent能干什么urlAgent的端点地址所有任务请求都发往这里version卡片的版本号判断接口是否升级、是否需要重新兼容capabilities支持的能力列表底层包含streamingSend、pushNotifications等子能力决定客户端能否走SSE推送而不是轮询skills每个技能的名称、描述和参数说明判断当前任务是否匹配某个技能authentication调用方需要使用的认证方案确定是匿名、Bearer还是OAuthdefaultInputModes / outputModes支持的消息内容类型决定可以发文本还是发文件我最关心的其实是skills和authentication这两块。skills直接决定我能不能把任务塞给这个远程Agentdescription写得好不好决定了模型判断时会不会出现“茫然不知该选谁”的情况authentication则决定了接入方的开发成本如果服务端要OAuth全套流程那客户端就得先跑通授权这对自动化链路来说不是小事。2.2 用一次真实请求读取Agent Card实际操作的时候读取Agent Card非常简单。协议支持从几个固定的路径发现卡片最常见的做法是直接请求Agent的根路径或/.well-known/agent-card.json。我拿本地起的一个“评论摘要Agent”演示服务举例它只是一个测试用的模拟服务用Python编写。启动之后我直接用curl看它暴露的Agent Cardcurl -s http://localhost:8000/ | python3 -m json.tool返回的卡片长这样{ name: Comment Summary Agent, description: 从产品评论中提取关键观点并按维度生成摘要适合做用户反馈分析, url: http://localhost:8000/, version: 0.1.0, capabilities: { streamingSend: true, pushNotifications: false }, skills: [ { id: summarize_comments, name: Summarize Comments, description: 输入一组评论输出按内容质量、服务态度等维度整理的摘要 } ], authentication: { schemes: [], credentials: none } }这段信息量其实很大。调用方看到streamingSend: true就知道可以建立事件流订阅看到capabilities.skills里有summarize_comments就可以放心发起任务看到authentication为空就知道暂时不需要处理Token。整个“发现-决策-调用”的过程完全不需要翻文档这就是标准化协议带来的好处。2.3 编写Agent Card时容易忽略的两个细节先说description。有些同学写卡片时喜欢写很抽象的话比如“智能处理平台”结果就是调用方的模型无法准确判断该不该调用。更好的做法是像写产品简介一样说清楚输入是什么、输出是什么、适合什么场景。因为我们实际测试中发现客户端经常会用LLM来解析Agent Card描述歧义会直接导致错误路由。再说URL一致性。Agent Card里写的url、实际接收请求的端点、事件流地址必须保持一致不能出现“卡片指向A地址、实际响应在B地址”的情况。我在调试时曾经犯过这个错卡片里url写的是服务网关注口但内部服务实际监听的是另一个端口结果请求发出来直接404。协议本身不会帮你校验这种一致性全靠自己小心。3. 任务的四步生命周期从发起到交付3.1 状态机与流转场景Agent Card解决了“这个Agent能不能做”真正干活要靠Task。在A2A里一次协作就是一个Task它有明确的状态。客户端发起任务后会看到Task在状态机之间流转submitted任务已提交等待处理。working远程Agent正在执行可能持续较久。input-required远程Agent需要调用方补充更多信息这时任务暂停。completed任务成功结束附带最终产物。canceled / failed / rejected任务被取消、执行失败或请求被拒绝。我实际跑过一个真实链路我先发一批评论给摘要Agent它很快进入“working”然后为了生成精细摘要会反问我“是否要按正负面分别输出”这时任务状态变成“input-required”。我补充信息后任务才继续走到“completed”。这个状态机带来的工程价值在于调用方不用自己发明一套任务流程管理逻辑只要围绕状态转移做处理看到working就等待看到input-required就补数据看到completed就取产物。协议把最麻烦的“协作进程同步”问题标准化了。3.2 消息、Part与Artifact三种数据形态的理解很多第一次接触A2A的人会被这个三层模型搞晕我用生活化方式拆一下。Message是一条消息相当于你和对方的一句话。每个Agent会话就是一组有序的Message。Message里包含一个或多个Part。Part是消息的具体内容块可以有三种类型text纯文本、file文件引用、data结构化数据。举个例子客户端发给摘要Agent的消息可以是一个包含文本评论的text Part也可以附带一个CSV文件引用作为file Part。Artifact则是任务完成后交付的结构化产物通常是一份文件或结构化JSON。它和Message的区别在于Message是过程中的对话Artifact是发呆到最后的交付物。你拿到Artifact任务才算真正完成。理解了这个模型写客户端就很清晰了拿产物看Artifact看过程进展就看Message列表里的Part。3.3 同步执行还是异步推送两种模式的分界协议支持两种执行模式选哪种直接影响客户端开发复杂度和用户体验。第一种是同步等待。客户端发请求后服务端在单个HTTP响应里就返回了完成的Task。适合那些几十上百毫秒内就能完成的小任务比如简单的文本分类、抽取关键字。第二种是异步长任务。服务端收到请求后立刻返回一个未完成的Task同时通过SSE推送事件流任务有进展就发来状态更新。适合生成式摘要、批量数据处理这些需要几秒甚至几十秒的耗时操作。具体怎么选我自己的判断标准很简单任务平均耗时超过2秒或者可能触发多轮交互就直接走异步推送否则可以先用同步试一试。但有一个坑要注意哪怕服务端支持同步返回客户端也要做好处理“Task尚未完成”这个响应的准备。因为服务端可能因为负载变化突然把一个原本秒回的任务改成异步执行这是协议允许的。也就是客户端必须具备消费事件流的能力不能赌它一定同步。4. 实战用Python走通一个A2A客户端4.1 先用手头的HTTP工具验证真实交互在动手写客户端之前我习惯先用curl把服务端的交互摸清楚这样后面代码才能写得一次通过。还是以那个“评论摘要Agent”为例先用curl发一个任务请求curl -X POST http://localhost:8000/agent/tasks/send \ -H Content-Type: application/json \ -d { id: task-001, message: { role: user, parts: [ { type: text, text: 请对以下评论做摘要物流很快但包装有破损客服态度不错退款速度一般。 } ] }, metadata: {} }服务端可能返回一个已经完成的任务也可能返回一个还在“working”的任务。我初测时碰到的就是第二种响应里带上了事件流地址{ id: task-001, status: working, eventsUrl: http://localhost:8000/agent/events/task-001 }接着我用curl连上事件流观察进展curl -N http://localhost:8000/agent/events/task-001事件流会持续推送类似TaskStatusUpdateEvent的事件直到出现completed状态。这一步走通了后面Python客户端的逻辑就只是把这套请求流程代码化而已。4.2 一个最小的Python客户端我用Python快速写了一个最小可用的A2A客户端只依赖requests库。核心逻辑分三步获取Agent Card、发送任务、按需监听事件或轮询。import json import time import requests class SimpleA2AClient: def __init__(self, base_url): self.base_url base_url self.agent_card None def load_agent_card(self): resp requests.get(self.base_url) resp.raise_for_status() self.agent_card resp.json() return self.agent_card def send_task(self, task_id, text): payload { id: task_id, message: { role: user, parts: [{type: text, text: text}] } } resp requests.post( f{self.base_url}/agent/tasks/send, jsonpayload, headers{Content-Type: application/json} ) resp.raise_for_status() task resp.json() if task.get(status) completed: return task return self._wait_for_completion(task) def _wait_for_completion(self, task): events_url task.get(eventsUrl) if events_url: # 简化为轮询获取任务状态 task_id task[id] while True: resp requests.get(f{self.base_url}/agent/tasks/get, json{id: task_id}) task resp.json() if task[status] in (completed, failed, canceled): return task time.sleep(1) return task if __name__ __main__: client SimpleA2AClient(http://localhost:8000/) card client.load_agent_card() print(Agent Name:, card.get(name)) task client.send_task( task-demo-001, 物流很快但包装有破损客服态度不错退款速度一般。 ) print(Task Status:, task[status]) if task.get(artifacts): for artifact in task[artifacts]: print(Artifact:, artifact[parts])这里我把事件流监听简化成了轮询因为轮询在任何环境下都不会出错。如果服务端支持并且你想追求实时性可以把_wait_for_completion改成对接SSE。不过要注意轮询间隔别设太短1秒左右在实际测试中已经够用太频繁反而容易触发服务端限流。4.3 把单轮变成多轮处理input-required只处理单轮任务其实是远远不够的。真实业务里远程Agent经常会要求补充信息比如摘要Agent需要你指定输出维度。这时客户端如果不会继续发送消息链路就卡死在“input-required”状态。处理逻辑也很直观当任务状态是input-required时客户端读取远程Agent发来的提示消息构造新的消息再发给它。继续上面的代码我加了一个简单的交互分支def send_task_with_feedback(self, task_id, text): task self.send_task(task_id, text) while task.get(status) input-required: # 读取Agent催问的提示这里简化处理取最后一条消息的文本 parts task[message][parts] prompt .join(p[text] for p in parts if p.get(text)) reply input(fAgent 想让你补充信息: {prompt}\n你的回复: ) resp requests.post( f{self.base_url}/agent/tasks/sendMessage, json{ id: task_id, message: { role: user, parts: [{type: text, text: reply}] } } ) task resp.json() if task.get(status) completed: return task return task这里有个小细节发送补充消息用的方法是/agent/tasks/sendMessage不是重新send因为任务已经存在了不能重复创建。我在第一次写的时候就在这里栽过跟头把回补信息当成新任务发出去结果远程Agent直接拒绝。4.4 我遇到的两个真实小坑第一个坑是事件流断线后没人管。当时我开了SSE监听任务中间网络抖动断了以为任务丢了其实服务端还在执行。正确做法是按协议规范通过resubscribe重新订阅而不是傻等或者直接判定失败。如果服务端不支持重订阅那就退化成轮询千万不要把“事件流断线”等价于“任务结束”。第二个坑是超时设置。Python的requests如果默认不设超时可能一直挂着等响应一旦服务端同步处理时间稍长客户端就像死掉一样。我现在所有的HTTP调用都会显式设置timeout比如发送任务给10秒轮询给5秒这样至少能保证错误快速暴露而不是卡到天荒地老。5. 集成经验与扩展方向5.1 什么样的场景值得用A2A调研的时候我专门做过对比想搞清楚“什么情况下值得引入A2A什么情况下直接写接口就行”。我的结论是如果远程Agent只服务于内部一个明确的调用方而且你们自己控制前后端两端直接写REST接口反而更省事A2A的收益不明显。但如果Agent是给多个团队共用、或者将来要开放给外部系统A2A的价值就会显现出来。它最大的好处是把“能力暴露”和“任务协作”的成本沉淀为标准协议。调用方不需要读你内部代码只要看Agent Card就能接入你也不需要为每个调用方各写一套对外接口只需要维护一份卡片和一套标准端点。尤其是平台化场景多个Agent要互相编排用非标准接口组合起来会是灾难而用A2A写出来的链路逻辑清晰度完全不一样。5.2 安全和治理的几个注意点安全方面A2A把认证方式放在Agent Card里声明所以接入方可以尽早拿到鉴权预期。我实际集成时重点做了三件事第一生产环境不要裸奔。本地测试时我用的是credentials: none但任何暴露给多方的远程Agent都应该启用Bearer Token或OAuth。我现在的做法是在Agent服务前面加一层统一鉴权网关所有A2A请求先经过网关校验再转发到业务服务。第二注意任务ID的幂等控制。同一个任务ID重复提交可能造成重复执行尤其是在重试场景下很危险。客户端生成任务ID时要用UUID或带业务语义的唯一标识服务端也要对已处理过的ID做好去重处理。第三日志审计要覆盖全链路。A2A通信中身份、请求体、状态变化和Artifact都要能追踪。多Agent链路上出了错没有日志根本定位不了是哪一跳出了问题。我现在会给每个任务维持一个链路ID贯穿所有Agent排查问题效率提升非常明显。5.3 后续场景我可以怎么扩展协议的价值不在于单次调用而在于可组合。我目前规划了三个方向。第一个是把历史API包装成A2A Agent。旧系统没有智能体概念但可以写一个适配层Agent对外通过Agent Card暴露能力对内调用旧接口。这样新系统里的其他Agent就能像调用智能体一样使用这些老服务不用改造老代码。第二个是做一个轻量级的任务编排器。不依赖复杂的编排框架用A2A把多个Agent串成流水线先安排一个规划Agent拆分任务再让执行Agent按步骤处理最后由汇总Agent归并结果。由于各步之间都用标准Task接驳编排器本身只需要纯粹的状态路由逻辑。第三个是探索人机协同场景。A2A天然支持input-required状态意味着Agent在执行过程中可以随时向人提问这个特性很适合做提交前审核、敏感操作二次确认。我打算把发布类流程设计成“Agent完成初稿人工确认后放行”的模式这样既有自动化效率又有了人类监督的保障。最后聊一点我自己的体会。协议还在快速演进阶段很多细节未来还会调整但“智能体之间应该用标准化协议对话”这个方向已经很明确了。我现在做内部集成时其实并不急着把所有Agent都推倒重来而是先把对外能力全部用Agent Card描述清楚再逐步把关键协作链路迁到A2A上。你有兴趣的话也可以先从自己的一个Agent开始写一份Agent Card跑通一次最简单的任务调用这个成本很低但对理解整套机制帮助很大。