1. Agent 通信协议 A2A 到底在解决什么问题第一次听到 A2A 协议这个词很多人会下意识把它和 MCP、LangChain 里的工具调用混在一起。我刚开始接触的时候也犯过这个迷糊后来在一个多 Agent 协作的项目里踩了坑才真正搞明白A2A 要解决的不是“一个 Agent 怎么调用工具”而是“多个 Agent 之间怎么互相说话”。举个生活里的例子。你家里有扫地机器人、智能音箱、智能灯它们各自都能独立干活。但如果你想让音箱听到“我要睡觉了”之后通知扫地机器人回充、让灯调暗这就需要一个它们之间能听懂的约定。A2A 协议扮演的就是这个“约定”的角色它规定了 Agent 之间用什么格式发消息、怎么描述自己的能力、怎么把任务委托出去、怎么把结果传回来。这里必须先厘清一个高频混淆点MCP 和 A2A 不是竞争关系。MCP 更像是“Agent 和工具/数据源之间的插头标准”解决的是 Agent 怎么拿到外部能力A2A 解决的是“Agent 和 Agent 之间的对话标准”解决的是协作编排。你可以把 MCP 理解成 USB 接口把 A2A 理解成两个人之间的通话礼仪。一个 Agent 完全可能一边用 MCP 接数据库一边用 A2A 和另一个 Agent 商量任务分工。那为什么现在需要 A2A因为单 Agent 的能力天花板很明显。一个 Agent 再强它的上下文窗口、工具集、专业领域都是有限的。真实业务里经常出现这种场景一个负责理解用户意图的 Agent需要把“帮我分析这份财报”拆成“抓数据”“算指标”“写结论”三件事分别交给三个更专业的 Agent。如果没有统一协议每对接一个新 Agent 就要重写一遍胶水代码维护成本会指数级上升。A2A 的价值就在于把这层胶水标准化让不同团队、不同框架、甚至不同公司开发的 Agent 能像积木一样拼起来。这篇文章适合三类人看正在做多 Agent 系统、被 Agent 之间通信搞得头大的开发者想搞清楚 A2A 和 MCP、LangGraph、AutoGen 这些概念边界的技术选型者以及准备把单 Agent 项目升级成协作系统的工程师。我会从设计思路、核心机制、实操落地到踩坑排查把 A2A 讲透尽量让你看完就能动手搭一个最小可用的多 Agent 通信demo。2. A2A 协议的整体设计与核心思路拆解2.1 为什么不是“再造一个 RPC”而是设计一套 Agent 语义很多人第一反应是Agent 之间通信直接用 HTTP JSON 不就行了为什么要搞个协议我一开始也这么想直到实际做项目才发现普通 RPC 缺的是“Agent 语义”。普通 RPC 调用是这样的我调用一个函数传参数拿返回值结束。但 Agent 之间的交互完全不是这个形态。一个 Agent 委托任务给另一个 Agent 时它需要知道对方能干什么能力发现、这个任务大概要多久长任务支持、中间能不能问进度流式更新、对方需不需要我补充信息多轮交互、任务做到一半能不能取消生命周期管理。这些在传统 RPC 里要么没有要么要自己造轮子。A2A 的设计思路可以概括成一句话把 Agent 当成一个有身份、有能力描述、能接受任务委托、能持续汇报状态的服务实体。围绕这句话它定义了几个核心概念。第一个是Agent Card也就是 Agent 名片。每个 Agent 对外暴露一个描述文件里面写清楚自己叫什么、能处理什么类型的任务、支持哪些输入输出格式、通信端点在哪。这解决了“我怎么知道对方能干嘛”的问题。没有这个多 Agent 系统就只能靠硬编码扩展性极差。第二个是Task任务对象。A2A 里的一次协作不是一次函数调用而是一个有状态的任务。任务有唯一 ID有状态机提交、处理中、需要补充输入、完成、失败、取消可以跨多次消息往返。这一点非常关键因为真实任务往往不是一问一答就结束的。第三个是Message 和 Part消息与内容块。A2A 的消息不是纯文本而是由多个 Part 组成Part 可以是文本、文件、结构化数据。这样设计是为了让 Agent 之间能传富内容比如一个 Agent 把生成的图表文件传给另一个 Agent。第四个是Streaming 和 Push流式与推送。长任务不可能让调用方一直阻塞等待所以 A2A 支持通过流式响应持续推送状态更新也支持在任务状态变化时主动回调通知。2.2 传输层选型为什么默认走 HTTP JSON-RPC 风格A2A 在传输层没有发明新东西而是复用了成熟的 HTTP 和 JSON。这个选择背后有很实际的考量。首先是穿透性和兼容性。HTTP 是互联网上最通用的传输方式任何语言、任何环境基本都能发 HTTP 请求。如果 A2A 自己搞一套二进制协议那跨语言、跨网络的成本会高很多。用 HTTP 意味着一个 Python 写的 Agent 可以很轻松地和 Java、Go、Node 写的 Agent 通信。其次是调试友好。JSON 可读用 curl 就能直接测日志里也能直接看。我在排查多 Agent 问题时最怕的就是那种二进制协议抓包出来一堆乱码。A2A 用 JSON出问题的时候直接看请求体就能定位大半。第三是和现有基础设施契合。HTTP 天然支持负载均衡、网关、鉴权、限流、可观测性。企业里已有的 API 网关、日志系统、监控体系都能直接复用不需要为 Agent 通信单独搭一套。不过要注意A2A 规范本身是传输无关的HTTP JSON 是它定义的标准绑定之一。理论上你也可以用 gRPC 或其他传输方式实现只是目前生态里 HTTP 绑定最成熟。选型的时候如果你的 Agent 都在同一个内网、对性能极度敏感可以考虑其他绑定但绝大多数场景HTTP JSON 是性价比最高的选择。2.3 和 MCP、LangGraph、AutoGen 的边界到底在哪这是被问得最多的问题我用一张表把边界说清楚。技术解决的问题通信对象典型场景MCPAgent 如何接入外部工具和数据Agent 与工具/资源让 Agent 读数据库、调 APIA2AAgent 之间如何协作通信Agent 与 Agent多 Agent 任务分工与编排LangGraph如何用图结构编排 Agent 流程流程节点之间单进程内的复杂工作流AutoGen多 Agent 对话式协作框架框架内的 Agent研究型多 Agent 对话关键区别在于跨边界能力。LangGraph 和 AutoGen 是框架它们定义的是“在同一个框架内Agent 怎么协作”。而 A2A 是协议它定义的是“不管你在哪个框架里Agent 之间怎么通信”。这意味着一个用 LangGraph 写的 Agent 可以通过 A2A 和一个用 AutoGen 写的 Agent 协作甚至和一个完全自研的 Agent 协作。我个人的经验是框架负责内部编排协议负责外部通信。你在一个系统内部用 LangGraph 编排流程完全没问题但当这个系统需要和另一个团队、另一个系统对接时A2A 就是那层标准接口。两者不冲突反而互补。3. A2A 核心机制与实操要点解析3.1 Agent Card能力发现的第一道门Agent Card 是 A2A 协作的起点。它通常是一个放在固定路径比如/.well-known/agent.json的 JSON 文件描述这个 Agent 的基本信息。一个典型的 Agent Card 长这样{ name: financial-analysis-agent, description: 负责财报数据抓取与指标计算, url: https://example.com/agents/finance, version: 1.0.0, capabilities: { streaming: true, pushNotifications: false }, skills: [ { id: fetch-financial-report, name: 抓取财报, description: 根据公司代码抓取指定季度的财报数据, inputModes: [text/plain, application/json], outputModes: [application/json] } ] }这里有几个实操要点必须注意。第一skills 的粒度要设计好。太粗调用方不知道你到底能干嘛太细Agent Card 会膨胀得没法维护。我的经验是按“业务能力”划分而不是按“函数”划分。比如“抓取财报”是一个 skill而不是“调用某 API 的某个 endpoint”。第二inputModes 和 outputModes 要如实填写。这直接决定了调用方怎么给你发数据。如果你声明支持application/json但实际只处理纯文本调用方发 JSON 过来就会失败。我踩过这个坑后来养成的习惯是声明什么就严格实现什么宁可少声明。第三Agent Card 的发现机制要考虑缓存。如果每次协作都去拉一遍 Agent Card在高频场景下会成为瓶颈。常见做法是本地缓存加定期刷新或者通过注册中心统一管理。3.2 Task 生命周期一次协作的完整状态流转A2A 里的一次协作是一个 Task它有明确的状态机。理解这个状态机是写好 Agent 的关键。任务状态大致包括submitted已提交、working处理中、input-required需要补充输入、completed完成、failed失败、canceled取消。状态之间的流转不是随意的而是有约束的。我重点说两个容易被忽略的状态。input-required状态是 A2A 的精髓之一。它让 Agent 之间的协作可以多轮进行。比如 A Agent 委托 B Agent 分析财报B 发现缺少公司代码就进入input-required状态回一条消息问“请提供公司代码”。A 补充后再继续。这个机制让 Agent 协作从“一次性调用”变成了“真正的对话”。canceled状态要考虑幂等。任务取消可能发生在任何阶段你的 Agent 实现必须能处理“任务已经完成但取消请求才到”这种情况。我的做法是在状态变更时加锁确保终态只能进入一次。下面是一个任务状态流转的参考表当前状态可流转到触发条件submittedworking / failed / canceled开始处理或直接失败workinginput-required / completed / failed / canceled需要输入或完成input-requiredworking / canceled收到补充输入completed / failed / canceled无终态不可再变3.3 消息与 Part富内容怎么传A2A 的消息由多个 Part 组成Part 类型包括文本、文件、结构化数据。这个设计看起来简单但实操里有讲究。文本 Part 适合传自然语言指令和结果。文件 Part 适合传二进制内容比如图片、PDF、生成的报告。结构化数据 Part 适合传 JSON 对象方便对方直接解析。我遇到过一个典型问题大文件传输。如果直接把一个大 PDF 塞进消息里消息体会非常大HTTP 传输和日志记录都会受影响。常见做法是文件走对象存储消息里只传引用 URL。A2A 的文件 Part 支持通过 URI 引用这个设计就是为这种场景准备的。另一个要点是Part 的顺序和组合。一条消息里可以有多个 Part接收方要能正确处理组合。比如“这是分析结果”文本 Part 结果 JSON Part接收方需要把两者关联起来理解。我的建议是如果多个 Part 有逻辑关联在文本 Part 里明确说明不要指望接收方能自动推断。3.4 流式与推送长任务怎么不阻塞长任务是 Agent 协作的常态。一个数据分析任务可能跑几分钟甚至几十分钟调用方不可能一直阻塞等待。A2A 提供了两种机制。流式响应是调用方保持连接服务方持续推送状态更新和中间结果。推送通知是调用方提供一个回调地址服务方在任务状态变化时主动通知。选型上我的经验是交互式场景用流式后台批处理场景用推送。用户在等结果的时候流式能让界面实时显示进度而夜间批量任务推送更省资源。流式实现里有个坑要注意连接中断的处理。网络不稳定时流式连接可能断掉你的实现要支持断线重连后继续获取任务状态而不是从头开始。这要求任务状态是持久化的不能只存在内存里。4. 从零搭建一个最小可用的 A2A 协作 Demo4.1 环境准备与依赖选型要动手跑一个 A2A demo先明确技术栈。我用 Python 举例因为生态最成熟。核心依赖其实很少因为 A2A 基于 HTTP JSON用 FastAPI 或 Flask 就能实现服务端用 httpx 或 requests 实现客户端。如果你想用官方 SDK可以装a2a-sdk但我的建议是第一遍先手写把协议细节吃透之后再上 SDK。pip install fastapi uvicorn httpx pydantic选 FastAPI 的理由是它原生支持异步和流式响应写 A2A 服务端很顺手。Pydantic 用来做消息结构的校验能帮你在开发阶段就发现格式错误。环境上本地开发用 localhost 就行。如果要模拟跨网络可以用两个不同的端口或者用 Docker 起两个容器。我建议一开始就用两个独立进程这样能真实模拟网络通信避免把进程内调用误当成协议通信。4.2 实现一个带 Agent Card 的服务端先写服务端。核心是三部分Agent Card 端点、任务提交端点、任务状态查询端点。from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional, List import uuid app FastAPI() # 内存里存任务生产环境要换数据库 tasks {} class TaskRequest(BaseModel): message: str task_id: Optional[str] None app.get(/.well-known/agent.json) def agent_card(): return { name: echo-agent, description: 一个简单的回声 Agent用于演示 A2A, url: http://localhost:8001, version: 1.0.0, capabilities: {streaming: False, pushNotifications: False}, skills: [{ id: echo, name: 回声, description: 把输入原样返回, inputModes: [text/plain], outputModes: [text/plain] }] } app.post(/tasks) def create_task(req: TaskRequest): task_id req.task_id or str(uuid.uuid4()) tasks[task_id] { id: task_id, status: completed, result: fecho: {req.message} } return tasks[task_id] app.get(/tasks/{task_id}) def get_task(task_id: str): if task_id not in tasks: raise HTTPException(status_code404, detailtask not found) return tasks[task_id]这段代码虽然简单但已经包含了 A2A 的核心骨架能力发现、任务创建、状态查询。你可以用uvicorn main:app --port 8001跑起来然后用 curl 测试。curl http://localhost:8001/.well-known/agent.json curl -X POST http://localhost:8001/tasks -H Content-Type: application/json -d {message:hello}4.3 实现调用方 Agent 并完成一次协作服务端有了现在写调用方。调用方的逻辑是先拉 Agent Card 确认对方能力再提交任务最后拿结果。import httpx class A2AClient: def __init__(self, base_url): self.base_url base_url self.card None def discover(self): resp httpx.get(f{self.base_url}/.well-known/agent.json) self.card resp.json() return self.card def send_task(self, message): if not self.card: self.discover() # 检查对方是否支持这个能力 skill_ids [s[id] for s in self.card[skills]] if echo not in skill_ids: raise ValueError(对方不支持 echo 能力) resp httpx.post(f{self.base_url}/tasks, json{message: message}) return resp.json() client A2AClient(http://localhost:8001) print(client.discover()) print(client.send_task(你好A2A))跑通这个 demo你就理解了 A2A 最核心的链路。但要注意这只是最小版本真实场景要补很多东西任务状态机、多轮交互、流式、错误处理、鉴权。4.4 加入多轮交互与状态机把上面的 demo 升级成支持多轮交互关键是在服务端引入状态机。当 Agent 发现信息不足时返回input-required状态并附带需要补充的信息。app.post(/tasks) def create_task(req: TaskRequest): task_id req.task_id or str(uuid.uuid4()) if req.task_id and req.task_id in tasks: # 这是补充输入继续处理 task tasks[req.task_id] task[status] completed task[result] f收到补充输入: {req.message}任务完成 return task # 新任务判断是否需要补充输入 if len(req.message) 5: tasks[task_id] { id: task_id, status: input-required, prompt: 输入太短请补充更多信息 } else: tasks[task_id] { id: task_id, status: completed, result: f处理完成: {req.message} } return tasks[task_id]调用方拿到input-required后带着同一个 task_id 再发一次请求就完成了多轮交互。这个模式在真实场景里非常常见比如表单填写、信息确认、分步任务。4.5 参数与超时设置的实操建议A2A 实操里超时和重试是最容易出问题的地方。我整理了一份经验参数表。场景连接超时读取超时重试次数说明短任务同步调用5s30s2适合快速响应长任务提交5s10s3提交后异步等结果流式连接5s无限制1断线靠重连状态查询3s10s3高频轮询要控制频率重试要注意幂等性。如果任务创建接口不是幂等的重试可能创建出重复任务。解决办法是让调用方生成 task_id 并传入服务端发现相同 ID 就返回已有任务而不是新建。5. 常见问题与排查技巧实录5.1 Agent Card 拉取失败怎么办这是最常见的入门问题。排查顺序我总结成三步。第一步确认路径。A2A 约定 Agent Card 放在/.well-known/agent.json但有些实现会放在别的路径。先用 curl 直接访问确认。第二步确认网络和端口。本地开发最常见的是端口写错或者服务没起来。用curl -v看详细连接过程能快速定位是连接问题还是路径问题。第三步确认返回格式。Agent Card 必须是合法 JSON且包含必要字段。我遇到过返回 HTML 错误页的情况客户端解析 JSON 直接崩。解决办法是在客户端加格式校验解析失败时给出明确错误。5.2 任务卡在 working 状态不结束这个问题的根因通常是状态没有正确更新。可能的情况有处理逻辑抛异常但没捕获导致状态没变或者异步任务没有正确回调更新状态。我的排查方法是先看服务端日志有没有异常再看任务状态存储里这个任务的实际状态。如果状态存储里是 working 但处理早就结束了那就是状态更新逻辑漏了。建议在状态变更的地方统一封装一个函数避免散落在各处。另一个可能是调用方没收到完成通知。如果是推送模式检查回调地址是否可达如果是流式检查连接是否还活着。5.3 多轮交互时 task_id 丢失多轮交互依赖 task_id 串联。常见错误是调用方在补充输入时没带上原 task_id导致服务端当成新任务处理。解决办法是在客户端封装时把 task_id 作为会话上下文的一部分管理起来。我的做法是维护一个本地会话表key 是业务标识value 是 task_id这样业务层不用关心 task_id 的传递。5.4 流式响应中断后状态不一致流式连接中断后调用方可能不知道任务到底完成没有。这时候需要靠状态查询接口兜底。我的经验是流式只作为进度展示最终结果以状态查询为准。也就是说即使流式断了调用方也能通过轮询状态接口拿到最终结果。这样设计能避免流式不稳定导致的结果丢失。5.5 常见问题速查表现象可能原因排查方向拉不到 Agent Card路径错误/服务未启动curl 直接访问任务一直 working状态未更新/异常未捕获查服务端日志多轮交互失败task_id 丢失检查请求体流式中断丢结果无状态兜底加状态查询重试产生重复任务接口非幂等调用方生成 task_id大文件传输失败消息体过大改用 URI 引用5.6 几个我踩过的坑坑一把 Agent Card 写死在代码里。早期为了省事调用方直接硬编码对方能力结果对方升级后能力变了调用方还在按老逻辑发请求。后来改成每次协作前动态拉取问题就没了。坑二忽略 input-required 状态。一开始只处理 completed 和 failed遇到 input-required 直接当失败处理导致多轮任务全部失败。这个状态必须显式处理。坑三任务状态只存内存。服务重启后任务全丢调用方查询直接 404。生产环境一定要持久化Redis 或数据库都行。坑四没有限制消息大小。有一次对方发了个超大 JSON服务端直接 OOM。后来加了消息大小限制和分片机制。6. A2A 在真实项目中的落地经验6.1 什么场景适合上 A2A不是所有多 Agent 场景都需要 A2A。我的判断标准是当 Agent 之间的协作需要跨进程、跨团队、跨框架时A2A 才有价值。如果所有 Agent 都在一个进程里用 LangGraph 或 AutoGen 内部编排就够了上 A2A 反而增加复杂度。但如果你的系统需要对接外部团队的 Agent或者你的 Agent 要作为服务对外提供能力那 A2A 就是必要的标准层。典型适合场景包括企业内部的 Agent 能力市场、跨系统的任务编排、SaaS 产品对外暴露 Agent 能力。这些场景的共同点是边界清晰、需要标准化对接。6.2 性能与可观测性怎么保障A2A 基于 HTTP性能瓶颈通常在网络和序列化。优化方向有几个。连接复用。用 httpx 的 AsyncClient 或连接池避免每次请求都建连接。这个优化在高频场景下效果很明显。消息体精简。大文件走 URI 引用消息里只传必要字段。我见过有人把整个数据集塞进消息性能直接崩。可观测性。A2A 的每次交互都应该有 trace_id方便串联日志。我的做法是在消息头里带 trace_id服务端和调用方都记录出问题时能快速定位。6.3 安全与鉴权的基本考虑A2A 规范本身不强制鉴权方式但生产环境必须考虑。常见做法是 OAuth2 或 API Key。我的建议是Agent Card 可以公开但任务接口必须鉴权。Agent Card 公开是为了让调用方能发现能力但真正提交任务时必须验证身份。另外任务数据可能包含敏感信息传输必须走 HTTPS。还有一个容易忽略的点是权限粒度。不同调用方可能只能访问部分 skill这需要在服务端做权限校验不能只靠 Agent Card 声明。6.4 后续扩展方向跑通基础 A2A 后可以往几个方向扩展。一是引入注册中心统一管理多个 Agent 的 Card支持动态发现。二是加入编排层让一个 Orchestrator Agent 根据任务自动选择合适的子 Agent。三是接入可观测平台把 A2A 交互纳入统一监控。我个人在实际项目里的体会是A2A 最大的价值不是技术本身而是它带来的标准化思维。当你用 A2A 的视角去设计系统时会自然地把每个 Agent 当成独立服务来对待这种解耦对系统的长期可维护性帮助很大。最后分享一个小技巧调试 A2A 时先把所有交互用 curl 跑通再写代码能省掉大量排查时间。