9.1 MCP 与 A2A：Agent 生态的两大标准协议
MCP 解决 Agent 与工具/数据的连接（纵向），A2A 解决 Agent 之间的发现与协作（横向）。两者互补，构成 Agent 生态的基础设施层。
MCP（Model Context Protocol）
协议概述
MCP 是 Anthropic 于 2024 年 11 月发起的开放标准协议，2025 年 12 月移交 Linux 基金会治理，成为中立的行业标准。
核心定位：AI 世界的「USB-C 接口」
传统方式：每个 AI 应用单独对接每个工具
App A ──┬── API 1
├── API 2
└── API 3
App B ──┬── API 1 （重复开发）
├── API 2
└── API 3
MCP 方式：工具写一次 Server，所有 AI 应用通用
App A ──┐
App B ──┤── MCP ──┬── Server 1
App C ──┘ ├── Server 2
└── Server 3架构详解
┌─────────────────────────────────────────────────────┐
│ Host（AI 应用：Claude Desktop / Cursor / 自定义 Agent）│
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ MCP Client 1 │ │ MCP Client 2 │ ... │
│ └──────┬───────┘ └──────┬───────┘ │
└─────────┼─────────────────┼──────────────────────────┘
│ │
stdio/SSE stdio/SSE
│ │
┌──────▼───────┐ ┌─────▼────────┐
│ MCP Server A │ │ MCP Server B │
│ (数据库) │ │ (GitHub) │
└──────────────┘ └──────────────┘三个核心角色：
| 角色 | 说明 | 示例 |
|---|---|---|
| Host | AI 应用，管理多个 Client | Claude Desktop、Cursor、你的 Agent |
| Client | Host 中与 Server 通信的组件 | 每个 Server 对应一个 Client |
| Server | 提供工具/资源/Prompt 的外部服务 | 数据库 Server、GitHub Server |
两种传输方式：
| 传输 | 适用场景 | 特点 |
|---|---|---|
stdio | 本地进程 | Server 作为子进程启动，通过标准输入/输出通信 |
SSE | 远程 HTTP | Server 独立部署，通过 HTTP + Server-Sent Events 通信 |
Server 开发实战
python
# pip install mcp
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-tools")
# === 1. 暴露工具（Tools）===
@mcp.tool()
async def query_database(sql: str) -> str:
"""执行 SQL 查询并返回结果
Args:
sql: SELECT 查询语句（仅支持只读查询）
"""
# 安全检查
if not sql.strip().upper().startswith("SELECT"):
return "错误：仅支持 SELECT 查询"
# 执行查询（示例）
import asyncpg
conn = await asyncpg.connect("postgresql://user:pass@localhost/mydb")
rows = await conn.fetch(sql)
await conn.close()
return str([dict(r) for r in rows])
@mcp.tool()
async def search_web(query: str, max_results: int = 5) -> str:
"""搜索互联网并返回结果摘要
Args:
query: 搜索关键词
max_results: 最大返回结果数
"""
import httpx
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.search.example.com/search",
params={"q": query, "limit": max_results}
)
return resp.text
# === 2. 暴露资源（Resources）===
@mcp.resource("docs://api-reference")
async def get_api_docs() -> str:
"""返回 API 参考文档"""
return open("docs/api.md").read()
@mcp.resource("config://app-settings")
async def get_config() -> str:
"""返回当前应用配置"""
import json
return json.dumps({"model": "gpt-4o", "temperature": 0.7})
# === 3. 暴露 Prompt 模板（Prompts）===
@mcp.prompt()
async def code_review_prompt(code: str, language: str = "python") -> str:
"""生成代码审查 Prompt"""
return f"""请审查以下 {language} 代码，关注：
1. 安全漏洞
2. 性能问题
3. 代码风格
```{language}
{code}
```"""
if __name__ == "__main__":
mcp.run() # stdio 模式
# mcp.run(transport="sse", port=8080) # SSE 模式在 Claude Desktop / Cursor 中使用
json
// Claude Desktop: ~/Library/Application Support/Claude/claude_desktop_config.json
// Cursor: .cursor/mcp.json
{
"mcpServers": {
"my-tools": {
"command": "python",
"args": ["/path/to/my_server.py"]
},
"postgres": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres",
"postgresql://user:pass@localhost/mydb"]
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": { "GITHUB_TOKEN": "ghp_xxxxx" }
}
}
}在自己的 Agent 中使用 MCP Client
python
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def agent_with_mcp():
"""Agent 通过 MCP Client 发现并调用工具"""
server_params = StdioServerParameters(
command="python",
args=["my_server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 1. 发现可用工具
tools = await session.list_tools()
print(f"可用工具: {[t.name for t in tools.tools]}")
# → ['query_database', 'search_web']
# 2. 将工具转为 OpenAI Function Calling 格式
openai_tools = []
for tool in tools.tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
# 3. LLM 决定调用哪个工具
from openai import AsyncOpenAI
client = AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "查询用户表有多少条记录"}],
tools=openai_tools
)
# 4. 通过 MCP 执行工具调用
tool_call = response.choices[0].message.tool_calls[0]
result = await session.call_tool(
tool_call.function.name,
arguments=json.loads(tool_call.function.arguments)
)
print(f"结果: {result.content[0].text}")2026 MCP 生态现状
已支持 MCP 的 AI 应用（Host）：
✅ Claude Desktop / Claude Code
✅ Cursor / Windsurf / Cline
✅ GitHub Copilot（预览）
✅ 自定义 Agent（Python/Node SDK）
常用社区 MCP Server：
📊 @modelcontextprotocol/server-postgres — PostgreSQL 查询
📁 @modelcontextprotocol/server-filesystem — 文件系统操作
🔍 @modelcontextprotocol/server-brave-search — Brave 搜索
🐙 @modelcontextprotocol/server-github — GitHub API
📝 @modelcontextprotocol/server-memory — 持久化记忆
🔧 @modelcontextprotocol/server-puppeteer — 浏览器自动化A2A（Agent-to-Agent Protocol）
协议概述
A2A 是 Google Cloud 于 2025 年 4 月发起的开放协议，解决不同 AI Agent 之间的发现、通信和协作问题。
核心定位：Agent 之间的「HTTP 协议」
MCP 解决的问题（纵向）： A2A 解决的问题（横向）：
Agent → 工具/数据 Agent ↔ Agent
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Agent │ │ Agent A │◄──►│ Agent B │
│ │ │ (研究员) │ │ (写手) │
│ MCP ↓ │ └────┬────┘ └────┬────┘
│ │ │ │
│ ┌─────┐ │ ┌────▼────┐ ┌────▼────┐
│ │Tool │ │ │ Agent C │◄──►│ Agent D │
│ └─────┘ │ │ (审核员) │ │ (设计师) │
└─────────┘ └─────────┘ └─────────┘核心概念
1. Agent Card（能力声明）
每个 Agent 通过 JSON 格式声明自己的能力，供其他 Agent 发现：
json
{
"name": "research-agent",
"description": "擅长搜索互联网、阅读论文并生成研究摘要",
"url": "https://research-agent.example.com",
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"skills": [
{
"id": "web-research",
"name": "互联网研究",
"description": "搜索并整理指定主题的最新信息"
},
{
"id": "paper-summary",
"name": "论文摘要",
"description": "阅读学术论文并生成结构化摘要"
}
]
}2. Task（任务协作）
Agent 之间通过标准化的 Task 对象协作：
python
# Agent A 委派任务给 Agent B
task = {
"id": "task-001",
"status": "submitted", # submitted → working → completed
"message": {
"role": "user",
"parts": [{"text": "请研究 MCP 协议的最新进展"}]
}
}
# Agent B 完成后返回
task_result = {
"id": "task-001",
"status": "completed",
"artifacts": [
{
"name": "research_report",
"parts": [{"text": "## MCP 最新进展\n\n1. Linux 基金会接管治理..."}]
}
]
}3. 跨框架协作
A2A 的核心价值——不同框架构建的 Agent 可以互操作：
┌────────────────┐ A2A ┌────────────────┐
│ LangGraph Agent│◄─────────►│ CrewAI Agent │
│ (Python) │ │ (Python) │
└────────────────┘ └────────────────┘
▲ ▲
│ A2A │
▼ ▼
┌────────────────┐ ┌────────────────┐
│ AutoGen Agent │◄─────────►│ Custom Agent │
│ (Python) │ │ (Node.js) │
└────────────────┘ └────────────────┘MCP vs A2A：完整对比
| 维度 | MCP | A2A |
|---|---|---|
| 发起者 | Anthropic（2024.11） | Google Cloud（2025.04） |
| 治理 | Linux 基金会 | 开放标准（社区驱动） |
| 解决问题 | Agent → 工具/数据 | Agent ↔ Agent |
| 通信方向 | 纵向连接 | 横向协作 |
| 核心概念 | Server / Client / Tool | Agent Card / Task / Artifact |
| 传输方式 | stdio / SSE | HTTP / WebSocket |
| 成熟度 | ⭐⭐⭐⭐⭐ 生产就绪 | ⭐⭐⭐ 快速发展中 |
| 典型用途 | 连接数据库、搜索引擎、文件系统 | 多 Agent 任务分工、跨组织协作 |
何时用哪个？
你需要 Agent 访问数据库？ → MCP
你需要 Agent 调用搜索 API？ → MCP
你需要多个 Agent 分工合作？ → A2A
你需要跨团队 Agent 互操作？ → A2A
你需要 Agent 既用工具又协作？ → MCP + A2A（两者互补）生产级多 Agent 系统的完整架构
┌───────────────────────────────────────────────┐
│ Orchestrator Agent │
│ （编排者：分解任务、分配 Agent） │
└──────┬────────────┬────────────┬──────────────┘
│ A2A │ A2A │ A2A
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Research │ │ Writer │ │ Reviewer │
│ Agent │ │ Agent │ │ Agent │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ MCP │ MCP │ MCP
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ 搜索引擎 │ │ 文档存储 │ │ 代码分析 │
│ MCP Srv │ │ MCP Srv │ │ MCP Srv │
└─────────┘ └─────────┘ └─────────┘学习资源