1. 从两个缩写说起MCP 与 A2A 到底在解决什么问题第一次看到 MCP 和 A2A 这两个词放在一起很多人会本能地以为是某种硬件通信协议比如 CAN、Modbus、IIC 那一类跑在物理总线上的东西。实际上它们都属于应用层的软件协议解决的是不同程序之间怎么把话说清楚的问题只不过一个面向的是模型和工具之间的连接另一个面向的是智能体与智能体之间的协作。MCP 全称 Model Context Protocol核心目标是给大模型应用提供一个标准化的方式去访问外部资源——文件、数据库、API、命令行工具都算。在它出现之前每接一个工具就要写一套适配代码工具一多维护成本直接爆炸。MCP 把这件事抽象成客户端-服务端模型工具方只需要实现一个 MCP Server任何支持 MCP 的客户端都能直接调用省掉了大量重复对接的工作。A2A 全称 Agent to Agent关注的是另一个层面当你有多个智能体各自负责不同任务时它们之间怎么协商、怎么传递任务、怎么汇报结果。MCP 解决的是智能体怎么用工具A2A 解决的是智能体怎么找智能体帮忙。两者不是竞争关系而是互补关系一个向内连接资源一个向外连接同伴。这篇内容适合谁看如果你正在做 AI 应用集成、想让自己的系统接入大模型能力、或者手上有多个自动化流程需要串起来那这两个协议值得花时间搞清楚。下面我会从设计思路、协议细节、实操落地到踩坑排查把整套东西拆开讲一遍尽量做到看完就能动手。2. 协议整体设计与选型思路拆解2.1 为什么是 JSON-RPC 作为底座MCP 和 A2A 在传输层都选择了 JSON-RPC 2.0 作为消息格式这个选择不是随便定的。JSON-RPC 的结构极其简单一个请求就是jsonrpc、method、params、id四个字段响应就是result或error二选一。这种极简结构带来的好处是任何语言都能在半小时内实现一个可用的编解码器不需要引入厚重的框架。对比一下其他选项就明白了。如果用 gRPC你得维护 proto 文件、处理代码生成、还要考虑 HTTP/2 的兼容性对于工具调用这种请求-响应模式来说太重了。如果用纯 REST虽然通用但缺少统一的错误码规范和请求 ID 机制做双向通信和异步任务时会很别扭。JSON-RPC 刚好卡在中间比 REST 规范比 gRPC 轻量。提示JSON-RPC 的id字段是关联请求和响应的关键异步场景下必须保证同一会话内 id 不重复否则响应会串线。2.2 Streamable HTTP 解决了什么痛点早期 MCP 的远程传输用的是 HTTP SSE 的组合服务端通过 SSE 长连接往客户端推消息。这套方案在本地跑没问题但一旦部署到有负载均衡、有网关的环境里就出问题SSE 连接容易被中间层掐断重连之后上下文丢失多实例部署时还会出现请求打到 A 实例、推送却从 B 实例发出的情况。Streamable HTTP 的思路是把流和请求统一到同一个 HTTP 端点上。客户端发一个 POST 请求服务端可以选择直接返回一个 JSON 响应也可以返回一个text/event-stream的流式响应在流里持续推送多条消息。关键在于这个流是可断可续的服务端可以给每条消息打上事件 ID客户端断线后用Last-Event-ID头重新发起请求就能从断点继续接收。这个设计对部署非常友好。因为所有交互都走标准 HTTP负载均衡器不需要做任何特殊配置会话状态可以通过事件 ID 在服务端重建多实例部署时只要共享会话存储就能正常工作。实测下来这套机制在容器化环境里的稳定性比老的 SSE 方案高出一个档次。2.3 MCP 与 A2A 的职责边界很多人会问既然 MCP 已经能让智能体调用工具了为什么还需要 A2A答案在于抽象层级不同。MCP 里的工具是无状态的能力单元比如查天气、读文件、执行 SQL它不关心谁在调用也不维护跨请求的上下文。而 A2A 里的智能体是有状态的协作方它可能记得之前的对话、有自己的任务队列、甚至需要反过来向调用方请求补充信息。举个实际场景你要做一个自动化的报表系统。用 MCP你可以让模型调用查询数据库和生成图表两个工具这是工具层面的组合。但如果报表生成需要经过数据校验智能体和格式排版智能体两道工序两者之间要传递中间结果、要处理校验失败后的重试这就属于 A2A 的范畴了。选型上的经验是单一职责、无状态、调用即返回的能力用 MCP 封装成工具需要多轮交互、有内部状态、可能反向提问的用 A2A 定义成智能体。两者可以在同一个系统里共存MCP Server 也可以被 A2A 智能体当作工具来调用。3. 核心细节解析与实操要点3.1 MCP 的三类核心原语MCP 协议里定义了三种服务端能力理解它们的区别是上手的关键。Tools工具是最常用的一类代表可以被模型主动调用的函数。每个工具需要提供名称、描述和 JSON Schema 格式的参数定义。模型根据描述决定要不要调用、传什么参数。这里有个容易踩的坑描述写得太模糊模型就不知道该在什么场景下用描述写得太长又会挤占上下文窗口。我的经验是描述控制在两句话以内第一句说清楚做什么第二句说清楚什么时候用。Resources资源代表可以被读取的数据比如文件内容、数据库记录、配置项。和工具不同资源是被动的通常由客户端或用户决定要不要加载而不是模型自主决定。资源用 URI 标识支持file://、db://这类自定义 scheme。Prompts提示模板是预定义的提示词模板服务端可以提供一组参数化的 prompt客户端调用时填入参数就能得到完整的提示词。这个能力在需要固定输出格式的场景下特别有用比如代码审查模板、周报生成模板。原语触发方典型用途是否消耗上下文Tools模型执行操作、查询数据是需加载 schemaResources客户端/用户加载文件、读取配置按需加载Prompts用户复用提示模板是加载模板内容3.2 工具定义的 Schema 设计要点工具的参数 Schema 直接决定了模型能不能正确调用。我见过太多因为 Schema 设计不当导致调用失败的案例这里总结几条硬性经验。参数类型尽量用基础类型string、number、boolean、array这几种就够了。避免用嵌套很深的 object模型在深层嵌套上出错率明显更高。如果确实需要复杂结构拆成多个扁平参数在服务端再组装。每个参数都要写description而且要写清楚格式要求。比如日期参数光写日期不够要写ISO 8601 格式例如 2024-01-15。枚举参数要把所有可选值列在enum里不要指望模型猜。必填参数用required数组明确标注。可选参数要给合理的默认值并且在描述里说明默认行为。我踩过的一个坑是某个可选参数没写默认值模型有时传有时不传导致服务端逻辑分支爆炸后来统一在服务端做了默认值兜底才稳定下来。{ name: query_orders, description: 查询指定时间范围内的订单。用于用户询问订单情况时。, inputSchema: { type: object, properties: { start_date: { type: string, description: 开始日期ISO 8601 格式例如 2024-01-15 }, end_date: { type: string, description: 结束日期ISO 8601 格式例如 2024-01-20 }, status: { type: string, enum: [pending, paid, shipped, completed], description: 订单状态筛选不传则返回全部状态 } }, required: [start_date, end_date] } }3.3 A2A 的任务生命周期A2A 的核心是任务Task这个概念。一个任务从创建到结束会经历若干状态submitted、working、input-required、completed、failed、canceled。理解状态流转是排查问题的前提。input-required这个状态是 A2A 区别于普通 RPC 的关键。当智能体执行到一半发现信息不足时它可以把任务置为这个状态并附带需要补充的信息说明。调用方看到这个状态后补充信息再继续推进任务。这种反向提问的能力让智能体之间的协作更接近人类之间的沟通。任务的状态变更通过流式事件推送。每次状态变化都会产生一个事件事件里包含任务 ID、当前状态和相关的消息内容。调用方根据事件类型决定下一步动作。这里要注意事件可能乱序到达处理逻辑必须基于任务 ID 做幂等不能假设事件严格有序。3.4 会话与上下文管理MCP 的会话通过Mcp-Session-Id头来标识。客户端首次连接时服务端返回一个会话 ID后续请求都带上这个头。服务端根据会话 ID 维护上下文比如已加载的资源、已建立的连接。A2A 的上下文管理更复杂一些因为涉及多个智能体。每个任务有独立的上下文任务之间通过contextId关联。同一个 context 下的多个任务可以共享历史消息这样智能体在处理新任务时能参考之前的交互。实操中一个常见问题是会话过期。服务端通常会设置会话超时时间超时后会话 ID 失效客户端需要重新初始化。如果客户端没处理好这个逻辑就会出现突然所有请求都返回会话无效的情况。我的做法是在客户端加一层自动重连捕获会话失效错误后自动重新初始化并重放当前请求。4. 实操过程与核心环节实现4.1 搭建一个最小可用的 MCP Server先从一个最简单的 MCP Server 开始实现一个查询当前时间的工具。用 Python 的官方 SDK 来做整个代码不到五十行。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import datetime app Server(time-server) app.list_tools() async def list_tools(): return [ Tool( nameget_current_time, description获取当前时间。用于用户询问现在几点时。, inputSchema{ type: object, properties: { timezone: { type: string, description: 时区名称例如 Asia/Shanghai默认 UTC } } } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name get_current_time: tz arguments.get(timezone, UTC) now datetime.datetime.now() return [TextContent(typetext, textf当前时间{now.isoformat()}时区{tz})] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这段代码用的是 stdio 传输也就是通过标准输入输出通信适合本地进程调用。如果要部署成远程服务需要换成 Streamable HTTP 传输。4.2 切换到 Streamable HTTP 传输远程部署时把传输层换成 HTTP。下面是基于 Starlette 的实现骨架。from mcp.server.streamable_http import StreamableHTTPServerTransport from starlette.applications import Starlette from starlette.routing import Route transport StreamableHTTPServerTransport( mcp_session_idNone, is_json_response_enabledTrue ) async def handle_mcp(request): return await transport.handle_request(request) app Starlette(routes[ Route(/mcp, handle_mcp, methods[POST, GET, DELETE]) ])这里有几个关键点。is_json_response_enabled设为 True 时服务端对简单请求直接返回 JSON对需要流式的请求返回 SSE 流。路由要同时支持 POST、GET、DELETE 三种方法POST 用于发送请求GET 用于建立流式连接DELETE 用于主动关闭会话。部署到容器环境时要注意会话存储的问题。如果服务端是多实例的会话状态必须放在共享存储里比如 Redis。否则客户端第一次请求打到实例 A第二次打到实例 BB 找不到会话就会报错。我一般会在会话存储里存三样东西会话 ID、已初始化的能力列表、最后活跃时间。4.3 实现一个 A2A 智能体A2A 智能体的实现比 MCP Server 复杂一些因为要处理任务状态流转。下面是一个处理文本摘要任务的智能体骨架。from a2a.server import A2AServer from a2a.types import Task, TaskState, Message class SummaryAgent(A2AServer): async def on_task(self, task: Task): text task.message.parts[0].text if len(text) 50: await self.update_task( task.id, TaskState.INPUT_REQUIRED, Message(text文本太短请提供至少 50 字的文本) ) return await self.update_task(task.id, TaskState.WORKING) summary await self.summarize(text) await self.update_task( task.id, TaskState.COMPLETED, Message(textsummary) ) async def summarize(self, text: str) - str: # 实际摘要逻辑 return text[:100] ...这个例子里智能体先检查输入长度不够就返回input-required状态要求补充。够的话进入working状态处理完置为completed。调用方通过流式事件感知这些状态变化。4.4 客户端如何同时对接 MCP 和 A2A实际系统里客户端往往需要同时管理 MCP 连接和 A2A 连接。我的做法是抽象一个统一的能力注册表把 MCP 工具和 A2A 智能体都注册进去对上层暴露统一的调用接口。class CapabilityRegistry: def __init__(self): self.mcp_clients {} self.a2a_clients {} async def register_mcp(self, name, url): client await MCPClient.connect(url) tools await client.list_tools() self.mcp_clients[name] {client: client, tools: tools} async def register_a2a(self, name, url): client await A2AClient.connect(url) card await client.get_agent_card() self.a2a_clients[name] {client: client, card: card} async def invoke(self, target, method, params): if target in self.mcp_clients: return await self.mcp_clients[target][client].call_tool(method, params) elif target in self.a2a_clients: return await self.a2a_clients[target][client].send_task(params)这层抽象的好处是上层业务代码不用关心底层是 MCP 还是 A2A统一按目标 方法 参数的方式调用。新增能力时只需要注册不用改业务逻辑。4.5 参数计算与超时设置超时设置是实操中最容易被忽视的环节。MCP 工具调用和 A2A 任务执行的耗时差异很大不能用一个统一的超时值。我的经验值是这样的MCP 工具调用超时设 30 秒因为大部分工具是查询类操作超过 30 秒基本可以判定有问题。A2A 任务超时设 5 分钟因为涉及多轮交互和复杂处理。流式连接的空闲超时设 60 秒超过这个时间没有事件推送就认为连接已死主动重连。重连策略用指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒最多退避到 30 秒。连续失败 5 次后停止重连并告警。这个策略在实测中能有效应对服务端短暂重启的情况又不会在服务端彻底挂掉时无限重试。5. 常见问题与排查技巧实录5.1 工具调用返回方法不存在这是最常见的问题通常有三个原因。第一是工具名拼写不一致客户端注册时用的名字和服务端定义的名字大小写不同。第二是会话未初始化服务端还没收到initialize请求就收到了tools/call。第三是服务端版本不匹配客户端按新协议发请求服务端还是老版本。排查顺序先看服务端日志有没有收到initialize再看工具列表里有没有目标工具最后对比客户端和服务端的协议版本号。我一般会在客户端启动时打印一次完整的工具列表出问题时对照这个列表就能快速定位。5.2 流式连接频繁断开流式连接断开的原因比较多按概率排序中间层网关的空闲超时、服务端心跳没发、客户端没处理重连。网关空闲超时是最常见的。很多云厂商的负载均衡默认 60 秒无数据就断连接。解决办法是服务端定期发送心跳事件比如每 30 秒发一个空的事件。这样连接始终有数据流动不会被判定为空闲。客户端侧要正确处理Last-Event-ID。断线重连时带上最后收到的事件 ID服务端就能从断点继续推送。如果客户端没实现这个逻辑重连后要么丢消息要么重复处理。5.3 A2A 任务卡在 working 状态任务一直停在working不结束通常是智能体内部抛了异常但没捕获导致状态没更新。排查方法是看智能体的日志找有没有未处理的异常。另一个可能是任务处理时间确实很长超过了客户端的超时设置。这种情况要么调大超时要么让智能体定期发送进度事件让客户端知道任务还在推进。还有一种隐蔽的情况智能体在处理过程中调用了另一个 A2A 智能体但那个智能体没响应导致当前智能体一直等待。这种级联等待很难排查我的做法是给每个智能体设置内部调用超时超时后主动失败并上报避免无限等待。5.4 常见问题速查表现象可能原因排查方法解决方式方法不存在名称不一致/未初始化对比工具列表统一命名确保先初始化连接频繁断开网关超时/无心跳抓包看断开时机加心跳实现断点续传任务卡住异常未捕获/级联等待查智能体日志加异常捕获和内部超时会话失效超时/多实例不共享看会话存储共享存储客户端自动重连参数校验失败Schema 不匹配对比请求和 Schema严格按 Schema 传参5.5 几个独家避坑技巧第一个技巧在开发阶段把 MCP 和 A2A 的原始报文都打到日志里。JSON-RPC 的报文不大全量记录对性能影响很小但排查问题时能省下大量时间。我一般会用一个环境变量控制开关生产环境关掉出问题时临时打开。第二个技巧给每个工具和智能体加一个健康检查方法。客户端启动时先调一遍健康检查确认对端可用再注册。这样能避免注册了一堆不可用的能力调用时才发现问题。第三个技巧A2A 任务的contextId要设计得有语义。不要用随机 UUID用业务类型-日期-序号这种格式比如report-20240115-001。这样排查问题时一眼就能看出这个任务属于哪个业务、什么时候创建的。第四个技巧MCP 工具的返回值尽量结构化。返回纯文本虽然简单但客户端要解析就很麻烦。返回 JSON 字符串客户端直接反序列化省去正则匹配的麻烦。如果确实需要返回文本在 Schema 里明确说明格式。6. 协议组合使用的进阶思路6.1 用 MCP 封装底层能力用 A2A 编排业务流程一个比较成熟的架构是分两层底层用 MCP 把各种原子能力封装成工具上层用 A2A 定义业务流程智能体。业务智能体不直接调用底层 API而是通过 MCP 工具来操作数据。这样分层的好处是职责清晰。底层能力的变化只影响 MCP Server业务流程的变化只影响 A2A 智能体两者解耦。新增一个数据源时只需要加一个 MCP Server所有业务智能体都能用上。6.2 智能体之间的任务分解与结果聚合复杂任务往往需要分解成多个子任务分给不同的智能体处理最后聚合结果。A2A 本身不提供任务分解能力这部分逻辑要自己实现。我的做法是定义一个协调者智能体它接收大任务后根据任务类型拆成子任务通过 A2A 分发给对应的执行智能体收集所有结果后聚合返回。协调者本身也是一个 A2A 智能体对外表现和普通智能体一样。任务分解的粒度要把握好。拆得太细协调开销大拆得太粗单个智能体负担重。经验值是每个子任务的处理时间控制在 10 到 60 秒之间这个区间内并行收益比较明显协调开销也可控。6.3 错误传播与降级策略多智能体协作时一个环节出错可能影响整个流程。错误传播的处理策略有三种快速失败、重试、降级。快速失败适合关键路径上的错误比如数据校验不通过直接终止整个流程并返回错误。重试适合临时性错误比如网络抖动重试几次通常能成功。降级适合非关键环节比如某个增强功能不可用跳过它继续走主流程。实现上我会给每个子任务标注关键性和可重试性两个属性。协调者根据这两个属性决定错误处理策略。关键且不可重试的直接失败非关键或可重试的按策略处理。7. 我在实际项目中的几点体会这套协议组合用下来最大的感受是标准化带来的复用价值远超预期。以前每接一个新系统都要写适配层现在只要对方实现了 MCP 或 A2A对接工作量能减少七成以上。特别是 MCP 的工具体系一旦积累了一批常用工具新项目启动时直接复用效率提升非常明显。另一个体会是协议版本管理要提前规划。MCP 和 A2A 都还在演进字段和语义可能变化。我的做法是在客户端和服务端都记录协议版本握手时校验版本兼容性不兼容就明确报错而不是带着隐患继续跑。这样虽然前期麻烦一点但能避免很多诡异的问题。最后分享一个小技巧调试 MCP 和 A2A 时用一个通用的回显工具或智能体作为参照。先确认回显能正常工作再排查具体业务逻辑。这样能把协议层问题和业务层问题快速区分开排查效率能提高不少。很多时候折腾半天最后发现是协议层没通而不是业务代码写错了。