MCP vs A2A：AI Agent 通信协议的双子星之争——当工具调用遇上多 Agent 协作

前言：当 AI Agent 生态从"各自为战"走向"标准大一统"

2024 年底，Anthropic 发布 MCP（Model Context Protocol）时，整个 AI 社区还在讨论"又一个协议标准"。2026 年中，MCP 已随 Linux Foundation 的 AI Agents 基金会一起，成为几乎所有主流 AI 编程工具和 Agent 框架的标配。几乎同一时间，Google 主导的 A2A（Agent-to-Agent）协议悄然落地，目标直指多 Agent 系统之间的协作通信问题。

这两个协议，一经推出就被频繁对比，但很多开发者的理解还停留在"MCP 是给 Agent 调工具的，A2A 是让 Agent 之间互相聊天的"——这个认知没错，却太浅了。

本文要做的，是把这两个协议从底层架构到生产落地全部拆透：

• 它们各自解决了什么问题？
• 协议层的设计哲学有什么本质差异？
• 当一个 Agent 通过 A2A 调用另一个 Agent，而被调用的 Agent 又通过 MCP 调了工具——这整条链路的错误处理、上下文传递、安全审计怎么做？
• 生产环境中，哪些场景用 MCP，哪些用 A2A，哪些需要两者协同？

读完这篇文章，你将不只是"知道这两个协议"，而是真正理解它们在 AI Agent 架构体系中的各自定位，以及如何在实际项目中做出正确的技术选型。

一、背景：AI Agent 生态的"巴别塔之痛"

1.1 碎片化困局：一个真实的开发地狱

让我们先还原一个你在 2025 年可能经历过的真实场景：

你的团队在构建一个"智能代码审查 Agent"，底层用了 LangChain。LLM 需要：

• 访问 GitHub 仓库读取代码（GitHub API）
• 查询 PostgreSQL 数据库中的历史审查记录
• 调用 Slack 通知审查结果
• 访问 Confluence 读取团队规范文档

在没有统一协议的时代，你得为每一个数据源写一套适配代码：

# GitHub 集成 - 专用适配器
class GitHubAdapter:
    async def get_file_content(self, repo, path, token):
        # GitHub API 签名、认证、分页、异常处理...
        pass

# PostgreSQL 集成 - 又是另一套
class PostgresAdapter:
    async def query(self, sql, params):
        # 连接池、超时、重试...
        pass

# Slack 集成 - 再来一套
class SlackAdapter:
    async def send_message(self, channel, text, token):
        # OAuth、速率限制、消息格式化...
        pass

当你把这个 Agent 迁移到另一个框架（比如 LlamaIndex），或者换一个 LLM 提供商（比如从 Claude 换成 GPT-5），这十套适配器全部要重写。更痛苦的是：如果团队里有 5 个不同的 Agent，每个 Agent 都需要 10 种工具——那就是 50 套适配器，维护成本爆炸。

这还没算上另一个维度的问题：如果你需要让多个 Agent 协作（比如一个"前端 Agent"和一个"后端 Agent"配合完成一个功能），它们之间怎么通信？用 REST API？用消息队列？还是用 WebSocket？每个团队的答案都不一样。

1.2 MCP 和 A2A 的诞生背景

MCP（Model Context Protocol） 由 Anthropic 于 2024 年 11 月发布，2025 年底正式移交 Linux Foundation 旗下的 Agentic AI Foundation 治理。它的核心使命是解决第一个问题：让每个 Agent 能以统一的方式调用任意外部工具和数据源。

A2A（Agent-to-Agent）协议 由 Google 于 2025 年中主导推出，与 Anthropic、OpenAI、Cohere 等厂商共建。它的核心使命是解决第二个问题：让不同框架、不同厂商构建的 Agent 能互相发现对方的能力、传递任务、协同工作。

两个协议几乎在同一时间窗口被主流厂商采用，不是巧合——而是 AI Agent 从"单兵作战"走向"协同网络"的必然产物。

二、MCP 深度解析：AI Agent 的工具调用标准

2.1 协议架构：三层模型的精确拆解

MCP 采用了经典的客户端-服务器架构，但它的"客户端"和"服务器"与日常理解略有不同：

┌─────────────────────────────────────────────────────┐
│                   MCP Host（宿主）                   │
│  Claude Desktop / Cursor / 自研 Agent / Dify 等     │
│         负责管理多个 MCP Client 连接                 │
└───────────────────────┬─────────────────────────────┘
                        │ MCP 协议（JSON-RPC 2.0 over SSE/HTTP）
                        ▼
┌─────────────────────────────────────────────────────┐
│                   MCP Server（服务）                 │
│  一个 Server 对应一类外部资源（GitHub/Postgres/本地FS）│
│  暴露: Resources、Tools、Prompts 三类能力            │
└─────────────────────────────────────────────────────┘

Host 端（即 MCP Client）：

• 负责任务分发和结果聚合
• 一个 Host 可以同时连接多个 MCP Server
• 支持 stdio（本地进程）和 HTTP+SSE（远程服务）两种传输方式

Server 端：

• 每个 Server 专注于一类资源或工具集
• 遵循 MCP 规范的 Server 可以被任意 MCP Host 直接使用
• 即插即用，无需 Host 端修改代码

2.2 四大核心能力：从"工具箱"到"记忆层"

MCP v1.0 定义了四种核心能力，这是理解整个协议的关键：

能力一：Resources（资源）——给 AI 一双眼睛

Resources 提供只读数据访问能力，类似于"动态配置文件"。Server 通过 resources/list 公告自己可提供哪些资源，Host 可以通过 resources/read 读取具体内容。

// MCP Server 端：定义一个 GitHub 仓库资源的 provider
// TypeScript + @modelcontextprotocol/sdk

import { MCPServer } from "@modelcontextprotocol/sdk/server";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse";

const server = new MCPServer({
  name: "github-mcp-server",
  version: "1.0.0",
});

// 定义 resources 列表（在初始化时告知 Host）
server.setRequestHandler("resources/list", async () => {
  return {
    resources: [
      {
        uri: "github://owner/repo/READMEs",
        name: "Repository README Files",
        description: "List of all README files in the repository",
        mimeType: "application/json",
      },
      {
        uri: "github://owner/repo/issues",
        name: "Open Issues",
        description: "All open issues with labels and assignees",
        mimeType: "application/json",
      },
    ],
  };
});

// 处理具体的资源读取请求
server.setRequestHandler("resources/read", async (request) => {
  const { uri } = request.params;

  if (uri.startsWith("github://owner/repo/READMEs")) {
    const files = await fetchReadmeFiles();
    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(files),
        },
      ],
    };
  }

  throw new Error(`Unknown resource URI: ${uri}`);
});

# MCP Server 端：Python 实现
# 使用 mcp Python SDK

from mcp.server import MCPServer
from mcp.types import Resource, TextContent

server = MCPServer(name="file-system-server", version="1.0.0")

@server.list_resources()
async def list_resources():
    """公告本 Server 可提供的资源"""
    return [
        Resource(
            uri="fs://project/src/**/*.py",
            name="Python Source Files",
            description="All Python source files in the project",
            mimeType="text/x-python"
        ),
        Resource(
            uri="fs://project/config.yaml",
            name="Project Configuration",
            description="Main configuration file",
            mimeType="application/yaml"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> TextContent:
    """读取具体资源内容"""
    if uri == "fs://project/config.yaml":
        with open("config.yaml", "r") as f:
            return TextContent(type="text", text=f.read())
    # ...

能力二：Tools（工具）——给 AI 一双手

Tools 是 MCP 最核心的能力，赋予 AI 主动执行操作的能力。与 Resources 的只读不同，Tools 允许 LLM 调用函数并修改外部状态。

// MCP Server 端：定义工具
// 典型场景：数据库写操作

server.setRequestHandler("tools/list", async () => {
  return {
    tools: [
      {
        name: "postgres_query",
        description: "Execute a read-only SQL query against the production database",
        inputSchema: {
          type: "object",
          properties: {
            sql: {
              type: "string",
              description: "SQL SELECT query (UPDATE/DELETE are not allowed)",
            },
            params: {
              type: "array",
              description: "Query parameter values",
              items: {},
            },
          },
          required: ["sql"],
        },
      },
      {
        name: "create_github_issue",
        description: "Create a new issue in the GitHub repository",
        inputSchema: {
          type: "object",
          properties: {
            title: { type: "string" },
            body: { type: "string" },
            labels: {
              type: "array",
              items: { type: "string" },
            },
          },
          required: ["title", "body"],
        },
      },
    ],
  };
});

server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "postgres_query") {
    // SQL 注入防护：参数化查询
    const result = await db.query(args.sql, args.params || []);
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(result.rows, null, 2),
        },
      ],
    };
  }

  if (name === "create_github_issue") {
    const issue = await githubClient.createIssue({
      title: args.title,
      body: args.body,
      labels: args.labels || [],
    });
    return {
      content: [
        {
          type: "text",
          text: `Issue created: ${issue.html_url}`,
        },
      ],
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

能力三：Prompts（提示模板）——给 AI 专家经验

Prompts 允许 MCP Server 预定义结构化的提示模板，Host 端的 LLM 可以直接调用。这相当于把"领域专家的经验"编码成可复用的模板。

server.setRequestHandler("prompts/list", async () => {
  return {
    prompts: [
      {
        name: "code-review",
        description: "Perform a thorough code review with security focus",
        arguments: [
          { name: "file_path", description: "Path to the file to review", required: true },
          { name: "language", description: "Programming language", required: false },
        ],
      },
      {
        name: "architecture-analysis",
        description: "Analyze system architecture and identify bottlenecks",
        arguments: [
          { name: "system_description", description: "Description of the system", required: true },
        ],
      },
    ],
  };
});

server.setRequestHandler("prompts/get", async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "code-review") {
    return {
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `You are a senior security-focused code reviewer. Review the following ${args.language || "code"} file at ${args.file_path}.\n\nFocus on:\n1. Security vulnerabilities (injection, auth bypass, secrets in code)\n2. Performance issues (N+1 queries, missing indexes, memory leaks)\n3. Error handling completeness\n4. Code maintainability\n\nProvide specific line numbers and fix suggestions.`,
          },
        },
      ],
    };
  }
});

能力四：Sampling（采样）——让 Server 回调 Host

这是 MCP v1.0 中最容易被忽视的能力：Server 可以反过来要求 Host 调用 LLM 生成内容。这打破了纯 C/S 的单向关系，实现了真正的双向通信。

// MCP Server 端：请求 Host 生成内容
// 典型场景：需要 LLM 判断数据质量后再处理

server.setRequestHandler("sampling/createMessage", async (request) => {
  // Server 需要 Host 的 LLM 来做某个判断
  const response = await request.server.request({
    method: "sampling/createMessage",
    params: {
      systemPrompt: "You are a data quality classifier.",
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `Classify this data record: ${JSON.stringify(dataRecord)}`,
          },
        },
      ],
      maxTokens: 256,
    },
  });
  return response;
});

2.3 协议传输层：两种模式的选择艺术

MCP 支持两种传输模式，选择取决于你的部署场景：

实战建议：

• 开发阶段用 stdio 模式，调试方便
• 生产环境优先 HTTP+SSE，支持水平扩展和集中日志
• 混合模式：一个 Host 连接多个 Server，部分本地（stdio）、部分远程（HTTPSSE）

2.4 MCP 生态现状（2026年7月）

截至本文发稿，MCP 生态已有：

• 官方/社区 Server 超过 3000 个，覆盖 GitHub、Slack、PostgreSQL、Redis、文件系统等主流工具
• 主要框架支持：Claude Desktop、Cursor、Windsurf/Devin Desktop、Codex、Cline、OpenClaw 均已内置 MCP Client
• SDK 生态：Python、TypeScript、Go、Rust、Java 均有官方或社区 SDK
• 治理：Linux Foundation AI Agents Foundation 负责协议演进，版本稳定性有保障

三、A2A 深度解析：多 Agent 协作的通信标准

3.1 协议定位：A2A 解决的不是工具问题

如果说 MCP 是让单个 Agent 调用工具，那么 A2A 解决的是更高层次的问题：Agent 之间的协作语言。

一个典型场景：

你有一个"后端架构 Agent"，能设计数据库 schema 和 API 接口。
你有一个"前端开发 Agent"，能把设计转化成 React 组件。
你有一个"测试 Agent"，负责生成单元测试和集成测试。

这三个 Agent 之间怎么协作？
后端 Agent 完成后，怎么通知前端 Agent 开始工作？
前端 Agent 需要什么信息才能准确工作？
测试 Agent 怎么知道从哪里读取接口文档？

在 A2A 出现之前，每个团队用自己定义的 API 解决这个问题——用 RESTful JSON、GraphQL、甚至 WebSocket。不同团队的 Agent 互操作等于零。

A2A 的目标：让任何框架构建的 Agent，只要遵循协议，就能互相发现、协商、协作。

3.2 核心概念：从 Agent Card 到任务生命周期

Agent Card：Agent 的"身份证"

A2A 协议要求每个 Agent 暴露一个 Agent Card（JSON-LD 格式），描述自己的能力和元数据：

{
  "name": "backend-architect-agent",
  "description": "Expert backend architect specializing in database design and API contracts",
  "version": "2.1.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "schema-design",
      "name": "Database Schema Design",
      "description": "Design PostgreSQL schemas with proper indexing",
      "tags": ["postgresql", "database", "schema", "indexing"]
    },
    {
      "id": "api-contract",
      "name": "REST API Contract Design",
      "description": "Design OpenAPI 3.1 compliant REST contracts",
      "tags": ["rest", "openapi", "api-design"]
    }
  ],
  "endpoints": {
    "default": "https://agent-cluster.internal/v1/agents/backend-architect"
  },
  "authentication": {
    "type": "bearer",
    "credentials": "required"
  }
}

这个 Agent Card 让其他 Agent（或 Agent 编排层）可以动态发现谁适合处理某个任务，而不需要硬编码 Agent 地址。

任务生命周期：A2A 的状态机

A2A 定义了标准化的任务状态流转：

  submit (创建任务)
      │
      ▼
  working ────► completed (任务完成，正常结果)
      │              │
      │              ▼
      │           failed (任务失败，异常退出)
      │
      ▼
  input-required (需要人工或外部 Agent 补充信息)
      │
      ▲
      └─────────── (外部补充后，继续 working)

核心状态解释：

• working：Agent 正在处理任务
• completed：任务成功完成，返回结构化结果
• failed：任务失败，包含错误原因和可重试标识
• input-required：Agent 发现自己缺少必要信息，主动暂停等待补充
• canceled：任务被主动取消

这个状态机看起来简单，但它解决了多 Agent 协作中最头疼的问题之一：任务跟踪和容错恢复。当一个 Agent 在 working 状态崩溃了，其他 Agent 或编排层可以通过状态查询发现这一点，并决定是否重试或降级。

3.3 消息协议：任务提交与结果聚合

A2A 的消息格式基于 JSON-RPC 2.0，但扩展了任务语义：

// A2A 任务提交请求
interface TaskSubmitRequest {
  jsonrpc: "2.0";
  method: "tasks/submit";
  params: {
    id: string;              // 全局唯一任务 ID
    sessionId: string;       // 会话 ID，支持关联多轮交互
    priority?: number;       // 优先级，0-100

    // 任务输入：支持多种格式
    input: {
      // 文本任务
      text?: string;
      // 结构化任务
      data?: Record<string, unknown>;
      // 多模态任务
      attachments?: Attachment[];
    };

    // 能力协商
    acceptedOutputModes: ("text" | "data" | "streaming" | "artifact")[];
    // 结果格式偏好（text/json/streaming）
  };
  id: string;
}

// A2A 任务结果响应
interface TaskResultResponse {
  jsonrpc: "2.0";
  result: {
    id: string;
    status: "completed" | "failed" | "input-required";
    // 任务结果
    output?: {
      // 支持多种输出模式
      text?: string;
      data?: Record<string, unknown>;
      artifact?: {
        name: string;
        mimeType: string;
        url?: string;
        content?: string;
      }[];
    };
    // 错误信息（如果 failed）
    error?: {
      code: string;
      message: string;
      recoverable: boolean;
    };
    // 历史记录（如果开启 stateTransitionHistory）
    history?: StatusUpdate[];
  };
  id: string;
}

3.4 流式响应：Server-Sent Events 实现

对于长时间运行的任务，A2A 支持通过 SSE（Server-Sent Events）推送实时状态更新：

# Python A2A Server 实现（简化版）
# 使用 a2a-python-sdk

from a2a.server import A2AServer
from a2a.types import (
    AgentCard,
    TaskStatusUpdateEvent,
    TaskArtifactUpdateEvent,
)
import sse_starlette.sse as sse

class BackendArchitectAgent(A2AServer):
    """后端架构 Agent：A2A Server 实现"""

    async def get_agent_card(self) -> AgentCard:
        return AgentCard(
            name="backend-architect-agent",
            description="Expert backend architect specializing in database design",
            version="2.1.0",
            capabilities={
                "streaming": True,
                "pushNotifications": True,
                "stateTransitionHistory": True,
            },
            skills=[
                Skill(id="schema-design", name="Database Schema Design", tags=["postgresql"]),
            ],
        )

    async def handle_task(self, task: Task):
        """处理来自其他 Agent 的任务请求"""
        input_text = task.input.get("text", "")

        # 步骤1：发送 working 状态
        await self.send_status_update(
            task_id=task.id,
            status="working",
            message="Analyzing requirements..."
        )

        # 步骤2：生成数据库 schema（流式输出）
        schema_result = await self.generate_schema_streaming(input_text)
        for chunk in schema_result:
            await self.send_artifact_update(
                task_id=task.id,
                artifact={
                    "name": "database_schema.sql",
                    "mimeType": "application/sql",
                    "content": chunk
                }
            )

        # 步骤3：生成 API 合约
        await self.send_status_update(
            task_id=task.id,
            status="working",
            message="Designing API contracts..."
        )
        api_contract = await self.generate_api_contract(input_text)

        # 步骤4：完成任务
        await self.send_task_result(
            task_id=task.id,
            status="completed",
            output={
                "schema": schema_result.finalize(),
                "api_contract": api_contract,
                "recommendations": ["Use connection pooling", "Add read replicas"]
            }
        )

// A2A Client：前端 Agent 调用后端 Agent
// TypeScript 实现

import { A2AClient } from "@a2a-agent/a2a-client";

const client = new A2AClient({
  agentUrl: "https://agent-cluster.internal/v1/agents/backend-architect",
  token: process.env.A2A_BEARER_TOKEN,
});

// 发现 Agent 能力
const agentCard = await client.getAgentCard();
const schemaSkill = agentCard.skills.find(s => s.id === "schema-design");
console.log(`Agent supports: ${schemaSkill.name}`);

// 提交任务
const task = await client.submitTask({
  input: {
    text: "Design a database schema for a multi-tenant SaaS project management tool. " +
          "Requirements: users belong to organizations, projects belong to organizations, " +
          "tasks belong to projects, tasks can have comments and attachments."
  },
  acceptedOutputModes: ["data", "artifact"],
});

// 订阅 SSE 流式更新
const eventSource = client.subscribeToTask(task.id);

eventSource.on("status", (update: TaskStatusUpdateEvent) => {
  console.log(`[${update.status}] ${update.message}`);
});

eventSource.on("artifact", (update: TaskArtifactUpdateEvent) => {
  console.log(`📄 Artifact: ${update.artifact.name}`);
  // 逐步接收 schema 内容...
});

// 等待最终结果
const result = await client.waitForCompletion(task.id);
console.log("Final output:", result.output);

3.5 错误传播：跨 Agent 异常的完整链路

这是 A2A 相比自定义 API 的核心优势之一：标准化的错误传播机制。

// 当后端 Agent 遇到不可恢复的错误时
interface A2AError {
  code: string;       // 标准化错误码
  message: string;   // 人类可读描述
  details?: unknown;  // 附加调试信息

  // 关键字段：决定上游 Agent 如何处理
  recoverable: boolean;      // 是否可以重试
  retryAfter?: number;       // 建议的重试等待时间（秒）
  fallbackAgent?: string;    // 推荐的降级 Agent ID
}

const standardErrorCodes = {
  "AUTHENTICATION_FAILED": "认证失败，Token 无效或已过期",
  "CAPABILITY_NOT_SUPPORTED": "目标 Agent 不支持此技能",
  "INPUT_VALIDATION_ERROR": "输入数据格式校验失败",
  "AGENT_UNAVAILABLE": "目标 Agent 暂时不可用",
  "TASK_TIMEOUT": "任务执行超时",
  "CONTEXT_OVERFLOW": "上下文超出长度限制",
  "RATE_LIMITED": "请求频率超限",
  "INTERNAL_ERROR": "Agent 内部错误（可重试）",
};

有了标准化的错误码，上游 Agent 的容错逻辑可以写得很优雅：

# A2A Client 容错处理
async def call_backend_agent(user_requirement: str):
    try:
        result = await client.submit_task({
            "input": {"text": user_requirement},
            "acceptedOutputModes": ["data", "artifact"],
        })
        return result.output

    except A2AError as e:
        if e.code == "AGENT_UNAVAILABLE" and e.recoverable:
            # 等一下再试
            await asyncio.sleep(e.retryAfter or 5)
            return await call_backend_agent(user_requirement)

        elif e.code == "CAPABILITY_NOT_SUPPORTED" and e.fallbackAgent:
            # 切换到降级 Agent
            return await call_fallback_agent(user_requirement, e.fallbackAgent)

        elif e.code in ("INPUT_VALIDATION_ERROR", "TASK_TIMEOUT"):
            # 不可恢复的错误，抛出给用户
            raise AgentExecutionError(f"Agent failed: {e.message}")

        else:
            # 通用内部错误，尝试重试 3 次
            for attempt in range(3):
                await asyncio.sleep(2 ** attempt)
                try:
                    return await client.submit_task({...})
                except A2AError:
                    continue
            raise AgentExecutionError("Max retries exceeded")

四、MCP vs A2A：深度对比与架构哲学

4.1 定位对比：不是竞争，是正交

这是理解两个协议最关键的一点：MCP 和 A2A 解决的是不同维度的问题，它们是正交的，不是替代关系。

                    AI Agent 架构分层

        ┌──────────────────────────────────┐
        │     Agent 与 Agent 协作层          │  ← A2A：多 Agent 通信
        │  "谁来做？怎么分工？结果交给谁？"    │
        └──────────────┬───────────────────┘
                       │ A2A 协议
        ┌──────────────▼───────────────────┐
        │     单个 Agent 能力增强层          │  ← MCP：工具调用 + 上下文
        │  "我有什么工具和数据可以调用？"     │
        └──────────────┬───────────────────┘
                       │ MCP 协议
        ┌──────────────▼───────────────────┐
        │        LLM 推理层                  │  ← 模型本身
        │  "我理解任务，生成决策"            │
        └──────────────────────────────────┘

MCP 的问题域：Agent 如何调用工具、获取数据？
A2A 的问题域：Agent 如何与同类通信、协作完成任务？

4.2 能力矩阵对比

4.3 当两者相遇：MCP + A2A 的混合架构

最有趣的实际场景：Agent A 通过 A2A 调用 Agent B，Agent B 在执行过程中通过 MCP 调了工具。这条链路的完整流程：

Agent A (A2A Client)
    │
    │ A2A: 提交任务 "设计一个电商数据库 schema"
    ▼
Agent B (A2A Server + MCP Client)
    │
    │ MCP: 调用 postgres-mcp-server 工具查询现有 schema
    ▼
PostgreSQL MCP Server
    │
    │ 返回现有 schema 信息
    ▼
Agent B 收到上下文，开始设计...
    │
    │ MCP: 调用 github-mcp-server 提交设计文档
    ▼
GitHub MCP Server
    │
    │ 返回 PR 链接
    ▼
Agent B (A2A Server)
    │
    │ A2A: 任务完成，返回 schema + PR 链接
    ▼
Agent A 收到结果

这条链路带来了三个必须解决的实际问题：

问题一：上下文传播
Agent A 传给 Agent B 的上下文，到了 Agent B 的 MCP 调用中是否可见？答案是：取决于 Agent B 的架构设计。好的实现会在内部维护一个上下文传递机制，把 A2A 任务输入的文本作为背景信息传给 MCP 工具调用。

问题二：错误传播
如果 GitHub MCP Server 返回 403 错误，这个错误如何传递回 Agent A？MCP 的工具错误是局部的，A2A 的任务错误是全局的。Agent B 需要做"翻译层"：把 MCP 错误转换为 A2A 任务失败，并附上 recoverable 标记。

问题三：安全边界
Agent B 调用 MCP 工具时，工具的权限来自哪里？是从 Agent A 的上下文中继承，还是 Agent B 有独立的权限体系？这在企业多租户场景下是核心问题，需要在 A2A 协议之上叠加额外的权限上下文。

五、生产实战：MCP + A2A 的工程落地模式

5.1 模式一：MCP 优先——单 Agent 能力扩展

适合场景：单个 Agent 需要连接大量外部工具，但 Agent 间协作需求不强。

┌──────────────────┐
│   Claude Desktop │
│   (MCP Host)     │
└────┬─────────────┘
     │ MCP stdio
     ├── github-mcp-server     (本地进程)
     ├── postgres-mcp-server   (本地进程)
     ├── filesystem-mcp-server (本地进程)
     └── slack-mcp-server      (HTTP SSE，远程)

代码实现（MCP Server 快速启动）：

# 使用官方 CLI 快速启动一个 MCP Server
pip install "mcp[cli]"

# 使用 mcp CLI 启动 filesystem server
mcp server filesystem --path /workspace/project

# 使用 mcp CLI 启动 GitHub server
mcp server github --token $GITHUB_TOKEN --repo owner/repo

# 在 Claude Desktop 配置文件中注册
# ~/.claude/desktop_settings.json
{
  "mcpServers": {
    "filesystem": {
      "command": "mcp",
      "args": ["server", "filesystem", "--path", "/workspace/project"]
    },
    "github": {
      "command": "mcp",
      "args": ["server", "github", "--token", "$GITHUB_TOKEN"]
    }
  }
}

5.2 模式二：A2A 优先——多 Agent 编排网络

适合场景：需要多个专业 Agent 协同工作，每个 Agent 内部可能用 MCP 调用工具。

                    ┌──────────────┐
                    │  Orchestrator │ ← A2A Client（编排层）
                    │  Agent        │
                    └──────┬───────┘
                           │ A2A
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ Backend  │    │ Frontend │    │ QA       │
    │ Architect│    │ Developer│    │ Engineer │
    │ Agent    │    │ Agent    │    │ Agent    │
    └─────┬────┘    └─────┬────┘    └─────┬────┘
          │               │               │
          │ MCP           │ MCP          │ MCP
          ▼               ▼               ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ GitHub   │    │ Figma    │    │ Playwright│
    │ MCP      │    │ MCP      │    │ MCP       │
    └──────────┘    └──────────┘    └──────────┘

5.3 模式三：混合部署——企业级 Agent 网络

                        ┌────────────────────────┐
                        │   Enterprise Gateway    │
                        │  (A2A + MCP 双协议代理) │
                        │  - 统一认证/鉴权       │
                        │  - 流量控制            │
                        │  - 审计日志            │
                        └───────────┬────────────┘
                                    │
         ┌──────────────────────────┼──────────────────────────┐
         │                          │                          │
         ▼                          ▼                          ▼
  ┌─────────────┐           ┌─────────────┐           ┌─────────────┐
  │  Agent Pool │           │  Agent Pool │           │  Agent Pool │
  │  (研发域)    │           │  (运维域)    │           │  (数据域)    │
  │             │           │             │           │             │
  │ - 编码Agent │           │ - 监控Agent │           │ - 分析Agent │
  │ - 审查Agent │           │ - 部署Agent │           │ - 报表Agent │
  │ - 重构Agent │           │ - 告警Agent │           │ - ETL Agent  │
  └─────────────┘           └─────────────┘           └─────────────┘

这是 2026 年头部科技企业实际采用的架构模式。Gateway 层的核心职责：

# 企业级 Agent Gateway 伪代码

class AgentGateway:
    def __init__(self):
        self.mcp_registry = MCPServiceRegistry()   # MCP Server 注册表
        self.a2a_discovery = A2ADiscoveryService() # A2A Agent 发现

    async def route_task(self, task: TaskRequest):
        # 1. 认证与鉴权
        identity = await self.authenticate(task.token)
        permissions = await self.authorize(identity, task)

        # 2. 如果是 A2A 任务，通过 Agent Card 发现合适的 Agent
        if task.protocol == "a2a":
            target_agent = await self.a2a_discovery.find_agent(
                skill_required=task.skill_id,
                domain=task.domain,
                max_cost=task.budget
            )

            # 3. 注入权限上下文
            authorized_task = task.with_context({
                "permissions": permissions,
                "audit_id": self.generate_audit_id(),
                "cost_limit": task.budget
            })

            # 4. 路由到目标 Agent
            result = await self.forward_to_agent(target_agent, authorized_task)

        # 5. 如果是 MCP 任务，直接路由到对应的 MCP Server
        elif task.protocol == "mcp":
            mcp_server = self.mcp_registry.get(task.server_name)
            result = await mcp_server.call_tool(
                task.tool_name,
                task.arguments,
                context={"permissions": permissions}
            )

        # 6. 记录审计日志
        await self.audit_log.record(identity, task, result)
        return result

六、安全与治理：企业落地的必备考量

6.1 MCP 安全模型

MCP 的安全风险集中在三个层面：

数据传输层：stdio 模式下，Host 和 Server 之间通过进程内存传递数据，操作系统级别的进程隔离提供基本保护。HTTP+SSE 模式下，必须启用 TLS，否则 Bearer Token 在网络上明文传输。

工具权限层：MCP Server 以哪个身份运行，决定了它能访问什么资源。建议为 MCP Server 配置最小权限原则：

// postgres-mcp-server 的最小权限配置
{
  "server": "postgres-mcp-server",
  "connection": {
    "host": "internal-db.company.com",
    "database": "app_production",
    "user": "mcp_readonly_user"   // 只读账号，不是 superuser
  },
  "allowed_operations": ["SELECT", "EXPLAIN"],
  "forbidden_operations": ["INSERT", "UPDATE", "DELETE", "DROP", "TRUNCATE"],
  "allowed_tables": ["code_reviews", "metrics"],
  "query_timeout_ms": 5000
}

提示注入层：当 MCP 资源内容来自外部（用户上传文件、第三方 API），恶意内容可能被注入到 LLM 的上下文中。2026 年初，GitHub MCP Server 就曾报告过通过仓库 README 注入恶意指令的安全漏洞。缓解方案：在 MCP Server 层面对输入内容做清理，禁止或限制动态 prompt 构造。

6.2 A2A 安全模型

A2A 的安全挑战更大，因为它涉及多个 Agent 之间的信任传递：

身份认证：A2A 要求每个 Agent 持有 Bearer Token，且 Agent Card 中声明了认证要求。但 Bearer Token 的生命周期管理（轮换、撤销、泄露检测）没有在协议规范中定义，需要部署方自行实现。

权限继承：当 Agent A 将任务委托给 Agent B 时，Agent B 应该获得多少权限？最简单的方案是：Agent B 使用自己的权限体系，与 Agent A 完全独立。更严格的方案是：A2A 任务携带委托上下文，其中包括 Agent A 的权限子集（最小权限原则）。

// A2A 任务中的权限委托上下文
interface A2ATaskContext {
  taskId: string;
  submittingAgent: {
    id: string;
    domain: string;  // 研发域/运维域/数据域
  };
  delegatedPermissions: {
    allowedResources: string[];    // 比如 ["github://org/repo/read"]
    allowedTools: string[];        // 比如 ["postgres_query"]
    maxDuration: number;           // 秒
    maxCost: number;               // 预估 Token 成本上限
  };
  auditContext: {
    auditId: string;
    parentAuditId?: string;         // 追溯完整调用链
    dataClassification: "public" | "internal" | "confidential" | "restricted";
  };
}

审计链路：A2A 的多跳特性让审计变得复杂。假设任务链路是 A → B → C → D，每一个 Agent 的操作都应该被记录，且需要通过 auditId 串联成完整链路。这在合规要求严格的金融和医疗行业是刚需。

七、实战代码：从零搭建一个同时支持 MCP 和 A2A 的 Agent

这一节给出一个完整的生产级示例：一个"代码审查 Agent"，它通过 A2A 接收来自"代码提交 Agent"的任务，通过 MCP 调用 GitHub API 获取代码，然后进行审查，最后通过 A2A 返回结果。

7.1 完整架构

代码提交 Agent
  (A2A Client)
      │
      │ A2A: 提交审查任务
      ▼
代码审查 Agent
  (A2A Server + MCP Client)
      │
      │ MCP: github-mcp-server.getFile()
      ▼
GitHub MCP Server
      │
      │ 返回 diff 内容
      ▼
代码审查 Agent 推理（Claude API）
      │
      │ MCP: github-mcp-server.createReviewComment()
      ▼
GitHub MCP Server
      │
      │ 返回 Review 链接
      ▼
代码审查 Agent (A2A Server)
      │
      │ A2A: 任务完成 + Review URL
      ▼
代码提交 Agent 收到结果

7.2 代码实现

# review_agent.py
# 代码审查 Agent：同时扮演 A2A Server 和 MCP Client

import asyncio
from typing import AsyncGenerator

from a2a.server import A2AServer, AgentCard
from a2a.types import (
    TaskStatusUpdateEvent,
    TaskArtifactUpdateEvent,
)
from mcp import Client as MCPClient
from anthropic import AsyncAnthropic


class CodeReviewAgent(A2AServer):
    """同时支持 A2A（接收任务）和 MCP（调用工具）的代码审查 Agent"""

    def __init__(self):
        self.mcp_client = MCPClient()
        self.claude = AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"]))

        # MCP Server 连接池
        self.mcp_servers = {
            "github": "http://localhost:3000/mcp",  # 自定义 GitHub MCP Server
        }

    async def get_agent_card(self) -> AgentCard:
        """公告本 Agent 的能力和入口"""
        return AgentCard(
            name="code-review-agent",
            description="Expert code reviewer specializing in security and performance",
            version="1.2.0",
            capabilities={
                "streaming": True,
                "pushNotifications": True,
                "stateTransitionHistory": True,
            },
            skills=[
                {
                    "id": "security-review",
                    "name": "Security Code Review",
                    "description": "Find security vulnerabilities in code",
                    "tags": ["security", "owasp", "injection", "xss"],
                },
                {
                    "id": "performance-review",
                    "name": "Performance Review",
                    "description": "Identify performance bottlenecks",
                    "tags": ["performance", "database", "caching", "async"],
                },
            ],
        )

    async def setup_mcp_servers(self):
        """初始化所有 MCP Server 连接"""
        await self.mcp_client.connect_to_server(
            name="github",
            url=self.mcp_servers["github"],
            auth_token=os.environ["MCP_GITHUB_TOKEN"]
        )

    async def handle_task(self, task) -> AsyncGenerator:
        """
        处理来自 A2A 的审查任务
        task.input 包含: repo, pr_number, file_paths, review_type
        """
        repo = task.input["repo"]
        pr_number = task.input["pr_number"]
        file_paths = task.input["file_paths"]
        review_type = task.input.get("review_type", "security,performance")

        # ====== 步骤1：通过 MCP 获取 PR diff ======
        await self.send_status_update(
            task_id=task.id,
            status="working",
            message=f"Fetching code from {repo}/PR #{pr_number}..."
        )

        diff_content = await self.mcp_client.call_tool(
            server="github",
            tool="get_pull_request_diff",
            arguments={"repo": repo, "pr_number": pr_number}
        )

        # 发送 artifact（diff 文件）
        await self.send_artifact_update(
            task_id=task.id,
            artifact={
                "name": f"PR-{pr_number}-diff.txt",
                "mimeType": "text/plain",
                "content": diff_content
            }
        )

        # ====== 步骤2：逐文件审查（流式输出）======
        findings = []
        for file_path in file_paths:
            await self.send_status_update(
                task_id=task.id,
                status="working",
                message=f"Reviewing {file_path}..."
            )

            # 获取单个文件内容
            file_content = await self.mcp_client.call_tool(
                server="github",
                tool="get_file_content",
                arguments={"repo": repo, "path": file_path, "ref": f"refs/pull/{pr_number}/head"}
            )

            # 调用 Claude 进行审查
            review_result = await self.claude.messages.create(
                model="claude-opus-4-5",
                max_tokens=4096,
                system="""You are an expert code reviewer. Review the code for:
1. Security: SQL injection, XSS, auth bypass, secrets in code
2. Performance: N+1 queries, missing indexes, inefficient algorithms
3. Correctness: race conditions, error handling, edge cases

Respond in JSON format with findings array. Each finding includes:
- severity: "critical" | "high" | "medium" | "low"
- line: line number (or null)
- category: "security" | "performance" | "correctness"
- description: plain text explanation
- suggestion: specific fix recommendation""",
                messages=[
                    {
                        "role": "user",
                        "content": f"Review this file ({file_path}):\n\n{file_content}"
                    }
                ]
            )

            import json
            result_text = review_result.content[0].text
            findings.extend(json.loads(result_text)["findings"])

            # 流式发送审查中间结果
            await self.send_status_update(
                task_id=task.id,
                status="working",
                message=f"Found {len(findings)} issues so far in {file_path}..."
            )

        # ====== 步骤3：通过 MCP 提交 GitHub Review ======
        await self.send_status_update(
            task_id=task.id,
            status="working",
            message="Submitting review to GitHub..."
        )

        review_body = self._format_github_review(findings)
        review_result = await self.mcp_client.call_tool(
            server="github",
            tool="create_review_comment",
            arguments={
                "repo": repo,
                "pr_number": pr_number,
                "body": review_body,
                "event": "COMMENT" if findings else "APPROVE"
            }
        )

        # ====== 步骤4：完成任务 ======
        await self.send_task_result(
            task_id=task.id,
            status="completed",
            output={
                "review_url": review_result["url"],
                "total_findings": len(findings),
                "severity_breakdown": {
                    "critical": len([f for f in findings if f["severity"] == "critical"]),
                    "high": len([f for f in findings if f["severity"] == "high"]),
                    "medium": len([f for f in findings if f["severity"] == "medium"]),
                    "low": len([f for f in findings if f["severity"] == "low"]),
                },
                "findings": findings
            }
        )

    def _format_github_review(self, findings: list) -> str:
        """将审查结果格式化为 GitHub Review 评论"""
        lines = ["## 🔍 Code Review Report\n"]

        severity_emojis = {
            "critical": "🚨",
            "high": "⚠️",
            "medium": "📝",
            "low": "💡"
        }

        for severity in ["critical", "high", "medium", "low"]:
            filtered = [f for f in findings if f["severity"] == severity]
            if not filtered:
                continue

            lines.append(f"\n### {severity_emojis[severity]} {severity.upper()} ({len(filtered)} issues)\n")

            for finding in filtered:
                line_ref = f"Line {finding['line']}" if finding.get("line") else "General"
                lines.append(f"**[{line_ref}]** {finding['description']}")
                lines.append(f"```suggestion\n{finding['suggestion']}\n```\n")

        lines.append("\n---\n*Reviewed by AI Code Review Agent*")
        return "\n".join(lines)

八、性能优化与生产调参

8.1 MCP 性能优化

连接复用：MCP Client 应当维护长连接池，避免每次调用都建立新连接。对于 HTTP+SSE 模式，建议心跳间隔设为 30 秒，超时时间设为 60 秒。

// MCP 连接池配置
const mcpPoolConfig = {
  // 最大并发连接数
  maxConnections: 10,
  // 连接复用超时（毫秒）
  connectionTTL: 300000, // 5分钟
  // 单 Server 最大并发请求数（避免过载）
  maxConcurrentRequestsPerServer: 5,
  // 请求超时
  requestTimeout: 30000,
  // 重试策略
  retry: {
    maxAttempts: 3,
    backoffMs: [100, 500, 2000],
    retryableErrors: ["TIMEOUT", "SERVER_BUSY", "RATE_LIMITED"],
  },
};

上下文压缩：当 MCP Resource 返回大量数据时（如完整的 GitHub 仓库文件列表），应在 Server 端实现流式返回或在 Client 端做智能截断。

# MCP Server 端：智能截断大量数据
@server.read_resource()
async def read_resource(uri: str, max_tokens: int = 4000):
    content = await fetch_full_content(uri)

    # 估算 token 数量（粗略：中文 ~2字符/token，英文 ~0.75字符/token）
    estimated_tokens = estimate_tokens(content)

    if estimated_tokens > max_tokens:
        # 返回压缩版本 + 摘要
        summary = await summarize(content, max_tokens // 2)
        return TextContent(
            type="text",
            text=f"[内容已截断，原文约 {estimated_tokens} tokens]\n\n摘要:\n{summary}\n\n完整内容请通过文件工具查看。"
        )
    return TextContent(type="text", text=content)

8.2 A2A 性能优化

任务优先级队列：当 A2A Server 同时收到多个任务时，高优先级任务（如 P0 故障处理）应优先调度：

import heapq

class PriorityTaskQueue:
    def __init__(self):
        self.heap: list[tuple[int, int, Task]] = []  # (priority, counter, task)
        self.counter = 0
        self.lock = asyncio.Lock()

    async def enqueue(self, task: Task, priority: int = 50):
        """priority: 0-100，100 最高"""
        async with self.lock:
            heapq.heappush(self.heap, (priority, self.counter, task))
            self.counter += 1

    async def dequeue(self) -> Task | None:
        async with self.lock:
            if not self.heap:
                return None
            return heapq.heappop(self.heap)[2]

    async def reprioritize(self, task_id: str, new_priority: int):
        """动态调整优先级（如工单紧急度变更）"""
        async with self.lock:
            for i, (p, c, t) in enumerate(self.heap):
                if t.id == task_id:
                    self.heap[i] = (new_priority, c, t)
                    heapq.heapify(self.heap)
                    break

流式结果缓存：对于重复的 A2A 查询（如同一个代码仓库的多次审查请求），可以在 Server 端实现结果缓存：

from functools import lru_cache
import hashlib

# 基于输入 hash 的轻量缓存（30秒 TTL）
class CachedA2AServer(A2AServer):
    def __init__(self):
        super().__init__()
        self._cache: dict[str, tuple[float, Any]] = {}
        self._cache_ttl = 30.0  # 秒

    async def handle_task(self, task: Task):
        # 生成缓存 key（基于关键输入字段）
        cache_key = hashlib.sha256(
            f"{task.input['repo']}:{task.input['pr_number']}:{task.input.get('commit_sha', 'latest')}"
            .encode()
        ).hexdigest()

        now = asyncio.get_event_loop().time()

        if cache_key in self._cache:
            cached_time, cached_result = self._cache[cache_key]
            if now - cached_time < self._cache_ttl:
                # 缓存命中，直接返回（但仍更新 A2A 任务状态）
                await self.send_task_result(task.id, cached_result)
                return

        result = await self._do_review(task)

        # 更新缓存
        self._cache[cache_key] = (now, result)
        await self.send_task_result(task.id, result)

九、未来展望：协议演进与生态预测

9.1 MCP 的下一步：MCP v1.1 和 MCP Registry

截至 2026 年中，MCP 正在推进几个关键演进方向：

• MCP Registry：类似 npm/PyPI 的官方 Server 注册表，解决"MCP Server 发现"问题
• MCP Proxy：企业可以部署统一的 MCP 代理，统一管理认证、审计、流量控制
• Streaming Tools：支持工具调用的流式响应（适合大文件处理、LLM 流式输出等场景）
• MCP Over gRPC：可选的 gRPC 传输层，提供更高的性能和双向流支持

9.2 A2A 的下一步：联邦 Agent 发现

A2A 协议目前依赖中心化的 Agent Card 注册表。未来的演进方向包括：

• 联邦发现：类似 DNS 的分布式 Agent 发现机制
• 协议桥接：A2A ↔ MCP 的标准化桥接层
• 多模态任务：支持图像、视频、3D 模型作为任务输入和输出
• 成本计量：标准化的 Agent 资源消耗计量接口（Token 成本、计算时间等）

9.3 两个协议会合并吗？

短期内不会。MCP 和 A2A 的问题域正交，强行合并只会增加协议的复杂度。但可以预见的是：

• 会有越来越多的"协议桥接层"出现，让 Agent 开发者不需要关心底层细节
• Agent 编排框架（LangGraph、AutoGen、CrewAI）会内置对两个协议的支持
• 企业 Gateway 产品会把两者封装成统一的管理平面

十、总结：工程师的选型决策树

最后，给你一个实际可操作的决策框架：

遇到新的工具/数据源集成需求
         │
         ▼
   MCP Server 是否已存在？
    ├─ 是 → 直接接入，零开发
    └─ 否 → 评估开发成本
              ├─ 工具简单（CRUD） → 自己写 MCP Server
              └─ 工具复杂（多步骤） → 考虑 MCP Server vs REST API 的取舍

遇到多 Agent 协作需求
         │
         ▼
   是否需要跨框架互操作？
    ├─ 否（团队内同一框架） → 框架自带编排机制即可
    └─ 是（跨框架/跨厂商） → 评估 A2A 的收益 vs 引入的复杂度

最终架构推荐：
  单 Agent 工具集 → MCP
  多 Agent 同框架协作 → 框架内置机制
  多 Agent 跨框架协作 → A2A
  企业级混合场景 → Gateway（统一管理 MCP + A2A）

回到文章开头的核心问题：

MCP 解决的是"Agent 怎么用工具"——它让 AI 从"什么都做不了"变成"能调用一切"。

A2A 解决的是"Agent 怎么找队友"——它让 Agent 从"单兵作战"变成"协同网络"。

两者不是竞争关系，而是 AI Agent 架构体系中两条不可或缺的支柱。2026 年，如果你正在构建 AI Agent 系统，理解这两个协议的区别和联系，已经不是"加分项"而是"必备知识"。

MCP 让你知道工具在哪里，A2A 让你知道队友在哪里。只有两者结合，AI Agent 才能真正从 Demo 走向生产，从单点智能走向群体智能。

本文涉及的协议规范基于 MCP v1.0 和 A2A Protocol 2026 年中版本。协议持续演进中，建议在实际使用时查阅最新规范：

• MCP 规范：modelcontextprotocol.io
• A2A 规范：a2aprotocol.org