几乎所有做 AI Agent 的人都会遇到同一个断头路Agent 单独跑没问题一旦要让两个 Agent 协作立刻陷入接口地狱。我在公司落地跨部门 Agent 集成时对这种痛苦记忆犹新——A 团队用 LangGraphB 团队自研了一套调度框架两边都无法直接调用对方的 Agent最后只能写一堆互相调用的胶水接口业务一变动就全部返工。A2A 协议Agent2Agent Protocol就是为解决这件事而设计的开放通信协议。它把 Agent 之间的对话方式统一成一套规范让不同厂商、不同技术栈的 Agent 彼此发现能力、创建任务、交换成果物。这篇不是官方文档的翻译我把从读协议到实际跑通一个最小服务端的完整过程、关键设计取舍、还有踩过的坑整理一遍给正在集成多 Agent 的同学当个参考。1. 多 Agent 协作的断头路A2A 到底解决什么问题1.1 每个 Agent 都是信息孤岛现在的 Agent 生态工具层已经通过 MCP 逐渐打通了但 Agent 与 Agent 之间仍然是孤岛。LangGraph、AutoGen、CrewAI 这些框架都有各自的协作编排方式可它们只能在框架内部协作。跨框架、跨团队、跨组织你要让一个 Agent 调用另一个 Agent只能对着别人封装出来的 HTTP API 写客户端代码对方换个字段你就要改代码。更麻烦的是你不知道对端 Agent 到底有什么能力、期望什么输入、返回什么格式——除非对方给你一份文档而文档几乎总是滞后于代码。我在实际项目里碰到过一个很典型的场景客服 Agent 需要实时查询物流状态而数据在另一个组的物流分析 Agent 手里。理论上一个 Context 传递就能完成但两个团队为这个接口来回开了三次评审会。每次对参数、对返回结构、对异常语义本质上是手工做一次协议设计。如果 Agent 数量从两个增长到十个这种手工协议是不可维护的。1.2 协议的目标不是发明新传输而是定义协作语义A2A 的定位很明确它不发明新的传输协议而是定义 Agent 之间协作的语义层。底层继续使用 HTTP消息格式基于 JSON-RPC 2.0这些都是行业里跑了多年的存量标准。协议要解决的是如何描述 Agent 能力如何发起任务如何追踪任务状态如何交付成果物这一套应用层问题。类比一下HTTP 定义了 Web 页面的传输方式但真正让不同网站能够互通的是 HTML 这个语义标准。A2A 在 Agent 生态里扮演的角色就像 HTML 之于 Web 页面——它规定了一个 Agent 该以什么结构介绍自己、该用哪些方法响应请求、该怎样表达任务状态。只要两端都遵循这套约定双方内部用什么框架、什么语言、什么模型都不重要。1.3 先划清概念边界提到通信协议很容易和硬件层的 I2C、UART、SPI、CAN 这类术语混淆。那些是设备与设备之间的传输层协议解决的是数据在物理线路上怎么走的问题A2A 是应用层协议解决的是 Agent 之间如何请求、响应、流转状态的问题。两者抽象层次完全不同。我见过有人把这两件事放在一起聊结果讨论半天发现不在一个频道。搞清楚这个定位再看 A2A 的设计就顺了——它聚焦的是协作而非传输。那协作的第一步是什么是让一个 Agent 知道另一个 Agent 是谁、能干什么。2. 认识 Agent Card一场协作从自我介绍开始2.1 客户端如何知道对面是谁、能干什么先说一个最基础的问题如果我现在要调用你的 Agent我怎么知道它存在怎么知道它擅长什么这就是 A2A 里 Agent Card 要解决的。Agent Card 是一份 JSON 文档描述一个 Agent 的基本信息、提供者、能力和认证方式。客户端拿到这份文档才知道该向哪里发请求、能请求什么服务。协议把 Agent 的名片标准化了这就是协作的第一步——能力发现。一个典型的 Agent Card 结构以 0.2.x 版本的常见字段为例大致如下{ identifier: supply-chain-agent-v2, name: SupplyChain Analyzer, description: 负责供应链数据分析、异常检测和配送方案重算的 Agent, url: https://agent.example.com/, version: 2.1.0, skills: [ { id: recompute_plan, name: Recompute Delivery Plan, description: 根据新的约束条件重新计算配送方案返回新的路线和时效预估, tags: [supply-chain, delivery], examples: [供应商延迟两天重新安排华东区域的配送计划], inputModes: [text, file], outputModes: [text, file] } ], provider: { organization: Example Corp, url: https://example.com }, capabilities: { streaming: true, pushNotifications: false } }name、description、url、version 这些字段都很直觉。真正值得关注的是 skills 字段——它定义了这个 Agent 能执行哪些技能以及每个技能的输入输出模式。客户端拿到 skills就能判断这个 Agent 能不能帮我干这件事以及我该怎么组织我的请求。2.2 skills 里藏着协作能不能成的关键很多第一次接触 A2A 的人会把 skills 当成接口文档来看其实更准确的类比是它是一份面向 LLM 的能力说明书。description 和 examples 写得好不好直接决定调用方的模型能不能理解这个技能的含义。我自己的经验是描述要写清楚三件事这个技能解决什么问题、期望什么输入、会返回什么类型的成果。如果可能给一两个非常具体的 example。比如recompute_plan的 description 写清楚根据新的约束条件重新计算配送方案比你只写配送方案重算有用得多——因为调用方 Agent 的模型是拿着这段描述去理解语义的描述模糊协作就会卡在第一步。输入输出模式也值得注意。inputModes和outputModes支持 text、file、data 几种类型这决定了协同时能交换什么形态的内容。如果一个技能输出的是 PDF 报告那outputModes至少要包含 file如果只是文本摘要text 就够了。声明得越清楚调用方的判断越准确。2.3 发现路径well-known 与兜底 URLAgent Card 放到哪里协议也有约定推荐放到/.well-known/agent.json这是沿用了 Web 生态里 well-known URI 的惯例同时很多实现也支持GET /agent.json作为兜底。客户端通过 HTTPS 访问 Agent 的域名先尝试 well-known 路径再回退到 agent.json拿到文档后按 schema 校验。选择 well-known 路径的用意是让 Agent 的能力发现变成一件完全无副作用的只读操作。调用方不发起任何任务只是获取一份公开声明就像我们先访问一个网站的 robots.txt 再决定要不要爬取。这种设计让 Agent 之间的集成从找负责人要接口文档变成了访问一个固定路径就能看到全部信息对整个集成成本的影响是巨大的。我在联调时踩过一个很低级的坑Agent Card 里填的 url 是http://localhost:8000本机 curl 完全正常结果让另一台服务器的客户端访问时它拿到 Card 后去请求 localhost当然是请求自己直接报连接拒绝。这个小小字段在生产环境里要填对端能访问到的真实地址不是本机调试地址。如果要通过反向代理暴露也记得一并把路径配好。3. 任务与消息Agent 协作的基本单元3.1 Task 状态机一条任务从生到死A2A 里最核心的概念是 Task任务。一次 Agent 间的协作本质就是一个或多个 Task 的创建和推进。为什么强调任务而不是对话因为任务有明确的边界和状态适合工程化管理。Task 有状态机常见状态包括submitted任务已提交等待处理workingAgent 正在执行中input-required需要调用方补充信息completed任务已完成failed任务失败canceled任务已取消unknown状态未知比如服务端重启后任务上下文丢失客户端通过 Task 的 id 进行后续操作查询状态、追加上下文、取消任务。状态机的存在让一次 Agent 调用不是一个返回后就消失的黑盒而是一个可以追踪、可以恢复、可以取消的实体。这个特性对生产环境特别重要一次跨 Agent 的执行可能持续几分钟调用方必须能随时确认这事到底干到哪一步了。3.2 Message 与 Part协作里传递的不只是文本Task 的执行过程通过 Message 来传递。Message 有 role要么是 user调用方要么是 agent服务方和聊天消息里用户/助手的结构类似。不同的是每条 Message 内部是若干 PartPart 有类型TextPart纯文本最常见FilePart文件内容可以是文件 URL 或 base64 编码的数据DataPart结构化数据比如 JSON这个 Part 机制解决了一个很实际的问题Agent 之间协作不可能只传文字。比如供应链 Agent 要传一份 Excel 报表给分析 Agent可以用 FilePart 携带文件引用要传一批结构化订单可以用 DataPart 直接携带 JSON。Part 的存在让 A2A 既能处理轻声细语的要求也能处理文件级别的交付物。3.3 Artifact任务的成果物任务执行完成后产出物叫 Artifact。Artifact 和 Message 的关系容易混淆——简单理解Message 是沟通过程中的对话记录Artifact 是最终交付的成品。一次完成的任务里可以包含多个 Artifact每个 Artifact 又有自己的名字和 Part 列表。比如你让一个 Agent 生成一份市场分析 PPT它的 Artifact 可能是一个 FilePart指向 PPT 文件的下载地址同时它的 messages 里可能保留着它过程中的思考摘要。调用方关心的是 Artifact过程记录可以按需查看。3.4 为什么协作要任务化而不是纯对话回到开头那个问题直接做一个类似 ChatGPT 的多轮对话接口不就行了为什么要搞出一套 Task 状态机原因在于生产环境里 Agent 协作很少是一问一答。一次业务协作可能要持续几分钟甚至更长比如一个 Agent 检索完数据、调用另一个 Agent 生成报告、再等第三个 Agent 做校验。这么长的链路调用方必须有办法确认任务进行到哪一步了才有可能做超时管理、重试、恢复。如果只是简单的对话接口一断线就全丢了这在企业内部工作流里是不可接受的。任务化把协作变成可追踪、可干预、可恢复的实体这才是它能走上生产线的关键。4. 通信语义与传输机制send、stream、轮询与推送4.1 协议骨架HTTP JSON-RPC 2.0A2A 的方法调用基于 JSON-RPC 2.0。每个 Agent 暴露一个 HTTP 端点通常是 POST /客户端发 JSON-RPC 请求Agent 返回 JSON-RPC 响应。协议里定义了几组方法覆盖了从发送消息到管理任务的全流程方法作用message/send向 Agent 发送一条消息返回任务当前状态message/stream以服务端流式方式发送消息持续推送任务状态和产出task/get根据 task_id 查询任务当前状态task/cancel取消一个任务task/pushNotificationConfiguration配置任务状态变化的推送通知task/resubscribe在 stream 连接断开后恢复订阅这里有个细节值得留意message/send 和 message/stream 的入参结构基本一致差异只在返回方式。send 是同步返回一次响应stream 会把后续的状态变化通过 SSE 持续推给客户端。这个设计让协议既能满足简单同步调用的场景也能服务长时运行任务的场景。4.2 简单场景用 message/send 完成一次同步调用如果任务很快就能完成用 message/send 最直接。请求长这样{ jsonrpc: 2.0, id: req-001, method: message/send, params: { task_id: task-123, message: { role: user, parts: [ { kind: text, text: 请分析这份订单数据返回异常项清单 } ] } } }服务端会创建或续接一个 Task返回该 Task 的当前快照。如果任务已经完成可以直接在响应里看到 messages 和 artifacts。这是一个非常标准的请求-响应模型和调普通的 REST API 没什么两样适合上下游都是轻量业务的场景。4.3 长任务场景stream、轮询与断线恢复但真实业务里Agent 干活往往不是一瞬间的事。比如让一个 Agent 去全网收集竞品信息再生成报告可能要两三分钟。如果客户端一直拿着 HTTP 连接等同步响应中间隔着负载均衡、网关很容易超时。A2A 的解法是 message/stream服务端用 SSEServer-Sent Events方式流式返回。客户端发同样的请求但服务端会先返回一个 working 状态的 Task 快照然后在任务推进过程中不断推送新的任务状态和消息增量最后在完成时推送含 Artifact 的结果。这样客户端能在几秒内感知到任务开始执行了而不是死死等一个未知时长的响应。SSE 事件里 data 部分仍然是完整的 JSON-RPC 响应结构协议没有因为流式传输就改变消息语义。这一点对客户端实现很友好——解析逻辑和同步模式基本一致只是从读一次响应变成了循环读事件流。推送通知也是协议内的可选能力。客户端在首次请求时通过A2A-Client-Negotiation-Strategy头声明自己支持 default轮询还是 push-notifications推送服务端在响应头里返回实际采用的策略。如果协商为推送客户端可以通过task/pushNotificationConfiguration注册回调地址任务状态变化时服务端主动通知。生产环境里最现实的问题不是协议好不好用而是断线。SSE 连接一旦断开对客户端来说任务状态就变成了未知。A2A 给了两条退路一是用 task/get 轮询查询任务状态二是如果之前协商支持推送可以用 task/resubscribe 恢复对一个任务的订阅。我建议真实项目里把 task/get 作为兜底校验手段哪怕主链路用的 stream断线发生后也用 task/get 确认一下任务真实状态再决定是继续等还是重新提交。5. A2A 与 MCP分工边界与组合落地思路5.1 MCP 管工具接入A2A 管 Agent 协作MCPModel Context Protocol和 A2A 常被放在一起比较但它们解决的问题不在一个层面。MCP 是 Agent 连接外部工具、数据源的协议解决的是Agent 怎么使用工具A2A 是 Agent 与 Agent 之间的协议解决的是Agent 怎么委托另一个 Agent 完成工作。打个比方MCP 像是给 Agent 配了一套标准化的外设接口让它能插上计算器、数据库、API 这些外部设备A2A 则是给 Agent 之间定了一套工作交接流程让它们能互相派活、汇报进度、交付成品。一个管设备和 Agent 的连接一个管 Agent 和 Agent 的协作。我听到过一个很普遍的说法工具都用 MCP 接好了Agent 之间直接通过 MCP 互相调用不就行了吗这个想法表面上成立实际会遇到问题MCP 里的 Server 本质是工具是没有自主性的。你通过 MCP 调用一个 Server执行的是确定的函数但 Agent 不是一个确定函数它会基于当前上下文自己做决策、可能跨多个步骤运行、过程中还可能需要向调用方请求补充信息。这些行为用 MCP 的工具调用语义来表达会非常别扭。更关键的是MCP 没有定义任务状态和成果物这些语义。一个 Agent 内部调用的 MCP 工具是即时返回结果的而 Agent 之间的协作是长期任务需要状态跟踪、结果交付、异常恢复。这部分恰恰是 A2A 的核心能力。所以两者不是替代关系是互补关系。5.2 一个可落地的组合架构示例拿我熟悉的场景举例企业里有一个客服 Agent和一个供应链分析 Agent。客服 Agent 通过 MCP 接入 CRM 系统能拿到客户信息和历史工单供应链分析 Agent 通过 MCP 接入数据仓库能查库存和配送数据。现在客户投诉物流延迟了两天能不能重新安排配送。客服 Agent 先通过 MCP 拿到客户上下文判断这个问题需要供应链数据于是它作为 A2A 客户端去供应链分析 Agent 的 well-known 地址拿到 Agent Card发现有一个recompute_plan技能。它通过 A2A 发起任务供应链 Agent 收到后再通过自己的 MCP 工具查询最新库存和配送约束计算结果把新方案作为 Artifact 返回。整条链路里每个 Agent 的手都是 MCP 提供的嘴是 A2A 提供的各司其职。这个架构的价值在于任何一端的 Agent 换成另一家厂商的实现只要它还遵循 A2A 和 MCP整个链路不需要改代码。5.3 落地顺序建议先通工具再通 Agent如果你马上要在自己团队里推进我的建议是先别急着上 A2A。第一步把每个 Agent 的工具接入统一到 MCP 上让每个 Agent 都能稳定执行自己的专业动作第二步再把这些 Agent 封装成 A2A 服务端写好 Agent Card第三步在一个编排层比如主 Agent里做 A2A 客户端统一调度。如果连工具接入都没统一让一堆空壳 Agent 互相通信只是在放大混乱。另外生产环境落地 A2A 时别忽略认证。Agent Card 里有 security 字段可以声明认证方式实际通信可以用 OAuth 2.0、API Key 等现有体系协议本身不强制特定方案但两端必须提前协商一致。这个细节和大多数企业内部系统对接类似属于不做会出事、做了不显眼的环节。6. 从零跑通最小 A2A Server实现、验证与踩坑6.1 用 FastAPI 搭建最小 A2A 服务端说了这么多概念最终还是要落到代码。我用 FastAPI 写了一个最小可运行的服务端只实现 Agent Card 和 message/send 两个能力足够跑通一次完整协作。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import Any import uuid app FastAPI() app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) AGENT_CARD { identifier: hello-a2a-001, name: Hello A2A Agent, description: 一个用于协议演示的最小 A2A 服务端会原样返回收到的消息, url: http://localhost:8000, version: 1.0.0, skills: [ { id: echo, name: Echo, description: 接收用户输入的文本返回一条包含原文的确认消息, tags: [demo], examples: [帮我确认: 你好], inputModes: [text], outputModes: [text], } ], } class JsonRpcRequest(BaseModel): jsonrpc: str 2.0 id: Any None method: str params: dict | None None tasks: dict[str, dict] {} def jsonrpc_result(request_id, result): return {jsonrpc: 2.0, id: request_id, result: result} def jsonrpc_error(request_id, code, message): return {jsonrpc: 2.0, id: request_id, error: {code: code, message: message}} app.get(/.well-known/agent.json) app.get(/agent.json) def get_agent_card(): return AGENT_CARD app.post(/) async def jsonrpc_endpoint(req: JsonRpcRequest): if req.method message/send: params req.params or {} message params.get(message, {}) text .join( part.get(text, ) for part in message.get(parts, []) if part.get(kind) text ) task { id: str(uuid.uuid4()), status: completed, messages: [ {role: user, parts: [{kind: text, text: text}]}, {role: agent, parts: [{kind: text, text: f收到{text}}]}, ], artifacts: [ {name: ack, parts: [{kind: text, text: fhello: {text}}]} ], } tasks[task[id]] task return jsonrpc_result(req.id, task) if req.method task/get: task_id (req.params or {}).get(id) task tasks.get(task_id) if task is None: return jsonrpc_error(req.id, -32602, task not found) return jsonrpc_result(req.id, task) return jsonrpc_error(req.id, -32601, fmethod {req.method} not supported)代码里我保留了两个方法message/send 直接建任务并返回 completed 快照task/get 用来查任务。麻雀虽小但 JSON-RPC 的请求响应结构、Agent Card 的暴露、任务状态管理这几件事都覆盖了。6.2 用 curl 验证 Agent Card 与 message/send服务跑起来之后先验证 Agent Cardcurl http://localhost:8000/.well-known/agent.json能拿到刚才那份 JSON说明能力发现这一步通了。再发一个 message/sendcurl -X POST http://localhost:8000/ \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: req-001, method: message/send, params: { message: { role: user, parts: [{kind: text, text: 你好 A2A}] } } }如果返回里有status: completed且 artifact 里带上了原文那这条最小链路就算跑通了。有了这个最小服务端做底子再去看协议里其他方法会轻松很多——你不会被概念绕晕因为你知道这些方法最终都落在这套请求响应结构上。6.3 长任务与流式处理的实现边界message/send 实现简单但生产环境里真正用得多的还是 stream。流式实现的复杂度比 send 高一截服务端要维护一个异步队列把任务的状态变更逐步推给 SSE 客户端还要处理客户端断开连接后的清理。我做过一个最小实现核心是利用 FastAPI 的异步生成器把 yield 出来的内容交给StreamingResponsefrom fastapi.responses import StreamingResponse import asyncio import json def process_long_task(say: str): # 模拟一个长任务每步产出一条任务快照 yield {status: working, message: {role: agent, parts: [{kind: text, text: 开始处理}]}} asyncio.sleep(1) yield {status: working, message: {role: agent, parts: [{kind: text, text: 正在查询数据}]}} asyncio.sleep(1) yield {status: completed, message: {role: agent, parts: [{kind: text, text: f完成{say}}]}} def sse(payloads): for payload in payloads: data json.dumps({jsonrpc: 2.0, id: req-001, result: payload}) yield fevent: message\ndata: {data}\n\n # 路由中这样返回 # return StreamingResponse(sse(process_long_task(你好)), # media_typetext/event-stream)这里最关键的一点SSE 每条事件里的 data 必须是完整的 JSON-RPC 响应结构而不是只把 result 字段裸着发出去。很多第一次写的人会在这一步踩坑客户端 SDK 解析直接失败。6.4 实测踩坑记录从联调到上线的六个教训把我在跑这个最小实现时遇到的坑集中列一下按杀伤力排序Agent Card 里的 url 写了 localhost。单机测试一切正常换服务器调用就找不到服务。这个字段从第一天就应该填真实的对外访问地址。响应里漏了 jsonrpc 字段。最初在 message/send 的响应里只写 result没带jsonrpc: 2.0官方 SDK 解析时直接报 invalid response。JSON-RPC 的响应规范是硬性的别省。跨域问题。在浏览器端测试客户端时FastAPI 必须开 CORS否则请求被浏览器拦截。服务端之间的调用虽然不受浏览器限制但很多公司内部的反向代理会拦截不带预检的请求头联调时要一起排查。大文件用 base64 data URI 直接塞进 Part。几千字以内的数据没问题一旦超过几 MBHTTP 请求体膨胀得很夸张。生产环境应该把文件放到对象存储用 FilePart 传 URL 引用。任务状态机不严格。出现过任务到 completed 后又往 messages 里追加内容的情况客户端 SDK 直接丢弃了追加内容。状态一旦终止就不要再去修改任务数据这是协议层面的硬约束。SSE 断线后任务状态丢失。客户端如果中途断开服务端任务还在跑但客户端已经失去感知。我后来统一用 task/get 做兜底判断避免任务其实成功、客户端却以为失败的乌龙。6.5 关于跨团队契约维护的一点体会最后说点代码之外的感受。A2A 把通信层的复杂度消掉了但真正让跨 Agent 协作稳定运行的是两端对语义契约的一致理解。协议只管消息怎么传不管你的 skill 描述是否含糊、你的 Artifact 是否真的满足调用方预期。我在维护多个 Agent 的过程中发现Agent Card 的 description 和 examples 要像对待接口文档一样认真维护每次技能行为有变化第一时间更新 Card 和版本号否则一段时间后你会发现调用方的模型还在按旧描述组织请求而服务端的行为已经变了。协议能解决的是怎么通话的问题通话内容是否靠谱还是要靠人来维护。想清楚这一点A2A 在你项目里才不会只是一堆规则而能真正变成跨团队协作的基础设施。