ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

MCP与A2A协议实战:大模型工具调用与智能体协作的边界与组合

MCP与A2A协议实战:大模型工具调用与智能体协作的边界与组合 最近在做一个多智能体协作的小项目核心需求就两条让大模型能调用外部工具让两个智能体之间能互相传递任务。起初我以为这只是简单调接口的事结果真正动手才发现工程上已经围绕这两条需求各自长出了一套协议——MCPModel Context Protocol模型上下文协议和 A2AAgent-to-Agent智能体间通信协议。如果只埋头写代码不把这两个协议的关系弄清楚后面大概率会把工具调用的职责和智能体间通信的职责混在一起项目越写越乱。这篇实践指南不打算从理论文档开始抄概念而是直接回答三个问题MCP 到底解决什么问题A2A 又解决什么问题一个实际项目里该分别怎么落地、怎么组合、怎么排查故障。内容面向两类读者一是刚接触 AI Agent 开发、想把手上的模型接到工具上的新手二是已经在用 LangChain 或各类 Agent 框架、想搞明白底层协议标准的开发者。后续所有代码和配置都是我实际验证过的精简版本可以当成脚手架直接抄。1. 内容整体设计与思路拆解两个协议各管哪一段1.1 MCP 是做什么的给大模型接上工具和数据源MCP 本质上是一个“把外部能力变成标准接口”的协议。你可以把它理解为给大模型装了一排 USB 接口每个接口对应一个工具、一份文档或一类数据库资源。以前要让 GPT、Claude 这类模型去调用某个函数你得专门写一段函数调用逻辑模型厂商不同调用方式也不同换一个模型就要重写一层适配代码。有了 MCP 之后模型应用变成统一的 Client工具提供方变成 ServerClient 通过协议去发现 Server 上有哪些工具、有哪些资源再把大模型的意图转换成具体的工具调用请求。这个思路和浏览器很像。你不需要知道键盘鼠标的 USB 协议具体怎么实现插上就能用MCP 想做的就是把 AI 生态里的“外设”也统一成一套插拔标准。Server 端只需要实现工具列表、资源列表和提示词列表的暴露方式Client 端负责和模型交互并在合适的时机调用这些工具。这样一来同一个 MCP Server 可以被 Claude Desktop、Cursor、自研应用等各种客户端复用工具开发者写一次就能到处用。1.2 A2A 是做什么的让智能体之间直接对话A2A 协议解决的是另一个问题当系统里有多个智能体时彼此之间怎么发现对方、怎么传递任务、怎么拿回结果。它由 Google 在 2025 年推动设计目标是让不同团队、不同框架写的 Agent 能够跨平台协作。这里有个容易混淆的点很多人以为 A2A 是把两个大模型直接连起来其实不是。A2A 规定的是智能体“外壳”之间的通信方式。你可以把每个智能体看成一家独立的公司公司内部用什么系统管理不重要重要的是对外要有一份公开的业务能力介绍AgentCard要能接收对方的订单Task要能回复订单状态和交付物。A2A 管的是这些公司之间的商务往来规则不是公司内部的业务流程。1.3 为什么实践时要把两者放在一起看在实际项目里MCP 和 A2A 经常被当作二选一的技术方案这是一个典型的理解偏差。MCP 主要解决“模型到工具”的纵向连接A2A 解决“智能体到智能体”的横向连接。一个完整的多智能体系统通常是两层都有的智能体 A 接到用户任务后可能需要通过 MCP 调用天气查询、数据库查询等工具做准备工作再通过 A2A 把准备好的子任务分发给智能体 B。我见过不少团队一开始只用 MCP 硬扛在 Server 里写死远程调用逻辑最后把工具协议改成了私有接口导致系统扩展性很差。正确的方式是先明确边界——工具调用走 MCP智能体协作走 A2A两者各司其职中间用业务代码串起来。2. 核心细节解析与实操要点2.1 MCP 协议的核心概念Server、Client、Tool、Resource、Prompt上手 MCP 之前我建议先记住四个概念之间的关系而不是去背整个规范。概念作用生活化类比Client连接模型和 Server 的中间层持有协议会话你手机上的 AppServer暴露工具、资源、提示词的进程某个硬件设备的驱动服务Tool可被大模型调用的函数有输入输出 schemaApp 里的一个按钮Resource可读的数据内容比如文件、数据库记录App 里的一个文件浏览器Prompt预定义的提示词模板App 里的“快捷回复模板”我在本地开发时习惯把 MCP Server 启动成 stdio 模式也就是通过标准输入输出和 Client 通信调试起来非常方便不需要开端口。生产环境再切换成 SSE 或 HTTP 模式让多个客户端通过 TCP 访问同一个 Server。写 MCP Server 时最需要注意的是工具参数的 schema 一定要写清楚。大模型是根据 schema 来决定传什么参数的如果你只写city: str模型可能传“北京”也可能传“北京市朝阳区”。我是因为偷懒吃过亏的后来每个参数都会在描述里写清楚格式要求比如“城市名如 北京不要带省市后缀”调用成功率立刻提升。2.2 A2A 协议的核心概念AgentCard、Task、Message、ArtifactA2A 并不是一个全新的 RPC 协议它建立在 JSON-RPC 2.0 之上传输层走 HTTP。这样设计的好处是生态复杂度低任何能发 HTTP 请求的客户端都能接入。A2A 的核心对象有四个AgentCard每个智能体对外暴露的“名片”通常放在/.well-known/agent-card.json路径。卡片里写着智能体名称、支持的能力、通信端点、认证方式、所支持的 MCP 工具列表等。其他智能体通过读取名片来决定要不要把任务交给它。Task一次协作的最小工作单元。发送方创建 Task接收方返回 Task 的状态可能是pending、working、completed、failed等。Message任务里的具体内容。一条消息可以是纯文本也可以是文件。消息由role区分是谁说的user代表任务发起方agent代表执行方。Artifact智能体执行任务后产生的可交付物比如生成的图片、代码、报表文件。Artifact 通常挂载在 Message 上接收方可以通过 URL 或内嵌数据获取。我最初写 A2A 实现时踩过一个理解上的坑以为 Task 和 Message 是一回事。其实 Task 是会话的骨架Message 是骨架上流动的内容。同一个 Task 可以有多轮 Message比如发起方说“请分析这份数据”执行方回复“数据缺少一列”发起方又补充“用上一份文件补上”这些都在同一个 Task 下推进。如果你把每轮一问一答都新建一个 TaskId 链路就断了后续拿结果和状态追踪都会非常痛苦。2.3 两个协议的边界与取舍按我的经验判断一个功能该放进 MCP 还是 A2A有一个很简单的标准看调用方是不是“需要大模型意图理解”的客户端。如果调用方是模型本身或者一个面向模型的工具调度层就放 MCP如果调用方是另一个独立的智能体系统需要的是任务级协作就放 A2A。有些项目会把“查询另一个 Agent 的数据”直接做成一个 MCP Tool。比如 Agent A 的 Server 里加一个call_agent_b()工具底层封装 HTTP 请求。短期看没毛病长期看会有一个问题MCP 的工具生命周期是一个请求一个响应很难表达多轮任务协商、进度推送、任务取消这类协作语义。A2A 则专门定义了task/cancel、task/pushNotification等语义。所以我的建议是临时互通可以用 MCP Tool 顶着正式的智能体协作还是要让 A2A 进入架构否则你会发现自己不断在工具协议里补协作文法最后写出一套没人懂的四不像协议。3. 实操过程与核心环节实现3.1 环境准备与依赖安装这次实践我用 Python 3.11相关库如下mcp官方 Python SDK用于搭建 MCP Server 和 Client。fastapiuvicorn用来托管 A2A 的 HTTP 端点。httpx用于 A2A 客户端发送请求。pydantic用来做 A2A 消息模型校验。安装命令很简单pip install mcp fastapi uvicorn httpx pydantic如果你是第一次搭 MCP我建议先不要上任何框架直接用官方 SDK 写个最小 Server确认协议链路通了你再往里面加业务这样排查问题范围小很多。3.2 用 Python 写一个最小 MCP Server这里我用 FastMCP 这个高阶封装它比手工实现协议类型节省很多代码。创建一个weather_server.pyfrom mcp.server.fastmcp import FastMCP mcp FastMCP(weather-demo) mcp.tool() def get_weather(city: str) - str: 获取指定城市的天气情况。城市名请使用中文名称例如北京。 # 这里替换成你的真实天气服务调用 return f{city} 当前天气晴气温 26 摄氏度空气质量良。 if __name__ __main__: mcp.run(transportstdio)启动一下python weather_server.py这个进程不会退出它会等待标准输入内容。如果直接手动跑看起来像卡住了其实是正常的。再加上一个可以读取本地文件的 Resourcemcp.resource(file:///tmp/notes.txt) def read_notes() - str: with open(/tmp/notes.txt, r, encodingutf-8) as f: return f.read()MCP Client 发现资源之后模型可以把资源内容当作上下文的一部分。我在实际项目里常用这个能力给模型喂配置文档它比自己拼 prompt 要干净得多。3.3 用官方 SDK 写一个 MCP Client 测试连接Server 跑起来之后另一个终端里运行客户端import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[weather_server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(发现工具:, [tool.name for tool in tools.tools]) result await session.call_tool(get_weather, {city: 上海}) print(调用结果:, result.content) if __name__ __main__: asyncio.run(main())运行后你应该能看到控制台打印出工具列表和调用结果。这一步的意义不仅仅是“能跑”它验证了 MCP Client 端的发现机制和调用机制都通了。记住如果你在某个 Chat 产品里配了自定义 MCP但模型不调用工具原因 80% 出在这个环节——工具列表压根没暴露出来。3.4 实现一个 A2A Agent 并完成通信A2A 的传输层比较直接。我写一个简单的 Agent B它通过 FastAPI 暴露两个端点一个用于返回 AgentCard一个用于接收 Task。先定义 AgentCard 响应。按照协议卡片要通过GET /.well-known/agent-card.json暴露from fastapi import FastAPI from pydantic import BaseModel app FastAPI() app.get(/.well-known/agent-card.json) def agent_card(): return { name: report-agent, description: 生成项目周报的智能体, url: http://localhost:8000, version: 1.0.0, capabilities: { message: { inputModes: [text], outputModes: [text] } }, skills: [ {id: generate_report, name: 生成周报} ] }然后处理任务请求。实际 A2A 使用 JSON-RPC 2.0 格式这里做一个简化的接收和响应class TaskRequest(BaseModel): jsonrpc: str 2.0 id: str method: str params: dict {} app.post(/messages) def handle_message(req: TaskRequest): if req.method message/send: message_text req.params.get(message, {}).get(parts, [{}])[0].get(text, ) task_id req.params.get(task, {}).get(id, unknown) if 生成周报 in message_text: # 实际逻辑可以是调用大模型生成内容 return { jsonrpc: 2.0, id: req.id, result: { taskId: task_id, status: completed, messages: [ { role: agent, parts: [{text: 周报已生成本周完成协议实践下周计划做性能测试。}] } ] } } return { jsonrpc: 2.0, id: req.id, error: {code: -32000, message: 不支持的任务类型} }客户端调用时用 httpx 发起 POSTimport httpx card httpx.get(http://localhost:8000/.well-known/agent-card.json).json() print(发现 Agent:, card[name]) payload { jsonrpc: 2.0, id: 123, method: message/send, params: { message: { role: user, parts: [{text: 请生成周报}] } } } resp httpx.post( http://localhost:8000/messages, jsonpayload, headers{Content-Type: application/json} ).json() print(resp[result][status]) print(resp[result][messages][0][parts][0][text])这个简化实现虽然砍掉了一部分协议细节但完整体现了 A2A 的交互模式先通过 AgentCard 发现再通过 Message/Task 交互。真实项目里你需要补上任务状态查询端点task/get、取消端点task/cancel以及异步处理逻辑。A2A 规范文档里有完整的 JSON-RPC 方法列表按需实现即可。3.5 把 MCP 和 A2A 组合在一个小型系统里只看单个协议是不过瘾的下面我画一个实际项目里的组装方式不用文字描述用代码结构说明层次User Request | v Coordinator Agent | \ | (A2A) | (MCP) v v Report Agent Weather Tool Server | | (A2A) v Search Agent在这个结构里Coordinator Agent 负责接收用户输入是唯一直接面对用户入口的智能体。Weather Tool Server 是个 MCP Server服务天气查询、地点查询等工具。Report Agent 和 Search Agent 是独立部署的 A2A 智能体各自也可能通过 MCP 调用自己的内部工具。组合的关键在 Coordinator Agent 的逻辑里它先用 MCP Client 连接 Weather Tool Server把用户请求拆解成工具调用如果发现任务需要另一个智能体处理就通过 A2A Client 把子任务发给对应 Agent并等待回包。这两套协议栈在代码上互不依赖只在业务编排层交汇。我用的是同一个进程里同时创建ClientSession和httpx.Client分别维护两套连接。不要试图把 MCP 的 Session 塞进 A2A 的消息体里协议边界一旦被打破后续升级和排查都会很难受。4. 常见问题与排查技巧实录4.1 MCP 连接失败、工具没加载症状一客户端初始化时报MCPError或者直接超时。排查顺序我固定为三步先确认 Server 进程有没有正常启动。在 bash 里直接运行python weather_server.py然后随便输入一行字符看看有没有报错输出。如果进程秒退大概率是代码里某个装饰器绑定的函数有问题先在本端用普通函数调用方式测一遍。再确认传输方式。stdio 模式要求 Client 和 Server 在同一台机器的同一用户下并且启动命令里的路径真实存在。我遇到过在 Windows 里用commandpython找不到解释器的情况解决办法是写绝对路径或者用.venv/Scripts/python.exe。最后用官方 SDK 的list_tools做连通性测试。这一步能区分“协议没通”和“工具列表为空”两种错误。工具列表为空多半是工具函数没有被装饰器绑定或者工具函数写在了if __name__下面忘记了装饰。症状二模型知道某个工具存在但调用时总是传错参数。这种情况下先看工具 schema。在 Client 端打印session.list_tools()检查工具的inputSchema描述是否足够明确。我习惯把每个参数的示例值直接写在描述里比如city字段写“示例北京不要带省市后缀”。别嫌啰嗦大模型对格式的敏感度超过你的想象。4.2 A2A 握手超时、任务状态卡住A2A 在实践中最容易出问题的是两端协议版本不一致。如果你本地用的是新版规范而对端 Agent 实现的是旧版字段名可能完全不同。我遇到过messageId变短、parts改为content这种不兼容变更后来统一做法是在 AgentCard 里声明协议版本protocolVersion发送方在每次请求时先读卡片再基于版本组装 payload。手写 A2A 客户端时千万别写死字段一定要动态读卡片。任务状态卡住也是一个常见问题多半发生在异步执行场景。发起方通过message/send发送任务后执行方返回working状态但发起方没有继续轮询task/get导致任务永远停在中间态。排查思路是查看执行方的日志看它是不是真的启动了一个后台任务还是异常退出后状态没更新。如果是后台线程崩溃前端的working永远不会变成completed这类问题日志里最容易看出来。4.3 协议选型时的几个坑第一个坑用 MCP 做多智能体通信。我见过有人把另一个 Agent 的地址封装成一个 MCP Tool理由是“省事”。这种方案在演示阶段没问题但 MCP Tool 本身是无状态的无法表达任务的取消和进度推送。一旦两个 Agent 之间需要长时间协作或者执行中途需要用户确认这个方案就会变得特别难维护。所以我的原则是能用 MCP 解决的问题不用 MCP 解决 Agent 协作问题优先把 Agent 协作放在 A2A 层。第二个坑在 A2A 消息里夹带私有二进制数据。协议传输层是 JSON二进制数据哪怕 base64 之后体积也很可观。建议把文件数据作为 Artifact 放在可下载的 URL 里消息里只传引用 URL。这样不仅减小了实时消息体积还能配合对象存储做权限控制。第三个坑忽视认证授权。本地 demo 可以裸奔但如果有两个 Agent 跑在不同网络区域一定要在 A2A 端点上加认证。A2A 规范里 AgentCard 支持声明认证方式但很多开源 demo 根本不实现。我的建议是至少加一个 Bearer Token用环境变量下发不要在代码里写死。MCP 这边则要关注工具权限不是所有工具都应该被任意客户端调用尤其是写操作类的工具最好在 Server 端做一次白名单判断。最后说点个人体会。MCP 和 A2A 这两个协议解决的是不同层的问题但我明显感到在实际工程里真正难的不是协议本身而是边界划分。手写一个 MCP Server 只要半小时难的是想清楚哪些能力应该暴露成工具、哪些职责应该拆成独立 AgentA2A 的通信样例也很短难的是设计任务粒度、状态流转和失败重试策略。我做完这个实践项目后的一个感受是AI Agent 系统正在从单体走向分布式而这种分布式不是靠框架的私有能力而是靠像 MCP、A2A 这样的开放协议慢慢形成的。如果你现在还没动手写过一个 MCP Server我建议你花一个晚上照着上面的最小示例跑一遍如果你已经在用 MCP那么下一步试着给系统加一个 A2A 协作链路那种“两个独立智能体通过标准协议对话”的感觉和自己在代码里硬拼私有接口是完全不同的体验。
返回列表