1. MCP 是什么：从 N×M 到 N+M
1.1 痛点：N×M 集成噩梦
在 MCP 出现之前，如果你有 10 个 AI 应用和 20 个工具，你需要写多少个集成？
答案：200 个。每个 AI 应用对每个工具都需要一个定制化的连接器。
┌─────────────────────────────────────────────────────┐
│ N × M 集成噩梦（Before MCP） │
│ │
│ AI App 1 ──→ Tool A, Tool B, Tool C ... Tool T │
│ AI App 2 ──→ Tool A, Tool B, Tool C ... Tool T │
│ AI App 3 ──→ Tool A, Tool B, Tool C ... Tool T │
│ ... │
│ AI App N ──→ Tool A, Tool B, Tool C ... Tool T │
│ │
│ 总连接数 = N × M = 10 × 20 = 200 │
└─────────────────────────────────────────────────────┘
每条连接线都包含：认证逻辑、错误处理、数据格式转换、版本兼容维护。新增一个模型？重写所有连接。新增一个工具？适配所有模型。这就是所谓的 N×M 问题。
1.2 MCP 的解法：USB-C for AI
MCP（Model Context Protocol，模型上下文协议）是 Anthropic 于 2024 年 11 月 25 日发布的开放标准。它的核心思路极其简单：
USB-C 统一了设备接口，MCP 统一了 AI 接口。
┌─────────────────────────────────────────────────────┐
│ N + M 模式（After MCP） │
│ │
│ AI App 1 ──┐ │
│ AI App 2 ──┤ ┌───────────┐ ┌──────────┐ │
│ AI App 3 ──┼────▶│ MCP 协议 │◀────│ Tool A │ │
│ ... │ │ (标准接口) │◀────│ Tool B │ │
│ AI App N ──┘ └───────────┘◀────│ Tool C │ │
│ │ ... │ │
│ │ Tool M │ │
│ └──────────┘ │
│ 总连接数 = N + M = 10 + 20 = 30 │
└─────────────────────────────────────────────────────┘
每个 AI 应用实现一次 MCP Client，每个工具实现一次 MCP Server，任何 Client 可以连接任何 Server。
1.3 关键数字
据《MCP hits 97M downloads — Anthropic just gave it away》（https://aiforautomation.io/news/2026-03-27-mcp-linux-foundation-97m-downloads-anthropic-donates）报道，MCP 的增长速度是 React npm 的 2 倍——React 用了约 3 年才达到类似的月下载规模。
1.4 治理演进
2024.11 ── Anthropic 发布 MCP（MIT License）
│
2025.03 ── OpenAI 采纳 MCP
│
2025.04 ── Google DeepMind 采纳 MCP
│
2025.06 ── VS Code、Cursor、JetBrains 原生支持
│
2025.12 ── Anthropic 将 MCP 捐赠给 Linux 基金会
│ → 成立 Agentic AI Foundation (AAIF)
│ → 联合创始方：Anthropic, OpenAI, Block
│ → 支持方：Google, Microsoft, AWS, Cloudflare
│
2026.03 ── 9700 万月下载，行业事实标准
据《MCP Hits 97 Million Installs: The Protocol Powering Agentic AI》（https://ai2.work/blog/mcp-hits-97-million-installs-the-protocol-powering-agentic-ai）报道，OpenAI 甚至宣布将在 2026 年中废弃其 Assistants API，全面转向 MCP 架构——这相当于用自家标准给 MCP 投票。
2. 核心架构：三层模型与 JSON-RPC 2.0
2.1 Host / Client / Server 三层
MCP 的架构由三个角色组成：
┌──────────────────────────────────────────────────────────────┐
│ MCP Host │
│ （AI 应用，如 Claude Desktop、VS Code、Cursor） │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ MCP Client │ │ MCP Client │ │ MCP Client │ ... │
│ │ (连接 A) │ │ (连接 B) │ │ (连接 C) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
└─────────┼────────────────┼────────────────┼─────────────────┘
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ MCP Server │ │ MCP Server │ │ MCP Server │
│ (GitHub) │ │ (PostgreSQL) │ │ (文件系统) │
│ │ │ │ │ │
│ Tools: │ │ Tools: │ │ Tools: │
│ - create_ │ │ - query │ │ - read_file │
│ issue │ │ - insert │ │ - write_file│
│ - list_prs │ │ - update │ │ - list_dir │
│ │ │ │ │ │
│ Resources: │ │ Resources: │ │ Resources: │
│ - repo_info │ │ - schema │ │ - file_tree │
└──────────────┘ └──────────────┘ └──────────────┘
角色说明
关键设计：一个 Host 可以同时连接多个 Server，每个 Client 维护独立的连接和会话状态。Server 之间互不感知，Host 负责协调。
2.2 JSON-RPC 2.0 通信
MCP 使用 JSON-RPC 2.0 作为消息格式，定义了三种消息类型：
# 1. Request（请求）—— 期望响应
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {"path": "/data/report.csv"}
}
}
# 2. Response（响应）—— 匹配请求 ID
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{"type": "text", "text": "date,revenue\n2026-01,150000"}
]
}
}
# 3. Notification（通知）—— 不期望响应
{
"jsonrpc": "2.0",
"method": "notifications/tools/list_changed"
}
2.3 传输层：Stdio vs Streamable HTTP
MCP 定义了两种传输方式：
┌──────────────────────────────────────────────────────────────┐
│ 传输方式对比 │
├──────────────┬──────────────────┬───────────────────────────┤
│ 维度 │ Stdio │ Streamable HTTP │
├──────────────┼──────────────────┼───────────────────────────┤
│ 通信方式 │ stdin/stdout 管道 │ HTTP POST + SSE │
├──────────────┼──────────────────┼───────────────────────────┤
│ 部署位置 │ 本地同一台机器 │ 本地或远程 │
├──────────────┼──────────────────┼───────────────────────────┤
│ 认证 │ 无需（本地信任） │ OAuth 2.0 / API Key │
├──────────────┼──────────────────┼───────────────────────────┤
│ 多客户端 │ 不支持 │ 支持 │
├──────────────┼──────────────────┼───────────────────────────┤
│ 网络延迟 │ 零（进程间通信） │ 有（HTTP 往返） │
├──────────────┼──────────────────┼───────────────────────────┤
│ 典型场景 │ IDE 插件、本地开发 │ 云部署、生产环境 │
├──────────────┼──────────────────┼───────────────────────────┤
│ 会话管理 │ 隐式（进程生命周期）│ Mcp-Session-Id Header │
└──────────────┴──────────────────┴───────────────────────────┘
2.4 能力协商
Client 和 Server 在初始化时交换能力声明：
Client Server
│ │
│ ── initialize ──────────────────────▶ │
│ {protocolVersion: "2025-03-26", │
│ capabilities: {roots: ...}, │
│ clientInfo: {name: "my-app"}} │
│ │
│ ◀── initialize response ───────────── │
│ {protocolVersion: "2025-03-26", │
│ capabilities: { │
│ tools: {listChanged: true}, │
│ resources: {subscribe: true}, │
│ prompts: {listChanged: true} │
│ }, │
│ serverInfo: {name: "postgres-mcp"}}│
│ │
│ ── initialized (notification) ──────▶ │
│ │
│ === 会话建立，开始通信 === │
注意 listChanged 字段——这意味着 Server 可以在运行时通知 Client 工具列表发生了变化。这在动态环境中非常关键。
3. 三大原语：Tools / Resources / Prompts
MCP 定义了三个核心原语，分别对应 AI Agent 的三种基本交互模式：
┌──────────────────────────────────────────────────────────────┐
│ MCP 三大原语 │
├──────────────┬──────────────────┬───────────────────────────┤
│ 原语 │ 类比 │ 核心特征 │
├──────────────┼──────────────────┼───────────────────────────┤
│ Tools │ AI 的"手" │ 可执行操作，有副作用 │
│ │ (POST 请求) │ 写数据库、发邮件、调 API │
├──────────────┼──────────────────┼───────────────────────────┤
│ Resources │ AI 的"眼" │ 只读数据，无副作用 │
│ │ (GET 请求) │ 读文件、查数据库 schema │
├──────────────┼──────────────────┼───────────────────────────┤
│ Prompts │ AI 的"模板" │ 可复用的提示词模板 │
│ │ (模板引擎) │ 系统提示、少样本示例 │
└──────────────┴──────────────────┴───────────────────────────┘
3.1 Tools（工具调用）
Tools 是最常用的原语。它让 AI 能够执行操作。
# 工具定义示例
{
"name": "query_database",
"description": "Execute a SQL query on the PostgreSQL database",
"inputSchema": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "The SQL query to execute"
},
"database": {
"type": "string",
"description": "Database name",
"default": "production"
}
},
"required": ["sql"]
}
}
3.2 Resources（数据读取）
Resources 提供只读的上下文数据，不需要消耗一个 Tool Call 的轮次：
# 资源定义示例
{
"uri": "postgres://production/schema",
"name": "Database Schema",
"description": "Current schema of the production database",
"mimeType": "application/json"
}
为什么 Resources 比 Tools 更重要？ 很多开发者只关注 Tools，但 Resources 能让你在不消耗 Tool Call 的情况下注入上下文。这意味着 LLM 可以更快获得关键信息，减少不必要的往返。
3.3 Prompts（提示模板）
Prompts 是最被低估的原语，但在生产中极其有用：
# 提示模板示例
{
"name": "code_review",
"description": "Review code changes with specific criteria",
"arguments": [
{
"name": "language",
"description": "Programming language",
"required": true
},
{
"name": "severity",
"description": "Review strictness level",
"required": false
}
]
}
4. 实战1：用 Python 写一个 MCP Server（文件管理工具）
下面我们用 Python SDK 写一个完整的文件管理 MCP Server。它支持列出目录、读取文件、写入文件、搜索文件四个工具。
4.1 项目结构
plaintext
file-manager-mcp/
├── server.py # MCP Server 主文件
├── pyproject.toml # 项目配置
└── README.md # 说明文档
4.2 安装依赖
bash
pip install mcp
4.3 完整代码
python
#!/usr/bin/env python3
"""
文件管理 MCP Server
提供目录列表、文件读写、文件搜索四个工具
作者：PySuper | 来源：zhengxingtao.com
"""
import os
import json
import fnmatch
from pathlib import Path
from typing import Optional
from mcp.server.fastmcp import FastMCP
# 创建 MCP Server 实例
mcp = FastMCP(
name="file-manager",
version="1.0.0",
description="File management MCP server for reading, writing, and searching files"
)
# 安全限制：只允许访问指定目录
ALLOWED_BASE_DIR = os.environ.get("MCP_FILE_BASE_DIR", os.path.expanduser("~/mcp-workspace"))
# 最大文件大小限制（10MB）
MAX_FILE_SIZE = 10 * 1024 * 1024
def _validate_path(path: str) -> Path:
"""验证路径是否在允许的目录内，防止路径遍历攻击"""
resolved = Path(path).resolve()
base = Path(ALLOWED_BASE_DIR).resolve()
if not str(resolved).startswith(str(base)):
raise ValueError(
f"路径 '{path}' 超出允许的范围。"
f"只允许访问 '{base}' 及其子目录。"
)
return resolved
def _safe_read(path: Path) -> str:
"""安全读取文件内容，带大小检查"""
if not path.exists():
raise FileNotFoundError(f"文件不存在：{path}")
if not path.is_file():
raise ValueError(f"路径不是文件：{path}")
if path.stat().st_size > MAX_FILE_SIZE:
raise ValueError(
f"文件过大（{path.stat().st_size} 字节），"
f"最大允许 {MAX_FILE_SIZE} 字节"
)
return path.read_text(encoding="utf-8", errors="replace")
# ============================================================
# Tool 1: 列出目录内容
# ============================================================
@mcp.tool()
def list_directory(
path: str = ".",
pattern: str = "*",
show_hidden: bool = False
) -> str:
"""
列出指定目录下的文件和子目录。
Args:
path: 目录路径（相对于工作空间根目录）
pattern: 文件名匹配模式，支持通配符，如 *.py, *.md
show_hidden: 是否显示隐藏文件（以 . 开头的文件）
Returns:
JSON 格式的目录列表
"""
target = _validate_path(os.path.join(ALLOWED_BASE_DIR, path))
if not target.exists():
raise FileNotFoundError(f"目录不存在：{target}")
if not target.is_dir():
raise ValueError(f"路径不是目录：{target}")
entries = []
for entry in sorted(target.iterdir()):
# 过滤隐藏文件
if not show_hidden and entry.name.startswith("."):
continue
# 过滤匹配模式
if not fnmatch.fnmatch(entry.name, pattern):
continue
entry_info = {
"name": entry.name,
"type": "directory" if entry.is_dir() else "file",
"size": entry.stat().st_size if entry.is_file() else None,
"modified": entry.stat().st_mtime,
}
entries.append(entry_info)
result = {
"path": str(target),
"total": len(entries),
"entries": entries
}
return json.dumps(result, indent=2, ensure_ascii=False)
# ============================================================
# Tool 2: 读取文件内容
# ============================================================
@mcp.tool()
def read_file(
path: str,
encoding: str = "utf-8",
start_line: Optional[int] = None,
end_line: Optional[int] = None
) -> str:
"""
读取指定文件的内容。
Args:
path: 文件路径（相对于工作空间根目录）
encoding: 文件编码，默认 utf-8
start_line: 起始行号（从1开始），不指定则从第1行开始
end_line: 结束行号（包含），不指定则到文件末尾
Returns:
文件内容（可能被截断）
"""
target = _validate_path(os.path.join(ALLOWED_BASE_DIR, path))
content = _safe_read(target)
# 按行范围截取
if start_line is not None or end_line is not None:
lines = content.splitlines(keepends=True)
start = (start_line or 1) - 1 # 转为0-based索引
end = end_line or len(lines)
lines = lines[start:end]
content = "".join(lines)
# 截断超长内容
if len(content) > 50000:
content = content[:50000] + "\n\n... [文件内容已截断，共 {} 字符]".format(len(content))
return content
# ============================================================
# Tool 3: 写入文件
# ============================================================
@mcp.tool()
def write_file(
path: str,
content: str,
mode: str = "overwrite"
) -> str:
"""
写入文件内容。
Args:
path: 文件路径（相对于工作空间根目录）
content: 要写入的内容
mode: 写入模式，overwrite（覆盖）或 append（追加）
Returns:
操作结果信息
"""
target = _validate_path(os.path.join(ALLOWED_BASE_DIR, path))
# 自动创建父目录
target.parent.mkdir(parents=True, exist_ok=True)
if mode == "append":
if target.exists():
existing = _safe_read(target)
content = existing + "\n" + content
with open(target, "w", encoding="utf-8") as f:
f.write(content)
elif mode == "overwrite":
with open(target, "w", encoding="utf-8") as f:
f.write(content)
else:
raise ValueError(f"不支持的写入模式：{mode}，请使用 overwrite 或 append")
result = {
"status": "success",
"path": str(target),
"mode": mode,
"bytes_written": len(content.encode("utf-8")),
"lines": content.count("\n") + 1
}
return json.dumps(result, indent=2, ensure_ascii=False)
# ============================================================
# Tool 4: 搜索文件
# ============================================================
@mcp.tool()
def search_files(
pattern: str,
path: str = ".",
max_depth: int = 5,
file_type: Optional[str] = None
) -> str:
"""
递归搜索匹配的文件。
Args:
pattern: 搜索模式，支持通配符，如 *.py, report*.csv
path: 起始目录路径（相对于工作空间根目录）
max_depth: 最大搜索深度
file_type: 文件类型过滤，file 或 directory
Returns:
JSON 格式的搜索结果
"""
target = _validate_path(os.path.join(ALLOWED_BASE_DIR, path))
if not target.is_dir():
raise ValueError(f"路径不是目录：{target}")
matches = []
base_depth = len(target.parts)
for root, dirs, files in os.walk(target):
current_depth = len(Path(root).parts) - base_depth
if current_depth > max_depth:
dirs.clear() # 不再深入
continue
# 搜索文件
for name in files:
if fnmatch.fnmatch(name, pattern):
full_path = Path(root) / name
rel_path = full_path.relative_to(target)
if file_type and file_type != "file":
continue
matches.append({
"name": name,
"path": str(rel_path),
"size": full_path.stat().st_size,
"type": "file"
})
# 搜索目录
for name in dirs:
if fnmatch.fnmatch(name, pattern):
full_path = Path(root) / name
rel_path = full_path.relative_to(target)
if file_type and file_type != "directory":
continue
matches.append({
"name": name,
"path": str(rel_path),
"type": "directory"
})
result = {
"pattern": pattern,
"base_path": str(target),
"total_matches": len(matches),
"matches": matches[:100] # 最多返回100条
}
return json.dumps(result, indent=2, ensure_ascii=False)
# ============================================================
# Resource: 工作空间概览
# ============================================================
@mcp.resource("workspace://overview")
def workspace_overview() -> str:
"""获取工作空间概览信息"""
base = Path(ALLOWED_BASE_DIR)
if not base.exists():
return json.dumps({"error": f"工作空间目录不存在：{base}"})
file_count = 0
dir_count = 0
total_size = 0
for root, dirs, files in os.walk(base):
dir_count += len(dirs)
for f in files:
file_count += 1
total_size += (Path(root) / f).stat().st_size
return json.dumps({
"base_dir": str(base),
"total_files": file_count,
"total_directories": dir_count,
"total_size_bytes": total_size,
"total_size_human": f"{total_size / (1024*1024):.2f} MB"
}, indent=2, ensure_ascii=False)
# ============================================================
# Prompt: 代码审查模板
# ============================================================
@mcp.prompt()
def code_review(language: str, focus: str = "all") -> str:
"""生成代码审查提示词模板"""
return f"""请对以下 {language} 代码进行审查。
审查重点：{focus}
请从以下维度评估：
1. **代码质量**：命名规范、代码风格、可读性
2. **潜在 Bug**：边界条件、空指针、类型错误
3. **性能问题**：不必要的计算、内存泄漏、N+1 查询
4. **安全风险**：SQL 注入、XSS、路径遍历
5. **可维护性**：耦合度、测试覆盖、文档完整性
请按严重程度排序（Critical > Warning > Info），给出具体修改建议。"""
# ============================================================
# 启动 Server
# ============================================================
if __name__ == "__main__":
# 确保工作空间目录存在
Path(ALLOWED_BASE_DIR).mkdir(parents=True, exist_ok=True)
print(f"[file-manager-mcp] 工作空间：{ALLOWED_BASE_DIR}")
print(f"[file-manager-mcp] 启动 MCP Server...")
# 以 stdio 模式启动
mcp.run(transport="stdio")
4.4 测试运行
bash
# 设置工作空间目录
export MCP_FILE_BASE_DIR=/tmp/mcp-workspace
# 启动 Server
python server.py
4.5 用 MCP Inspector 调试
MCP 提供了一个可视化调试工具——Inspector：
bash
npx @modelcontextprotocol/inspector python server.py
打开浏览器访问 Inspector，你可以：
查看所有注册的 Tools / Resources / Prompts
手动调用工具并查看返回结果
检查 JSON-RPC 消息流
验证能力协商过程
5. 实战2：用 Python 写一个 MCP Client 对接 Claude/GPT
5.1 Client 代码
python
#!/usr/bin/env python3
"""
MCP Client：连接 MCP Server 并与 LLM 交互
作者：PySuper | 来源：zhengxingtao.com
"""
import asyncio
import json
from typing import Any, Optional
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# ============================================================
# MCP Client 封装
# ============================================================
class MCPClient:
"""MCP Client 封装，支持工具发现和调用"""
def __init__(self, server_command: list[str], server_env: Optional[dict] = None):
"""
Args:
server_command: 启动 MCP Server 的命令
server_env: 传递给 Server 的环境变量
"""
self.server_params = StdioServerParameters(
command=server_command[0],
args=server_command[1:],
env=server_env
)
self.session: Optional[ClientSession] = None
self._available_tools: list[dict] = []
self._available_resources: list[dict] = []
self._available_prompts: list[dict] = []
async def connect(self):
"""连接到 MCP Server"""
self._stdio_context = stdio_client(self.server_params)
streams = await self._stdio_context.__aenter__()
self._session_context = ClientSession(*streams)
self.session = await self._session_context.__aenter__()
# 初始化会话
result = await self.session.initialize()
print(f"[MCP Client] 连接成功")
print(f" Server: {result.serverInfo.name} v{result.serverInfo.version}")
print(f" 协议版本: {result.protocolVersion}")
print(f" 能力: {result.capabilities}")
# 获取可用工具、资源、提示
await self._discover_capabilities()
async def _discover_capabilities(self):
"""发现 Server 提供的所有能力"""
# 发现工具
try:
tools_result = await self.session.list_tools()
self._available_tools = [
{
"name": t.name,
"description": t.description,
"inputSchema": t.inputSchema
}
for t in tools_result.tools
]
print(f"\n[MCP Client] 发现 {len(self._available_tools)} 个工具:")
for t in self._available_tools:
print(f" - {t['name']}: {t['description'][:60]}...")
except Exception as e:
print(f"[MCP Client] 工具发现失败: {e}")
# 发现资源
try:
resources_result = await self.session.list_resources()
self._available_resources = [
{
"uri": r.uri,
"name": r.name,
"description": r.description,
"mimeType": r.mimeType
}
for r in resources_result.resources
]
print(f"\n[MCP Client] 发现 {len(self._available_resources)} 个资源:")
for r in self._available_resources:
print(f" - {r['name']} ({r['uri']})")
except Exception as e:
print(f"[MCP Client] 资源发现失败: {e}")
# 发现提示模板
try:
prompts_result = await self.session.list_prompts()
self._available_prompts = [
{
"name": p.name,
"description": p.description,
"arguments": p.arguments
}
for p in prompts_result.prompts
]
print(f"\n[MCP Client] 发现 {len(self._available_prompts)} 个提示模板:")
for p in self._available_prompts:
print(f" - {p['name']}: {p['description'][:60]}...")
except Exception as e:
print(f"[MCP Client] 提示发现失败: {e}")
async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
"""调用 MCP 工具"""
if not self.session:
raise RuntimeError("未连接到 MCP Server")
result = await self.session.call_tool(name, arguments)
# 提取文本内容
text_parts = []
for content in result.content:
if content.type == "text":
text_parts.append(content.text)
elif content.type == "image":
text_parts.append(f"[图片数据: {content.mimeType}]")
elif content.type == "resource":
text_parts.append(f"[资源: {content.resource.uri}]")
return "\n".join(text_parts)
async def read_resource(self, uri: str) -> str:
"""读取 MCP 资源"""
if not self.session:
raise RuntimeError("未连接到 MCP Server")
result = await self.session.read_resource(uri)
text_parts = []
for content in result.contents:
if hasattr(content, "text"):
text_parts.append(content.text)
elif hasattr(content, "blob"):
text_parts.append(f"[二进制数据: {content.mimeType}]")
return "\n".join(text_parts)
async def get_prompt(self, name: str, arguments: dict[str, str] = None) -> str:
"""获取 MCP 提示模板"""
if not self.session:
raise RuntimeError("未连接到 MCP Server")
result = await self.session.get_prompt(name, arguments)
text_parts = []
for msg in result.messages:
if hasattr(msg.content, "text"):
text_parts.append(msg.content.text)
return "\n".join(text_parts)
@property
def tools_for_llm(self) -> list[dict]:
"""返回 LLM 可用的工具定义（兼容 OpenAI Function Calling 格式）"""
return [
{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t["inputSchema"]
}
}
for t in self._available_tools
]
async def disconnect(self):
"""断开连接"""
if self._session_context:
await self._session_context.__aexit__(None, None, None)
if self._stdio_context:
await self._stdio_context.__aexit__(None, None, None)
# ============================================================
# 演示：连接文件管理 MCP Server
# ============================================================
async def demo():
"""演示 MCP Client 的完整使用流程"""
client = MCPClient(
server_command=["python", "server.py"],
server_env={"MCP_FILE_BASE_DIR": "/tmp/mcp-demo"}
)
try:
# 1. 连接
await client.connect()
# 2. 调用工具 - 写入文件
print("\n" + "="*60)
print("调用工具：write_file")
result = await client.call_tool("write_file", {
"path": "hello.txt",
"content": "Hello, MCP! 这是通过 MCP 协议写入的文件。\n第二行内容。"
})
print(f"结果: {result}")
# 3. 调用工具 - 列出目录
print("\n" + "="*60)
print("调用工具：list_directory")
result = await client.call_tool("list_directory", {"path": "."})
print(f"结果: {result}")
# 4. 调用工具 - 读取文件
print("\n" + "="*60)
print("调用工具：read_file")
result = await client.call_tool("read_file", {"path": "hello.txt"})
print(f"结果: {result}")
# 5. 调用工具 - 搜索文件
print("\n" + "="*60)
print("调用工具：search_files")
result = await client.call_tool("search_files", {"pattern": "*.txt"})
print(f"结果: {result}")
# 6. 读取资源
print("\n" + "="*60)
print("读取资源：workspace://overview")
result = await client.read_resource("workspace://overview")
print(f"结果: {result}")
# 7. 获取提示模板
print("\n" + "="*60)
print("获取提示模板：code_review")
result = await client.get_prompt("code_review", {"language": "Python", "focus": "security"})
print(f"结果: {result}")
# 8. 打印 LLM 工具定义
print("\n" + "="*60)
print("LLM 工具定义（OpenAI 格式）:")
for tool in client.tools_for_llm:
print(f" - {tool['function']['name']}")
finally:
await client.disconnect()
if __name__ == "__main__":
asyncio.run(demo())
5.2 对接 OpenAI API 的完整流程
python
#!/usr/bin/env python3
"""
MCP + OpenAI 完整集成示例
作者：PySuper | 来源：zhengxingtao.com
"""
import asyncio
import os
import json
from openai import OpenAI
from mcp_client import MCPClient
async def chat_with_mcp_tools(user_message: str):
"""使用 MCP 工具与 OpenAI 对话"""
# 1. 连接 MCP Server
client = MCPClient(
server_command=["python", "server.py"],
server_env={"MCP_FILE_BASE_DIR": "/tmp/mcp-demo"}
)
try:
await client.connect()
# 2. 初始化 OpenAI Client
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# 3. 构造消息
messages = [
{"role": "system", "content": "你是一个文件管理助手，可以帮助用户管理文件。"},
{"role": "user", "content": user_message}
]
# 4. 第一次调用 LLM，带上 MCP 工具
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=client.tools_for_llm,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# 5. 处理工具调用循环
max_iterations = 10
for _ in range(max_iterations):
if not message.tool_calls:
break
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
print(f"\n[工具调用] {function_name}({function_args})")
# 调用 MCP 工具
result = await client.call_tool(function_name, function_args)
print(f"[工具结果] {result[:200]}...")
# 将结果回填给 LLM
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 再次调用 LLM
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=client.tools_for_llm,
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# 6. 输出最终回复
print(f"\n[AI 回复] {message.content}")
return message.content
finally:
await client.disconnect()
# ============================================================
# 运行示例
# ============================================================
async def main():
# 示例1：简单文件操作
await chat_with_mcp_tools("帮我在工作空间创建一个 config.json 文件，内容是默认配置")
print("\n" + "="*60 + "\n")
# 示例2：多步操作
await chat_with_mcp_tools("列出工作空间的所有文件，然后告诉我有哪些是 JSON 文件")
if __name__ == "__main__":
asyncio.run(main())
6. 安全模型：权限控制、沙箱执行、Prompt 注入防御
6.1 三层权限模型
plaintext
┌──────────────────────────────────────────────────────────────┐
│ MCP 安全架构 │
│ │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ 第一层：Host 控制（用户授权） │ │
│ │ - 用户决定是否允许连接某个 MCP Server │ │
│ │ - 用户审批工具调用请求 │ │
│ │ - 类似：手机 App 的权限弹窗 │ │
│ └────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ 第二层：Client 控制（连接限制） │ │
│ │ - Client 决定发送什么请求 │ │
│ │ - Client 过滤敏感参数 │ │
│ │ - 类似：浏览器 CORS 策略 │ │
│ └────────────────────────────────────────────────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────────┐ │
│ │ 第三层：Server 控制（能力暴露） │ │
│ │ - Server 决定暴露哪些工具和资源 │ │
│ │ - Server 实现访问控制和数据过滤 │ │
│ │ - 类似：API 的鉴权和脱敏 │ │
│ └────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
6.2 Prompt 注入防御
MCP 的 Server 返回数据会被注入到 LLM 的上下文中，这带来了 Prompt 注入风险。防御策略：
python
# 在 MCP Server 中实现输出过滤
import re
def sanitize_output(text: str) -> str:
"""过滤潜在的 Prompt 注入内容"""
# 1. 移除可能的系统提示伪装
patterns = [
r"(?i)ignore\s+(previous|above|all)\s+instructions",
r"(?i)you\s+are\s+now\s+",
r"(?i)system\s*:\s*",
r"(?i)<\s*/?\s*(system|instruction|prompt)\s*>",
]
for pattern in patterns:
text = re.sub(pattern, "[FILTERED]", text)
# 2. 限制输出长度
if len(text) > 100000:
text = text[:100000] + "\n[输出已截断]"
return text
6.3 沙箱执行
对于需要执行代码的 MCP Server，推荐使用沙箱：
python
import subprocess
import tempfile
def execute_in_sandbox(code: str, timeout: int = 30) -> dict:
"""在 Docker 沙箱中执行代码"""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
f.flush()
try:
result = subprocess.run(
[
"docker", "run", "--rm",
"--network", "none", # 无网络访问
"--memory", "512m", # 内存限制
"--cpus", "1.0", # CPU 限制
"--read-only", # 只读文件系统
"-v", f"{f.name}:/tmp/code.py:ro",
"python:3.12-slim",
"python", "/tmp/code.py"
],
capture_output=True,
text=True,
timeout=timeout
)
return {
"stdout": result.stdout[:5000],
"stderr": result.stderr[:5000],
"exit_code": result.returncode
}
except subprocess.TimeoutExpired:
return {"error": f"执行超时（{timeout}秒）"}
finally:
os.unlink(f.name)
7. 生产部署：SSE 传输、OAuth 认证、水平扩展、监控
7.1 Streamable HTTP 传输
生产环境必须使用 Streamable HTTP 传输：
python
#!/usr/bin/env python3
"""
生产级 MCP Server（Streamable HTTP 传输）
作者：PySuper | 来源：zhengxingtao.com
"""
import os
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
name="file-manager-prod",
version="1.0.0",
description="Production file management MCP server"
)
# ... (工具定义与前面相同，省略)
if __name__ == "__main__":
# 以 Streamable HTTP 模式启动
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8080
)
7.2 OAuth 2.0 认证
python
from authlib.integrations.starlette_client import OAuth
from starlette.middleware import Middleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import JSONResponse
# OAuth 配置
oauth = OAuth()
oauth.register(
name="github",
client_id=os.environ.get("GITHUB_CLIENT_ID"),
client_secret=os.environ.get("GITHUB_CLIENT_SECRET"),
authorize_url="https://github.com/login/oauth/authorize",
access_token_url="https://github.com/login/oauth/access_token",
userinfo_endpoint="https://api.github.com/user",
)
# 在 MCP Server 前加认证中间件
async def auth_middleware(request, call_next):
"""验证请求中的 Bearer Token"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse(
{"error": "unauthorized", "message": "Missing or invalid token"},
status_code=401
)
token = auth_header[7:]
# 验证 token（这里简化，实际应查询 OAuth Provider）
if not validate_token(token):
return JSONResponse(
{"error": "forbidden", "message": "Invalid or expired token"},
status_code=403
)
response = await call_next(request)
return response
def validate_token(token: str) -> bool:
"""验证 OAuth Token"""
# 实际实现：查询 OAuth Provider 的 /introspect 端点
# 或验证 JWT 签名
return len(token) > 10 # 简化示例
7.3 水平扩展
plaintext
┌──────────────────────────────────────────────────────────────┐
│ MCP Server 水平扩展架构 │
│ │
│ ┌──────────────┐ │
│ │ 负载均衡器 │ │
│ │ (Nginx/ALB) │ │
│ └──────┬───────┘ │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ │ │ │ │
│ ┌──────▼─────┐ ┌────▼──────┐ ┌────▼──────┐ │
│ │ MCP Server │ │ MCP Server│ │ MCP Server│ │
│ │ Instance 1│ │ Instance 2│ │ Instance 3│ │
│ └──────┬─────┘ └────┬──────┘ └────┬──────┘ │
│ │ │ │ │
│ └─────────────┼─────────────┘ │
│ │ │
│ ┌──────▼───────┐ │
│ │ 共享存储层 │ │
│ │ (Redis/DB) │ │
│ └──────────────┘ │
│ │
│ 注意：Streamable HTTP 的会话亲和性 │
│ - 同一个 Mcp-Session-Id 必须路由到同一个实例 │
│ - 使用 Nginx 的 hash 负载均衡策略 │
└──────────────────────────────────────────────────────────────┘
Nginx 配置示例：
nginx
upstream mcp_servers {
# 基于 Session ID 的会话亲和
hash $http_mcp_session_id consistent;
server mcp-server-1:8080;
server mcp-server-2:8080;
server mcp-server-3:8080;
}
server {
listen 443 ssl;
server_name mcp.example.com;
location /mcp {
proxy_pass http://mcp_servers;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header Host $host;
# SSE 支持
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 86400s;
}
}
7.4 监控
python
import time
import logging
from collections import defaultdict
from dataclasses import dataclass, field
logger = logging.getLogger("mcp-monitor")
@dataclass
class MCPMetrics:
"""MCP Server 监控指标"""
tool_calls: dict = field(default_factory=lambda: defaultdict(int))
tool_errors: dict = field(default_factory=lambda: defaultdict(int))
tool_latency: dict = field(default_factory=lambda: defaultdict(list))
active_sessions: int = 0
total_requests: int = 0
def record_call(self, tool_name: str, latency_ms: float, success: bool):
self.tool_calls[tool_name] += 1
self.total_requests += 1
if not success:
self.tool_errors[tool_name] += 1
self.tool_latency[tool_name].append(latency_ms)
# 只保留最近 1000 条延迟记录
if len(self.tool_latency[tool_name]) > 1000:
self.tool_latency[tool_name] = self.tool_latency[tool_name][-1000:]
def get_summary(self) -> dict:
summary = {}
for tool, count in self.tool_calls.items():
latencies = self.tool_latency[tool]
summary[tool] = {
"calls": count,
"errors": self.tool_errors[tool],
"error_rate": self.tool_errors[tool] / max(count, 1),
"avg_latency_ms": sum(latencies) / max(len(latencies), 1),
"p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] if latencies else 0,
}
return summary
metrics = MCPMetrics()
# 在工具调用前后记录
def track_tool(func):
"""装饰器：记录工具调用的延迟和成功率"""
async def wrapper(*args, **kwargs):
start = time.time()
success = True
try:
result = await func(*args, **kwargs)
return result
except Exception as e:
success = False
raise
finally:
latency = (time.time() - start) * 1000
metrics.record_call(func.__name__, latency, success)
logger.info(f"Tool {func.__name__}: {latency:.1f}ms, success={success}")
return wrapper
8. 与 Skill 的关系：MCP 是"连工具的手"，Skill 是"存能力的脑"
这是很多人混淆的概念，让我用一张图说清楚：
plaintext
┌──────────────────────────────────────────────────────────────┐
│ MCP 与 Skill 的定位对比 │
│ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ AI Agent │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────────┐ │ │
│ │ │ Skill │ │ MCP │ │ │
│ │ │ "存能力的脑"│ │ "连工具的手" │ │ │
│ │ │ │ │ │ │ │
│ │ │ • 知道"做什么"│ │ • 知道"怎么连" │ │ │
│ │ │ • 封装流程 │ │ • 标准化接口 │ │ │
│ │ │ • 行业知识 │ │ • 发现工具 │ │ │
│ │ │ • 最佳实践 │ │ • 调用工具 │ │ │
│ │ │ │ │ │ │ │
│ │ │ 格式： │ │ 格式： │ │ │
│ │ │ SKILL.md │ │ JSON-RPC 2.0 │ │ │
│ │ └──────┬──────┘ └────────┬────────┘ │ │
│ │ │ │ │ │
│ │ │ Skill 指导 Agent │ │ │
│ │ │ 何时调用哪个工具 │ │ │
│ │ └──────────────────────────────┘ │ │
│ │ │ │ │
│ └────────────────────────┼────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ MCP Server │ │
│ │ (外部工具) │ │
│ └─────────────────┘ │
└──────────────────────────────────────────────────────────────┘
场景举例：
"审查这段 Python 代码"
1. Agent 匹配到 code-review Skill → 知道要检查安全、性能、可维护性
2. Skill 指导 Agent 调用 MCP 工具 → read_file 读取代码
3. MCP Server 执行 read_file → 返回代码内容
4. Agent 按照 Skill 的规则分析代码 → 生成审查报告
一句话总结：MCP 解决的是"能不能连上"的问题，Skill 解决的是"会不会用"的问题。两者互补，不是竞争关系。
9. 踩坑记录：5 个常见问题
坑1：Stdio 模式下 Server 的 stdout 污染
问题：在 Stdio 模式下，MCP 通过 stdin/stdout 传递 JSON-RPC 消息。如果你的 Server 代码里有 print() 调试输出，它会混入 JSON-RPC 消息流，导致 Client 解析失败。
python
# ❌ 错误：print 输出会污染 JSON-RPC 消息流
print("Server started!") # Client 会收到：Server started!\n{"jsonrpc":"2.0"...}
# → JSON 解析失败！
# ✅ 正确：所有日志输出到 stderr
import sys
print("Server started!", file=sys.stderr)
# ✅ 更好的做法：使用 logging
import logging
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
logger = logging.getLogger(__name__)
logger.info("Server started!")
排查方法：如果 Client 报 Unexpected token 或 JSON 解析错误，检查 Server 是否有 stdout 输出。
坑2：能力协商不匹配
问题：Client 请求了 Server 没有声明的能力，导致静默失败。
python
# ❌ 错误：没有在 capabilities 中声明 tools 支持
# Client 调用 list_tools() 时可能得到空结果
# ✅ 正确：在 Server 初始化时声明完整能力
# FastMCP 自动处理能力声明，但如果你手动实现，必须显式声明
坑3：Session-Id 丢失导致状态丢失
问题：在 Streamable HTTP 模式下，如果 Client 没有在后续请求中携带 Mcp-Session-Id Header，Server 会认为这是新会话，之前的所有状态（订阅、上下文）都丢了。
python
# ✅ 正确：每次请求都携带 Session ID
import httpx
async def call_with_session(url: str, session_id: str, payload: dict):
headers = {
"Content-Type": "application/json",
"Mcp-Session-Id": session_id # 必须携带！
}
async with httpx.AsyncClient() as client:
response = await client.post(url, json=payload, headers=headers)
return response.json()
坑4：工具描述不够具体导致 LLM 误调
问题：工具的 description 太模糊，LLM 在不该调用的时候调用了，或者该调用的时候没调用。
python
# ❌ 错误：描述太模糊
@mcp.tool()
def search(query: str) -> str:
"""Search for something"""
...
# ✅ 正确：描述包含触发条件和使用场景
@mcp.tool()
def search_files(pattern: str, path: str = ".", max_depth: int = 5) -> str:
"""
在工作空间中递归搜索文件。
当用户需要查找特定文件、按名称模式搜索文件时使用。
触发词：'找文件', '搜索文件', 'where is', 'find file'。
不适用于：搜索文件内容（请用 grep_tool）。
"""
...
坑5：资源泄漏——忘记关闭连接
问题：MCP Client 使用 async context manager，如果忘记关闭，会留下僵尸进程。
python
# ❌ 错误：忘记关闭连接
async def bad_example():
client = MCPClient(...)
await client.connect()
result = await client.call_tool("read_file", {"path": "test.txt"})
return result
# Server 进程永远不会被终止！
# ✅ 正确：使用 try/finally 确保清理
async def good_example():
client = MCPClient(...)
try:
await client.connect()
result = await client.call_tool("read_file", {"path": "test.txt"})
return result
finally:
await client.disconnect()
10. 总结与展望
MCP 的现状
MCP 已经从一个 Anthropic 的内部项目，成长为 AI 工具集成的事实标准。9700 万月 SDK 下载量、10,000+ 公开 Server、Linux 基金会治理——这些数字说明了一切。
选择 MCP 的时机
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 是否需要 MCP？决策树 │
│ │
│ 你需要连接外部工具吗？ │
│ │ │
│ ├── 否 → 不需要 MCP │
│ │ │
│ └── 是 → 需要连接的工具数量 > 3？ │
│ │ │
│ ├── 否 → 手动集成可能更简单 │
│ │ │
│ └── 是 → 需要支持多个 AI 应用？ │
│ │ │
│ ├── 否 → MCP 也值得考虑 │
│ │ │
│ └── 是 → ✅ 必须用 MCP │
└──────────────────────────────────────────────────────────────┘
MCP 的未来
据《MCP (Model Context Protocol): A Developer's Guide for 2026》（https://www.marsdevs.com/blog/model-context-protocol-mcp）分析，MCP 的演进方向包括：
Tool Search：Server 可以声明工具的语义标签，Client 可以按功能搜索而非按名称
Async Operations：长时间运行的工具调用支持异步结果通知
Server Identity Verification：企业级的服务器身份验证
与 A2A 协议互补：MCP 负责 Agent→Tool（垂直），A2A 负责 Agent→Agent（水平）
MCP 不是万能药，但它是 2026 年 AI 工程化基础设施的基石。掌握它，你就掌握了 AI 应用与外部世界连接的标准语言。
本文作者：PySuper | 来源：zhengxingtao.com
参考资源：
MCP 官方规范：https://modelcontextprotocol.io
MCP GitHub 组织：https://github.com/modelcontextprotocol
Anthropic MCP 公告：https://www.anthropic.com/news/model-context-protocol
评论区