引言：为什么需要 MCP？
在 AGI 智能体生态高速发展的今天，一个尴尬的问题始终困扰着开发者：每接入一个新工具或数据源，就需要为特定的 LLM 平台编写定制化的适配代码。Claude 有 Function Calling，OpenAI 有 Tools API，Gemini 有 Function Declarations——协议碎片化严重制约了智能体的互操作性。
2024 年底，Anthropic 发布了 Model Context Protocol（MCP），一个开放标准协议，旨在统一 LLM 与外部工具、数据源之间的通信接口。正如 USB-C 统一了物理接口那样，MCP 试图统一智能体的"能力插拔"层。
本文将从协议架构、消息格式、传输层到代码实现，全面拆解 MCP 的核心机制。
MCP 架构总览
MCP 采用经典的 Client-Server 架构，但在其上引入了三个关键抽象：
┌─────────────────┐ JSON-RPC 2.0 ┌─────────────────┐
│ MCP Client │ ◄──────────────────► │ MCP Server │
│ (LLM Host App) │ stdio / SSE / WS │ (Tool Provider) │
└────────┬────────┘ └────────┬────────┘
│ │
▼ ▼
┌───────────┐ ┌───────────────┐
│ LLM Engine│ │ External Tools│
│ (Claude等)│ │ (DB/API/File) │
└───────────┘ └───────────────┘
三大核心原语
MCP 定义了三种核心原语（Primitives），所有功能都围绕它们构建：
- Tools（工具）：可被 LLM 调用的函数，类似于 Function Calling。例如查询数据库、调用 API、执行代码。
- Resources（资源）：可被 LLM 读取的数据源，以 URI 标识。例如文件内容、数据库记录、日志流。
- Prompts（提示模板）：预定义的提示词模板，支持参数化注入。例如代码审查模板、SQL 生成模板。
通信协议：JSON-RPC 2.0
MCP 选择 JSON-RPC 2.0 作为消息格式，这是一个轻量级、语言无关的远程过程调用协议。核心消息类型包括：
// 请求（Client → Server）
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "query_database",
"arguments": { "sql": "SELECT * FROM users LIMIT 10" }
}
}
// 响应（Server → Client）
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{ "type": "text", "text": "[{...10 rows...}]" }
]
}
}
传输层
MCP 支持两种传输方式：
| 传输方式 | 场景 | 特点 |
|---|---|---|
| stdio | 本地进程通信 | 零配置、低延迟，适合本地工具 |
| HTTP+SSE | 远程服务通信 | 支持认证、跨网络，适合云端服务 |
在 2026 年的最新规范中，MCP 还引入了 Streamable HTTP 传输模式，替代了原有的 SSE 长连接方式，支持更灵活的流式通信。
生命周期与握手流程
MCP 连接的建立遵循严格的生命周期管理：
Client Server
│ │
│──── initialize ───────────────►│
│◄─── initialize result ────────│
│──── initialized ──────────────►│
│ │
│ (正常通信阶段) │
│──── tools/list ───────────────►│
│◄─── tools/list result ────────│
│──── tools/call ───────────────►│
│◄─── tools/call result ────────│
│ │
│──── shutdown ─────────────────►│
│◄─── shutdown result ──────────│
│ │
握手阶段，Client 会发送自己的协议版本和能力声明：
{
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolVersion": "2025-06-18",
"capabilities": {
"sampling": {},
"roots": { "listChanged": true }
},
"clientInfo": { "name": "my-agent", "version": "1.0.0" }
}
}
Server 则回应自己支持的协议版本和能力：
{
"jsonrpc": "2.0",
"id": 0,
"result": {
"protocolVersion": "2025-06-18",
"capabilities": {
"tools": { "listChanged": true },
"resources": { "subscribe": true },
"prompts": { "listChanged": true }
},
"serverInfo": { "name": "my-tools-server", "version": "2.1.0" }
}
}
从零实现一个 MCP Server
下面用 Python 实现一个提供代码执行工具的 MCP Server：
import json
import subprocess
from mcp.server import Server
from mcp.types import Tool, TextContent
server = Server("code-executor")
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="execute_python",
description="在沙箱中执行 Python 代码并返回输出",
inputSchema={
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "要执行的 Python 代码"
},
"timeout": {
"type": "integer",
"description": "超时时间（秒）",
"default": 10
}
},
"required": ["code"]
}
),
Tool(
name="execute_shell",
description="执行 Shell 命令并返回输出",
inputSchema={
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "要执行的 Shell 命令"
}
},
"required": ["command"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "execute_python":
code = arguments["code"]
timeout = arguments.get("timeout", 10)
try:
result = subprocess.run(
["python3", "-c", code],
capture_output=True,
text=True,
timeout=timeout
)
output = result.stdout
if result.stderr:
output += f"\n[STDERR]\n{result.stderr}"
return [TextContent(type="text", text=output)]
except subprocess.TimeoutExpired:
return [TextContent(type="text", text=f"执行超时（{timeout}s）")]
elif name == "execute_shell":
command = arguments["command"]
result = subprocess.run(
command, shell=True, capture_output=True, text=True, timeout=30
)
return [TextContent(type="text", text=result.stdout + result.stderr)]
raise ValueError(f"未知工具: {name}")
if __name__ == "__main__":
import asyncio
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
asyncio.run(main())
实现 MCP Client
Client 端的核心职责是连接 Server、发现工具、并将工具调用结果桥接给 LLM：
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def connect_to_server():
# 配置 Server 启动参数
server_params = StdioServerParameters(
command="python3",
args=["mcp_server.py"],
env={"PYTHONPATH": "/opt/tools"}
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
# 发现可用工具
tools_result = await session.list_tools()
tools = tools_result.tools
print(f"发现 {len(tools)} 个工具:")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
# 调用工具
result = await session.call_tool(
"execute_python",
{"code": "print(sum(range(100)))"}
)
print(f"结果: {result.content[0].text}")
import asyncio
asyncio.run(connect_to_server())
与 LLM 的集成模式
MCP Client 通常嵌入在 LLM Host Application 中。以与 Claude 集成为例：
import anthropic
from mcp import ClientSession
async def agent_loop(session: ClientSession, user_query: str):
client = anthropic.Anthropic()
# 1. 获取 MCP 工具列表
tools_result = await session.list_tools()
mcp_tools = [
{
"name": t.name,
"description": t.description,
"input_schema": t.inputSchema
}
for t in tools_result.tools
]
# 2. 将工具信息发送给 LLM
messages = [{"role": "user", "content": user_query}]
while True:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=mcp_tools,
messages=messages
)
# 3. 检查是否有工具调用
if response.stop_reason == "tool_use":
tool_calls = [
b for b in response.content if b.type == "tool_use"
]
# 4. 通过 MCP 执行工具调用
for tc in tool_calls:
mcp_result = await session.call_tool(
tc.name, tc.input
)
tool_result = mcp_result.content[0].text
messages.append({"role": "assistant", "content": response.content})
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tc.id,
"content": tool_result
}]
})
else:
# 5. 返回最终回答
return response.content[0].text
MCP 生态现状（2026 年中）
截至 2026 年 6 月，MCP 生态已初具规模：
- 官方 SDK：Python、TypeScript、Go、Rust、Java 五种语言
- 社区 Server 数量：超过 3,000 个公开可用的 MCP Server
- 集成平台：Claude Desktop、Cursor、Zed、Windsurf、VS Code 等主流 IDE 均已支持
- 企业采用：包括数据库（PostgreSQL、MongoDB）、云平台（AWS、Cloudflare）、开发工具（GitHub、GitLab）等
MCP 相比传统 Function Calling 的优势
| 维度 | Function Calling | MCP |
|---|---|---|
| 标准化 | 各厂商私有格式 | 开放标准协议 |
| 工具发现 | 硬编码在 prompt 中 | 动态发现（list_tools） |
| 传输方式 | API 内嵌 | 独立传输层（stdio/HTTP） |
| 复用性 | 与特定 LLM 绑定 | 跨 LLM 复用 |
| 状态管理 | 无状态 | 支持有状态会话 |
常见陷阱与最佳实践
1. 工具 Schema 设计
陷阱：Schema 描述模糊，导致 LLM 错误调用。
# ❌ 不好的做法
Tool(
name="search",
description="搜索",
inputSchema={"type": "object", "properties": {"q": {"type": "string"}}}
)
# ✅ 好的做法
Tool(
name="search_web",
description="在互联网上搜索给定关键词，返回相关网页标题和摘要",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，支持自然语言"
},
"max_results": {
"type": "integer",
"description": "返回结果数量上限（1-20）",
"default": 5
}
},
"required": ["query"]
}
)
2. 错误处理
MCP Server 应返回结构化的错误信息，而非抛出异常：
@server.call_tool()
async def call_tool(name: str, arguments: dict):
try:
result = await execute(name, arguments)
return [TextContent(type="text", text=json.dumps(result))]
except ValidationError as e:
return [TextContent(type="text", text=f"[参数错误] {e}")]
except TimeoutError:
return [TextContent(type="text", text="[超时] 请求处理超时，请重试")]
except Exception as e:
return [TextContent(type="text", text=f"[错误] {type(e).__name__}: {e}")]
3. 安全边界
- 沙箱隔离：代码执行类工具必须在容器或沙箱中运行
- 权限控制：通过 MCP 的
roots机制限制文件系统访问范围 - 审计日志：记录所有工具调用，便于事后追溯
展望
MCP 正在向以下方向演进：
- Agent-to-Agent 通信：基于 MCP 扩展的智能体间协作协议
- 流式工具调用：支持 Streaming Response，适配长任务场景
- OAuth 2.1 集成：标准化远程 MCP Server 的认证授权流程
- MCP Registry：去中心化的 Server 发现与注册机制
MCP 的愿景是让智能体的能力扩展像安装浏览器插件一样简单。随着协议规范的持续迭代和生态的繁荣，MCP 正在成为 AGI 时代的基础设施级标准。
总结
MCP 通过定义统一的协议层，解决了 LLM 与外部世界交互的碎片化问题。其 JSON-RPC 2.0 消息格式、Client-Server 架构、三大核心原语（Tools/Resources/Prompts）的设计简洁而有力。对于智能体开发者而言，掌握 MCP 不仅意味着减少重复工作，更意味着融入一个正在快速成长的开放生态。
在 AGI 的技术栈中，MCP 之于智能体，正如 HTTP 之于 Web——它不是某个厂商的私产，而是整个行业的公共财富。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。