简介A2A协议与MCP协议是当前多Agent系统绕不开的两个关键话题这份由蒋俊整理的PPT从协议基础概述讲起逐步对比技术架构、功能特性与应用场景并延伸至性能优化与发展前景适合AI开发者、架构师及需要做方案选型的技术人员快速建立整体认知。资源共1个pptx文件压缩包约2.42MB图文页紧凑可配合正文直接学习目前已有331人浏览学习。内容结合物流智能体、药物模拟实验、工业物联网、医疗影像分析等案例清晰梳理了两者的分工A2A通过“Agent卡”和任务生命周期管理支持智能体自然语言协作MCP则以类似USB-C集线器的统一连接方式帮助模型即插即用调用外部工具理解二者如何互补对构建可扩展的多Agent技术栈很有帮助。1. A2A 协议与 MCP 协议两套协议管的是 AI 协作的哪一段一个常见场景你正在搭多智能体平台主 Agent 要调用另一个 Agent 做文件分析还要让主 Agent 能直接读数据库、发消息。市场上被频繁提到的 A2A 协议和 MCP 协议正好落在两个层次上A2A 管 Agent 与 Agent 之间的发现、任务下发和状态追踪MCP 管模型与工具、数据源之间的上下文连接。两者不是二选一而是上下层关系。这篇文章从协议拆解、最小可运行实现到联调踩坑按一线落地顺序写适合正在做多 Agent 编排或把 AI 接入内部系统的工程师。2. A2A 协议拆开看AgentCard、任务卡片与最小可运行服务A2A 协议把 Agent 协作抽象成三个对象AgentCard 是名片描述这个 Agent 的能力和入口Task 是一次可追踪的协作单元Message 是任务里的一条消息。这套模型的好处是调用方不需要提前知道 Agent 的实现语言或框架只要拿到 AgentCard 就能按统一格式发请求。下面按“发现 → 建任务 → 查状态”的顺序拆。2.1 AgentCard 与任务状态先看懂协议怎么描述能力一个最小 AgentCard 在 HTTP 的/.well-known/agent.json上返回内容大致是这样{ name: invoice-processor, description: 处理并校验发票文件, url: https://agent.example.com/rpc, version: 1.0, capabilities: { streaming: true, pushNotifications: false } }这里有几个容易误读的字段。name和description是给人和编排器看的url才是真正的 RPC 入口后续所有 JSON-RPC 请求都要发到这个地址。capabilities声明的是传输能力不是业务能力——streaming表示能否返回流式结果pushNotifications表示能否在长任务结束后主动通知。A2A 没有把“这个 Agent 能干什么”写进卡片业务能力靠协作时的内容协商或独立的能力服务发现这一点新手经常栽跟头。任务侧的状态机更简单一个 Task 包含id、status.state、artifacts和messages。status.state常见值是submitted、working、input-required、completed、failed、canceled。服务端在message/send的响应里返回 Task客户端根据state决定是继续轮询还是直接取结果。2.2 用 Python 跑通一个最小的 A2A 客户端与服务端我一般不先引 SDK而是用标准库写一个最小服务跑通协议再换框架。下面这个服务只实现了两个关键端点GET /.well-known/agent.json返回名片POST /rpc处理message/send。# server.py -- 一个符合 A2A 核心交互的最小 HTTP 服务 import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer AGENT_CARD { name: mock-agent, description: A minimal A2A agent for demo, url: http://localhost:8080/rpc, capabilities: {streaming: False} } class Handler(BaseHTTPRequestHandler): def _send_json(self, obj, status200): body json.dumps(obj).encode() self.send_response(status) self.send_header(Content-Type, application/json) self.send_header(Content-Length, str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path /.well-known/agent.json: self._send_json(AGENT_CARD) else: self.send_error(404) def do_POST(self): if self.path ! /rpc: return self.send_error(404) size int(self.headers.get(Content-Length, 0)) req json.loads(self.rfile.read(size)) method req.get(method) if method message/send: # 构造一个已完成的任务 task { id: req[params][message][taskId], status: {state: completed}, artifacts: [{parts: [{text: hello from mock agent}]}] } self._send_json({jsonrpc: 2.0, id: req[id], result: {task: task}}) else: self.send_error(400) if __name__ __main__: ThreadingHTTPServer((0.0.0.0, 8080), Handler).serve_forever()代码里的AGENT_CARD是协议中的“名片”本体capabilities.streaming设为false时message/send会直接返回完整 Task。如果你要模拟长任务需要把state改成working再在后续请求中实现tasks/get轮询。这里taskId直接取了客户端消息里的字段是简化行为真实协议建议由服务端生成。对应客户端也很短# client.py -- 读取 AgentCard 并发起任务 import json import urllib.request def get_agent_card(base_url): with urllib.request.urlopen(base_url /.well-known/agent.json) as resp: return json.load(resp) def send_message(rpc_url, task_id, text): payload { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, taskId: task_id, parts: [{text: text}] } } } req urllib.request.Request( rpc_url, datajson.dumps(payload).encode(), headers{Content-Type: application/json} ) with urllib.request.urlopen(req) as resp: return json.load(resp) if __name__ __main__: base http://localhost:8080 card get_agent_card(base) print(agent:, card[name]) result send_message(card[url], task_idtask-001, textping) print(state:, result[result][task][status][state])客户端的重点是card[url]是 RPC 入口不是名片地址。整个调用链从发现文档开始之后客户端不再需要知道 Agent 内部实现。参数说明task_id在简化版里由客户端传入实际协作中应由任务发起方在内的调用链生成parts是消息内容可以是纯文本也可以是结构化数据。如果打印出的 state 是working就要转入轮询或流式等待。2.3 A2A 的认证与安全边界你需要知道的几个参数A2A 规范本身没有强制认证但实际部署时你至少要考虑三层传输层 TLS、请求头认证、任务级权限。目前主流做法是 OAuth 2.0 Bearer TokenAgentCard 里的security字段可以声明支持的认证方案。我自己的项目起步阶段都用 Bearer Token服务端做一个中间件检查Authorizationheader客户端把 token 放在请求头里。认证方式AgentCard 配置适合场景Bearer Token在 security 字段声明 Bearer 方案内部服务间调用OAuth 2.0声明授权服务器地址面向外部开发者mTLS网关层终止 TLS 并校验证书敏感数据服务另一个常见参数是超时。A2A 的message/send是同步 HTTP 请求但任务本身可能异步完成所以 HTTP 超时和任务超时要分开设。我通常把 HTTP 超时设为 5 秒任务轮询间隔 12 秒如果capabilities.streaming开了则用 SSE 通道接收增量结果。注意不要把 AgentCard 的url写成卡片自身路径那是第 4 章里最经典的联调失败原因。3. MCP 协议落地从 Resources 到 Tools 的本地 Server 搭建MCP 的定位与 A2A 完全不同它解决的是模型 Host 与外部系统之间的“上下文插头”。一个 MCP Server 可以暴露三类能力Prompts、Resources、Tools。模型客户端通过 JSON-RPC 2.0 向 Server 发起调用Server 通过 stdin/stdout 或 HTTP 响应。下面先讲清楚三个原语再给一个不依赖 SDK 的最小实现。3.1 MCP 的三大原语Prompts、Resources、ToolsMCP 把可共享能力分成三类理解它们的区别是选型的前提。我用一张表总结原语作用关键方法能否产生外部副作用Prompts预置可复用的提示模板prompts/list, prompts/get否Resources向模型提供只读上下文resources/list, resources/read否Tools可执行函数模型可调用tools/list, tools/call是生产系统里 Tools 用得最多因为模型需要读数据库、发消息、执行计算。Resources 适合把文档、配置、状态快照提供给模型比如把一份崩溃日志作为只读上下文喂给模型。Prompts 则适合团队统一 prompt 风格避免每个调用的人写出的提示词差异过大。注意这三类原语不是互斥的。同一个 Server 往往同时提供 Resources 和 Tools资源负责喂数据工具负责产生动作。设计时我会先列出模型的真实需求如果只是“看”数据优先 Resources如果需要“改”数据必须走 Tools这样权限边界才清晰。3.2 不依赖 SDK 的最小 MCP Serverstdio 的初始化与 tools/callMCP 在本地开发最常用的是 stdio 模式Host 把 MCP Server 当作子进程启动通过标准输入输出交互。stdout 是协议通道所有日志必须写到 stderr。下面这个代码只实现initialize、tools/list、tools/call但足够验证协议。# mcp_mini_server.py -- 只实现核心方法的最小 MCP 服务 import sys import json def send(msg): # stdio 简化帧每行一个 JSON-RPC 消息 sys.stdout.write(json.dumps(msg) \n) sys.stdout.flush() def handle(req): method req.get(method) if method initialize: return { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: mini-mcp, version: 0.1.0} } if method tools/list: return { tools: [ { name: add, description: Add two integers, inputSchema: { type: object, properties: {a: {type: integer}, b: {type: integer}}, required: [a, b] } } ] } if method tools/call: name req[params][name] args req[params][arguments] if name add: return { content: [{type: text, text: str(args[a] args[b])}] } return {error: {code: -32601, message: method not found}} for line in sys.stdin: req json.loads(line) if req.get(method) notifications/initialized: continue result handle(req) if result and error in result: send({jsonrpc: 2.0, id: req[id], error: result[error]}) else: send({jsonrpc: 2.0, id: req[id], result: result})运行方式是python3 mcp_mini_server.py。Host 会以子进程拉起它并通过管道写入 JSON-RPC 请求。代码里忽略notifications/initialized因为通知不需要响应每个响应都带idHost 用id匹配请求和响应。参数说明capabilities.tools表示这个 Server 支持工具调用protocolVersion使用常见的2024-11-05不同 Host 可能要求不同版本联调时以 Host 日志为准。inputSchema必须是 JSON Schema 而不是 Python 类型标注模型端会按照 Schema 做参数校验这里add工具要求 a、b 都是整数如果模型拿了字符串进来Host 会先报错。3.3 MCP 的传输层选型stdio 与 HTTPSSE 的区别MCP 的 JSON-RPC 消息格式相同但传输层不同直接影响部署和认证方式。本地开发选 stdio远程多租户选 HTTP。维度stdioHTTPSSE进程部署本地子进程远程服务认证依赖父进程权限可加 OAuth适用场景本地脚本、桌面端多客户端共享stdio 的最大优势是无认证配置只要文件系统权限和父进程权限控制好。缺点是 Server 和 Host 必须同机无法做成中心化服务。HTTPSSE 可以把 MCP Server 部署成独立服务但要注意 SSE 是单向流服务端推送消息需要额外通道或者选用后来 MCP 社区更常用的 Streamable HTTP 方式。我自己的经验是先按 stdio 把协议逻辑跑通再根据部署环境换传输层业务处理代码可以做到不用大改。4. A2A 与 MCP 对接踩坑5 个让联调翻车的真实原因两套协议单独跑通不难难在把它们接到一个系统里。下面 5 个坑是我在联调中反复遇到的按协议生命周期排列每条都是“现象 → 原因 → 解决”的结构。4.1 AgentCard 的 url 指向错误任务发不出去现象客户端拿到 AgentCard 后把请求发到了http://host/.well-known/agent.json结果收到 404 或 JSON-RPC 错误任务始终发不出去。原因AgentCard 里的url是协议入口也就是 RPC endpoint而不是发现文档地址。发现文档只负责返回名片不接收 RPC 请求。解决始终用card[url]作为 POST 目标。验证时可以先curl -s http://host/.well-known/agent.json | jq .确认 url 字段指向的是/rpc或者你自定义的端点然后再向该端点发message/send。4.2 把 Task 和 Message 混为同一个对象现象服务端在处理message/send时返回了一条普通消息客户端解析result.task.status报错任务状态显示不出来。原因A2A 中message/send的响应结构是result.taskTask 包含status、artifacts等而 Message 只是 Task 里的候选字段。两者是包含关系不是同一种结构。解决服务端先构造 Task 对象再把 Message 放进去。客户端始终读result.task.status.state不要读result.message。可以给响应结构打印一次 JSON确认层级再写解析逻辑。4.3 MCP 日志打到 stdoutHost 连接失败现象MCP Server 启动后Host 立刻报 “not a valid JSON” 或一直转圈代码逻辑检查没问题但程序里有print调试信息。原因stdio 传输层把 stdout 当作协议通道任何非 JSON-RPC 输出都会破坏帧边界。这是 MCP 最常见的翻车点。解决所有日志和print改到sys.stderr或者用 Python logging 配置StreamHandler指向 stderr。我一般会先写一个最小客户端手动向 stdin 写入initialize观察 stdout 是否只有 JSON。如果 stdout 里出现了普通文本Host 解析必然失败。4.4 MCP inputSchema 与实际参数类型不一致现象tools/list能看到工具但模型一调用就报argument a must be integer, got 1或者干脆把字符串传进了算术逻辑。原因MCP Host 会对入参做 JSON Schema 校验和类型转换。Schema 声明 string 但实现代码用 int 处理或者 schema 少写了required字段导致运行时空数据参与逻辑。解决把inputSchema当作服务契约严格按 JSON Schema 写清楚类型、枚举、默认值。上云环境建议用代码生成 schema不要手写调试时先在tools/call入口打印arguments再走业务逻辑。4.5 长任务没有轮询客户端卡在 working现象服务端返回 Task 状态为working客户端一直等待最终结果最终超时。AgentCard 的streaming是 false也没开pushNotifications。原因A2A 不要求单次响应返回最终结果。当状态是working时客户端应该按协议发起tasks/get轮询或者通过流式/推送接收。解决根据 AgentCard 的capabilities决定模式。如果既不能流式也不能推送客户端在拿到working后用tasks/get轮询间隔建议 12 秒。服务端必须实现tasks/get否则状态就永远是黑匣子。5. 验证与进阶用 curl 和一个小网关同时驾驭两套协议协议落地后我习惯用三条命令做冒烟验证避免一上来就写编排代码。A2A 侧先验证发现和消息curl -s http://localhost:8080/.well-known/agent.json | jq .name curl -s -X POST http://localhost:8080/rpc -H Content-Type: application/json -d - EOF {jsonrpc:2.0,id:1,method:message/send,params:{message:{role:user,taskId:smoke-1,parts:[{text:ping}]}}} EOFMCP 侧因为默认是 stdio不能直接 curl可以模拟 Host 发一条initializeprintf {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:smoke,version:0}}}\n \ | python3 mcp_mini_server.py看到返回serverInfo和capabilities说明传输和握手都正常。这个冒烟流程只需 30 秒比看日志定位快得多。真正把两种协议放在一个系统里时我建议用统一网关隔离外部与内部A2A 负责外部 Agent 间协作MCP 负责内部工具调用。网关收到 A2A 任务后把需要工具的子任务转成 MCP 请求再把结果包装成 A2A 的 Task 返回。核心转换逻辑可以这样组织def handle_a2a(task_id, text): mcp_result call_mcp(read_database, {query: text}) a2a_task { id: task_id, status: {state: completed}, artifacts: [{parts: [{text: mcp_result[content][0][text]}]}] } return a2a_task注意不要透传两套协议的响应结构A2A 的 Task 包含状态MCP 的响应是 content 列表网关负责把 MCP content 聚合后塞进 A2A 的artifacts.parts。我现在的习惯是每接一个 Agent先跑冒烟命令再写编排逻辑通了之后才放心做复杂业务。这个习惯帮我少翻了很多车希望帮到你。本文还有配套的精品资源点击获取 A2A与MCP协议落地实战：从Agent协作到工具调用
简介A2A协议与MCP协议是当前多Agent系统绕不开的两个关键话题这份由蒋俊整理的PPT从协议基础概述讲起逐步对比技术架构、功能特性与应用场景并延伸至性能优化与发展前景适合AI开发者、架构师及需要做方案选型的技术人员快速建立整体认知。资源共1个pptx文件压缩包约2.42MB图文页紧凑可配合正文直接学习目前已有331人浏览学习。内容结合物流智能体、药物模拟实验、工业物联网、医疗影像分析等案例清晰梳理了两者的分工A2A通过“Agent卡”和任务生命周期管理支持智能体自然语言协作MCP则以类似USB-C集线器的统一连接方式帮助模型即插即用调用外部工具理解二者如何互补对构建可扩展的多Agent技术栈很有帮助。1. A2A 协议与 MCP 协议两套协议管的是 AI 协作的哪一段一个常见场景你正在搭多智能体平台主 Agent 要调用另一个 Agent 做文件分析还要让主 Agent 能直接读数据库、发消息。市场上被频繁提到的 A2A 协议和 MCP 协议正好落在两个层次上A2A 管 Agent 与 Agent 之间的发现、任务下发和状态追踪MCP 管模型与工具、数据源之间的上下文连接。两者不是二选一而是上下层关系。这篇文章从协议拆解、最小可运行实现到联调踩坑按一线落地顺序写适合正在做多 Agent 编排或把 AI 接入内部系统的工程师。2. A2A 协议拆开看AgentCard、任务卡片与最小可运行服务A2A 协议把 Agent 协作抽象成三个对象AgentCard 是名片描述这个 Agent 的能力和入口Task 是一次可追踪的协作单元Message 是任务里的一条消息。这套模型的好处是调用方不需要提前知道 Agent 的实现语言或框架只要拿到 AgentCard 就能按统一格式发请求。下面按“发现 → 建任务 → 查状态”的顺序拆。2.1 AgentCard 与任务状态先看懂协议怎么描述能力一个最小 AgentCard 在 HTTP 的/.well-known/agent.json上返回内容大致是这样{ name: invoice-processor, description: 处理并校验发票文件, url: https://agent.example.com/rpc, version: 1.0, capabilities: { streaming: true, pushNotifications: false } }这里有几个容易误读的字段。name和description是给人和编排器看的url才是真正的 RPC 入口后续所有 JSON-RPC 请求都要发到这个地址。capabilities声明的是传输能力不是业务能力——streaming表示能否返回流式结果pushNotifications表示能否在长任务结束后主动通知。A2A 没有把“这个 Agent 能干什么”写进卡片业务能力靠协作时的内容协商或独立的能力服务发现这一点新手经常栽跟头。任务侧的状态机更简单一个 Task 包含id、status.state、artifacts和messages。status.state常见值是submitted、working、input-required、completed、failed、canceled。服务端在message/send的响应里返回 Task客户端根据state决定是继续轮询还是直接取结果。2.2 用 Python 跑通一个最小的 A2A 客户端与服务端我一般不先引 SDK而是用标准库写一个最小服务跑通协议再换框架。下面这个服务只实现了两个关键端点GET /.well-known/agent.json返回名片POST /rpc处理message/send。# server.py -- 一个符合 A2A 核心交互的最小 HTTP 服务 import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer AGENT_CARD { name: mock-agent, description: A minimal A2A agent for demo, url: http://localhost:8080/rpc, capabilities: {streaming: False} } class Handler(BaseHTTPRequestHandler): def _send_json(self, obj, status200): body json.dumps(obj).encode() self.send_response(status) self.send_header(Content-Type, application/json) self.send_header(Content-Length, str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): if self.path /.well-known/agent.json: self._send_json(AGENT_CARD) else: self.send_error(404) def do_POST(self): if self.path ! /rpc: return self.send_error(404) size int(self.headers.get(Content-Length, 0)) req json.loads(self.rfile.read(size)) method req.get(method) if method message/send: # 构造一个已完成的任务 task { id: req[params][message][taskId], status: {state: completed}, artifacts: [{parts: [{text: hello from mock agent}]}] } self._send_json({jsonrpc: 2.0, id: req[id], result: {task: task}}) else: self.send_error(400) if __name__ __main__: ThreadingHTTPServer((0.0.0.0, 8080), Handler).serve_forever()代码里的AGENT_CARD是协议中的“名片”本体capabilities.streaming设为false时message/send会直接返回完整 Task。如果你要模拟长任务需要把state改成working再在后续请求中实现tasks/get轮询。这里taskId直接取了客户端消息里的字段是简化行为真实协议建议由服务端生成。对应客户端也很短# client.py -- 读取 AgentCard 并发起任务 import json import urllib.request def get_agent_card(base_url): with urllib.request.urlopen(base_url /.well-known/agent.json) as resp: return json.load(resp) def send_message(rpc_url, task_id, text): payload { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, taskId: task_id, parts: [{text: text}] } } } req urllib.request.Request( rpc_url, datajson.dumps(payload).encode(), headers{Content-Type: application/json} ) with urllib.request.urlopen(req) as resp: return json.load(resp) if __name__ __main__: base http://localhost:8080 card get_agent_card(base) print(agent:, card[name]) result send_message(card[url], task_idtask-001, textping) print(state:, result[result][task][status][state])客户端的重点是card[url]是 RPC 入口不是名片地址。整个调用链从发现文档开始之后客户端不再需要知道 Agent 内部实现。参数说明task_id在简化版里由客户端传入实际协作中应由任务发起方在内的调用链生成parts是消息内容可以是纯文本也可以是结构化数据。如果打印出的 state 是working就要转入轮询或流式等待。2.3 A2A 的认证与安全边界你需要知道的几个参数A2A 规范本身没有强制认证但实际部署时你至少要考虑三层传输层 TLS、请求头认证、任务级权限。目前主流做法是 OAuth 2.0 Bearer TokenAgentCard 里的security字段可以声明支持的认证方案。我自己的项目起步阶段都用 Bearer Token服务端做一个中间件检查Authorizationheader客户端把 token 放在请求头里。认证方式AgentCard 配置适合场景Bearer Token在 security 字段声明 Bearer 方案内部服务间调用OAuth 2.0声明授权服务器地址面向外部开发者mTLS网关层终止 TLS 并校验证书敏感数据服务另一个常见参数是超时。A2A 的message/send是同步 HTTP 请求但任务本身可能异步完成所以 HTTP 超时和任务超时要分开设。我通常把 HTTP 超时设为 5 秒任务轮询间隔 12 秒如果capabilities.streaming开了则用 SSE 通道接收增量结果。注意不要把 AgentCard 的url写成卡片自身路径那是第 4 章里最经典的联调失败原因。3. MCP 协议落地从 Resources 到 Tools 的本地 Server 搭建MCP 的定位与 A2A 完全不同它解决的是模型 Host 与外部系统之间的“上下文插头”。一个 MCP Server 可以暴露三类能力Prompts、Resources、Tools。模型客户端通过 JSON-RPC 2.0 向 Server 发起调用Server 通过 stdin/stdout 或 HTTP 响应。下面先讲清楚三个原语再给一个不依赖 SDK 的最小实现。3.1 MCP 的三大原语Prompts、Resources、ToolsMCP 把可共享能力分成三类理解它们的区别是选型的前提。我用一张表总结原语作用关键方法能否产生外部副作用Prompts预置可复用的提示模板prompts/list, prompts/get否Resources向模型提供只读上下文resources/list, resources/read否Tools可执行函数模型可调用tools/list, tools/call是生产系统里 Tools 用得最多因为模型需要读数据库、发消息、执行计算。Resources 适合把文档、配置、状态快照提供给模型比如把一份崩溃日志作为只读上下文喂给模型。Prompts 则适合团队统一 prompt 风格避免每个调用的人写出的提示词差异过大。注意这三类原语不是互斥的。同一个 Server 往往同时提供 Resources 和 Tools资源负责喂数据工具负责产生动作。设计时我会先列出模型的真实需求如果只是“看”数据优先 Resources如果需要“改”数据必须走 Tools这样权限边界才清晰。3.2 不依赖 SDK 的最小 MCP Serverstdio 的初始化与 tools/callMCP 在本地开发最常用的是 stdio 模式Host 把 MCP Server 当作子进程启动通过标准输入输出交互。stdout 是协议通道所有日志必须写到 stderr。下面这个代码只实现initialize、tools/list、tools/call但足够验证协议。# mcp_mini_server.py -- 只实现核心方法的最小 MCP 服务 import sys import json def send(msg): # stdio 简化帧每行一个 JSON-RPC 消息 sys.stdout.write(json.dumps(msg) \n) sys.stdout.flush() def handle(req): method req.get(method) if method initialize: return { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: mini-mcp, version: 0.1.0} } if method tools/list: return { tools: [ { name: add, description: Add two integers, inputSchema: { type: object, properties: {a: {type: integer}, b: {type: integer}}, required: [a, b] } } ] } if method tools/call: name req[params][name] args req[params][arguments] if name add: return { content: [{type: text, text: str(args[a] args[b])}] } return {error: {code: -32601, message: method not found}} for line in sys.stdin: req json.loads(line) if req.get(method) notifications/initialized: continue result handle(req) if result and error in result: send({jsonrpc: 2.0, id: req[id], error: result[error]}) else: send({jsonrpc: 2.0, id: req[id], result: result})运行方式是python3 mcp_mini_server.py。Host 会以子进程拉起它并通过管道写入 JSON-RPC 请求。代码里忽略notifications/initialized因为通知不需要响应每个响应都带idHost 用id匹配请求和响应。参数说明capabilities.tools表示这个 Server 支持工具调用protocolVersion使用常见的2024-11-05不同 Host 可能要求不同版本联调时以 Host 日志为准。inputSchema必须是 JSON Schema 而不是 Python 类型标注模型端会按照 Schema 做参数校验这里add工具要求 a、b 都是整数如果模型拿了字符串进来Host 会先报错。3.3 MCP 的传输层选型stdio 与 HTTPSSE 的区别MCP 的 JSON-RPC 消息格式相同但传输层不同直接影响部署和认证方式。本地开发选 stdio远程多租户选 HTTP。维度stdioHTTPSSE进程部署本地子进程远程服务认证依赖父进程权限可加 OAuth适用场景本地脚本、桌面端多客户端共享stdio 的最大优势是无认证配置只要文件系统权限和父进程权限控制好。缺点是 Server 和 Host 必须同机无法做成中心化服务。HTTPSSE 可以把 MCP Server 部署成独立服务但要注意 SSE 是单向流服务端推送消息需要额外通道或者选用后来 MCP 社区更常用的 Streamable HTTP 方式。我自己的经验是先按 stdio 把协议逻辑跑通再根据部署环境换传输层业务处理代码可以做到不用大改。4. A2A 与 MCP 对接踩坑5 个让联调翻车的真实原因两套协议单独跑通不难难在把它们接到一个系统里。下面 5 个坑是我在联调中反复遇到的按协议生命周期排列每条都是“现象 → 原因 → 解决”的结构。4.1 AgentCard 的 url 指向错误任务发不出去现象客户端拿到 AgentCard 后把请求发到了http://host/.well-known/agent.json结果收到 404 或 JSON-RPC 错误任务始终发不出去。原因AgentCard 里的url是协议入口也就是 RPC endpoint而不是发现文档地址。发现文档只负责返回名片不接收 RPC 请求。解决始终用card[url]作为 POST 目标。验证时可以先curl -s http://host/.well-known/agent.json | jq .确认 url 字段指向的是/rpc或者你自定义的端点然后再向该端点发message/send。4.2 把 Task 和 Message 混为同一个对象现象服务端在处理message/send时返回了一条普通消息客户端解析result.task.status报错任务状态显示不出来。原因A2A 中message/send的响应结构是result.taskTask 包含status、artifacts等而 Message 只是 Task 里的候选字段。两者是包含关系不是同一种结构。解决服务端先构造 Task 对象再把 Message 放进去。客户端始终读result.task.status.state不要读result.message。可以给响应结构打印一次 JSON确认层级再写解析逻辑。4.3 MCP 日志打到 stdoutHost 连接失败现象MCP Server 启动后Host 立刻报 “not a valid JSON” 或一直转圈代码逻辑检查没问题但程序里有print调试信息。原因stdio 传输层把 stdout 当作协议通道任何非 JSON-RPC 输出都会破坏帧边界。这是 MCP 最常见的翻车点。解决所有日志和print改到sys.stderr或者用 Python logging 配置StreamHandler指向 stderr。我一般会先写一个最小客户端手动向 stdin 写入initialize观察 stdout 是否只有 JSON。如果 stdout 里出现了普通文本Host 解析必然失败。4.4 MCP inputSchema 与实际参数类型不一致现象tools/list能看到工具但模型一调用就报argument a must be integer, got 1或者干脆把字符串传进了算术逻辑。原因MCP Host 会对入参做 JSON Schema 校验和类型转换。Schema 声明 string 但实现代码用 int 处理或者 schema 少写了required字段导致运行时空数据参与逻辑。解决把inputSchema当作服务契约严格按 JSON Schema 写清楚类型、枚举、默认值。上云环境建议用代码生成 schema不要手写调试时先在tools/call入口打印arguments再走业务逻辑。4.5 长任务没有轮询客户端卡在 working现象服务端返回 Task 状态为working客户端一直等待最终结果最终超时。AgentCard 的streaming是 false也没开pushNotifications。原因A2A 不要求单次响应返回最终结果。当状态是working时客户端应该按协议发起tasks/get轮询或者通过流式/推送接收。解决根据 AgentCard 的capabilities决定模式。如果既不能流式也不能推送客户端在拿到working后用tasks/get轮询间隔建议 12 秒。服务端必须实现tasks/get否则状态就永远是黑匣子。5. 验证与进阶用 curl 和一个小网关同时驾驭两套协议协议落地后我习惯用三条命令做冒烟验证避免一上来就写编排代码。A2A 侧先验证发现和消息curl -s http://localhost:8080/.well-known/agent.json | jq .name curl -s -X POST http://localhost:8080/rpc -H Content-Type: application/json -d - EOF {jsonrpc:2.0,id:1,method:message/send,params:{message:{role:user,taskId:smoke-1,parts:[{text:ping}]}}} EOFMCP 侧因为默认是 stdio不能直接 curl可以模拟 Host 发一条initializeprintf {jsonrpc:2.0,id:1,method:initialize,params:{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:smoke,version:0}}}\n \ | python3 mcp_mini_server.py看到返回serverInfo和capabilities说明传输和握手都正常。这个冒烟流程只需 30 秒比看日志定位快得多。真正把两种协议放在一个系统里时我建议用统一网关隔离外部与内部A2A 负责外部 Agent 间协作MCP 负责内部工具调用。网关收到 A2A 任务后把需要工具的子任务转成 MCP 请求再把结果包装成 A2A 的 Task 返回。核心转换逻辑可以这样组织def handle_a2a(task_id, text): mcp_result call_mcp(read_database, {query: text}) a2a_task { id: task_id, status: {state: completed}, artifacts: [{parts: [{text: mcp_result[content][0][text]}]}] } return a2a_task注意不要透传两套协议的响应结构A2A 的 Task 包含状态MCP 的响应是 content 列表网关负责把 MCP content 聚合后塞进 A2A 的artifacts.parts。我现在的习惯是每接一个 Agent先跑冒烟命令再写编排逻辑通了之后才放心做复杂业务。这个习惯帮我少翻了很多车希望帮到你。本文还有配套的精品资源点击获取