多智能体协作的“通用语“：MCP + A2A 协议栈实战解析-平芜编程栈

多智能体协作的"通用语"：MCP + A2A 协议栈实战解析

![多智能体协作](https://picsum.photos/seed/17860690546182/800/400)

2026 年，大模型竞争的主战场正在从"单模型能力"转向"多智能体协作"。当中国日均词元调用量突破 140 万亿、AI Agent 从演示走向生产环境时，一个关键问题浮出水面：**来自不同厂商、运行在不同框架上的智能体，如何像人类团队一样分工协作？** 答案就是 MCP 与 A2A 组成的"分层协议栈"——MCP 给智能体"动手能力"，A2A 给智能体"社交网络"。

一、为什么需要协议栈？

回顾 2025 到 2026 年的演进脉络：MCP（Model Context Protocol，模型上下文协议）由 Anthropic 于 2024 年底开源，迅速成为智能体连接工具与数据的事实标准；2025 年 4 月，Google 联合 50+ 企业发布 A2A（Agent2Agent）协议，专攻智能体之间的互联互通。到了 2026 年，两者已经形成稳定互补：

• **MCP**：解决"智能体 → 工具/数据"的单向能力扩展问题；

• **A2A**：解决"智能体 ↔ 智能体"的双向协作问题；

• **OpenAPI 认证**：解决身份与授权问题，凭证通过标准 HTTP Header 传递，协议本身与认证机制解耦。

打个比方：MCP 是给每个智能体装上"手"，A2A 是让这些智能体彼此"开口说话"。没有 MCP 的智能体是"残疾人"，没有 A2A 的智能体是"孤岛"。

二、MCP：给智能体"动手能力"

MCP 采用经典的Host / Client / Server三层架构。Host（如 Claude Desktop、Cursor、自研应用）中嵌入 MCP Client，通过 JSON-RPC 2.0 与 MCP Server 通信。Server 暴露三类核心原语：Tools（可执行动作）、Resources（可读取数据）、Prompts（可复用提示词）。

下面是一个极简的 MCP Server 示例（Python，使用官方 SDK）：

from mcp.server.fastmcp import FastMCP mcp = FastMCP("order-service") @mcp.tool() def query_order(order_id: str) -> str: """查询订单状态（模拟工具）""" status_map = {"A1001": "已发货", "A1002": "待支付"} return status_map.get(order_id, "订单不存在") @mcp.tool() def calc_shipping(weight_kg: float, express: str = "standard") -> float: """计算运费：standard 8元/首重，express 15元/首重""" base = 8.0 if express == "standard" else 15.0 extra = max(0.0, weight_kg - 1.0) * 2.0 return round(base + extra, 2) if __name__ == "__main__": mcp.run(transport="stdio") # 支持 stdio / sse / streamable-http

客户端侧，一个 Agent 可以同时挂载多个 MCP Server，像拼积木一样获得"查订单 + 算运费 + 查物流"等能力：

from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def call_tool(server_cmd, tool_name, args): params = StdioServerParameters(command=server_cmd[0], args=server_cmd[1:]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result = await session.call_tool(tool_name, arguments=args) return result.content[0].text # 同一个 Agent 同时使用多个工具 print(await call_tool(["python", "order_server.py"], "query_order", {"order_id": "A1001"})) print(await call_tool(["python", "order_server.py"], "calc_shipping", {"weight_kg": 3.2}))

MCP 生态在 2026 年已经极其繁荣：从 GitHub、Slack 到 Elasticsearch，主流 SaaS 纷纷提供官方 MCP Server；远程 MCP（streamable-http）支持也让"客户端连远程工具"成为常态。这正是"词元经济"落地的基础设施——模型负责"想"，MCP 负责"做"。

三、A2A：给智能体"社交网络"

如果说 MCP 是"人与工具的契约"，A2A 就是"智能体之间的社交协议"。它基于 HTTP/1.1 或 HTTP/2 + JSON-RPC 2.0 + SSE（Server-Sent Events），核心概念只有四个：

1.Agent Card：智能体的"名片"，JSON 格式声明自己是谁、能干什么、认证方式；

2.Task：任务对象，携带状态机（submitted → working → input-required → completed / failed）；

3.Message / Part / Artifact：消息与产物，支持文本、图片、表单等多模态；

4.Streaming / Push Notification：面向长任务的实时进度推送。

先看一张"名片"——Agent Card 长什么样：

{ "name": "reservation-agent", "description": "处理餐厅订座、改签与取消", "url": "https://agents.example.com/reservation", "version": "1.0.0", "capabilities": { "streaming": true, "pushNotifications": true, "stateTransitionHistory": true }, "security": { "schemes": ["bearer"] }, "skills": [ { "id": "book_table", "name": "在线订座", "description": "按人数、时间与偏好预订餐厅桌位" } ], "defaultInputModes": ["text/plain"], "defaultOutputModes": ["text/plain", "application/json"] }

再写一个极简的 A2A Server（FastAPI 实现 `tasks/send` 端点）：

from fastapi import FastAPI, Request from pydantic import BaseModel app = FastAPI() BOOKINGS: dict[str, str] = {} class TaskSendRequest(BaseModel): id: str message: dict metadata: dict = {} @app.get("/.well-known/agent.json") def agent_card(): return { "name": "reservation-agent", "description": "处理订座请求", "url": "https://agents.example.com/reservation", "version": "1.0.0", "capabilities": {"streaming": False}, } @app.post("/tasks/send") async def tasks_send(req: TaskSendRequest): parts = req.message.get("parts", []) text = "".join(p.get("text", "") for p in parts) # 解析意图：book/cancel if "订" in text and "座" in text: task_id = f"task-{req.id}" BOOKINGS[task_id] = "已预订 4 人桌，今晚 19:00" return { "id": task_id, "status": {"state": "completed"}, "artifacts": [{"parts": [{"text": BOOKINGS[task_id]}]}], } return {"id": req.id, "status": {"state": "failed"}, "statusMessage": "无法识别的意图"}

客户端智能体只需把请求 POST 过去，就能把"订座"这件事委托出去——不需要知道对方是用 LangChain、ADK 还是自研框架写的。

四、实战：MCP + A2A 分层协同

在真实系统里，两者是分层配合而非二选一。以一个"智能客服工单系统"为例：

用户提问 │ ▼ ┌─────────────┐ A2A(JSON-RPC+SSE) ┌─────────────────┐ │ 主控 Agent │ ──────────────────► │ 订单 Agent │ │ (orchestrator)│ ◄────────────────── │ (MCP: 查订单/退款) │ └──────┬──────┘ └─────────────────┘ │ MCP (工具调用) ▼ ┌─────────────────┐ │ CRM / 物流 / 知识库 │ └─────────────────┘

主控 Agent 的伪代码清晰展示了这种"脑手分离"：

async def handle(user_msg: str): # 1) 本地判断：是否需要求助其他智能体？ if "订单" in user_msg: # 2) A2A：委托给订单 Agent（远程协作） resp = await a2a_send("https://agents.example.com/order", user_msg) return resp.artifacts[0].parts[0].text # 3) MCP：本地调用工具（查询知识库） docs = await mcp_call_tool("kb-server", "search", {"q": user_msg}) return f"根据知识库：{docs}" # A2A 客户端最小实现 async def a2a_send(agent_url: str, text: str): payload = { "id": uuid4().hex, "message": {"role": "user", "parts": [{"text": text}]}, "metadata": {}, } async with httpx.AsyncClient() as client: r = await client.post(f"{agent_url}/tasks/send", json=payload) return TaskResponse.model_validate(r.json())

Google ADK（Agent Development Kit）已经原生支持"在 A2A Agent 里挂载 MCP 工具"，Elasticsearch 官方也给出了"LLM Agent 新闻室"的完整参考实现——一个记者 Agent 通过 A2A 把稿件交给编辑 Agent，编辑再通过 MCP 调用 Anthropic 分析内容。协议栈的价值在于：每个团队只维护自己的智能体，全局却像一个团队在运转。

五、挑战与展望

协议有了，工程化才刚刚开始，2026 年最受关注的是三个问题：

1.安全与信任：英国 AI 安全研究所 8 月发布的报告显示，测试中 AI 智能体出现了 19 项超出设定范围的自主行为。多智能体放大了攻击面——A2A 必须与 OAuth2/JWT 结合，且对每个请求做鉴权与审计；

2.可观测性：跨智能体的链路追踪（trace）、任务状态机与成本归因，需要一套类似 OpenTelemetry 的标准；

3.协议生态之争：MCP、A2A、AGNTCY（思科主导）仍在争夺生态位，短期内"分层兼容"是最务实的策略——MCP 管工具、A2A 管协作，二者都基于 JSON-RPC/HTTP，迁移成本极低。

可以预见，未来 12 个月我们会看到更多"智能体团队"进入生产：科研场景里检索智能体数秒出结果、模拟智能体跑数天任务并实时推送进度；企业场景里"主控 + 专家"的架构成为标配。就像 TCP/IP 定义了互联网设备如何互联，MCP + A2A 正在定义 AI 智能体如何协作——这是词元经济时代最重要的基础设施。