MCP 协议深度实战:AI Agent 的万能连接器——从架构设计到生产级 Server 开发的全链路解析
前言
2024年11月,Anthropic 扔出了一颗"深水炸弹"——开源了 Model Context Protocol(MCP),并将其定义为"AI 世界的 USB-C 接口"。彼时 AI 圈正陷入一个令人窒息的困境:每个 AI 应用想要接入外部工具,都得从头写一遍对接代码。数据库有数据库的 SDK,文件系统有文件系统的接口,GitHub 有 GitHub 的 REST API,Slack 有 Slack 的 Bot 框架……每个工具都像一根独立的"充电线",而 AI 应用则像一部没有统一充电口的老款手机——你得随身带一堆转接头。
MCP 的出现,彻底改变了这个局面。它不是又一个"更好的 SDK",而是一套协议层抽象——只要工具方实现了 MCP Server,应用方就能用统一的 MCP Client 接入,不需要关心对方是数据库还是 Slack。这种"即插即用"的设计思路,在软件工程史上有着悠久的传统:USB 统一了物理接口,TCP/IP 统一了网络通信,而 MCP 正在统一 AI Agent 与外部世界的交互方式。
本文将深入剖析 MCP 的设计哲学、核心架构、协议细节,并通过完整的代码实战,展示如何从零构建一个生产级的 MCP Server。全文约12000字,建议配合代码边看边实践。
一、问题背景:为什么 AI 应用需要一个"通用插座"
1.1 当代 AI 应用的连接困境
在 MCP 出现之前,AI 应用对接外部工具的方式大致可以分为三类:
第一类:硬编码集成。 开发者直接在 AI 应用中为每个工具写死对接代码。代价是每当工具方更新 API,应用方就得跟着改。这种模式在工具数量少时还能接受,一旦需要接入十几个外部服务,代码就变成了一团乱麻。
第二类:Function Calling 规范。 GPT-4 带来了 Function Calling 能力,让 LLM 可以根据对话上下文决定调用哪些函数。这确实前进了一步,但问题是 Function Calling 只是一个"调用约定",它没有定义工具如何暴露自己的能力、如何描述输入输出schema、如何处理认证和权限。每个工具仍然需要自己实现这些基础设施。
第三类:Prompt Engineering 包装。 最原始的做法——在 System Prompt 里写一长串指令,告诉 AI 怎么用某个工具。这本质上是把工具协议"塞进"了 Prompt 里,不仅脆弱,而且无法利用类型系统和工具描述的可验证性。
举一个具体的痛点:假设你在开发一个 AI 编程助手,需要同时接入 GitHub(查代码)、Jira(看任务)、Slack(发消息)、PostgreSQL(查数据)。在 MCP 出现之前,这四个接入需要四套完全不同的代码:GitHub 要处理 OAuth + REST API,Jira 要处理 Basic Auth + REST API,Slack 要处理 Bot Token + WebSocket,PostgreSQL 要处理连接池 + SQL 解析。你的 AI 应用里90%的代码其实跟"AI"没关系,全是在写各种 API 的适配层。
1.2 MCP 的设计目标
MCP 的设计目标非常清晰:在 LLM 和外部工具之间,建立一个标准化的双向通信协议。
具体来说,它要解决三个核心问题:
能力发现(Capability Discovery): 工具方需要用一种标准格式描述自己有哪些能力、支持哪些操作、输入输出的 schema 是什么。AI 应用只需要解析这个描述,就能知道如何调用该工具,不需要预先写死代码。
标准化通信(Standardized Communication): 所有的工具调用都走统一的协议格式(基于 JSON-RPC 2.0),而不是每个工具自己定义的私有协议。协议支持两种传输模式:本地进程通信(stdio)和远程 HTTP 服务(SSE)。
认证与权限(Authentication & Permissions): MCP 定义了一套基于 OAuth 2.0 的认证机制,支持用户授权和作用域控制。AI 应用在调用工具时,协议会携带用户授权的令牌,工具方据此判断是否放行。
1.3 与 Function Calling 的关系
这里需要澄清一个常见误解:MCP 不是 Function Calling 的替代品,而是 Function Calling 的扩展和标准化。
Function Calling 定义的是"LLM 决定调用哪个函数",MCP 定义的是"这个函数如何被描述、如何被发现、如何被调用"。两者是正交的——你可以在 MCP Server 的 schema 中描述工具能力,然后在支持 MCP 的 AI 应用(如 Claude Desktop、Cursor IDE)中,让 LLM 通过 Function Calling 机制决定调用哪个 MCP 工具。
打个比方:Function Calling 是"你要去哪个目的地",MCP 是"这条路怎么走、收费站在哪里、加油站怎么用"。两者结合,AI 才能真正像人一样自然地使用工具。
二、架构解析:三层组件的协作模型
MCP 的架构可以用一句话概括:一个 Host、多个 Client、一个或多个 Server。
2.1 核心组件
┌──────────────────────────────────────────────────────┐
│ MCP Host │
│ (Claude Desktop / Cursor / 自研应用) │
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Client │ │ Client │ │ Client │ │
│ │ (GitHub)│ │ (Jira) │ │ (Slack) │ │
│ └───┬─────┘ └───┬─────┘ └───┬─────┘ │
└──────│───────────│───────────│──────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Server │ │ Server │ │ Server │
│(stdio) │ │ (HTTP) │ │ (HTTP) │
└─────────┘ └─────────┘ └─────────┘
2.1.1 MCP Host
MCP Host 是用户直接操作的 AI 应用,它是整个架构的协调中枢。常见的 Host 包括:
- Claude Desktop:Anthropic 官方桌面客户端,通过 MCP 接入本地文件和数据库
- Cursor IDE:AI 代码编辑器,通过 MCP 接入 GitHub、Jira、Figma 等工具
- VS Code 插件:微软的 Copilot 也在跟进 MCP 支持
- 自研应用:任何基于 Claude SDK 或 OpenAI SDK 构建的 AI 应用,只要集成了 MCP Client SDK
Host 的职责是:
- 维护多个 MCP Client 实例,每个 Client 对应一个 Server
- 向 LLM 提供工具列表(从各 Server 的 schema 中聚合)
- 处理 LLM 的工具调用请求,路由到对应的 Client
- 管理用户认证和授权流程
2.1.2 MCP Client
MCP Client 运行在 Host 内部,是 Host 与外部 Server 之间的代理。每个 Client 严格一一对应一个 Server。
Client 的职责是:
- 与对应的 Server 建立和维护长连接
- 接收 Host 转发的工具调用请求
- 处理认证令牌和请求签名
- 将 Server 的响应(结果或错误)返回给 Host
Client 和 Server 之间使用 JSON-RPC 2.0 通信,具体协议我们在第三节详细展开。
2.1.3 MCP Server
MCP Server 是外部工具的 MCP 接口层。它暴露工具的能力描述(schema),处理来自 Client 的调用请求,并返回结果。
Server 有两种运行模式:
stdio 模式(本地进程): Server 作为子进程运行,Client 通过标准输入/输出与其通信。适合本地工具,如文件系统、本地数据库、CLI 工具等。这种模式的优势是简单、安全——工具的代码完全在本地运行,不需要暴露网络端口。
HTTP + SSE 模式(远程服务): Server 作为 HTTP 服务运行,Client 通过 HTTP POST 请求发送调用,通过 Server-Sent Events(SSE)接收响应。适合需要远程访问的工具,如 GitHub API、Slack API 等。
2.2 通信协议:JSON-RPC 2.0 + 传输层适配
MCP 的应用层协议基于 JSON-RPC 2.0,这是一套成熟的无状态远程过程调用(RPC)规范。相比于 gRPC,它更轻量,不需要预编译 proto 文件,直接用 JSON 文本即可。
JSON-RPC 2.0 的请求格式:
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "github_search_repos",
"arguments": {
"query": "MCP protocol",
"language": "Python"
}
}
}
响应格式:
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "Found 42 repositories..."
}
]
}
}
错误格式:
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32600,
"message": "Invalid request: missing required parameter 'query'"
}
}
2.3 握手流程:从陌生到信任
MCP Client 和 Server 建联时,需要经历一个握手过程,确保双方使用相同版本的协议,并同步能力描述:
Step 1:Initialize(Client → Server)
Client 发送自己支持的协议版本和客户端信息:
{
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"clientInfo": {
"name": "claude-desktop",
"version": "1.0.0"
},
"capabilities": {}
}
}
Step 2:Initialized(Server → Client)
Server 响应自己的协议版本和提供的能力:
{
"jsonrpc": "2.0",
"id": 0,
"result": {
"protocolVersion": "2024-11-05",
"serverInfo": {
"name": "github-mcp-server",
"version": "1.2.0"
},
"capabilities": {
"tools": {}
}
}
}
Step 3:工具发现(Client → Server)
握手完成后,Client 可以随时向 Server 请求工具列表:
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
Server 返回:
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [
{
"name": "github_search_repos",
"description": "Search GitHub repositories by keyword",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
},
"language": {
"type": "string",
"description": "Filter by programming language"
}
},
"required": ["query"]
}
}
]
}
}
这个 schema 遵循 JSON Schema Draft-07 规范,LLM 可以直接从中理解每个工具的用途、参数类型和必填项,无需任何额外文档。
三、能力发现机制:工具的"自我介绍"
MCP 最有价值的设计之一,是它的自描述能力。Server 不需要预先与 Client 约定好"你知道有哪些工具",而是运行时动态暴露自己的能力。这带来了一个巨大的工程优势:Server 升级添加新工具时,Client 无需重新配置,只要重新调用一次 tools/list,就能看到新工具。
3.1 工具的 Schema 结构
每个工具在 MCP 中的描述由以下字段构成:
| 字段 | 类型 | 说明 |
|---|---|---|
name | string | 工具的唯一标识符,AI 调用时的函数名 |
description | string | 工具的自然语言描述,供 LLM 理解工具用途 |
inputSchema | object | JSON Schema,定义参数类型和约束 |
其中 inputSchema 是最关键的部分。一个设计良好的 schema 能让 LLM 准确理解如何调用工具:
{
"name": "database_query",
"description": "Execute a read-only SQL query against the analytics database. " +
"Only SELECT statements are allowed for security reasons.",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "SQL SELECT statement (no INSERT/UPDATE/DELETE/DROP)",
"pattern": "^\\s*SELECT\\s+"
},
"max_rows": {
"type": "integer",
"description": "Maximum number of rows to return",
"default": 100,
"minimum": 1,
"maximum": 1000
}
},
"required": ["query"]
}
}
通过 description 中的安全提示("Only SELECT")、pattern 约束(正则强制以 SELECT 开头)、max_rows 的范围限制,这个 schema 不仅让 LLM 知道怎么用,还隐性植入了安全约束。
3.2 多工具聚合:Host 的工具目录
当一个 Host 连接多个 Server 时,它会将所有 Server 返回的工具列表聚合在一起,生成一个统一的工具目录,提交给 LLM。这个过程对用户完全透明——LLM 看到的不是"GitHub 的 search_repos"、"Jira 的 get_issue",而是一组统一命名的工具列表,LLM 根据描述自行决定调用哪个。
这带来了一个有趣的工程问题:当两个 Server 提供了同名工具时怎么办? MCP 规范中,工具名在单一 Server 内唯一,但不同 Server 之间不保证不冲突。主流 Host 的做法是:聚合时给工具名加上 Server 前缀,如 github.search_repos vs gitlab.search_repos。
四、安全模型:OAuth 2.0 + 作用域控制
AI Agent 访问外部工具,涉及用户数据的读写,安全问题不容忽视。MCP 的安全模型建立在 OAuth 2.0 之上,实现了精细化的权限控制。
4.1 认证流程
MCP 支持 OAuth 2.0 的授权码模式(Authorization Code Flow),流程如下:
用户授权: 用户在 Host 中安装某个 Server 时,Host 会引导用户到 Server 的 OAuth 授权页面。用户登录并同意授权后,获得授权码。
令牌交换: Host 用授权码向 Server 的令牌端点换取访问令牌(Access Token)和刷新令牌(Refresh Token)。
请求携带令牌: 每次 Client 向 Server 发起工具调用时,在 HTTP Header 中携带 Bearer Token:
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...令牌刷新: 当访问令牌过期时,Host 用刷新令牌自动换取新令牌,用户无需重新授权。
4.2 作用域(Scopes)
OAuth 2.0 的作用域机制在 MCP 中被充分利用。每个 Server 可以定义多个作用域,代表不同的权限等级:
{
"scopes": {
"read:repositories": "Read repository metadata (not code)",
"read:issues": "Read issue tracker",
"write:issues": "Create and modify issues",
"admin:settings": "Access administrative settings"
}
}
用户在授权时可以选择授权哪些作用域,而不是"全有或全无"。这种最小权限原则在企业场景下尤为重要——一个 AI 助手可能只需要"读"权限,不需要"写"权限。
4.3 安全边界:stdio vs HTTP
stdio 模式的本质是一个本地子进程。 Client 通过 stdin/stdout 与 Server 通信,数据不经过网络。这意味着 stdio Server 不需要也不应该暴露网络端口——它就是一个被调用即启动、用完即退出的工具进程。
这种设计天然安全:stdio Server 只能访问运行主机的本地资源,且只能在 Child Process 的生命周期内运行,无法维持后台监听。但如果 Host 本身被攻破(如恶意 Prompt 注入),stdio Server 的隔离优势会被绕过。
HTTP 模式的 Server 暴露在公网上,需要更严格的保护。 除了 OAuth 令牌,还应配合 TLS 加密、IP 白名单、请求频率限制等手段。
五、代码实战:从零构建一个生产级 MCP Server
5.1 技术选型
MCP 官方提供了多语言 SDK:
- Python SDK:
mcp(官方维护,最完善) - TypeScript/JS SDK:
@modelcontextprotocol/sdk(官方维护) - 其他语言:社区实现(C++、Rust、Go 等)
本文选择 Python SDK,因为 Python 在 AI/数据领域的普及度最高,且 SDK 的抽象程度最好。
5.2 项目结构
mcp-weather-server/
├── pyproject.toml
├── src/
│ └── weather_server/
│ ├── __init__.py
│ ├── server.py # MCP Server 核心
│ ├── resources.py # 资源定义
│ ├── tools.py # 工具定义
│ ├── prompts.py # 提示模板
│ └── weather_api.py # 天气 API 封装
└── README.md
5.3 核心代码实现
5.3.1 Server 骨架
# src/weather_server/server.py
from mcp.server.fastmcp import FastMCP
# FastMCP 是官方提供的简化 Server 框架,
# 封装了协议处理、连接管理和工具注册的底层细节。
# 开发者只需要关注业务逻辑,不需要手动处理 JSON-RPC。
mcp = FastMCP(
name="weather-server",
# 可选:版本信息,用于握手时告知 Client
version="1.0.0"
)
# 初始化资源(Resource)
@mcp.resource("weather://current/{city}")
def get_weather_resource(city: str) -> str:
"""
将天气数据暴露为 Resource。
Resource 与 Tool 的区别:
- Resource: AI 可以主动读取的数据源,类似文件系统
- Tool: AI 可以执行的操作,会改变系统状态
"""
return fetch_weather(city)
# 定义工具(Tool)
@mcp.tool(
name="get_current_weather",
description=(
"Get the current weather for a city. "
"Use this when user asks about today's weather. "
"Returns temperature, humidity, wind speed, and conditions."
)
)
def get_current_weather(
city: str,
country: str = "CN",
units: Literal["celsius", "fahrenheit"] = "celsius"
) -> str:
"""
获取指定城市的当前天气。
Args:
city: 城市名(拼音或中文均可,取决于 API 支持)
country: 国家代码,默认 CN(中国)
units: 温度单位,默认摄氏度
Returns:
格式化的天气报告,包含温度、体感温度、湿度、风速、天气状况
"""
data = fetch_weather_api(city, country, units)
report = f"""\
## {data['name']}, {data['sys']['country']} 实时天气
🌡️ 温度:{data['main']['temp']}°{"C" if units == "celsius" else "F"}
🌡️ 体感:{data['main']['feels_like']}°{"C" if units == "celsius" else "F"}
💧 湿度:{data['main']['humidity']}%
🌬️ 风速:{data['wind']['speed']} m/s
🌤️ 状况:{data['weather'][0]['description']}
📊 气压:{data['main']['pressure']} hPa
> 数据来源:OpenWeatherMap | 更新时间:{datetime.fromtimestamp(data['dt'])}
"""
return report
# 定义提示模板(Prompt)
@mcp.prompt(
name="weather_report",
description="Generate a daily weather report summary"
)
def weather_report_prompt(cities: list[str]) -> str:
"""\
生成多城市天气对比报告的 Prompt 模板。
Usage:
/weather_report ["北京", "上海", "广州"]
"""
cities_str = "、".join(cities)
return f"""\
请生成{cities_str}的今日天气对比报告。
报告应包含:
1. 各城市温度对比(最高/最低)
2. 各城市天气状况汇总
3. 出行建议(基于天气数据)
4. 特别注意(极端天气预警)
"""
5.3.2 天气 API 封装
# src/weather_server/weather_api.py
import os
import httpx
from typing import Literal
from dataclasses import dataclass
from functools import lru_cache
@dataclass
class WeatherData:
city: str
country: str
temp: float
feels_like: float
humidity: int
wind_speed: float
description: str
pressure: int
dt: int
class WeatherAPIError(Exception):
"""天气 API 调用失败时抛出的异常"""
pass
@lru_cache(maxsize=128)
def fetch_weather_api(
city: str,
country: str = "CN",
units: Literal["celsius", "fahrenheit"] = "celsius"
) -> dict:
"""
调用 OpenWeatherMap API 获取天气数据。
使用 lru_cache 缓存结果,同一城市在 10 分钟内不重复请求。
这对于 AI Agent 多次调用同一工具的场景非常重要,
既减少了 API 消耗,又避免了 API 限流问题。
"""
api_key = os.environ.get("OPENWEATHER_API_KEY")
if not api_key:
raise WeatherAPIError(
"OPENWEATHER_API_KEY environment variable not set. "
"Get a free API key at https://openweathermap.org/api"
)
unit = "metric" if units == "celsius" else "imperial"
url = "https://api.openweathermap.org/data/2.5/weather"
try:
with httpx.Client(timeout=10.0) as client:
response = client.get(url, params={
"q": f"{city},{country}",
"appid": api_key,
"units": unit
})
if response.status_code == 404:
raise WeatherAPIError(f"City not found: {city}, {country}")
elif response.status_code == 401:
raise WeatherAPIError("Invalid API key")
elif response.status_code != 200:
raise WeatherAPIError(
f"Weather API returned {response.status_code}: {response.text}"
)
return response.json()
except httpx.TimeoutException:
raise WeatherAPIError(f"Request timeout for city: {city}")
except httpx.RequestError as exc:
raise WeatherAPIError(f"Network error: {exc}")
5.3.3 入口文件
# src/weather_server/__main__.py
from weather_server.server import mcp
# FastMCP 的 run() 方法会:
# 1. 启动事件循环(asyncio)
# 2. 注册所有 @mcp.tool / @mcp.resource / @mcp.prompt 装饰的函数
# 3. 根据环境变量 MC_PROTOCOL 或 --protocol 选择传输模式:
# - MC_TRANSPORT=stdio:标准输入输出(默认,推荐)
# - MC_TRANSPORT=sse:HTTP + SSE
# 4. 开始处理来自 Client 的 JSON-RPC 请求
if __name__ == "__main__":
mcp.run()
5.3.4 配置文件
# pyproject.toml
[project]
name = "weather-mcp-server"
version = "1.0.0"
description = "MCP Server for real-time weather data"
requires-python = ">=3.10"
dependencies = [
"mcp>=1.0.0",
"httpx>=0.27.0",
]
[project.scripts]
weather-mcp = "weather_server.__main__:main"
[tool.mypy]
python_version = "3.10"
strict = true
[tool.ruff]
line-length = 88
target-version = "py310"
5.4 在 Claude Desktop 中配置
在 Claude Desktop 的 MCP 配置文件中添加:
// ~/Library/Application Support/Claude/claude_desktop_config.json
{
"mcpServers": {
"weather": {
"command": "uvx",
"args": ["--from", "git+https://github.com/your-name/weather-mcp-server", "weather-mcp"],
"env": {
"OPENWEATHER_API_KEY": "your_api_key_here"
}
}
}
}
或者,如果你是在本地开发和调试(用源码运行):
{
"mcpServers": {
"weather-local": {
"command": "python",
"args": ["-m", "weather_server"],
"cwd": "/path/to/your/weather-mcp-server",
"env": {
"OPENWEATHER_API_KEY": "your_api_key_here"
}
}
}
}
配置完成后重启 Claude Desktop,在设置中应该能看到 "weather" Server 已连接。此时,Claude 就能在对话中主动调用 get_current_weather 工具了。
5.5 测试工具调用
# tests/test_server.py
import pytest
from weather_server.server import mcp
from weather_server.weather_api import WeatherAPIError
class TestWeatherTool:
"""测试天气工具的基本行为"""
def test_city_not_found(self):
"""测试无效城市名"""
with pytest.raises(WeatherAPIError, match="City not found"):
# 模拟传入一个不存在的城市
# 实际测试需要 mock httpx.Client
pass
def test_schema_generation(self):
"""验证工具的 schema 是否符合 MCP 规范"""
tools = mcp.list_tools()
weather_tool = next(t for t in tools if t.name == "get_current_weather")
# Schema 必须包含 name、description、inputSchema
assert weather_tool.name == "get_current_weather"
assert "temperature" in weather_tool.description.lower()
assert weather_tool.inputSchema["type"] == "object"
assert "city" in weather_tool.inputSchema["properties"]
# 参数必须有 description,供 LLM 理解
city_schema = weather_tool.inputSchema["properties"]["city"]
assert "description" in city_schema
assert city_schema["description"].strip() != ""
六、高级主题:MCP 的生态与演进
6.1 官方 MCP Server 生态
Anthropic 和社区已经维护了一个丰富的官方 Server 生态,覆盖了 AI 应用最常见的使用场景:
| Server | 功能 | Star 数 | 传输模式 |
|---|---|---|---|
filesystem | 本地文件系统读写 | 官方 | stdio |
github | GitHub API(Repo、Issue、PR) | 官方 | stdio |
sqlite | SQLite 数据库查询 | 官方 | stdio |
brave-search | 网页搜索 | 官方 | stdio |
sentry | Sentry 错误追踪 | 官方 | stdio |
以 GitHub Server 为例,它暴露了约30个工具,覆盖了 GitHub REST API 的核心能力:
github_search_repositories:搜索代码仓库github_get_issue:获取 Issue 详情github_create_issue:创建 Issuegithub_list_pull_requests:列出 PR 列表github_review_pull_request:审查 PR
6.2 社区 Server:百花齐放
MCP 的开放性催生了大量社区 Server:
PostgreSQL Server:直接用 SQL 查询数据库,AI 可以自主写 SELECT 语句查数据。Schema 自动从数据库表结构生成,不需要手动维护。
Slack/Discord Server:让 AI 在 Slack 频道发消息、管理线程、查询历史记录。
Figma Server:让 AI 读取 Figma 设计文件、导出设计规范。
Airtable Server:读写 Airtable 表格数据。
Notion Server:读写 Notion 页面和数据库。
有意思的是,MCP 生态中出现了一个独特的现象:开发者用 MCP Server 来"封装"MCP Server 本身。 例如社区里有人写了 mcp-server-registry——一个专门用于管理其他 MCP Server 配置的 Server。AI 可以通过它来查询"当前有哪些可用的 MCP 工具",甚至可以"安装新的 MCP Server"。这种自举能力非常符合 AI Agent 的自进化理念。
6.3 MCP 协议的演进方向
截至 2026年5月,MCP 协议已经从 2024年11月 的 0.1.0 版本演进到 2025年11月 的 1.0.0 版本。协议在以下方向持续演进:
采样(Sampling): 这是最革命性的新特性——允许 Server"反过来"调用 LLM。在传统架构中,只有 Host 可以调用 LLM,Server 只能被动处理请求。Sampling 特性让 Server 可以主动发起 LLM 调用,实现 Server 端的推理能力。例如,一个数据库 Server 可以自动生成 SQL 查询,无需依赖 Client 侧的 LLM 来理解自然语言。
根资源(Roots): 支持 Server 声明自己管理的根目录(如工作区路径),使多租户场景下的资源隔离成为可能。
多级进度(Progress Tokens): 长时操作(如文件批量上传)支持进度回调,Client 可以向用户展示操作进度。
服务端工具(Server-side Tools): Server 可以反过来注册工具供 Client 调用,形成双向能力交换。
七、性能优化与生产实践
7.1 减少 LLM 调用次数:批量操作
MCP 工具调用每次都需要 LLM 生成一个请求并等待响应。如果 AI 需要对一个列表中的每个元素执行类似操作,应该支持批量接口:
@mcp.tool(name="batch_weather", description="Get weather for multiple cities at once")
def batch_weather(cities: list[str]) -> list[dict]:
"""批量查询多个城市的天气,减少 LLM 交互次数"""
results = []
for city in cities[:10]: # 限制最多10个城市,防滥用
try:
data = fetch_weather_api(city)
results.append({
"city": data["name"],
"temp": data["main"]["temp"],
"status": "success"
})
except WeatherAPIError as e:
results.append({
"city": city,
"status": "error",
"error": str(e)
})
return results
7.2 缓存策略
如前文的 lru_cache 示例,合理的缓存策略可以显著减少外部 API 调用。生产环境中推荐使用 Redis:
import redis
redis_client = redis.Redis(host="localhost", port=6379, db=0)
def cached_fetch_weather(city: str) -> dict:
cache_key = f"weather:{city}"
# 先查缓存
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 未命中,发起 API 请求
data = fetch_weather_api(city)
# 写入缓存,TTL = 10 分钟
redis_client.setex(cache_key, 600, json.dumps(data))
return data
7.3 限流与降级
当 MCP Server 被多个 Client 同时调用时,需要限流保护下游 API:
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=60, period=60) # OpenWeatherMap 免费版限制:60 calls/minute
def fetch_weather_api_ratelimited(city: str, country: str, units: str) -> dict:
return fetch_weather_api(city, country, units)
# 当限流触发时,返回友好的降级信息
def fetch_weather_api(city: str, country: str = "CN", units: str = "celsius") -> dict:
try:
return fetch_weather_api_ratelimited(city, country, units)
except Exception:
return {
"error": True,
"message": "Weather service temporarily unavailable. "
"Please try again in a few minutes.",
"city": city
}
7.4 日志与可观测性
import structlog
logger = structlog.get_logger()
@mcp.tool(name="get_current_weather", description="...")
def get_current_weather(city: str, country: str = "CN", units: str = "celsius") -> str:
# 结构化日志,方便接入 Datadog / Grafana
logger.info(
"weather_request_started",
city=city,
country=country,
units=units
)
try:
data = fetch_weather_api(city, country, units)
logger.info(
"weather_request_success",
city=city,
temp=data["main"]["temp"]
)
return format_weather_report(data)
except WeatherAPIError as e:
logger.error(
"weather_request_failed",
city=city,
error=str(e)
)
raise
八、架构反思:MCP 的边界在哪里
MCP 解决了 AI Agent 与外部工具的通信问题,但它不是一个万能的"智能体编排框架"。在实际生产中,有几个问题值得思考:
8.1 MCP 不解决"做什么"的问题
MCP 定义的是"工具如何被调用",不涉及"什么时候该用什么工具"的决策。这些决策由 Host 中的 LLM 通过 Function Calling 机制完成。MCP 本身没有 Planner、没有工作流引擎、没有状态机——它只是一个通信协议。
这意味着,如果你需要复杂的多步骤工作流(如"先查邮件 → 再更新 CRM → 再发 Slack 通知"),你需要在 LLM 层面编排这些步骤,或者使用额外的 Agent 框架(如 LangGraph、AutoGen)。
8.2 Server 治理的挑战
当 MCP 生态中有成百上千个 Server 时,如何管理它们的版本、安全性、质量?这是一个尚未被很好解决的问题。Anthropic 正在推进 MCP Registry 建设,但目前还是社区驱动的状态。
8.3 跨 Host 兼容性
MCP 规范本身是 Host 中立的,但各 Host 厂商对规范的实现完整度不同。一些 Host 可能不支持 Sampling、新版 Progress Token 等高级特性。Server 在生产部署前,需要针对目标 Host 做兼容性测试。
九、总结与展望
Model Context Protocol 的出现,标志着 AI Agent 生态进入了一个新的阶段——从"各自为战"的私有协议,走向"互联互通"的标准化时代。
回顾本文的核心要点:
MCP 的定位:AI Agent 与外部工具之间的"通用插座",基于 JSON-RPC 2.0,支持 stdio 和 HTTP 两种传输模式。
三层架构:Host(协调中枢)→ Client(路由代理)→ Server(工具接口),各司其职,协同工作。
自描述能力:Server 运行时动态暴露工具 schema,使工具发现和升级完全自动化,无需 Client 重新配置。
安全模型:基于 OAuth 2.0 + 作用域控制,实现精细化的权限管理,最小权限原则保护用户数据。
生产开发实践:通过 FastMCP 框架,可以快速构建功能完善的 MCP Server,配合缓存、限流、日志等工程实践,满足生产环境需求。
展望未来,MCP 的 Sampling 特性(Server 反向调用 LLM)可能是最具颠覆性的方向——它让 Server 拥有了"主动思考"的能力,模糊了工具和智能体之间的边界。当每一个工具都能做推理、每一个数据源都能生成洞察,AI Agent 的能力边界将被极大拓展。
而对于开发者而言,现在正是入局 MCP 生态的最佳时机:规范已经成熟、SDK 已经完善、社区正在快速扩张。无论是构建自己的 MCP Server 接入已有工具链,还是基于 MCP 设计新的 AI Native 应用,这套协议都值得你深入了解。
毕竟,当 USB-C 出现时,最先上车的人,总是能最快触达新世界的人。
参考链接
- MCP 官方规范:https://modelcontextprotocol.io
- MCP Python SDK:https://github.com/modelcontextprotocol/python-sdk
- MCP Server 示例集:https://github.com/modelcontextprotocol/servers
- Anthropic 官方 MCP Server:https://github.com/anthropics/claude-desktop/tree/main/src/mcp
本文作者:程序员茄子 | 首发于程序员茄子 | 原创不易,转载请注明出处