最近有个事在技术圈里讨论度挺高——Google 把 A2AAgent2Agent协议开源了。我花了两周时间把它跑通做了几个多 Agent 协同的试验项目这篇文章把整个过程中关键的技术点和踩坑记录整理出来希望能给正在调研或已经上手的朋友一些参考。先说结论A2A 协议解决的是不同框架、不同厂商的 Agent 之间“怎么互相发现、怎么通信、怎么协同”的问题。用大白话说以前每个 Agent 都是信息孤岛各自用各自的协议互相听不懂A2A 就是给它们定了一套通用的“普通话”和“握手规则”。我实测下来A2A 协议最核心的价值有三个一是标准化了 Agent 之间的通信方式不再需要为每个对接方单独写适配层二是引入了 AgentCard 机制Agent 可以自动广播自身能力服务发现变得非常简单三是定义了完整的任务生命周期管理能把长期运行的任务状态同步得明明白白。这篇文章会围绕这几个点展开结合我的实战代码把协议细节和实现方案掰开揉碎讲清楚。1. A2A 协议核心设计思路与架构拆解1.1 为什么需要 A2AAgent 协作的“巴别塔”困境在深入协议细节之前想先聊聊我为什么会关注 A2A。过去一年我做了不少 LLM 应用从简单的 RAG 问答到复杂的多步骤任务编排越往后做越发现一个问题单体 Agent 的能力上限很明显。你需要让“写代码的 Agent”和“查数据库的 Agent”协作你会发现它们各自用的工具调用协议五花八门有的走 Function Calling有的走 MCP有的干脆自己定义了 JSON 格式。这就很尴尬了。每接入一个新的 Agent就要为它写一套定制化的通信代码而且这种代码往往是一次性的换个场景就报废。MCPModel Context Protocol解决的是 Agent 访问外部工具时的标准化问题但 Agent 和 Agent 之间的对话仍然是各自为政。当时我脑子里就一直有个念头如果 Agent 之间也能有一套类似 HTTP 协议这种“约定俗成”的通信标准就好了。后来我看到 A2A 协议第一反应是“终于有人来解决这个问题了”。它不是要取代 MCP而是和 MCP 互补MCP 管 Agent 与工具之间A2A 管 Agent 与 Agent 之间。整个体系分成了三层——应用层通过 A2A 通信能力层通过 MCP 调用工具模型层各自用各自的 LLM。这个分层逻辑非常清晰每个协议只解决自己那一层的问题而不是试图包揽一切。1.2 A2A 的四种核心原语发现、任务、消息、推送A2A 协议设计的巧妙之处在于它把 Agent 间通信抽象成了非常少的几个核心概念。我梳理了一下整个规范最基础的就是这四个AgentCard、Task、Message 和 Push Notification。理解了这四个概念整个协议就等于掌握了七成。AgentCard 是 Agent 的“名片”以 JSON 格式暴露在一个约定的路径上。客户端可以通过 HTTP 请求获取名片然后就知道这个 Agent 支持哪些能力、接受什么输入类型、输出什么格式、支持哪些协议版本、有哪些安全验证方式甚至还能带上厂商信息和图标。这个设计很大程度上简化了服务发现——以前你要对接一个 Agent得先翻它的文档看 API 格式现在只要拿一张名片所有信息一目了然。Task 是整个协作流程的中枢。每次调用客户端创建一个 Task然后对它持续进行更新。Task 生命周期覆盖了从提交、进行中、完成到失败、取消的全部状态。最有价值的是一个 Task 可以对应多个 Message消息双方可以像聊天一样交替发送消息整个过程是状态化的。这就和普通的 REST API 有了本质区别——REST API 是“发一个请求拿一个响应”而 A2A 是“创建一个任务来回交流直到达成目标”。Message 是 Task 内部的信息单元。每个 Message 包含角色用户或代理、内容文本或文件和元数据。Protocol 规范对这部分做了标准化所以 Agent A 发出的消息Agent B 一定能解析能识别出哪些是文本、哪些是文件数据。Push Notification 解决的问题是异步通知。远程 Agent 处理一个任务可能要几十秒甚至几分钟客户端如果一直轮询会浪费大量的 HTTP 请求。A2A 协议里定义了 PUSH 通知机制任务状态一变化服务端会主动通知客户端。这个机制对构建响应式协作系统非常重要实操中我甚至觉得它是影响多 Agent 协同体验的关键因素之一。1.3 通信模式请求-响应与流式响应并存A2A 协议支持两种通信模式同步模式和流式模式。同步模式就是普通的 HTTP 请求——客户端发一个请求服务端处理完直接返回结果适用于耗时较短的任务。流式模式基于 SSEServer-Sent Events服务端可以持续推送消息流给客户端适用于长耗时任务或者需要实时展示进度的场景。我在实操时发现流式模式特别适合用在内容生成任务上。比如我们的写作 Agent 和审校 Agent 协作时写作 Agent 一句一句生成内容通过 SSE 推给审校 Agent审校 Agent 可以边收边看不用等全部生成完才开始处理。这种交互方式在传统的 REST API 模式里是很难实现的——你会看到超时时间被拉满或者干脆被网关切断。所以我的一个经验是只要是 AI 生成类任务优先用流式模式任务简单、耗时短、响应数据小的才用同步模式。1.4 协议的分层设计让我们回看这波设计决策为什么 A2A 要专门定义一套协议而不是直接用 MCP 或者 REST OpenAPI 文档凑合这里其实有很深的考量。以我个人的经验来看MCP 是面向“工具调用”设计的Agent 调用 MCP 工具时是严格的请求-响应模式目的是获取外部数据或者执行外部动作。但 Agent 之间的对话往往是多轮的、有上下文的、需要持续更新的。比如说客户服务 Agent 和信息查询 Agent 之间可能会来回沟通好几轮用户的需求也在不断调整。这种场景用 MCP 那套机制会非常别扭因为 MCP 没有一个“任务”的概念来承载这种长对话状态。而如果直接上裸的 REST API问题就更多了。每个 Agent 的服务端实现千差万别没有统一的状态模型没有统一的任务管理客户端对接每一个 Agent 都要写一大堆硬编码逻辑。A2A 相当于在 REST 之上做了一个抽象层封装了任务状态和消息交换的通用语义这样不同实现之间就有了互操作性。对比之下A2A 更贴近业务层MCP 更贴近工具层。现在的生态里两者并不是竞争关系而是互补关系。后面我会用一个具体的例子说明你的 Agent 要访问数据库走 MCP要跟另一个 Agent 协作完成任务走 A2A。2. 核心细节解析与实操要点2.1 AgentCard 内容解析Agent 的身份证AgentCard 是整个 A2A 协定的第一个落地环节。它通过 HTTP 协议的 GET 方法暴露路径约定是/.well-known/agent.json或/.well-known/agent-card.json。客户端拿到这个 JSON 后就能知道目标 Agent 的完整能力画像。我们来看一个实际的 AgentCard 示例这是我为了联调写的一个“新闻聚合 Agent”的配置{ name: NewsAggregator Agent, description: 从多个数据源聚合新闻并按主题分类, url: https://api.example.com/agents/news-aggregator, version: 1.0.0, iconUrl: https://api.example.com/agents/news-aggregator/icon.png, capabilities: { streaming: true, pushNotifications: false, stateTransitionHistory: true }, security: { authentication: { schemes: [bearer], credentials: https://api.example.com/agents/news-aggregator/auth } }, defaultInputModes: [text/plain, text/markdown], defaultOutputModes: [text/plain, application/json], skills: [ { id: fetch_news, name: 拉取新闻, description: 按关键词拉取最新新闻条目, tags: [news, search], inputModes: [text/plain], outputModes: [application/json] } ], provider: { organization: Example Corp, url: https://example.com } }注意几个 Key Point。capabilities字段决定了调用方可以使用的通信模式如果目标 Agent 不支持流式你就别用 SSE否则会报错。skills字段是给 Agent 的能力打标签类似于函数签名客户端可以根据技能 ID 做路由。security字段非常重要如果配置了 bearer 认证客户端请求时必须带上 Token否则握手直接失败。我在第一个版本里就吃了这个坑——服务端 AgentCard 里声明了 pushNotifications 支持但我实际上没实现推送接口结果客户端一直等推送任务状态停在“进行中”永远不动。后来我只能把 AgentCard 改成pushNotifications: false客户端自动切回轮询模式问题才解决。2.2 任务生命周期状态的每一步都算数A2A 协议把 Task 的状态定义得非常明确这是它最接近“标准协议”气质的地方。我实测过程中对状态机的理解直接决定了我写的调用逻辑是否正确。核心状态分为submitted客户端已提交任务请求、working服务端正在处理、input-required服务端需要客户端补充信息、completed任务成功完成、failed任务失败、canceled任务被取消。需要注意的是任务过程中 Agent 可能会主动或被动发起多轮对话。比如新闻聚合 Agent 在聚合新闻时发现关键词“华为”有歧义需要客户端确认是要找“华为公司”还是“华为技术国内”此时任务状态就会变成input-required。客户端收到这个状态后应该补充消息然后任务重新回到working。我自己写过一个基于 A2A 的客服 Agent这个状态在整个流程里用了很多次它本质上提供了一种“动态澄清机制”让 Agent 协作更像真人团队而不是简单的一次性 API 调用。2.3 消息结构不要小看这部分设计Message 是 Task 内部传递的信息单元我把它理解成“带类型的内容块数组”。我们看一个具体的消息结构示例{ messageId: msg-1234, role: agent, taskId: task-5678, parts: [ { text: 根据您的需求我完成了三篇新闻的聚合分类如下, metadata: { style: markdown } }, { file: { fileId: file-001, fileName: aggregated_news.md, mimeType: text/markdown, bytes: 1024, uri: https://storage.example.com/aggregated_news.md } } ], metadata: { timestamp: 2025-04-01T10:30:00Z } }这里的parts数组可以理解为一条消息里可以包含多个类型的内容块文本、文件等。这种设计非常灵活Agent 可以一边输出文字一边附带文件就像微信里你既可以发文字又可以发图片。我在实际开发时发现一个细节file里的uri字段其实非常重要。如果给的是一个临时链接一定要设置好过期时间不然客户端延迟拉取时链接触发 403。另外bytes字段能帮客户端提前分配内存或做大小校验对构建稳定的传输管道很有用。2.4 两种任务提交方式的对比与选型A2A 协议为客户端提交任务提供了两种方式tasks/send和tasks/sendSubscribe。这两者的核心逻辑是一样的区别在于响应方式。tasks/send同步响应客户端发送一个任务请求服务端处理完后返回最终结果。tasks/sendSubscribe基于 SSE 的流式响应服务端一边处理一边把中间结果推给客户端。我建议优先用tasks/sendSubscribe因为你永远不知道一个“看起来很简单”的任务会不会因为外部 API 慢而拖到超时。用流式模式至少可以持续收到心跳和进度信息客户端不容易断连。如果 AgentCard 里声明的streaming是 false再退回到tasks/send。另外一个细节是轮询。如果服务端不支持推送也不支持流式客户端只能定期调用tasks/get去查询状态。轮询间隔我建议设置在 2~5 秒之间太频繁会白白消耗服务端资源太慢会影响用户体验。我通常的做法是前 30 秒每隔 2 秒轮询一次之后如果任务还在进行就逐步拉大到 5 秒这叫指数退避实测对性能优化很有效。3. 实操过程与核心环节实现3.1 环境准备与依赖安装这部分直接上干货。我的实验环境是 Python 3.10 FastAPI uvicornA2A SDK 用的是官方 Python 库a2a-sdk。我建议不要自己从头实现 A2A 协议栈直接基于官方 SDK 开发可以省去大量底层细节的调试时间。安装命令非常简单pip install a2a-sdk如果你需要用到 FastAPI 的 SSE 支持可能还要装一下 sse-starlette实际上 a2a-sdk 强依赖了它会自动装上。装完之后验证一下能否正常导入python -c from a2a.types import AgentCard, Task, Message; print(OK)如果输出OK说明环境已经准备好。注意这里导入的AgentCard, Task, Message就是上一章节介绍的数据模型SDK 已经帮你实现了序列化反序列化不用手动拼 JSON。初始化项目结构大概是这样my-a2a-lab/ ├── server.py # A2A 服务端入口 ├── client.py # A2A 客户端调用示例 ├── agent_cards/ │ └── news_agent.json # AgentCard 配置 └── requirements.txt # 依赖锁定3.2 服务端实现三步跑通一个 A2A Agent为了把 A2A 的核心流程讲透我写了一个极简的“信息查询 Agent”。它的作用很简单接收用户传入的一句话返回这句话是“新闻”“天气”还是“其他”同时随机返回两条相关数据模拟真实 Agent 的决策能力。步骤一定义 AgentCard。我直接把前面那张news_agent.json用 FastAPI 暴露在/.well-known/agent-card.json路径下。代码如下from fastapi import FastAPI from a2a.types import AgentCard, AgentCapabilities, AuthenticationInfo, SecurityScheme app FastAPI() card AgentCard( nameInfoClassifier Agent, description对一句输入进行分类返回分类结果和模拟上下文, urlhttp://localhost:8080, version1.0.0, capabilitiesAgentCapabilities(streamingTrue, pushNotificationsFalse), securitySecurityScheme(auth_schemes[AuthenticationInfo(schemes[bearer], credentials)]), default_input_modes[text/plain], default_output_modes[application/json], skills[] ) app.get(/.well-known/agent-card.json) async def get_card(): return card.model_dump()这里有个容易被忽略的地方url字段要填服务的根地址包括端口客户端后续构造回调 URL 时会基于它来拼接。而且如果你的服务是 HTTPS 部署url务必填 HTTPS否则客户端如果校验回调协议会直接拒绝。步骤二实现核心任务处理。A2A SDK 里最核心的类叫A2AHandler我们继承它并实现handle_message方法即可。处理函数接收的是Message对象里面包含parts消息块数组我们要从里面把文本内容取出来。from a2a.server import A2AHandler, TaskManager from a2a.types import Task, TaskState, Message, TextPart class InfoClassifierHandler(A2AHandler): def __init__(self, task_manager: TaskManager): super().__init__(task_manager) async def handle_message(self, request): # request 是一个 Message取文本内容 user_text extract_text_from_message(request) result classify_text(user_text) # 构造回复消息 reply_msg Message( message_idreply-001, roleagent, parts[TextPart(textresult)], metadata{} ) return Task( idtask-001, stateTaskState.COMPLETED, artifacts[reply_msg], )这里我故意省略了classify_text的实现它内部就是简单的关键词匹配。你可以换成任意 LLM 调用或推荐算法原理完全一致。步骤三挂载到 FastAPI 路由上。SDK 提供了现成的路由注册函数如下from a2a.server import TaskManager, run_server app.get(/) async def index(): return {status: running} # 创建任务管理器并注册路由 task_manager TaskManager() handler InfoClassifierHandler(task_manager) run_server(app, handler, task_manager, port8080)其实 a2a-sdk 提供了更高层的run_server方法可以自动把/、/.well-known/agent-card.json、/tasks/send、/tasks/sendSubscribe这些路由全部挂载好。如果你用 FastAPI可以手动挂载我在实践中的做法是from a2a.server.fastapi import A2ARoutes A2ARoutes(app, handler, task_manager)然后启动服务uvicorn server:app --host 0.0.0.0 --port 8080启动成功后访问http://localhost:8080/.well-known/agent-card.json应该能看到我们定义的 AgentCard。这一步跑通你的第一个 A2A Agent 服务端就上线了。3.3 客户端实现如何发现并调用远程 Agent服务端打开后客户端就简单了。我们用A2AClient来调用。先获取 AgentCard 再发送任务。from a2a.client import A2AClient from a2a.types import Message, TextPart async def main(): client A2AClient(http://localhost:8080) # 1. 获取 AgentCard 发现能力 card await client.agent_card() print(Agent name:, card.name) print(Streaming support:, card.capabilities.streaming) # 2. 发送任务同步模式 message Message( message_idmsg-001, roleuser, parts[TextPart(text今天天气怎么样)], metadata{} ) task await client.send_task(message) print(Task state:, task.state) if task.artifacts: print(Agent response:, task.artifacts[0].text)这里send_task对应的是tasks/send端点。如果我们要用流式模式就改成send_task_subscribe接收一个异步生成器async for event in client.send_task_subscribe(message): # event 里包含任务状态和增量消息 print(event)流式模式下你会看到状态从submitted逐步变为working最后到completed中间可能有多次消息推送。我建议你在自己的代码里也这么处理用异步生成器去消费这些事件这样做的好处是前端可以实时显示“Agent 正在处理中”这样的状态交互体验明显提升。3.4 双 Agent 协作让两个 Agent 真正“对话”起来单 Agent 的 A2A 对接算热身真正体现协议价值的是多 Agent 协作。我这里搭了一个简单的“写作→审校”双 Agent 流水线写作 Agent 负责生成一段产品介绍文案审校 Agent 负责检测文案是否合理并输出修改建议。架构图很简单一个主控程序通过 A2A 同时连接写作 Agent 和审校 Agent。主控先调用写作 Agent 生成初稿然后拿着初稿调用审校 Agent审校 Agent 输出结构化修改意见写回给写作 Agent 继续修改循环若干轮。核心代码如下省略部分细节async def run_collaboration(client_writer, client_reviewer): prompt 请生成一段关于智能手表的介绍文案 writer_msg Message(roleuser, parts[TextPart(textprompt)]) writer_task await client_writer.send_task(writer_msg) draft extract_text_from_task(writer_task) for round_idx in range(3): review_msg Message( roleuser, parts[TextPart(textdraft)] ) review_task await client_reviewer.send_task(review_msg) review_result extract_review_result(review_task) if review_result[score] 85: return draft, round_idx 1 revise_msg Message( roleuser, parts[TextPart(textf根据建议修改{review_result[advice]})] ) writer_task await client_writer.send_task(revise_msg) draft extract_text_from_task(writer_task) return draft, 3这个流程跑通以后你会非常直观地感受到 A2A 的价值两个 Agent 各自只暴露 A2A 接口主控程序不需要关心它们内部用了什么框架是 LangChain 还是自研的 RAG Pipeline通通无所谓。只要 AgentCard 能拿到、tasks/send能调通就能协作。我用这个双 Agent 流水线跑过很多次最终质量取决于每个 Agent 的提示词设计和模型能力而协作本身的稳定性A2A 协议保障得很扎实。3.5 多 Agent 并发调用的工程经验真实业务里多 Agent 往往不是一对一依次调用而是一对多并发调用。比如你要同时让 5 个 Agent 去分析不同类型的数据最后汇总。这时就需要控制并发度否则服务端可能被压垮。我常用的方案是asyncio.Semaphore控制并发上限。import asyncio semaphore asyncio.Semaphore(5) async def call_with_limit(client, msg): async with semaphore: return await client.send_task(msg) async def main(): tasks [call_with_limit(client, Message(parts[TextPart(textt)])) for t in workload] results await asyncio.gather(*tasks, return_exceptionsTrue)并发上限不是越大越好。Agent 服务端如果有外部依赖比如调用 LLM API并发过高会直接触发对方限流。我自己的经验是如果每个 Agent 都依赖同一个 LLM API并发控制在 3~5 个就差不多了。你可以在服务端监控响应时长的 95 分位数来判断自己的并发阈值。另一个很容易踩的坑是超时设置。A2AClient底层基于 HTTPX默认超时可能是 5 秒或 10 秒但 A2A 任务常常要跑 20 秒以上。我建议显式设置超时from a2a.client import A2AClient client A2AClient(urlhttp://localhost:8080, timeout120)这个配置对流式任务尤其重要流式模式下服务端会持续推送空闲时才可能触发超时但很多情况下客户端代理网关会在 60 秒截断连接所以超时设置最好大于最长可能的生成时长同时配合心跳消息来保护连接。4. 常见问题与排查技巧实录4.1 握手失败AgentCard 请求 404 或 403这是新人最常见的坑。我在一开始搭建服务时也遇到过客户端请求 AgentCard 返回 404。原因很简单路由没有正确注册或者路径写错。排查步骤用浏览器或 curl 直接访问http://localhost:8080/.well-known/agent-card.json看是否能返回 JSON。如果返回 404检查你的服务是否监听了正确端口以及路由注册是否生效。如果返回 403基本是服务端做了访问控制。加了认证的话客户端请求时要把 Token 放进 Header。注意A2A 协议规定Agent Card 端点不需要鉴权也能访问因为它是给所有客户端做发现用的。如果你想保护具体任务接口可以只对/tasks/*接口加认证而 AgentCard 保持开放。这是我后来在整理服务安全时的经验总结。4.2 任务状态一直停在 submitted 或 working如果你看到任务提交成功但一直没有后续问题通常出在服务端任务处理器没有正确触发或者推送通知没到达客户端。第一种情况很好排查看服务端日志。如果服务端日志显示任务处理函数没有被调用问题在手柄注册如果日志显示函数被调用但返回异常问题在业务代码。第二种情况需要检查 AgentCard 里的pushNotifications字段。如果客户端接收推送失败协议会自动回退到轮询模式但如果你把轮询间隔搞太长看起来就像“卡住了”。我把轮询间隔心理底线调到 10 秒超过 10 秒用户就会觉得不对劲。另外有一种特殊情况任务确实正在跑但算法内部有长循环或者调用了外部 API导致迟迟没有输出。这种情况建议在任务处理函数里增加进度反馈用sendSubscribe模式定时推送进度让客户端知道系统还活着。4.3 SSE 连接断开或数据截断流式模式下 SSE 中断是很头疼的问题。A2A 的 SSE 本质上是 chunked transfer encoding 的事件流。如果前面有代理服务器nginx、网关等很可能默认会缓冲响应导致客户端半天收不到数据。我的解决方案是在服务端代理配置里显式关闭缓冲并设置长超时。以 nginx 为例需要在location块里加proxy_buffering off; proxy_read_timeout 300s; proxy_send_timeout 300s;另外 SSE 事件的 header 有个retry字段可以告诉客户端如果断开后多久重连。我在服务端推荐设置为 3000 毫秒这样断线后客户端能较快恢复。如果客户端直接调用的本地服务没走代理还断连那大概率就是代码里没有正确释放生成器。使用async for消费 SSE 流时一定要确保循环体里没有异常导致提前退出否则连接会被 GC 掉。4.4 认证与鉴权如何安全暴露 AgentA2A 协议本身不带认证实现它只定义了 AgentCard 里的security字段。这意味着你可以接入任意你想用的认证方式只要双方约定一致即可。我最常用的是 Bearer Token。做法是在客户端请求头里带上Authorization: Bearer token。服务端用 FastAPI 的依赖注入做校验比如from fastapi import Header, HTTPException def verify_auth(authorization: str Header(None)): if authorization ! Bearer my-secret-token: raise HTTPException(status_code401, detailUnauthorized)如果你想让一套 Token 对应多个 Agent 调用也可以做成集中式的 API Key 管理。但核心原则只有一个AgentCard 不需要鉴权任务接口必须鉴权。这个原则能保证服务发现机制不被破坏同时任务调用不会被滥用。4.5 多 Agent 循环调用时的死锁问题这个我踩过很深。多 Agent 协作时如果 A 调 B、B 调 A两边的 HTTP 连接同时挂着等对方返回会造成死锁。比如一个“规划 Agent”调用一个“执行 Agent”执行 Agent 由于某种原因反过来又调用了规划 Agent两个任务都互相等着最终超时。解决思路有三种取消同步调用改成事件驱动。通过消息队列Redis Stream、RabbitMQ、Kafka把 Agent 间的调用变成异步消息不持有 HTTP 连接。增加递归深度限制。如果 Agent 之间的调用超过 N 层直接拒绝再往下调防止环形依赖无限膨胀。设置全局任务超时和最大待处理数。一旦积压任务超过阈值新任务直接返回“忙”让上层 Agent 等待一段时间后重试。我的建议是如果多 Agent 是一个长期存在的系统尽量采用异步消息驱动架构HTTP 仅仅用于客户端到网关的通信Agent 之间的内部调用全部走消息队列。这样扩展性最好也最容易排查问题。5. 进阶优化与实用建议5.1 用 A2A 做服务发现Agent 注册中心随着 Agent 数量增多你不可能让每个客户端都记住所有 Agent 的地址。我的做法是搭一个轻量级的 Agent 注册中心每个 Agent 启动时把自己的 AgentCard 注册到中心中心维护一个可查询的列表。实现逻辑非常简单写一个维护数组的 FastAPI 服务暴露GET /agents和POST /registry。Agent 启动时向POST /registry提交自己的 AgentCard客户端需要时调用GET /agents查询。这样做的意义在于当客户端需要找一个“能查天气”的 Agent 时只需要遍历注册中心里的 AgentCard根据skills字段匹配目标就可以自动拉取地址并调用不再需要硬编码。我用这个方式接入了 6 个不同的内部 Agent效果相当不错新增 Agent 时只需要编写 AgentServer 并注册客户端的调用零修改。5.2 结合 MCPA2A MCP 的混合架构模式严格来说A2A 和 MCP 并不是二选一。我个人的最佳实践是每个 Agent 对外用 A2A 暴露能力对内用 MCP 调用外部工具和知识库。比如我的“数据分析 Agent”它对外接受任务请求时走 A2A处理过程中需要访问数据库时走 MCP需要读取文件时走 MCP。这种组合的好处是外层 A2A 保证了跨 Agent 协作的互操作性内层 MCP 保证了工具接入的标准化两个协议各管一层边界非常清晰。如果你还在纠结接入哪个协议建议都学你可以在一个 Agent 里同时实现 A2A Server 和 MCP Server一个对外一个对内。5.3 性能优化Agent 间通信的三大瓶颈多 Agent 协同的性能问题和单体应用完全不同。我实测下来三大瓶颈往往出现在序列化开销、网络往返、任务排队。序列化开销方面A2A 每次传输都要做 JSON 序列化反序列化。如果你传递的数据很大比如几 MB 的文件直接把内容嵌在 Message 的parts里会很吃亏。我的建议是大数据文件放在对象存储如 S3/MinIO里Message 中只放uri指针Agent 需要时再按需拉取。这样整个消息体保持小体积序列化成本可控。网络往返方面Agent 之间的每次调用都要消耗一次 HTTP 往返。如果一个协作流程有 10 步那就是 10 次以上的网络延迟。可以把多个步骤合并成一个 Agent 内部的子流程减少对外暴露的调用次数。本质上和微服务里避免连环调用是同一个道理。任务排队方面当多个客户端同时向同一个 Agent 发起任务时服务端默认是并发处理的。但如果 Agent 背后依赖的是 LLM API而 LLM API 有速率限制就会导致大量超时。这时建议在服务端引入线程池或信号量控制并发数超出的任务先放入待处理队列客户端侧就可以看到submitted状态但迟迟不进展。队列越长时间越长该限流还是要限流。5.4 从 demo 到生产落地 A2A 时的工程建议最后聊点实际的。如果你和我一样准备把一个 A2A demo 变成生产系统有几件事越早做越好。第一把 AgentCard 放到配置中心不要硬编码。服务端地址变更时只需要更新配置中心里的卡片所有客户端自动感知。第二写好日志和可观测性。A2A 的 taskId 是绝佳的链路追踪标识建议把它透传到内部的日志系统里每个任务的处理耗时、消息交互过程都要有迹可循。我在生产环境用 OpenTelemetry 做了全链路追踪一个 taskId 就能看到整个协作链路。第三做好重试和幂等设计。网络问题在分布式系统中不可避免客户端调用 Agent 时要有超时重试服务端处理任务时要以 taskId 为纬度做幂等判断防止同一个任务被重复处理。第四给 Agent 加“心跳”。生产环境中 Agent 可能会因为各种原因崩溃有一个监控系统定时检查每个 Agent 的 AgentCard 是否可访问能帮助运维快速发现问题而不是等客户的调用全部超时后再被动响应。写在最后的一点体会A2A 协议目前还在快速演进中。从发布到现在我眼瞅着官方 SDK 从 alpha 到 beta接口也在调。如果你现在就想上手我建议跟着官方 GitHub 仓库的示例做并且把 SDK 版本 pin 死免得升级后破坏现有代码。从实际效果来说我觉得 A2A 最大的价值不是让你多做几个 Agent而是把所有 Agent 装进同一套标准里让“Agent 团队”像“人团队”一样协作。它把“通信成本”从“开发成本”变成了“约定成本”——这正是标准化协议的意义所在。如果你已经在用 MCP再花一到两周时间把 A2A 的收发链路跑通绝对是值得的。后续如果我自己在真实业务中积累更多数据比如任务吞吐量、平均协作次数、失败率这些指标再写一篇后续分享给大家。