一文搞懂 MCP 与 A2A:AI 多智能体通信的两种协议,附 Python 双协议实战

📖 摘要:当单个大模型不够用时,我们开始把任务拆给多个 Agent 协作完成。但 Agent 之间、Agent 与工具之间"怎么说话"却成了新痛点。本文用一线的工程视角,讲清 MCP(模型连工具)与 A2A(Agent 连 Agent)两种主流协议的设计哲学与差异,并给出可直接跑的 Python 双协议最小实现,帮你在半小时内搭出"工具 + 多智能体"的混合编排骨架。

🏷️ 关键词:MCP,A2A,AI Agent,多智能体,协议对比

目录

一、背景:为什么需要 Agent 通信协议

早期我们把 LLM 当"超级函数":传 prompt,拿回答。但当业务变复杂,单个模型开始力不从心——它不会查实时天气、不会读写你的数据库、也不会把"写报告"拆成"检索→分析→排版"三步并交给不同角色。

于是出现了两条协作需求:

  1. 模型要连工具:让 LLM 能调用外部 API、查库、执行命令。这是"模型能力的延伸"。
  2. Agent 要连 Agent:让多个各有所长的智能体互相委派任务、交换结果。这是"智能体的组织化"。

这两条需求看似相近,但抽象层级不同。前者关注"模型怎么用一件工具",后者关注"一个团队怎么分工"。这恰好对应了当下最热的两个协议:MCPA2A

💡 一句话区分:MCP 解决"模型用什么",A2A 解决"谁来做、怎么做"。

二、MCP 与 A2A 到底是什么

2.1 MCP:给模型接工具的"USB-C"

MCP(Model Context Protocol)由 Anthropic 提出,核心思想是用一套标准协议,统一描述"工具长什么样、怎么调用、返回什么"。就像 USB-C 让不同设备共用一个接口,MCP 让不同模型共用同一批工具,而不必为每个模型重写一遍对接代码。

它的三个角色:

  • Host:运行模型的应用(如 IDE、聊天客户端)。
  • Client:Host 内嵌的 MCP 客户端,负责握协议。
  • Server:提供工具/资源的进程,暴露 tools/listtools/call 等接口。

通信默认用 JSON-RPC 2.0,传输层早期以 stdio 为主,现已支持 HTTP(Streamable HTTP / SSE)。

2.2 A2A:Agent 之间的"HTTP 协作层"

A2A(Agent2Agent)由 Google 牵头,定位在更高一层:它假设网络里已经存在多个独立的 Agent,每个 Agent 是黑盒,只暴露自己"能做什么"。A2A 让它们互相发现、委派、追踪任务状态。

它的关键概念:

  • Agent Card:放在 /.well-known/agent.json 的能力名片,声明本 Agent 的技能、端点、鉴权方式。相当于"我的简历"。
  • Task:一次协作的最小单位,有 submitted → working → completed/failed 等状态。
  • Message / Artifact:交互的消息与产出的制品(文本、文件等)。

A2A 天然面向"跨厂商、跨语言、互不信任"的多智能体网络,而 MCP 更偏向"同一应用内、模型与本地工具"的紧耦合集成。

三、核心差异对比

3.1 设计哲学

  • MCP:以"模型为中心",目标是把世界变成模型可调用的工具。强调标准化接入、上下文注入。
  • A2A:以"Agent 为中心",目标是让智能体像微服务一样互相组网。强调能力发现、异步任务、身份与治理。

3.2 通信模型

  • MCP:同步请求-响应为主(call 一个工具拿一个结果),偏"函数调用"。
  • A2A:任务导向,支持长任务、流式状态更新、推送通知,偏"工单流转"。

3.3 选型决策表

维度 MCP A2A
抽象层级 模型 ↔ 工具 Agent ↔ Agent
典型场景 给 LLM 接搜索/查库/执行命令 多角色分工、跨团队协作
发现机制 静态配置 Server 地址 Agent Card 动态发现
任务模型 单次 RPC 调用 有状态的 Task 生命周期
同步/异步 多为同步 原生支持异步、流式
治理重点 工具权限、输入输出 schema 身份、委派边界、审计
互补性 可与 A2A 共存:Agent 内部用 MCP 调工具 A2A 的 Agent 内部仍可跑 MCP

⚠️ 注意:二者不是二选一,而是互补。一个 A2A Agent 的内部,完全可以用 MCP 去连自己的工具。真实生产里常见组合是"MCP 做手脚,A2A 做团队"。

四、实战:用 Python 同时实现两套协议

下面用 FastAPI 各写一个最小可运行的骨架(教学用,省略鉴权与错误全量处理)。两个服务独立部署,再由一个 Orchestrator 统一调度,直观看到"MCP 调工具 + A2A 派任务"如何共存。

4.1 环境准备

# 新建虚拟环境(示例,环境路径自行调整)
python -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn httpx

假设我们用 127.0.0.1 本地演示,MCP 服务跑 8001,A2A Agent 跑 8002,Orchestrator 跑 8000。

4.2 实现 MCP 风格的工具服务

核心是用 JSON-RPC 暴露 initializetools/listtools/call 三个方法,让模型"列工具、调工具"。

# mcp_server.py —— 极简 MCP 风格工具服务(教学骨架,非官方 SDK)
from fastapi import FastAPI, Request
import uvicorn

app = FastAPI()

# 工具清单:真实 MCP 通过 tools/list 动态返回,这里用内存字典简化
TOOLS = {
    "get_weather": {"description": "查询某城市天气", "input_schema": {"city": "string"}},
    "calc": {"description": "简单四则运算", "input_schema": {"expr": "string"}},
}

def dispatch(name: str, args: dict) -> str:
    # 仅演示;生产环境严禁直接用 eval,应走安全表达式引擎
    if name == "get_weather":
        return f"{args['city']} 今天晴,26℃"
    if name == "calc":
        return str(eval(args["expr"], {"__builtins__": {}}))
    return "unknown tool"

@app.post("/mcp")
async def handle(request: Request):
    req = await request.json()
    method, req_id = req.get("method"), req.get("id")

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": req_id, "result": {"protocolVersion": "2025-06-18"}}
    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOLS}}
    if method == "tools/call":
        name = req["params"]["name"]
        result = dispatch(name, req["params"]["arguments"])
        return {"jsonrpc": "2.0", "id": req_id,
                "result": {"content": [{"type": "text", "text": result}]}}
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "method not found"}}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

4.3 实现 A2A 风格的 Agent

关键两步:暴露 /.well-known/agent.json 作为能力名片,以及 /tasks/send 接收并完成任务。

# a2a_agent.py —— 极简 A2A 风格 Agent(教学骨架)
from fastapi import FastAPI, Request
import uvicorn, uuid

app = FastAPI()

# Agent Card:声明"我是谁、能做什么",供 Orchestrator 动态发现
AGENT_CARD = {
    "name": "research-agent",
    "description": "负责检索资料并产出摘要",
    "url": "http://127.0.0.1:8002/",
    "skills": [{"id": "search", "description": "检索给定关键词并总结"}],
}

@app.get("/.well-known/agent.json")
def agent_card():
    return AGENT_CARD

@app.post("/tasks/send")
async def send_task(request: Request):
    body = await request.json()
    task_id = body.get("id") or str(uuid.uuid4())
    text = body["message"]["parts"][0]["text"]
    # 真实场景应丢进后台异步执行并回报状态;这里同步返回以示意
    summary = f"已检索:{text} -> 提炼出 3 条关键结论(示例数据,仅作演示)"
    return {
        "id": task_id,
        "status": {"state": "completed"},
        "artifacts": [{"parts": [{"text": summary}]}],
    }

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

4.4 在 Orchestrator 中统一调度

Orchestrator 自己是个 Agent:先用 MCP 调工具拿数据,再用 A2A 把数据交给研究 Agent 出报告。

# orchestrator.py —— 同时调度 MCP 工具与 A2A Agent
import httpx

def call_mcp(tool: str, args: dict) -> str:
    r = httpx.post("http://127.0.0.1:8001/mcp", json={
        "jsonrpc": "2.0", "id": 1, "method": "tools/call",
        "params": {"name": tool, "arguments": args},
    })
    return r.json()["result"]["content"][0]["text"]

def call_a2a(text: str) -> str:
    r = httpx.post("http://127.0.0.1:8002/tasks/send", json={
        "id": "t1", "message": {"parts": [{"text": text}]},
    })
    return r.json()["artifacts"][0]["parts"][0]["text"]

if __name__ == "__main__":
    weather = call_mcp("get_weather", {"city": "深圳"})   # 走 MCP:模型调工具
    report = call_a2a(f"基于天气【{weather}】给出出行建议")  # 走 A2A:Agent 派 Agent
    print(report)

跑起来后,orchestrator.py 先通过 MCP 拿到天气,再把天气作为任务委托给 A2A 研究 Agent——一次调用里同时走通了两条协议,这正是混合编排的雏形。

五、生产落地踩坑与优化

5.1 认证与安全

  • MCP Server 别裸奔在内网外:工具能执行命令、查库,必须加 Bearer Token 或 mTLS,并做工具级白名单。
  • A2A 的 Agent Card 要带鉴权声明:在 Card 里明确 authentication 方案,Orchestrator 据此选择凭据,避免越权委派。

5.2 长任务与状态机

  • MCP 的 tools/call 默认同步,超过几十秒的任务要改用 Streamable HTTP / SSE 流式返回,或改为"提交任务拿 ticket,轮询进度"。
  • A2A 的 Task 本就是异步友好的:生产里把 working 状态通过 tasks/sendSubscribe 推流给调用方,别用忙等轮询。

5.3 可观测性

  • 给每个调用注入 trace_id,MCP 与 A2A 共用一套链路追踪(OpenTelemetry),否则"Agent 调了谁、谁又调了工具"会成一团黑盒。
  • 记录工具/Agent 的入参出参(脱敏后)做复盘,是定位"幻觉链路"的关键。

六、总结

MCP 与 A2A 不是竞争关系,而是分层互补

  • 想给模型"长手脚"——接搜索、查库、执行命令,用 MCP
  • 想让多个智能体"组队干活"——发现能力、委派任务、跟踪状态,用 A2A
  • 真实系统里,A2A 的 Agent 内部往往再用 MCP 连自己的工具,形成"团队 + 手脚"的完整架构。

本文的 Python 骨架可直接跑通双协议最小闭环,建议你在本地按 4.4 节串起来,再逐步补上鉴权、异步任务与链路追踪,就能平滑演进成一个像样的多智能体生产系统。

觉得有用就点个赞 / 收藏,评论区聊聊你更看好 MCP 还是 A2A,或者你的 Agent 是怎么组网的。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐