适用场景与目标

适用场景：

• 单智能体需要调用多种外部工具（GitHub API、数据库、Slack 等）
• 多智能体协作流水线（如研究 Agent → 规划 Agent → 执行 Agent）
• 企业内部 AI 系统对接已有业务 API，需要统一接入标准
• AI 应用需要动态发现和组合能力，而非硬编码工具调用

目标： 通过 MCP（Model Context Protocol）实现智能体 → 工具的标准化连接，通过 A2A（Agent-to-Agent）实现智能体 → 智能体的协作编排，掌握两者结合的落地路径。

---

最小可行方案（MVP）步骤

工具准备

# Node.js MCP SDK（TypeScript/Node 生态最成熟）
npm install @modelcontextprotocol/sdk

# Python MCP SDK
pip install mcp

# A2A Python SDK（Linux Foundation）
pip install a2a-sdk

# 可选：已有 MCP Server 快速上手
# npx @modelcontextprotocol/server-github

Step 1：用 MCP 构建一个自定义工具 Server

MCP 的核心思路：把任意 API 封装成 LLM 可理解的工具。

// mcp-server-example.ts
import { McpServer, StdioServerTransport } from "@modelcontextprotocol/sdk/server";
import { z } from "zod";

const server = new McpServer({
  name: "company-tools",
  version: "1.0.0",
});

// 注册一个"查询库存"工具
server.tool(
  "query_inventory",
  "查询指定商品的当前库存",
  {
    product_id: z.string().describe("商品SKU编号"),
    warehouse: z.enum(["BJ", "SH", "GZ"]).optional().describe("仓库代码，默认全部"),
  },
  async ({ product_id, warehouse }) => {
    // 这里调用真实业务 API
    const result = await callInventoryAPI(product_id, warehouse);
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(result, null, 2),
        },
      ],
    };
  }
);

const transport = new StdioServerTransport();
await server.run(transport);

运行方式：

npx ts-node mcp-server-example.ts
# Server 通过标准输入/输出与 MCP Client 通信

Step 2：配置 MCP Client（以 LangChain 为例）

# python client 示例
from mcp import ClientSession, StdioServerParameters
from langchain_mcp.tools import MCPToolKit

server_params = StdioServerParameters(
    command="npx",
    args=["ts-node", "mcp-server-example.ts"],
)

async def main():
    async with ClientSession(server_params) as session:
        await session.initialize()
        toolkit = MCPToolKit(session)
        tools = toolkit.get_tools()
        # tools 现在可以被 LLM 直接调用了

Step 3：搭建 A2A 多智能体编排层

A2A 让多个 Agent 可以互相发现和通信，而不是硬编码调用链。

# a2a_orchestrator.py
from a2a import A2AServer, A2AClient, AgentCard
from a2a.server.apps import A2AStarletteApplication
from starlette.applications import Starlette
import uvicorn

# 定义研究 Agent
class ResearchAgent(A2AServer):
    async def handle(self, task):
        query = task.input["query"]
        # 调用 MCP 工具收集信息
        context = await mcp_tools.search_web(query)
        return {"context": context, "status": "done"}

# Agent Card：元数据声明，供其他 Agent 发现
research_agent_card = AgentCard(
    name="research-agent",
    description="研究型 Agent，负责信息检索与整理",
    skills=["web_search", "data_analysis"],
    url="http://localhost:8001",
)

# 编排层：根据任务类型路由到不同 Agent
class Orchestrator(A2AServer):
    async def handle(self, task):
        task_type = classify(task.input)
        if task_type == "research":
            # 通过 A2A 调用 Research Agent
            client = A2AClient("http://localhost:8001")
            result = await client.send(task.input)
        elif task_type == "code_review":
            client = A2AClient("http://localhost:8002")
            result = await client.send(task.input)
        return result

app = A2AStarletteApplication(
    agent=Orchestrator(),
    agent_cards=[research_agent_card],
).app()

uvicorn.run(app, host="0.0.0.0", port=8000)

Step 4：用 Agent Card 实现动态发现

A2A 的核心价值在于无需硬编码就能让 Agent 发现彼此：

// agent-card.json（每个 Agent 的自我介绍）
{
  "name": "code-review-agent",
  "description": "代码审查 Agent，发现安全漏洞和性能问题",
  "skills": ["security_scan", "performance_audit"],
  "url": "http://localhost:8002",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  }
}

---

关键实现细节

MCP 工具注册与发现机制

MCP 使用 JSON-RPC 2.0 协议，工具通过 tools/list 和 tools/call 两个核心接口暴露：

// Client 发送 tools/list 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

// Server 返回可用工具列表
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "query_inventory",
        "description": "查询指定商品的当前库存",
        "inputSchema": {
          "type": "object",
          "properties": {
            "product_id": { "type": "string" }
          }
        }
      }
    ]
  }
}

MCP 的核心优势：

• 零硬编码：只要 MCP Server 在运行，LLM 就能通过统一协议发现和调用工具
• 生态丰富：截至 2026 年 4 月已有 5,800+ 社区 MCP Server（GitHub、Slack、Postgres 等）

A2A 任务状态机

A2A 每个任务都有明确的状态流转，落地时要注意处理：

pending → working → input_required → completed/failed

# 正确处理 A2A 任务状态
from a2a.types import TaskStatus, TaskState

async def handle_task(task):
    if task.status.state == TaskState.INPUT_REQUIRED:
        # 需要人工介入或补充信息
        await send_for_human_review(task)
    elif task.status.state == TaskState.WORKING:
        # 正在处理，可以推送中间进度
        await send_progress_update(task, progress=0.5)

MCP + A2A 组合架构（推荐模式）

用户请求
    ↓
Orchestrator Agent（A2A Hub）
    ├── Research Agent ←──→ MCP: GitHub API / Web Search
    ├── Code Agent    ←──→ MCP: CI System / Code Review Tools
    └── Report Agent  ←──→ MCP: Docs API / Email

为什么不只用 MCP？
MCP 适合”1 个 Agent + N 个工具”的场景；一旦需要多 Agent 协作、动态任务分发、跨 Agent 结果汇总，就必须引入 A2A。

为什么不只用 A2A？
A2A 不定义工具调用规范，每个 Agent 如果自己实现工具访问，成本高且不统一。MCP 负责”工具层”，A2A 负责”协作层”，各司其职。

---

常见坑与规避清单

坑 1：MCP Server 启动超时 / 进程僵死

问题：MCP Server 通过 stdio 通信，如果 Server 端 API 调用慢，Client 会长时间阻塞。

规避：

• 所有外部 API 调用设置超时（建议 10s 以内）
• 使用 server.timeout_ms = 10000 配置
• 生产环境用 HTTP MCP Server（如 FastMCP）替代 stdio 模式

import { HttpServerTransport } from "@modelcontextprotocol/sdk/server-http";

const transport = new HttpServerTransport("/mcp", 3001);
await server.run(transport);

坑 2：A2A Agent 之间上下文丢失

问题：A2A 任务跨 Agent 传递时，上下文需要手动序列化传递，否则每个 Agent 都从零开始。

规避：

• 在 Orchestrator 层维护统一的上下文传递
• 使用 TaskContext 携带历史中间结果

# 在 Orchestrator 中显式传递上下文
async def route_to_agent(self, target_url, original_task):
    enriched_task = {
        **original_task,
        "context": original_task.get("context", {}),  # 保留上游结果
        "history": original_task.get("history", []) + [original_task]
    }
    client = A2AClient(target_url)
    return await client.send(enriched_task)

坑 3：MCP 工具 Schema 与 LLM 理解不匹配

问题：工具描述写得模糊，LLM 频繁调用错误参数。

规避：

• description 字段要具体说明”何时使用”和”返回什么”
• 使用 Zod（TS）或 Pydantic（Python）强制 Schema，减少 LLM 猜测空间

// ❌ 差：描述模糊
{ name: "search", description: "搜索" }

// ✅ 好：描述具体
{
  name: "search_inventory",
  description: "当用户询问商品库存、余量、补货时使用。返回 JSON 含 product_id、quantity、warehouse。",
  inputSchema: z.object({
    product_id: z.string().describe("必须是 SKU 格式，如 'SKU-2024-001'"),
    warehouse: z.enum(["BJ", "SH", "GZ"]).describe("不填默认查全仓库")
  })
}

坑 4：A2A 循环依赖导致死锁

问题：Agent A 调用 Agent B，Agent B 又回调 Agent A，形成死循环。

规避：

• 任务对象中加入 trace_id 和 depth 字段，禁止超过深度阈值（如 5 层）
• 关键路径加分布式锁

if task.get("depth", 0) > 5:
    raise RecursionError("Agent 递归深度超限，强制终止")

坑 5：忽视 MCP Server 的幂等性

问题：MCP 工具没有幂等设计时，LLM 重试会导致重复操作（如重复下单）。

规避：

• 工具实现要支持幂等（如通过 idempotency_key 防重）
• 在 MCP Server 层对写操作做重复检测

坑 6：生产环境 MCP Server 安全风险

问题：MCP 工具拥有业务系统权限，一旦被提示词注入绕过，可能导致未授权操作。

规避：

• MCP Server 实施最小权限原则，每个 Server 只暴露必要工具
• 关键操作（删除、支付等）走独立鉴权，不通过 MCP 直接暴露
• 对所有 MCP 工具调用做操作审计日志

---

成本 / 性能 / 维护权衡

成本

方案	MCP Server 数量	基础设施成本	开发成本
纯 HTTP + 同步调用	1-3 个集中式	低（单台 2C4G 足够）	中
分布式的 MCP + A2A	5-10 个微服务	中（需服务发现）	高
全自动多 Agent 流水线	10+ Agent	高（需要消息队列）	极高

成本优化建议：初期从单 MCP Server + 2 个 A2A Agent 开始，使用 stdio 模式本地运行；验证后再迁移到 HTTP 模式做水平扩展。

性能

• MCP stdio 模式：延迟低（进程内通信），但无法跨机器扩展
• MCP HTTP 模式：延迟略高（~20-50ms），支持分布式
• A2A 通信：每个 Agent 间有 ~50-200ms 的序列化/反序列化开销

关键瓶颈：Orchestrator Agent 是单点，所有任务都经过它。需要时可以做 Agent 池 + 负载均衡。

维护

• MCP Server 升级时需要同步更新 Client 侧的 Schema 定义（使用版本号管理）
• A2A Agent 发现依赖 Agent Card 的健康状态，需要实现心跳机制
• 推荐用 服务网格（Istio/Linkerd） 管理 MCP/A2A 服务的流量和可观测性

---

一周内可执行行动清单

Day 1-2：搭建 MCP 最小链路

• 在本地运行一个官方 MCP Server（如 GitHub MCP Server）
• 用 LangChain 或 OpenAI SDK 连接，验证”LLM → 工具调用”链路跑通
• 写一个最简单的自定义 MCP Server（暴露 1-2 个内部 API）

Day 3-4：接入 A2A 编排层

• 部署两个 A2A Agent（研究 Agent + 执行 Agent）
• 实现 Orchestrator 层，根据任务类型路由
• 验证 Agent 间上下文传递是否正确

Day 5：生产风险加固

• 给所有 MCP 工具添加超时和幂等性保护
• 实现 MCP 调用审计日志
• 给 A2A 任务加深度限制，防止循环调用
• 验证 Agent Card 服务发现是否正常

Day 6-7：性能验证与扩展

• 压力测试：10 并发任务下 MCP 调用延迟
• 从 stdio 模式迁移到 HTTP MCP Server
• 准备技术文档，记录所有已注册的 MCP 工具清单和 Schema

---

一句话总结：MCP 解决的是”智能体怎么说清楚工具”的问题，A2A 解决的是”智能体怎么协作”的问题。两者结合（工具层 MCP + 协作层 A2A）是 2026 年企业级 AI Agent 系统的事实标准，值得现在就开始落地。