过去几篇我们聊完了工具调用、路由分发和多Agent编排其实都是在解决一个核心问题如何让Agent真正协作起来。但之前那些方案都有一个隐含前提——所有Agent都住在同一个进程里共享同一套内存和函数调用。真实项目一跑起来你就会发现这个前提站不住脚。团队A用Python写了数据分析Agent团队B用Java做了订单处理Agent它们跑在不同的服务里各自维护自己的状态这时候让它们“直接互相调用”就成了灾难。A2AAgent-to-Agent间通信模式就是为了这类场景准备的它是一种让独立部署、异构技术栈的Agent之间通过标准化协议完成发现、协作和任务分发的通信范式。这篇文章我会把A2A模式的底层逻辑、协议设计、核心代码实现和踩坑经验一次讲透适合正在搭建多Agent系统、准备把Agent服务化、或者研究Agent互操作方案的开发者参考。从我个人的实践感受来说A2A模式并没有引入什么玄乎的新概念它本质上就是把分布式系统领域被反复验证过的经验——服务发现、异步任务、状态持久化、事件驱动——迁移到了Agent这个新实体上。真正让我觉得它值得单独写成一篇模式文章的是它对“Agent如何描述自己”和“Agent之间如何约定任务边界”这两个问题的回答方式和我们以前写REST接口、写消息队列消费者时的思维方式完全不同。下面我按从思路到落地的顺序把整个模式拆开讲。1. 为什么多Agent之间需要一张“通信协议”1.1 单体Agent到多Agent的必然演进做AI Agent开发的人大概都有同感单Agent从0到1容易但随着业务变复杂一个Agent包揽所有能力会迅速触到天花板。我想让Agent既会查库存、又会写营销文案、还会做数据分析结果就是prompt越堆越长、工具越挂越多、上下文被无关内容塞满每次调用都在昂贵的LLM接口上浪费token还经常出现工具调用互相干扰的情况。于是自然想到拆把“查库存”交给库存Agent把“写文案”交给文案Agent把“做分析”交给分析Agent。每个Agent只维护自己那一小块领域知识上下文短了、工具少了、准确率也上来了。但拆完第二天就会遇到新问题——这几个Agent之间怎么说话如果用直接HTTP调用那每个Agent都得知道对方的接口地址、鉴权方式、字段格式耦合度比单体时还高如果想通过消息队列解耦又得约定topic命名和消息体规范本质上是自己发明一套协议。1.2 A2A解决的是海量Agent的互操作问题A2A通信模式的核心思路是给所有Agent定义一套统一的“语言规范”让它们不再关心对方的内部实现只关心彼此暴露出来的能力和约定好的任务接口。打个比方公司里不同部门之间协作不需要每个人都知道其他人的手机号和性格只需要通过统一的工单系统提交请求、跟踪状态、交付结果。A2A模式里的Agent Card就是这个“工单系统的服务目录”Task就是那张“工单”。这套思路在分布式微服务里早就跑通了服务注册中心提供实例信息API规范定义调用方式消息队列解耦异步依赖。A2A协议的贡献在于它把这些已经验证过的模式针对Agent场景做了取舍Agent的能力描述要更语义化因为消费方可能是另一个LLM Agent人未必看得过来任务的粒度要更接近人类的工作流而不是函数签名状态流转要比普通的RPC更丰富因为Agent任务往往是异步长时运行。2025年4月Google把A2A协议贡献给了Linux基金会目前已经有好几家公司基于它实现了产品级的Agent服务。但我觉得更重要的是把这个“通信模式”本身理解透哪怕你不想引入任何外部协议只是在团队内部设计一套Agent协作机制A2A模式里的三件套——能力声明、任务模型、流式事件通道——依然是绕不开的基础设计。1.3 A2A与MCP不是替代关系经常有人把A2A和MCPModel Context Protocol放在一起比较然后问“我该选哪个”。我的理解是它们根本不在一个层面上。MCP解决的是“Agent如何调用外部工具/数据源”的问题方向是Agent到系统的A2A解决的是“Agent如何与其他Agent协作”的问题方向是Agent到Agent的。实际项目中往往是都需要的一个Agent对外用A2A声明自己“能写代码”对内通过MCP调用代码执行器和文件系统。理解这个区分对做架构特别重要因为把两者混用会让系统边界变得模糊后面定位问题会非常痛苦。2. 理解A2A模式的三个核心契约一个A2A通信模式能落地依赖三个层面的契约能力描述、任务模型、事件通道。下面逐个拆。2.1 Agent Card每个Agent的“能力名片”Agent Card用一句我的话说就是一个Agent的简历。它用JSON格式描述这个Agent是谁、能做什么、怎么联系它。A2A协议规范里Agent Card长这样{ name: data-analysis-agent, description: 负责数据分析、报表生成与趋势预测, url: https://agent.example.com/a2a, version: 1.0.0, skills: [ { id: csv_analysis, name: CSV数据文件分析, description: 接收CSV文件并对指定列做统计分析, inputModes: [application/json], outputModes: [text/markdown, application/json] }, { id: trend_forecast, name: 基于历史数据的趋势预测, description: 根据时间序列数据输出未来N天的预测值, inputModes: [application/json], outputModes: [application/json] } ] }这里有几个关键设计点值得展开说。首先url字段不是让你直接调用的接口地址而是这个Agent接收A2A消息的入口地址调度方拿到这个地址后统一走A2A协议发消息而不是根据不同的skill去拼不同的API。其次skills数组里每一项对应一个相对独立的能力单元这个粒度很重要——太粗会导致消费方不知道什么任务能发过来太细又会让Agent Card变成千行长的接口文档。从实践看把公证粒度控制在“用户可感知的一个操作级别”最合适类似“分析CSV”“预测趋势”而不是“读取文件头”“计算平均值”这种函数级描述。Agent Card一般通过一个固定的/.well-known/agent-card.json路径对外暴露消费方只需要知道一个Base URL就能获取到完整的Card继而判断“这个Agent是否适合处理我的任务”。这个流程和OAuth的发现文档、微服务的Swagger文档是一个思想。2.2 Task与生命周期把协作抽象成“工单”A2A模式里最核心的数据结构是Task。当我让数据Agent帮我分析一份CSV文件时我发给它的不是“分析一下这个文件”这种自然语言请求而是一个结构化的任务消息。Task的模型中包含了任务ID、输入内容、状态、产物Artifact和关联消息。一个任务的完整生命周期如下submitted - working - completed | | v v input-required failed | v canceled简单来说任务一提交就进入working过程中Agent可能需要我补充文件或澄清要求input-required干完了就是completed失败了就是failed被主动取消就是canceled。这个状态机看着简单但跟普通RPC的“请求-响应”二元结构比起来它解决了两个很实际的问题。第一个问题是长任务。LLM Agent执行任务动辄几十秒甚至几分钟你不可能让客户端一直保持一个HTTP连接傻等着Task模型允许客户端提交任务之后拿着task_id走人隔一会儿回来查一次。第二个问题是交互式任务。Agent在推理过程中需要外部输入的场景其实非常常见比如写代码的Agent写完一版要你确认后再跑测试。Task模型的input-required状态给这种“半路追问”留了标准位置不像传统RPC那样需要自己发明一套回调机制。2.3 事件通道流式输出与异步通知A2A的另一个容易被忽视的设计是消息推送机制。A2A协议推荐使用SSEServer-Sent Events服务器推送事件向客户端推送任务的状态变更、文本增量等内容。也就是说当我订阅一个Task之后服务器每产生一个内容片段都会通过SSE通道推给我而不是等我下一次轮询返回。这可能就是Agent通信和普通API最大的体验差异人类看Agent工作很像看一个实习生写报告最好能实时看到它写到哪一步了、下一步打算做什么、中途有没有卡壳。A2A把“流式状态反馈”作为协议层的能力来做而不是上层应用的优化手段这点对构建有用户参与感的Agent产品帮助很大。3. 站在模式角度理解A2A协议的关键选型前面讲的是模式层面的抽象现在落到协议层面。A2A协议本身是Google牵头开源的一套规范当前版本的协议文档定义了基于JSON-RPC 2.0和SSE标准的事件格式。这一节我从模式作者的角度聊聊它为什么做这些技术选型。3.1 为什么消息格式用JSON-RPC而不是REST我之前第一个想法是“Agent间通信为什么不用REST又简单又普及”。仔细看完协议文档才意识到REST的资源导向模型和Agent的任务导向模型并不吻合。REST要求你把所有东西都定义成资源POST /analysis-tasks、GET /tasks/{id}操作含义完全由URL决定而Agent间的通信动作高度一致就是“发消息、收消息、订阅状态”这几类原语用REST反而要维护一堆语义模糊的端点和状态码。JSON-RPC 2.0则是最小化的远程调用协议客户端发一个{method, params, id}结构服务端回一个{result, error, id}结构不定义资源不做状态管理恰好吻合A2A“把任务本身交给服务端客户端只负责交互”的定位。A2A协议核心就定义了三个方法message/send向Agent发送任务消息、message/reply回复Agent提出的问题、tasks/get查询任务状态。这套东西和Agent的认知模型是贴合的学起来也快。3.2 为什么用SSE而不是WebSocket在A2A协议里实时事件推送采用的是SSE。SSE的特点是单向的服务器到客户端、基于HTTP的、自动重连机制成熟WebSocket则是全双工、长连接、面向消息的。Agent协作场景里主导方向其实是“服务端向客户端汇报进展”客户端真正需要主动上行消息的场景并不多——提交任务可以用普通的JSON-RPC请求回复输入也可以用普通的JSON-RPC请求并不需要WebSocket的“双向实时”。加上A2A的客户端往往是另一个Agent或者服务端应用它们对实时性没有聊天软件那么苛刻而对“断线重连”“负载均衡兼容”这些工程特征要求很高。SSE天然跑在HTTP上对网关和日志系统都友好实现成本也低。这个选型给我的启示是协议选型要服务场景的真实需要而不是追新。3.3 传输层抽象协议与模式解耦理解A2A模式时容易犯的错是以为“用了A2A协议就是应用了这个模式”。实际上A2A协议只是这个模式在HTTP/JSON环境下的一种标准化落地。模式层面的东西是那些抽象概念——Agent Card、Task状态机、事件流——这些东西完全可以脱离协议在自己内部的微服务接口里实现。我自己在团队内部实践时就曾用gRPC实现过一版类似的Agent间通信框架Agent Card存进服务注册中心而不是暴露well-known路径Task状态存进Redis并配合网关做查询事件流用gRPC server streaming替代SSE。模式的核心思想牢牢抓住了最终业务跑得也很稳。我这么说不是让大家别用A2A协议而是想强调认知层面要分清“模式和协议”模式是你架构的骨架协议是你实现骨架时可选的现成材料。具体选型时如果你的Agent全部跑在纯内网、没有跨组织协作需求、且对性能极致敏感完全可以基于这个模式做裁剪反过来如果你要做跨公司、跨平台的Agent协作直接用开源A2A协议就是最省力的选择。4. 手把手落地一个A2A模式的最小系统光说不练假把式。这一节我带着你用Python实现一个极简但完整的A2A通信机制包含服务端Agent和消费方Agent两侧。代码基于FastAPI和标准asyncio实现核心是演示协议逻辑不引入过于复杂的框架。4.1 定义服务端Agent的Card服务端Agent就是一个FastAPI应用先定义Agent Card路由。按A2A协议的约定它的路径是/.well-known/agent-card.jsonfrom fastapi import FastAPI from pydantic import BaseModel from typing import Optional, List app FastAPI(titleData Analysis Agent) AGENT_CARD { name: data-analysis-agent, description: 提供CSV数据分析和趋势预测能力, url: http://localhost:8100/a2a, version: 1.0.0, skills: [ { id: csv_analysis, name: CSV数据分析, description: 接收CSV文件并输出统计指标, inputModes: [text/csv], outputModes: [application/json] } ] } app.get(/.well-known/agent-card.json) async def get_agent_card(): return AGENT_CARD这段代码几乎不需要解释就是把一个字典通过HTTP暴露出去。真正的消费方拿到这个Card后会用它来决定“要不要把任务发给这个Agent”。4.2 实现任务模型和状态存储A2A的Task是需要跨请求跟踪的所以需要一个存储层。最小实现里我用一个全局字典充当存储生产环境建议换成Redis或数据库。这一步定义了任务的核心数据结构# 简化版的内存任务存储 tasks: dict[str, dict] {} class TaskState: SUBMITTED submitted WORKING working COMPLETED completed FAILED failed INPUT_REQUIRED input-required CANCELED canceled def create_task(task_id: str, input_payload: dict) - dict: task { task_id: task_id, status: TaskState.SUBMITTED, input: input_payload, artifacts: [], messages: [] } tasks[task_id] task return task在实际项目里你很快会意识到这个存储层的重要性。因为Agent任务可能是分钟级的长任务进程一重启状态全没会让调用方直接血崩。哪怕是在原型阶段我也建议直接把这块存储指向Redis省得到后面再迁移。4.3 任务处理流程与状态流转接下来是核心逻辑的骨架接收任务、模拟异步处理、返回任务ID。关键是asyncio.create_task允许我们在返回请求后继续在后台处理import asyncio import uuid class MessageRequest(BaseModel): task_id: Optional[str] None message: dict app.post(/a2a/message/send) async def send_message(req: MessageRequest): task_id req.task_id or str(uuid.uuid4()) if task_id not in tasks: create_task(task_id, req.message) # 异步启动任务处理 asyncio.create_task(process_task(task_id)) return { task_id: task_id, status: TaskState.WORKING } async def process_task(task_id: str): task tasks[task_id] task[status] TaskState.WORKING try: # 模拟耗时的Agent推理过程 await asyncio.sleep(3) # 生成产物 artifact { artifact_id: str(uuid.uuid4()), content_type: application/json, data: {rows: 100, columns: 5, summary: ok} } task[artifacts].append(artifact) task[status] TaskState.COMPLETED except Exception as e: task[status] TaskState.FAILED task[messages].append({role: error, content: str(e)})这段代码省略了A2A协议规范的很多细节但模式的关键点已经体现出来任务提交后立刻返回不阻塞客户端Agent在后台推进状态最终落到completed或failed。4.4 支持SSE流式推送事件真正的A2A协议里客户端可以订阅Task的事件流实时看到Agent输出。我实现一个极简的SSE版本让读者直观感受流式的效果from fastapi.responses import StreamingResponse app.get(/a2a/tasks/{task_id}/events) async def task_events(task_id: str): async def event_generator(): # 客户端连接后每500ms推送一次当前状态 for i in range(20): task tasks.get(task_id) if not task: break yield fdata: {json.dumps({status: task[status]})}\n\n if task[status] in (TaskState.COMPLETED, TaskState.FAILED): break await asyncio.sleep(0.5) return StreamingResponse( event_generator(), media_typetext/event-stream, headers{Cache-Control: no-cache, Connection: keep-alive} )SSE格式本身很简单data: 内容\n\n这样一帧一帧地发就行。客户端就是用EventSource或者post-processing脚本去读。要注意的是真实生产环境里不能简单用全局字典做任务存储多个Worker实例之间的任务状态是不同步的这个问题我在第6节还会详细讲。4.5 消费方Agent的调用策略消费方Agent调用数据Agent时核心流程是一个“提交任务 轮询状态 收集产物”的循环。我封装成一个简单函数import requests import time def call_analysis_agent(payload: dict, base_url: str http://localhost:8100): # 1. 获取Agent Card card requests.get(f{base_url}/.well-known/agent-card.json, timeout5).json() print(f发现Agent: {card[name]}, 技能: {[s[name] for s in card[skills]]}) # 2. 提交任务 send_result requests.post( f{base_url}/a2a/message/send, json{message: {role: user, content: payload}}, timeout5 ).json() task_id send_result[task_id] # 3. 轮询或订阅任务状态 while True: task_info requests.get( f{base_url}/a2a/tasks/{task_id}, timeout5 ).json() status task_info[status] if status completed: return task_info[artifacts] elif status in (failed, canceled): raise RuntimeError(f任务失败状态: {status}) else: time.sleep(1)这里你会发现一个重要的工程细节消费方如果时间充裕应该优先用SSE订阅而不是轮询因为轮询的延迟取决于sleep间隔设得多小设得太小还会白白浪费请求。我通常的做法是两者的混合默认用SSE订阅流程失败或断连时退回轮询保证任务链路在任何情况下都能推进下去。5. 场景选型A2A与其他Agent通信方式的对比写到这里肯定有人会问市面上有AutoGen的对话模式、有LangGraph的图编排、有消息队列的发布订阅为什么我还要自己实现一套A2A不同方案有不同适用边界我做了个对比表给各位参考。通信方式适用场景优点缺点进程内函数调用单体Agent内部模块零开销、直观无法跨服务、耦合高直接HTTP API少量固定Agent协作简单直接接口耦合、能力发现差A2A协议通信异构、多实例、跨组织Agent标准化、可发现、异步长任务友好需要额外实现协议栈消息队列发布订阅高吞吐扇出解耦彻底、削峰填谷不适合需要响应结果的请求-响应协作编排框架LangGraph等单进程内复杂状态流状态管理强锁定技术栈、跨服务弱选型的核心判断标准是看你的Agent系统是“封闭式”还是“开放式”。如果所有Agent都由你一个人在一个进程里写死那用进程内函数调用就够了加什么A2A反而是过度设计。如果系统里有多个团队、多种技术栈、甚至可能引入第三方Agent服务那就得考虑A2A这种标准化通信方式否则每一次增加Agent都要重新签一遍接口协议体系的复杂度会指数级上升。我自己经历过的项目里最典型的误用是团队明明才两个Agent、天天泡在一起改代码却非要上一整套消息队列和事件总线结果查消息轨迹的成本比直接函数调用还高。所以架构选型先看范围再看复杂度A2A是为“独立部署、独立演进、独立运维”的Agent体系准备的。6. 实践中的坑与排查经验6.1 Task状态机的“悬挂”问题多Agent协作最让人崩溃的场景是消费方等了一个小时任务状态一直停在working既没有失败也没有完成。这种“悬挂”状态最隐蔽的成因是Agent进程在处理任务时抛了异常但异常没有被捕获进状态机导致后台任务直接退出状态永远停在working。排查利器就是给任务加“最后活跃时间”字段。每次状态变更都刷新它调度方可以定期扫描超过阈值未更新的任务把它们标记为failed或重启重新执行。这里分享一个实用经验对任何长时运行的Agent任务从一开始就把超时判定设计进状态机而不是事后补。判定规则做两层第一层是单个步骤的超时上限第二层是整个Task的总时间上限。我当时第一版没做这个结果线上Agent卡死了几轮才发现问题。6.2 SSE断连与重试抖动SSE看着简单但在公网环境下网关和代理的缓冲区配置很容易导致连接被意外断开。Auto reconnect是SSE规范自带的但很多客户端SDK对重连的处理是“立即重试”这会在Agent群体互相唤醒时产生可怕的请求风暴。我的修复策略是给所有重试加上指数退避初始等待1秒、递进到最多30秒并加上随机抖动。同样的原则也适用于消费方Agent的轮询频率如果100个Agent同时用1秒间隔轮询同一个服务几乎等价于一次小型DDoS。6.3 Agent Card粒度拿捏不准Agent Card的skills描述太粗消费方尤其是LLM驱动的消费方就不知道这个Agent能不能处理自己的特殊任务描述太细Agent Card会变成一份几百行的JSON消费方解析和匹配的token成本也上去了。试了几轮我的建议是每项skill的description控制在30-80字核心写清楚“输入是什么、输出是什么、解决什么问题”。这一条也值得留意如果description写得含糊遇到LLM消费方它经常会把不能干的任务硬发过来造成下游Agent大量报错。6.4 鉴权与身份信任A2A协议本身没有强制鉴权机制但在真实系统里不能让任何Agent随便声明一个url就加入协作网络。我的做法是引入一个简单的API Token体系以及每项skill单独配置可访问白名单。消费方调用时要带着TokenAgent发布者在处理任务前校验调用方身份。关于这个点你再留心一下Agent Card里不要暴露过于详细的内部IP和端口生产环境建议Gateway统一对外暴露Agent本身收在内网和普通微服务的网络治理原则完全一致。6.5 调试手段消息日志与任务回放A2A模式下排查问题最大的障碍是链路长消费方Agent、网络传输、服务方Agent、LLM推理任何一个环节出错都可能导致最终结果不对。我强烈建议在早期就建立一套“任务回放”日志记录每一条进入/离开Agent的消息都落库格式尽量和发送时保持完全一致。一旦出现问题可以按task_id把整条链路的消息按时间序列拉出来看。这个回放日志的成本非常低但它在多Agent系统里的价值等同于金融系统里的交易流水怎么强调都不过分。我记得第一个多Agent项目上线后的第二天就碰上一次生产事故上游Agent说“已完成任务”但下游Agent收到的产物是空的。当时如果没有消息流水日志我们根本不知道问题出在序列化过程中——服务端把一个空列表存进了artifacts字段。这种问题靠肉眼调代码很难发现但回放日志几乎是一步到位提示了异常位置。7. 从A2A模式延伸出去的一点心得这个系列写到这里我对“设计模式”本身的理解也在加深。最初接触23种经典设计模式时总觉得它们像是工厂、单例、观察者这种代码层面的技巧用在面向对象语言里很顺手。到了AI Agent时代设计模式的重心明显从“代码组织结构”移向了“系统架构与实体协作”。A2A模式所关注的不再是一个类怎么被实例化而是一群自治的Agent怎么在动态变化的网络里互相发现、理解、协作。从这个角度看A2A模式的价值其实藏在它定义的那几个核心契约里。Agent Card让系统的能力变得可发现、可扩展Task状态机让长时运行的智能任务有了可靠的生命周期管理事件流让Agent之间的协作过程对人和机器都透明。这些思想和我前几篇写的路由分发模式、编排模式能天然组合起来编排器充当中心调度者通过A2A协议向挂在网络里的各领域Agent分发任务每个Agent用MCP接入自己的工具集最终形成一套完整的企业级Agent服务体系。我个人这段实践的体会是别急着把整套A2A协议引入生产先在自己的系统边界内把模式的思想跑通一遍——定义好能力描述、任务流转、事件通道这三个核心机制你会立刻感觉到多Agent协作的复杂度被约束住了。等业务上确实出现异构系统、跨组织的协作诉求再平滑升级到标准协议也不迟。
