A2A协议深度解析：从Agent互操作到多Agent协同编排实战

1. A2A协议到底解决什么问题1.1 多Agent协作的三连问最近圈子里聊AI Agent的人明显多起来了但聊着聊着总会撞上同一个尴尬每个Agent都挺能干可一旦要让几个Agent互相配合就立刻变成“接口对接事故现场”。A方用JSON传消息B方要求XMLA方把任务结果放在HTTP Response里B方非要轮询一个任务ID更别提各家自定义的鉴权方式联调一次能磨掉半条命。这种局面其实特别像早期IM软件各自为战的时代——你装了A软件的聊天工具就没办法和装B软件的人直接对话。Google在2025年4月开源的这个A2A协议Agent-to-Agent Protocol目标就是给智能体之间定一套统一的“普通话”让不同团队、不同框架、不同技术栈开发出来的Agent能直接对话、派活、交付结果而不是每家都自说自话。这篇文章我会从协议的核心设计讲起再带着你手动跑通一个客户端和服务端最后聊一聊在多Agent协同场景里的真实编排方式和并发调优适合正在做Agent产品、或者准备在团队内部搭建多Agent协作体系的开发者参考。先说结论A2A解决的是“智能体与智能体之间的互操作”问题它和另一套很火的MCP协议解决的不是同一层问题。MCPModel Context Protocol把工具、数据源以标准化方式暴露给Agent解决的是“Agent怎么调用工具”A2A则是让Agent之间互相发现、互相委派任务并回收结果。如果把单个Agent比作一个员工MCP就是员工手里能用的各种办公工具和资料库A2A则是员工之间沟通协作的流程和语言规范。这两者不冲突反而天然互补。1.2 A2A协议给的标准答案A2A的核心设计目标用一句话概括就是让多Agent协作像HTTP调用一样简单。Google联合了超过五十家技术伙伴一起推动这套协议包括业内主流的大模型厂商、云计算平台和AI创业公司意图很明显——大家都别搞私有协议了不然整个生态会碎成一地。协议底层基于JSON-RPC 2.0传输层默认走HTTP消息结构全部用JSON表达这套技术选型非常务实任何会写HTTP接口的开发者都能在半小时内上手不需要引入特殊的消息中间件或SDK。协议里最关键的一个概念叫Agent Card。你可以把它理解成每个Agent对外发布的“名片”用JSON格式声明这个Agent叫什么、擅长什么、接受什么类型的任务、回调地址在哪里。客户端拿到这张名片就能判断要不要把某个任务交给它。这个设计和现实世界的招聘流程很像招聘方客户端先看简历Agent Card觉得匹配了再安排面试发送任务。另一个核心概念是任务Task。A2A把一次完整的协作定义成一个TaskTask有生命周期从submitted、working、completed到failed。客户端提交任务服务端处理任务并返回状态整个交互围绕任务ID展开而且支持同步返回和异步推送两种模式。后面我会展开讲这两个模式在实际场景中的取舍。2. A2A协议核心概念逐层拆解2.1 Agent Card智能体的“自我介绍”每一份Agent Card就是一份JSON文档通常会被放在Agent服务的一个固定路径下比如/.well-known/agent.json。客户端通过这个路径去发现和获取Agent的能力信息。一个基本的Agent Card长这样{ name: research-agent, description: 负责检索公司内部的文档库并生成摘要, url: https://agent.example.com, version: 1.0.0, skills: [ { id: document_search, name: 文档检索, description: 搜索内部Confluence和飞书文档 } ], capabilities: { streaming: true, pushNotifications: false } }字段看起来不多但每一个都在决定协作的顺畅程度。skills字段是Agent能力清单的“入口”客户端不需要提前写死每个Agent能干什么而是动态读取这份清单来做路由。当你有多个Agent同时在线最自然的做法就是把它们的Card收集到一个注册中心让调度层根据请求内容来选择匹配的Agent。capabilities则告诉调用方这个Agent支持什么交互模式支持streaming客户端就可以建一条流式连接持续接收输出支持pushNotifications服务端处理完任务后会主动回调。这两个能力直接影响你客户端代码怎么写也决定了任务耗时特别长时的交互方案。我建议从一开始就把Agent Card放在一个公开且稳定的URL下并且保持路径不变只更新内容。原因很简单A2A生态里会有各种Agent发现机制它们默认去/.well-known/agent.json抓取信息。如果你把路径随便改来改去会造成调度方的404和误判。2.2 Task与Message把一次协作变成一张“工作单”A2A协议里Task是协作的基本单元。整个生命周期是客户端创建Task服务端接收后进入working状态处理完成后置为completed或failed。这不就是我们日常工作中的工单系统吗你提交一个工单系统派发给处理人处理人更新状态你在工单里看到进度和结果。A2A把这一套搬到了智能体世界。一个Task下挂若干Message。Message是实际的内容载体有role字段user或agent有content字段支持纯文本和结构化JSON还可以带附件或部件。多个Message围绕同一个Task进行多轮对话。这里有个设计细节值得注意A2A的消息交互方式不是“客户端发一次、服务端回一次”就结束而是允许围绕同一个Task进行多轮、异步、双向的消息交换。任务执行到一半服务端可以给客户端发消息补充问题客户端也可以给服务端追加材料。这在实际业务中非常有用——比如一个Agent在执行“帮用户订机票”的任务时发现用户没有指定舱位它可以返回一个input-required的状态反过来向调用方要信息而不是干等或者报错。对于长任务比如生成一份行业研究报告、跑一个数据管道A2A的异步模式是标准解法。客户端提交任务后拿到taskId之后要么轮询任务状态要么等服务端回调通知不需要一直占着HTTP连接不放。这是协议拓展性的关键。2.3 认证与安全企业落地的关键一环任何协议只要走向真实业务认证和安全就是绕不开的坎。A2A在设计上走的是“实用主义”路线它没有重造一套复杂的认证加密体系而是支持标准HTTP认证方案。从最基础的API Key、Bearer Token到企业环境的OAuth2.0都可以根据场景灵活配置。你在部署时需要注意的是在微服务内部调用链路上可以考虑简单的共享密钥但一旦Agent对外部生态开放就必须上OAuth2或JWT并且要在Agent Card里声明支持的认证方式和凭证获取URL。官方文档里专门强调协议本身允许在WSS/HTTPS之上运行生产环境强制要走HTTPS防止任务内容和Agent Card被抓包窃取。安全这块我的建议是先想清楚你的Agent是“私有Agent”还是“开放Agent”。如果是服务公司内部多个团队共用的建议在网关层统一做身份认证和权限校验业务逻辑里不要重复造轮子如果是面向外部开发者开放的AgentAgent Card里的描述一定要精确到技能粒度防止别人拿你的Agent去做计划外的操作。3. 从零跑通一个A2A实例3.1 环境准备与服务端骨架理论讲再多不如直接跑一遍。我用的环境是Python 3.10 FastAPI为什么选FastAPI因为它配合uvicorn可以快速搭出一个异步HTTP服务和A2A的任务模型天然匹配。你也可以用Flask或Spring Boot协议本身不限语言。先装依赖pip install fastapi uvicorn httpx pydantic然后建一个目录结构如下。这里我刻意拆分了模块是为了后面扩展多Agent时更清晰单一Agent原型你完全可以把代码写进同一个文件里。a2a-demo/ ├── server.py # A2A服务端 ├── client.py # A2A客户端 └── agent_card.json # 名片文件服务端实现的核心是两件事暴露Agent Card以及处理JSON-RPC请求。先写一个最简版本的Agent Card文件{ name: summarize-agent, description: 接收一段文本返回200字以内的摘要, url: http://localhost:8000, version: 1.0.0, skills: [ { id: text_summarization, name: 文本摘要, description: 对输入的文本生成摘要 } ], capabilities: { streaming: false, pushNotifications: false } }3.2 用Agent Card暴露能力服务端代码里第一件事是把Agent Card挂到公开路径下。我习惯用/.well-known/agent.json这个标准路径后续任何A2A兼容客户端都能第一时间发现我。from fastapi import FastAPI, Request from fastapi.responses import JSONResponse import json app FastAPI() with open(agent_card.json, r, encodingutf-8) as f: agent_card json.load(f) app.get(/.well-known/agent.json) async def get_agent_card(): return JSONResponse(contentagent_card)这一步的原理很简单客户端调用这个端点获取Agent的自我介绍然后决定是否要跟它协作。如果你的Agent部署在内网也可以在服务注册中心里维护Agent Card列表由调度器统一管理并下发。3.3 客户端发现、连接、提交任务接下来是服务端的核心逻辑处理message/send这个JSON-RPC方法也就是接收任务并返回结果。我先实现一个同步版本客户端发消息过来服务端处理后直接返回不搞异步回调。from pydantic import BaseModel class JSONRPCRequest(BaseModel): jsonrpc: str 2.0 id: str method: str params: dict app.post(/) async def handle_rpc(request: JSONRPCRequest): if request.method message/send: params request.params task_id params[taskId] message params[message] text message[content][text] summary summarize(text) return { jsonrpc: 2.0, id: request.id, result: { task: { id: task_id, status: completed, artifacts: [ {content: {text: summary}} ] } } } return { jsonrpc: 2.0, id: request.id, error: {code: -32601, message: Method not found} }summarize函数这里我就用最朴素的实现策略按句子数量截取前若干句或者直接调用任意大模型接口。在企业项目里这里就是对接真实业务逻辑的地方。客户端的任务发现和提交也很简单我用httpx实现了一个最简客户端import httpx def discover_agent(base_url: str): resp httpx.get(f{base_url}/.well-known/agent.json) return resp.json() def send_task(base_url: str, text: str): payload { jsonrpc: 2.0, id: client-1, method: message/send, params: { taskId: task-001, message: { role: user, content: {type: text, text: text} } } } resp httpx.post(base_url, jsonpayload) return resp.json()3.4 运行效果与验证分别在两个终端运行uvicorn server:app --port 8000 python client.py客户端会先取到Agent Card打印出Agent的能力说明然后把“这篇文章实在太长了帮我总结一下核心观点”发给服务端拿到一个completed状态的任务结果。整个链路跑通之后你会发现A2A的本质一点也不神秘它就是一套约定好格式的HTTP接口规范但正是这套约定让不同系统间的Agent可以不再互相“开发适配器”。这里有个坑提醒一下JSON-RPC请求里的id字段建议用UUID不要用固定值。我见过有人为了调试方便把所有请求的id写死为1结果在追踪任务日志时完全分不清是哪一次调用排查问题直接变成一场灾难。4. 多Agent协同实战从单点调用到协同编排4.1 一个真实场景三条Agent的协同流水线单点调用是基本功但只调一个Agent根本算不上多Agent协同。我做一个你大概率有共鸣的场景假设你在做一个企业内部的“智能工作助理”需要完成一个需求“帮我的新业务写一份市场分析摘要并预约下周三和团队开会”。这个需求拆开至少涉及三个Agent的能力搜索Agent检索公开市场报告和行业动态输出原始材料分析Agent对检索到的材料进行整理、提炼形成一份分析摘要日历Agent根据团队忙闲状态预约会议室和发送邀请如果没有A2A实现方式就是在一片业务代码里按顺序依次调用三个Agent的私有SDK接口每个接口的鉴权方式、参数格式、错误处理都不一样代码里全是适配逻辑。有了A2A之后每个Agent只要实现标准的Agent Card和任务处理端点编排层就可以用统一的方式把它们串起来。4.2 协同编排的三种常见模式根据实际项目经验多Agent协同大约能归成三类模式分量对应三种代码组织方式。第一种是流水线模式前一个Agent的输出是后一个Agent的输入上面那个市场分析场景就是这种结构代码上一般用队列或链式调用来表达。第二种是扇出汇聚模式比如让三个不同行业的搜索Agent同时搜索材料全部结果回来后再汇总给分析Agent这种模型需要用并发调用来实现后面讲并发配置时细说。第三种是动态路由模式调度者先看任务内容再根据Agent Card里的skills动态选择合适的下一跳Agent本质上是把路由表隐藏在能力注册表里。三种模式可以混用。我在真实项目里的做法是用A2A作为统一通信层在编排层用一个简单的事件循环驱动具体流程用配置去描述而不是硬编码。比如用一份YAML定义“第一步调用search-agent第二步把所有结果拼起来调用analyze-agent第三步调用calendar-agent拿到可预约时间”。这样Agent升级和替换只需要改配置和Agent Card编排代码一行都不用动。4.3 并发配置与性能调优多Agent协同最头疼的是并发尤其是“扇出汇聚”模式。如果你的编排器串行调用三个搜索Agent总耗时会等于三个Agent耗时的累加而并发调用理论上只要一个最长耗时的Agent完成就能汇总。实际项目中我一般用两种手段做并发控制协程和信号量。Python的asyncio.gather是最直接的方式。把所有子任务丢给事件循环并发执行代码非常简洁import asyncio import httpx async def call_agent(base_url: str, task_id: str, text: str): async with httpx.AsyncClient() as client: resp await client.post(base_url, json{ jsonrpc: 2.0, id: task_id, method: message/send, params: { taskId: task_id, message: { role: user, content: {type: text, text: text} } } }) result resp.json() return result async def fan_out(text: str): agents [ (http://localhost:8001, task-search-1), (http://localhost:8002, task-search-2), (http://localhost:8003, task-search-3) ] results await asyncio.gather( *[call_agent(url, tid, text) for url, tid in agents] ) return results这里有个重点注意事项并发调用时最好对每个Agent设置超时和重试策略。在A2A的标准实践里任务模型支持taskId的幂等哪怕同一个请求因为超时被重复提交服务端也能识别出来不至于执行两遍业务逻辑。这在“预约会议”这类有副作用的场景里至关重要否则一次网络抖动可能导致会议室被重复预定。还有一个非常关键的经验A2A服务端在接收任务时要尽快返回一个working状态而不是傻傻地同步等待任务彻底完成后才返回结果。否则一旦某个Agent的执行时间超过HTTP连接的保持时间客户端那边就会报网关超时。正确的姿势是第一服务端接收任务后立刻返回working taskId第二任务真正执行完后如果Agent Card声明了pushNotifications就通过回调通知客户端。如果声明了streaming则可以用流式的通道把结果一点一点推给客户端。我自己的项目里通常优先用sse协议做流式输出不仅实时性好也天然兼容Httpx和浏览器。5. 经验总结与踩坑记录5.1 最容易踩的五个坑多Agent协同的坑多到可以单独写一本排错手册。我挑五个最高频的分享给大家。第一个坑是Agent Card的URL和实际服务地址不一致。很多人把Agent Card写到注册中心之后又换了服务部署的域名或端口没有同步更新导致客户端发现Agent成功真正调用时却连接失败。这个坑的排查成本极高因为报错信息往往只会显示“invalid agent url”你根本想不到是名片里的URL没同步。第二个坑是任务状态只置completed或failed缺少中间态。A2A设计的初衷是支持长任务如果你的Agent服务端没有把working状态回传给客户端那客户端收到的最自然结果就是等待超时然后莫名其妙地重复发送任务。这在企业级编排链路里会造成大量重复计算。第三个坑是消息内容结构不统一。A2A的Message.content既支持纯文本也支持structured结构。常见的团队协作事故是有人把内容包装成{type:text,text:...}格式有人直接传一个纯字符串结果服务端解析时报错。好的做法是团队内部约定好所有发送过来的消息统一先做格式校验再进业务逻辑。第四个坑是鉴权设计过度。A2A在HTTP协议层明文传递任务内容直接上生产环境又不走HSM加密风险很大但反过来说如果Agent本身只是内部工具却套了复杂的OAuth2授权码流程每次调用都要重新经历一次完整的认证会严重影响协同效率。我的建议是“按需分级”内部Agent用轻量API Key外部Agent用OAuth2一层网关搞定别在Agent里去实现完整的SSO逻辑。第五个坑是Agent路由缺乏兜底。动态路由模式很爽但一旦Agent Card的skill描述写得不够准确调度系统就可能把任务发给一个完全不合适的Agent。实战中要设置兜底策略匹配度低于阈值时把任务交给人工或退回给用户重新描述而不是强行塞给某个Agent去试错。5.2 什么时候选A2A什么时候不选A2A不是银弹这一点必须说清楚。如果你只是在做一个单Agent应用通过函数调用解决单一问题那根本不需要A2A直接写代码比引入一套协议划算得多。A2A的收益要等到Agent数量变多、团队之间需要共享Agent能力时才显现出来。我的判断标准很简单当你发现自己开始为每个Agent写不同的HTTP调用封装、并且组内每个工程师都在重复这套工作时就是接入A2A的合适时机了。反过来如果团队已经有了成熟的内部消息中间件和一套好用的Agent编排框架强行改成A2A可能得不偿失。协议的好处是标准化带来的互操作代价是额外的抽象层和性能开销。取舍的核心永远是业务场景而不是技术趋势。我在实际项目里体会最深的一件事是A2A真正降低的是“新Agent接入成本”。以前团队里某个同事训练好一个新Agent想让它被其他应用调用需要有人专门研读这个Agent的接口文档、写适配层、处理各种私有协议。现在只要这个Agent实现了A2A把Agent Card挂到注册中心整个体系里的其他组件就自动拥有了调用它的能力。这种“能力即插即用”的体验一旦适应就很难回去了。最后再分享一个小技巧多Agent协同调试时给每个Agent配一个统一的taskId前缀规则比如search-、analyze-、calendar-。配合日志系统之后你就可以用前缀快速追踪到某条链路上的所有任务记录。这个习惯帮我省了无数排查问题的时间也强烈推荐给你试试。