两个 Agent 之间怎么通信，你选哪条路？

前段时间我在折腾 MCP 协议做工具集成，效果确实不错。但还没高兴两天，Google 的 A2A 协议就杀出来了——而且来势汹汹，社区里两边支持者吵得不可开交。有人说 MCP 代表了未来，有人说 A2A 才是正道。

我干脆花了两天时间，把两个协议都撸了一遍，从原理到代码到实际跑通，做个彻底的对比。不吹不黑，结论可能跟你想象的不一样。

为什么突然冒出两个协议？

事情要从 2025 年说起。

Anthropic 推出了 MCP（Model Context Protocol），目标是给 LLM 一个统一的方式来调用外部工具、访问数据源。简单说就是：Agent <-> 工具 的标准化接口。这个想法很直接——既然每个模型都要重新写一套工具调用代码，为什么不统一成一个标准接口？

然后 Google 在 2025 年 4 月发布了自己的 A2A（Agent-to-Agent），目标完全不同：Agent <-> 另一个 Agent 的协作协议。Google 看到的痛点是：未来会有成千上万个 Agent，它们之间怎么发现彼此、怎么协商任务、怎么交接结果——这才是真正的难题。

所以第一件事就得理清楚——这俩不是竞品，是互补品。MCP 解决的是"Agent 怎么用工具"，A2A 解决的是"Agent 怎么找队友"。但实际落地的时候，边界并不是那么清晰。比如：如果 Agent_B 通过 A2A 调用 Agent_A，而 Agent_A 内部用 MCP 调了工具——那中间的信息流怎么追踪？错误怎么传播？这些都是生产环境必须面对的问题。

MCP：Agent 的工具箱标准

MCP 干了什么

没有 MCP 之前，想让 LLM 调用外部工具，你得给每个工具写适配器——一个 API 一套写法，一个数据库一套逻辑。MCP 统一了这件事：

Agent → MCP Client → MCP Server → 外部工具/数据源

MCP Server 暴露出三个核心能力：

• Resources：只读数据（文件、DB 记录、API 响应）
• Tools：可执行的函数/API（Agent 通过 LLM 的 function calling 调用）
• Prompts：预定义的提示模板（类似可复用的 prompt 片段）

我最喜欢的一点是：MCP 的 Transport 层支持 stdio 和 SSE 两种模式。stdio 适合本地工具，SSE 适合远程服务，同一个 Server 代码可以无缝切换。

写一个完整的 MCP Server

这是我实际写的一个日志分析 MCP Server，用来让 AI 助手直接查生产日志：

import asyncio, glob, json, os
from datetime import datetime, timedelta
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource

server = Server("log-analyzer")

@server.list_resources()
async def list_resources():
    """暴露日志文件列表作为只读资源"""
    resources = []
    for f in glob.glob("/var/log/*.log"):
        stat = os.stat(f)
        resources.append(Resource(
            uri=f"file://{f}",
            name=os.path.basename(f),
            description=f"日志文件, 大小 {stat.st_size} 字节, 修改于 {datetime.fromtimestamp(stat.st_mtime)}",
            mimeType="text/plain"
        ))
    return resources

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_logs",
            description="按关键词和时间范围搜索日志",
            inputSchema={
                "type": "object",
                "properties": {
                    "keyword": {"type": "string", "description": "搜索关键词（支持正则）"},
                    "hours_back": {"type": "integer", "description": "往回搜索的小时数", "default": 24},
                    "max_results": {"type": "integer", "default": 20}
                },
                "required": ["keyword"]
            }
        ),
        Tool(
            name="count_errors_by_hour",
            description="统计过去 N 小时每小时 ERROR 级别的日志数量",
            inputSchema={
                "type": "object",
                "properties": {
                    "hours": {"type": "integer", "default": 24}
                }
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_logs":
        keyword = arguments["keyword"]
        hours = arguments.get("hours_back", 24)
        max_results = arguments.get("max_results", 20)
        
        # 计算时间范围
        cutoff = datetime.now() - timedelta(hours=hours)
        results = []
        
        for f in glob.glob("/var/log/*.log"):
            try:
                with open(f, errors="ignore") as fh:
                    for line in fh:
                        if keyword in line:
                            results.append(f"{os.path.basename(f)}: {line.strip()}")
                            if len(results) >= max_results:
                                break
            except PermissionError:
                continue
            if len(results) >= max_results:
                break
        
        summary = f"搜索关键词 '{keyword}'（最近 {hours} 小时），找到 {len(results)} 条匹配\n"
        return [TextContent(type="text", text=summary + "\n".join(results))]
    
    elif name == "count_errors_by_hour":
        hours = arguments.get("hours", 24)
        cutoff = datetime.now() - timedelta(hours=hours)
        hourly_counts = {}
        
        for f in glob.glob("/var/log/*.log"):
            try:
                with open(f, errors="ignore") as fh:
                    for line in fh:
                        if "ERROR" in line or "error" in line:
                            # 简单匹配时间戳格式 "2026-05-12 20:"
                            for h in range(hours):
                                ts = (datetime.now() - timedelta(hours=h)).strftime("%H")
                                if f" {ts}:" in line or f":{ts}:" in line:
                                    hourly_counts[h] = hourly_counts.get(h, 0) + 1
            except PermissionError:
                continue
        
        result = json.dumps({
            "time_range": f"过去 {hours} 小时",
            "errors_by_hour_ago": hourly_counts
        }, ensure_ascii=False, indent=2)
        return [TextContent(type="text", text=result)]
    
    raise ValueError(f"未知工具: {name}")

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write)

if __name__ == "__main__":
    asyncio.run(main())

MCP 的核心理念是垂直集成——Agent 直接访问工具，中间没有中间人转手。响应快、延迟低，适合单个 Agent 需要大量工具调用的场景。

客户端调用

# mcp_client.py - 任意 AI 应用连接 MCP 工具
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client

async def main():
    async with stdio_client(["python", "mcp_log_server.py"]) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 1. 列出资源
            resources = await session.list_resources()
            print(f"可读日志文件: {len(resources)} 个")
            
            # 2. 列出工具
            tools = await session.list_tools()
            print(f"可用工具: {[t.name for t in tools.tools]}")
            
            # 3. 搜索错误日志
            result = await session.call_tool(
                "search_logs",
                {"keyword": "ERROR", "hours_back": 48, "max_results": 5}
            )
            print(f"\n搜索结果:\n{result.content[0].text}")
            
            # 4. 统计错误分布
            stats = await session.call_tool(
                "count_errors_by_hour",
                {"hours": 12}
            )
            print(f"\n错误统计:\n{stats.content[0].text}")

asyncio.run(main())

跑起来之后，Agent 就能通过 MCP 协议直接操作日志文件了。最爽的地方是：你不需要给 Agent 写任何 API 调用代码，所有工具调用通过 MCP 协议自动协商完成。

A2A：Agent 的社交网络

A2A 的思路完全不同

MCP 解决的问题比较直接，A2A 则是一个更大的野心。

想象一下你有一个研究助手 Agent、一个写代码 Agent、一个代码审查 Agent。它们怎么协作？用 MCP 没法直接解决，因为 MCP 是 Agent 对工具的接口，不是 Agent 对 Agent 的。

A2A 的方案是建立一个水平协作层，核心思想是：每个 Agent 都发布自己的能力名片（Agent Card），其他 Agent 发现这些名片后就可以互相调用。

Agent_A → A2A Client → [A2A over HTTP] → A2A Server → Agent_B
                    ↓
              Agent Card 发现机制

Agent Card 的设计

A2A 最聪明的设计就是 Agent Card。它像 LinkedIn 一样，把 Agent 的能力公开出来：

{
  "name": "code-reviewer",
  "description": "自动化代码审查，支持 Python/Go/Java",
  "url": "http://review-service:8080/a2a",
  "authentication": {
    "schemes": ["bearer"],
    "credentials": "需要在请求头带 token"
  },
  "skills": [
    {
      "id": "review_diff",
      "name": "审查代码变更",
      "input": {
        "type": "object",
        "properties": {
          "language": {"type": "string", "enum": ["python", "go", "java"]},
          "diff_content": {"type": "string"},
          "strictness": {"type": "string", "enum": ["low", "medium", "high"], "default": "medium"}
        },
        "required": ["language", "diff_content"]
      },
      "output": {
        "type": "object",
        "properties": {
          "issues": {"type": "array"},
          "score": {"type": "number"},
          "suggestions": {"type": "array"}
        }
      }
    }
  ]
}

这个设计巧妙在哪？它把 Agent 的能力发现和调用解耦了。Agent_A 不需要知道 Agent_B 的内部实现，只需要知道它的 Card 描述的能力接口。

写一个完整的 A2A Agent

下面这个是实际跑的代码审查 Agent：

# a2a_reviewer.py - A2A 代码审查服务
from a2a import A2AServer, AgentCard, Skill
import subprocess, re, asyncio

# 定义 Agent 能力卡片
card = AgentCard(
    name="code-reviewer",
    description="自动分析 Git PR 的代码质量，给出审查意见和改进建议",
    url="http://localhost:8080/a2a",
    skills=[
        Skill(
            id="analyze_pr",
            description="审查 PR 的代码质量",
            input={
                "type": "object",
                "properties": {
                    "repo_path": {"type": "string", "description": "仓库本地路径"},
                    "pr_id": {"type": "string", "description": "PR 编号或分支名"},
                    "review_depth": {
                        "type": "string", "enum": ["quick", "deep"],
                        "default": "quick"
                    }
                },
                "required": ["repo_path", "pr_id"]
            },
            output={
                "type": "object",
                "properties": {
                    "summary": {"type": "string"},
                    "issues": {"type": "array"},
                    "metrics": {"type": "object"}
                }
            }
        )
    ]
)

# 代码分析规则
RULES = [
    {"pattern": r"TODO|FIXME|HACK", "severity": "warning", "msg": "残留临时标记"},
    {"pattern": r"print\(|console\.log", "severity": "warning", "msg": "调试语句未清理"},
    {"pattern": r"(?<!def )\bpass\b", "severity": "info", "msg": "空实现体"},
    {"pattern": r"try:\s*\n\s+pass", "severity": "error", "msg": "空的 try 块"},
    {"pattern": r"import \*", "severity": "error", "msg": "禁止 import *"},
    {"pattern": r"exec\(|eval\(", "severity": "error", "msg": "禁止使用 exec/eval"},
]

async def analyze_diff(repo_path: str, pr_id: str) -> dict:
    """分析 PR diff 并返回审查结果"""
    try:
        # 获取 diff
        result = subprocess.run(
            ["git", "diff", f"origin/{pr_id}..." if pr_id.isdigit() else pr_id],
            capture_output=True, text=True, cwd=repo_path, timeout=30
        )
        diff = result.stdout
    except subprocess.TimeoutExpired:
        return {"summary": "❌ 超时：diff 生成超过 30 秒", "issues": [], "metrics": {}}
    
    if not diff:
        return {"summary": "⚠️ 无代码变更", "issues": [], "metrics": {}}
    
    # 统计变更
    added = len(re.findall(r'^\+\+', diff, re.MULTILINE))
    removed = len(re.findall(r'^--', diff, re.MULTILINE))
    changed_files = len(re.findall(r'^\+\+\+ b/', diff, re.MULTILINE))
    
    # 匹配规则
    issues = []
    for rule in RULES:
        matches = re.findall(rule["pattern"], diff)
        if matches:
            issues.append({
                "severity": rule["severity"],
                "message": f"{rule['msg']}（发现 {len(matches)} 处）",
                "count": len(matches)
            })
    
    # 计算评分
    base_score = 100
    for issue in issues:
        if issue["severity"] == "error":
            base_score -= 15 * issue["count"]
        elif issue["severity"] == "warning":
            base_score -= 5 * issue["count"]
    
    metrics = {
        "changed_files": changed_files,
        "added_lines": added,
        "removed_lines": removed,
        "quality_score": max(0, min(100, base_score))
    }
    
    summary = f"审查了 {changed_files} 个文件的 {added + removed} 行变更"
    if metrics["quality_score"] >= 80:
        summary += "，代码质量良好 👍"
    elif metrics["quality_score"] >= 60:
        summary += "，有改进空间 🔧"
    else:
        summary += "，建议修改后合并 ⚠️"
    
    return {"summary": summary, "issues": issues, "metrics": metrics}

# 处理 A2A 调用请求
async def handle_skill(skill_id: str, params: dict) -> dict:
    if skill_id == "analyze_pr":
        return await analyze_diff(params["repo_path"], params["pr_id"])
    raise ValueError(f"不支持的能力: {skill_id}")

# 启动服务
server = A2AServer(card=card, handler=handle_skill)
print(f"🚀 A2A Code Reviewer 启动于 http://localhost:8080/a2a")
print(f"📋 Agent Card: http://localhost:8080/.well-known/agent.json")
server.run(host="0.0.0.0", port=8080)

客户端发现机制

A2A 的客户端发现也很有意思。不需要硬编码目标地址，而是通过服务发现自动找到能处理任务的 Agent：

# a2a_client.py - 自动发现并调用 A2A Agent
import asyncio, json
from a2a import A2AClient

async def find_and_call():
    client = A2AClient()
    
    # 场景：从注册中心发现可用的 Agent
    registry_urls = [
        "http://localhost:8080/.well-known/agent.json",
        "http://review-service:8080/.well-known/agent.json",
        "http://localhost:8081/.well-known/agent.json"
    ]
    
    for url in registry_urls:
        try:
            agents = await client.discover(url)
            for agent in agents:
                print(f"发现 Agent: {agent.name}")
                print(f"  描述: {agent.description}")
                for skill in agent.skills:
                    print(f"  能力: {skill.id} -> {skill.input.get('properties', {}).keys()}")
        except Exception as e:
            print(f"  ❌ {url} 不可达: {e}")
    
    # 调用代码审查 Agent
    result = await client.call(
        agent_url="http://localhost:8080/a2a",
        skill_id="analyze_pr",
        params={
            "repo_path": "/home/dev/my_project",
            "pr_id": "feature/new-api",
            "review_depth": "quick"
        }
    )
    print(f"\n审查结果:")
    print(f"  摘要: {result['summary']}")
    print(f"  问题数: {len(result.get('issues', []))}")
    print(f"  质量评分: {result.get('metrics', {}).get('quality_score', 'N/A')}")

asyncio.run(find_and_call())

核心差异对比

我花了一天时间，用同一个需求——“自动分析部署日志并生成审查报告”——分别在两个协议上完整跑了一遍，以下是真实数据：

实现复杂度

对比维度	MCP	A2A
服务端代码行（最小可用）	~50 行	~70 行
通信方式	stdio / SSE	HTTP 短连接
协议版本成熟度	1.0（2026.3 已稳定）	0.9（草案阶段）
Python SDK 成熟度	完善（mcp-py 1.2）	可用但 API 不稳
调试工具	mcp-cli inspector	无官方调试工具
是否需要注册中心	不需要	推荐用 Agent Card 注册

基准性能

测试环境：同一台 Linux 服务器, Python 3.12
场景：查询日志 → 分析 → 返回结构化结果
各跑 10 次取中位数

MCP (stdio):   ~42ms  ← 最快，进程内通信
MCP (SSE):     ~118ms ← 多了一层 HTTP 流开销
A2A (HTTP):    ~95ms  ← 标准 HTTP 短连接

MCP stdio 模式有明显的性能优势，因为它在同一个进程空间里通信。但如果 MCP 走 SSE（Server-Sent Events），因为要维持一个长连接，反而比 A2A 的短连接稍复杂。

错误处理哲学

两个协议的错误处理方式反映了它们的设计哲学差异：

# MCP 的错误处理：简洁、直接、像函数调用
# Server 端：
try:
    result = query_database(sql)
except Exception as e:
    raise McpError(code=-32000, message=str(e))
    # → Client 端收到 McpError，需要自己处理重试

# A2A 的错误处理：协商式、带信息交换
# 客户端重试模式：
try:
    result = await client.call(agent_url, skill_id, params)
except A2ANotEnoughInformation as e:
    # Agent 说：信息不够，我还需要这些
    extra = await get_more_data(e.required_fields)
    # 带上补充信息重新协商
    result = await client.call(agent_url, skill_id, {**params, **extra})
except A2ATaskTimeout:
    # Agent 说：这活太久了，给我更多时间
    result = await client.wait(agent_url, task_id, timeout=120)

A2A 的协商机制很实用。Agent_B 发现参数不够时，不直接报错，而是告诉 Agent_A “我还需要 XXX”。这在跨团队多 Agent 协作时特别有用。

多 Agent 编排

这是两者最大的差异。

MCP 做多 Agent：MCP 本身不关心 Agent 间通信，你需要自己搭编排层。通常用 LangGraph 或 Dify 之类的框架来调度：

# MCP 下多 Agent 需要自建编排
from langgraph.graph import StateGraph, START

# 每个 Agent 内部使用 MCP 工具
def researcher(state):
    # 这个 agent 内部通过 MCP 调用搜索工具
    docs = mcp_client.call_tool("web_search", {"q": state["topic"]})
    return {"research": docs}

def writer(state):
    # 这个 agent 通过 MCP 调用文件工具
    mcp_client.call_tool("write_file", {"path": "/tmp/draft.md", "content": "..."})
    return {"draft": "/tmp/draft.md"}

# 手动定义工作流
graph = StateGraph(dict)
graph.add_node("research", researcher)
graph.add_node("writing", writer)
graph.add_edge("research", "writing")

A2A 做多 Agent：天然支持。不需要额外框架，Agent 之间通过 Agent Card 互相发现：

# A2A 下多 Agent 是原生能力
# Agent_A 完成研究后，直接调用 Agent_B 写文章
async def research_and_write():
    # 1. 研究 Agent 工作
    research = await research_agent.call(
        skill="search_papers",
        params={"topic": "agent protocols", "max_results": 10}
    )
    
    # 2. 发现能写文章的 Agent
    writer_agent = await discover_agent("article-writer")
    
    # 3. 把研究结果传给写文章 Agent
    article = await writer_agent.call(
        skill="write_article",
        params={
            "topic": "MCP vs A2A",
            "materials": research["papers"],
            "tone": "technical"
        }
    )
    return article

生态成熟度

生态维度	MCP	A2A
官方 SDK	Python, TypeScript, Java, Go	Python, TypeScript, Java
社区 Server 数量	500+（官方市场）	100+（分散在 GitHub）
主流框架支持	LangChain, LlamaIndex, Dify	CrewAI, Spring AI
CI/CD 集成	无官方方案	有，Agent Card 可内建
企业级支持	单进程，无 HA 方案	天生 HTTP，易做负载均衡

实战结论：什么场景用什么

两天的实操下来，我的建议非常明确：

选 MCP，如果你……

1. Agent 需要调用很多外部工具 —— 数据库、API、文件系统、命令行工具
2. 单 Agent 架构 —— 不需要多个 Agent 协作
3. 对延迟敏感 —— MCP stdio 模式延迟最低
4. 工具集成是主要需求 —— 让 LLM 能操作已有系统
5. 本地或单一节点部署 —— 同一个机器上运行的 Agent

选 A2A，如果你……

1. 需要多个 Agent 分工协作 —— 研究 Agent + 写代码 Agent + 审查 Agent
2. Agent 分布在不同服务 / 机器上 —— 微服务化的 Agent 架构
3. 需要动态发现可用 Agent —— 不知道有哪些 Agent 可用，需要运行时发现
4. Agent 之间需要协商 —— 任务参数不固定，需要来回沟通
5. 构建的是 Agent 生态系统 —— 而非单个 Agent 的能力增强

两者都用的最佳实践

如果构建的是复杂 Multi-Agent 系统，我的推荐组合是：

每个 Agent 内部：       用 MCP 调用工具（stdio 模式）
Agent 之间：            用 A2A 通信（Agent Card + HTTP）
编排调度层：            用 A2A 发现机制 + 自建 Workflow

这样每个 Agent 内部是快速的工具调用（MCP stdio），Agent 之间是松耦合的 HTTP 协作（A2A），各取所长。

踩坑记录

实操两天踩了不少坑，记下来省得你重蹈覆辙：

1. MCP SSE 连接会静默断开

MCP 的 SSE 实现没有内置心跳。如果 Agent 长时间没有工具调用，SSE 连接会静默断开，Agent 下一次调用时会超时。

# 解决方案：在 MCP Server 里加心跳
async def ping_loop(server, interval=30):
    while True:
        await asyncio.sleep(interval)
        try:
            # 发送空消息维持连接
            await server.send_ping()
        except ConnectionError:
            print("⚠️ SSE 连接断开，准备重连...")
            break

# 或者在客户端侧加重试
async def safe_call_tool(session, name, args, retries=3):
    for i in range(retries):
        try:
            return await session.call_tool(name, args)
        except (ConnectionError, TimeoutError):
            if i < retries - 1:
                await asyncio.sleep(2 ** i)  # 指数退避
                continue
            raise

2. A2A Agent Card 认证互操作性问题

A2A 规范支持 bearer token、mTLS、OAuth 等多种认证方式。但不同 SDK 的实现不一致——Python SDK 生成的 Agent Card 中 authentication.schemes 字段名，在 TypeScript SDK 里解析时字段名不统一。建议在所有 Agent 之间约定同一个认证方式，不要混用。

3. 生产环境不要依赖 stdio 做远程

MCP stdio 模式虽然快，但它走的是子进程管道。如果 Agent 进程重启，所有 MCP 子进程也会被干掉。在 Docker/K8s 环境下，必须用 MCP SSE 模式或者把每个工具独立部署成 HTTP 服务。

4. 版本迭代太快

写这篇文章时（2026年5月），MCP 最新是 1.0（已经稳定了），但 A2A 还是 0.9 草案阶段。A2A 的 API 每个月都有破坏性变更，建议锁死 SDK 版本：

# 锁死 a2a 版本，不要用 latest
pip install a2a==0.9.2  # 不是 pip install a2a

5. 不要被社区带偏

很多人非此即彼地站队。实际上 Anthropic 和 Google 已经在合作推动互操作性——MCP 负责工具集成层，A2A 负责 Agent 协作层，两者天然互补。非要选一个的话，先想清楚你是要做工具链还是做Agent 网络。

写在最后

2026 年的 Agent 协议生态，像极了 2015 年的容器编排——Docker Swarm、Mesos、Kubernetes 混战。当年大家也吵 Swarm 简单、K8s 复杂，最后赢的不是最简单的，而是生态最完善的。

MCP 和 A2A 也不是谁取代谁的关系。MCP 已经先发优势明显，生态丰富、稳定可靠；A2A 虽然还在草案阶段，但解决的是 Multi-Agent 这个核心里程碑问题。我个人的判断是：未来 6-12 个月内，两个协议会各自清晰定位，MCP 做工具集成标准，A2A 做 Agent 协作标准。

你现在开始上手哪个都不晚。我的建议是先跑 MCP，因为成熟度高、社区大、坑少。等 A2A 进入 1.0 稳定版后再接入，到时候两个协议已经可以无缝互通了。

代码全部开源在 github.com/xxx/mcp-a2a-comparison，包括完整可运行的 MCP Server、A2A Agent 和基准测试脚本。有任何问题评论区聊。