MCP 工具生态与高级开发教程
定位：进阶/高级教程 | 适合人群：有 MCP 基础的 AI 开发者与架构师 预计阅读时间：3-4 小时 | 关键词：MCP 工具生态、MCP Server 高级开发、MCP 安全、MCP 网关、Agent 集成、生产部署
目录
- 第一章 MCP 协议深度剖析
- 第二章 高级 Server 开发模式
- 第三章 自定义 Transport 实现
- 第四章 MCP 安全加固
- 第五章 生产级 Server 架构
- 第六章 主流 MCP 工具生态全景
- 第七章 MCP 与 Agent 框架集成
- 第八章 性能优化
- 第九章 MCP 网关与代理
- 第十章 实战项目：企业级 MCP Server 集群
- 第十一章 常见问题与解决方案
- 总结与展望
第一章 MCP 协议深度剖析
1.1 JSON-RPC 2.0 与 MCP 的关系
MCP（Model Context Protocol）建立在 JSON-RPC 2.0 协议之上，但做了大量扩展。理解底层协议是高级开发的基础。
JSON-RPC 2.0 核心结构：
// 请求
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": { "path": "/etc/config.json" }
}
}
// 响应
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [{ "type": "text", "text": "{\"key\":\"value\"}" }]
}
}
// 通知（无 id，无需响应）
{
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": { "progress": 50, "total": 100 }
}
MCP 的扩展点：
MCP 在 JSON-RPC 2.0 基础上增加了：
- 能力协商：初始化阶段双向声明能力
- 生命周期管理：
initialize/initialized/shutdown三阶段 - 进度通知：长时间操作的进度报告
- 资源订阅：资源变更的实时推送
1.2 MCP 消息类型详解
MCP 定义了四大核心原语，每个原语都有对应的请求/响应/通知消息：
初始化握手：
// 客户端 → 服务端：initialize
{
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {
"roots": { "listChanged": true },
"sampling": {}
},
"clientInfo": {
"name": "MyAgent",
"version": "1.0.0"
}
}
}
// 服务端 → 客户端：initialize 响应
{
"jsonrpc": "2.0",
"id": 0,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"tools": { "listChanged": true },
"resources": { "subscribe": true, "listChanged": true },
"prompts": { "listChanged": true },
"logging": {}
},
"serverInfo": {
"name": "AdvancedMCPServer",
"version": "2.0.0"
}
}
}
// 客户端 → 服务端：initialized 通知
{
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
能力矩阵：
| 能力 | 方向 | 说明 |
|---|---|---|
tools |
Server → Client | 服务端提供可调用工具 |
resources |
Server → Client | 服务端暴露可读资源 |
prompts |
Server → Client | 服务端提供提示模板 |
sampling |
Client → Server | 服务端可请求客户端进行 LLM 采样 |
roots |
Client → Server | 客户端提供文件系统根目录 |
logging |
Server → Client | 服务端发送日志消息 |
1.3 协议版本演进与兼容性
MCP 协议使用日期格式版本号（如 2024-11-05、2025-03-26），每次更新都需关注向后兼容性。
# 版本兼容性检查器
class MCPVersionNegotiator:
"""MCP 协议版本协商器"""
SUPPORTED_VERSIONS = [
"2025-03-26",
"2024-11-05",
]
def negotiate(self, client_version: str, server_versions: list[str]) -> str:
"""协商双方都支持的最高版本"""
# 客户端声明的版本优先
if client_version in server_versions:
return client_version
# 回退到服务器支持的最高版本
for sv in server_versions:
if sv in self.SUPPORTED_VERSIONS:
return sv
raise ValueError(f"无共同支持的协议版本: client={client_version}, server={server_versions}")
def check_feature(self, version: str, feature: str) -> bool:
"""检查某版本是否支持特定特性"""
feature_matrix = {
"2024-11-05": ["tools", "resources", "prompts"],
"2025-03-26": ["tools", "resources", "prompts", "sampling", "logging", "completions"],
}
return feature in feature_matrix.get(version, [])
第二章 高级 Server 开发模式
2.1 资源（Resources）高级用法
基础的资源暴露是静态的，高级场景需要动态资源、资源模板和变更通知。
动态资源与模板：
// TypeScript MCP Server - 动态资源模板
import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
const server = new McpServer({ name: "DynamicResourceServer", version: "1.0.0" });
// 使用 URI 模板（RFC 6570）定义动态资源
server.resource(
"database-records",
new ResourceTemplate("db://{table}/{id}", { list: async () => {
// 动态列举可用资源
const tables = await db.getTables();
const resources = [];
for (const table of tables) {
const rows = await db.query(`SELECT id FROM ${table} LIMIT 100`);
for (const row of rows) {
resources.push({ uri: `db://${table}/${row.id}`, name: `${table}#${row.id}` });
}
}
return resources;
}}),
async (uri, { table, id }) => {
// 安全校验
if (!isValidTableName(table)) {
throw new Error(`非法表名: ${table}`);
}
const record = await db.query(`SELECT * FROM ${table} WHERE id = ?`, [id]);
return {
contents: [{
uri: uri.href,
mimeType: "application/json",
text: JSON.stringify(record, null, 2)
}]
};
}
);
// 资源变更通知
server.resource(
"config",
"config://app/settings",
async (uri) => ({
contents: [{
uri: uri.href,
mimeType: "application/json",
text: JSON.stringify(await loadConfig())
}]
})
);
// 当配置文件变更时，通知客户端
function onConfigChanged() {
server.sendResourceListChanged(); // 通知资源列表可能变化
// 也可以通过 notification 通知特定资源变更
}
资源订阅机制：
# Python MCP Server - 资源订阅
import asyncio
from mcp.server import Server
from mcp.types import Resource, ResourceUpdatedNotification
server = Server("SubscriptionServer")
# 存储订阅关系
_subscriptions: dict[str, set[str]] = {} # uri -> session_ids
@server.list_resources()
async def list_resources():
return [
Resource(
uri="metrics://system/cpu",
name="CPU 使用率",
mimeType="application/json",
description="实时 CPU 使用率数据"
),
Resource(
uri="metrics://system/memory",
name="内存使用率",
mimeType="application/json"
),
]
@server.subscribe_resource()
async def subscribe_resource(uri: str, session_id: str):
"""客户端订阅资源变更"""
if uri not in _subscriptions:
_subscriptions[uri] = set()
_subscriptions[uri].add(session_id)
return {"success": True}
@server.unsubscribe_resource()
async def unsubscribe_resource(uri: str, session_id: str):
"""取消订阅"""
_subscriptions.get(uri, set()).discard(session_id)
return {"success": True}
# 后台监控任务：检测资源变更并推送
async def monitor_resources():
prev_cpu = None
while True:
current_cpu = await get_cpu_usage()
if prev_cpu is not None and abs(current_cpu - prev_cpu) > 5: # 变化超过5%
uri = "metrics://system/cpu"
for session_id in _subscriptions.get(uri, set()):
await server.send_notification(
session_id,
ResourceUpdatedNotification(uri=uri)
)
prev_cpu = current_cpu
await asyncio.sleep(1)
2.2 工具（Tools）高级设计模式
模式一：组合工具（Composite Tool）
将多个原子操作封装为一个高级工具，减少 LLM 的工具调用轮次：
// 组合工具：一次性完成代码分析
server.tool(
"analyze_codebase",
"对代码仓库进行全面分析，包括结构、依赖、质量和安全扫描",
{
repo_path: z.string().describe("仓库根路径"),
analysis_types: z.array(z.enum(["structure", "deps", "quality", "security"]))
.default(["structure", "deps", "quality"])
.describe("分析类型列表"),
depth: z.number().int().min(1).max(10).default(3).describe("目录扫描深度"),
},
async ({ repo_path, analysis_types, depth }) => {
const results: Record<string, any> = {};
// 并行执行各分析模块
const tasks = analysis_types.map(async (type) => {
switch (type) {
case "structure":
results.structure = await analyzeStructure(repo_path, depth);
break;
case "deps":
results.dependencies = await analyzeDependencies(repo_path);
break;
case "quality":
results.quality = await runQualityChecks(repo_path);
break;
case "security":
results.security = await runSecurityScan(repo_path);
break;
}
});
await Promise.all(tasks);
// 生成综合报告
const report = generateReport(results);
return {
content: [{
type: "text",
text: JSON.stringify(report, null, 2)
}]
};
}
);
模式二：流式工具（Streaming Tool）
对于耗时操作，通过进度通知实时报告状态：
# Python - 流式工具实现
from mcp.server import Server
from mcp.types import Tool, TextContent, ProgressNotification
server = Server("StreamingToolServer")
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "large_file_processor":
return await process_large_file(arguments)
async def process_large_file(args: dict):
file_path = args["file_path"]
total_lines = count_lines(file_path)
processed = 0
results = []
with open(file_path, 'r') as f:
for line_num, line in enumerate(f, 1):
result = process_line(line)
results.append(result)
# 每100行报告一次进度
if line_num % 100 == 0:
await server.send_notification(
"notifications/progress",
{
"progress": line_num,
"total": total_lines,
"message": f"已处理 {line_num}/{total_lines} 行"
}
)
return [TextContent(type="text", text=json.dumps({
"file": file_path,
"total_lines": total_lines,
"results_count": len(results),
"summary": summarize(results)
}))]
模式三：条件工具暴露
根据客户端能力或用户权限动态暴露工具：
// 根据权限动态暴露工具
server.setRequestHandler(ListToolsRequestSchema, async (request) => {
const userRole = request.meta?.userRole ?? "viewer";
const allTools = getToolDefinitions();
// 根据角色过滤工具
const filteredTools = allTools.filter(tool => {
const requiredRole = tool.annotations?.requiredRole ?? "viewer";
return hasPermission(userRole, requiredRole);
});
return { tools: filteredTools };
});
2.3 提示模板（Prompts）动态生成
// 动态提示模板 - 根据上下文生成
server.prompt(
"code_review",
"生成代码审查提示",
{
language: z.string().describe("编程语言"),
focus: z.enum(["security", "performance", "readability", "all"]).default("all"),
style: z.enum(["strict", "friendly", "mentor"]).default("friendly"),
},
async ({ language, focus, style }) => {
const guidelines = await loadCodingGuidelines(language);
const reviewFocus = {
security: "重点关注安全漏洞、输入验证、权限控制",
performance: "重点关注性能瓶颈、内存泄漏、算法复杂度",
readability: "重点关注命名规范、代码结构、注释质量",
all: "全面审查安全性、性能和可读性"
};
const styleMap = {
strict: "以严格的标准审查，指出所有不符合最佳实践的地方",
friendly: "以友好的语气指出改进点，同时肯定做得好的部分",
mentor: "以导师的角度，不仅指出问题，还解释为什么以及如何改进"
};
return {
messages: [{
role: "user",
content: {
type: "text",
text: `请审查以下 ${language} 代码。
审查重点：${reviewFocus[focus]}
审查风格：${styleMap[style]}
编码规范参考：
${guidelines}
请按以下格式输出审查结果：
1. 🔴 严重问题（必须修复）
2. 🟡 建议改进（推荐修复）
3. 🟢 代码亮点（做得好的地方）
4. 📝 总结评价`
}
}]
};
}
);
2.4 采样（Sampling）与人机协作
采样是 MCP 的独特能力——服务端可以反向请求客户端的 LLM 进行推理：
// 服务端请求客户端 LLM 采样
server.tool(
"smart_analyze",
"使用 AI 智能分析数据，必要时请求人类确认",
{
data: z.string().describe("待分析数据"),
auto_confirm: z.boolean().default(false).describe("是否自动确认"),
},
async ({ data, auto_confirm }) => {
// 第一步：请求 LLM 初步分析
const analysis = await server.requestSampling({
messages: [{
role: "user",
content: {
type: "text",
text: `请分析以下数据并提取关键信息：\n${data}`
}
}],
maxTokens: 1000,
modelPreferences: {
hints: [{ name: "claude-sonnet-4-20250514" }],
costPriority: 0.3,
speedPriority: 0.7,
}
});
// 第二步：如果分析结果不确定，请求人类确认
if (!auto_confirm && analysis.confidence < 0.8) {
const humanReview = await server.requestSampling({
messages: [{
role: "user",
content: {
type: "text",
text: `AI 分析结果置信度较低（${analysis.confidence}），请审查：\n${analysis.result}\n\n是否需要补充数据或调整分析方向？`
}
}],
includeContext: "thisServer",
maxTokens: 500,
});
return {
content: [{
type: "text",
text: JSON.stringify({
analysis: analysis.result,
confidence: analysis.confidence,
human_feedback: humanReview.content,
final_recommendation: mergeResults(analysis, humanReview)
}, null, 2)
}]
};
}
return {
content: [{
type: "text",
text: JSON.stringify({
analysis: analysis.result,
confidence: analysis.confidence
}, null, 2)
}]
};
}
);
第三章 自定义 Transport 实现
3.1 Transport 抽象层设计
MCP 的核心设计之一是 Transport 的可插拔性。理解 Transport 抽象层是构建自定义通信方式的基础。
// Transport 接口定义（TypeScript）
import { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js";
export interface Transport {
// 生命周期
start(): Promise<void>;
close(): Promise<void>;
// 消息收发
send(message: JSONRPCMessage): Promise<void>;
// 事件回调
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
// 会话标识
sessionId?: string;
}
3.2 WebSocket Transport
// WebSocket Transport 实现
import { WebSocketServer, WebSocket } from "ws";
import { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "@modelcontextprotocol/sdk/types.js";
export class WebSocketTransport implements Transport {
private wss: WebSocketServer | null = null;
private ws: WebSocket | null = null;
onclose?: () => void;
onerror?: (error: Error) => void;
onmessage?: (message: JSONRPCMessage) => void;
sessionId?: string;
constructor(
private options: {
port: number;
host?: string;
path?: string;
authenticate?: (req: IncomingMessage) => Promise<boolean>;
}
) {
this.sessionId = crypto.randomUUID();
}
async start(): Promise<void> {
return new Promise((resolve, reject) => {
this.wss = new WebSocketServer({
port: this.options.port,
host: this.options.host ?? "0.0.0.0",
path: this.options.path ?? "/mcp",
});
this.wss.on("connection", async (ws, req) => {
// 认证检查
if (this.options.authenticate) {
const authorized = await this.options.authenticate(req);
if (!authorized) {
ws.close(4001, "Unauthorized");
return;
}
}
this.ws = ws;
ws.on("message", (data) => {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(data.toString()));
this.onmessage?.(message);
} catch (err) {
this.onerror?.(new Error(`消息解析失败: ${err}`));
}
});
ws.on("close", () => this.onclose?.());
ws.on("error", (err) => this.onerror?.(err));
resolve();
});
this.wss.on("error", reject);
});
}
async send(message: JSONRPCMessage): Promise<void> {
if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
throw new Error("WebSocket 未连接");
}
this.ws.send(JSON.stringify(message));
}
async close(): Promise<void> {
this.ws?.close(1000, "Server shutting down");
this.wss?.close();
}
}
3.3 gRPC Transport
// mcp_transport.proto
syntax = "proto3";
package mcp;
service MCPService {
// 双向流式 RPC，用于 MCP 消息传递
rpc MessageStream (stream MCPMessage) returns (stream MCPMessage);
}
message MCPMessage {
bytes payload = 1; // JSON-RPC 消息的 JSON 字节
string session_id = 2;
int64 timestamp = 3;
}
# gRPC Transport 实现（Python）
import grpc
import json
from concurrent import futures
from mcp.server.transport import Transport
from mcp.types import JSONRPCMessage
import mcp_transport_pb2 as pb2
import mcp_transport_pb2_grpc as pb2_grpc
class GRPCServicer(pb2_grpc.MCPServiceServicer):
"""gRPC MCP 服务端"""
def __init__(self):
self._message_queue = asyncio.Queue()
self._response_streams = {}
async def MessageStream(self, request_iterator, context):
session_id = None
# 启动接收协程
async for msg in request_iterator:
session_id = msg.session_id
payload = json.loads(msg.payload.decode('utf-8'))
message = JSONRPCMessage(**payload)
# 放入消息队列等待处理
await self._message_queue.put((session_id, message))
# 等待响应
response = await self._wait_for_response(session_id, message.id)
yield pb2.MCPMessage(
payload=json.dumps(response).encode('utf-8'),
session_id=session_id,
timestamp=int(time.time() * 1000)
)
class GRPCTransport(Transport):
"""gRPC Transport 实现"""
def __init__(self, host: str = "0.0.0.0", port: int = 50051):
self.host = host
self.port = port
self.server = None
self._message_queue = asyncio.Queue()
async def start(self):
self.server = grpc.aio.server(
futures.ThreadPoolExecutor(max_workers=10),
options=[
('grpc.max_receive_message_length', 50 * 1024 * 1024), # 50MB
('grpc.max_send_message_length', 50 * 1024 * 1024),
]
)
servicer = GRPCServicer()
pb2_grpc.add_MCPServiceServicer_to_server(servicer, self.server)
self.server.add_insecure_port(f"{self.host}:{self.port}")
await self.server.start()
print(f"gRPC MCP Server 启动于 {self.host}:{self.port}")
async def send(self, message: JSONRPCMessage):
# 通过 gRPC 流发送
pass
async def close(self):
if self.server:
await self.server.stop(grace=5)
3.4 消息队列 Transport
适用于分布式场景，多个 MCP Server 通过消息队列解耦：
# RabbitMQ Transport 实现
import asyncio
import json
import aio_pika
from mcp.server.transport import Transport
from mcp.types import JSONRPCMessage
class RabbitMQTransport(Transport):
"""基于 RabbitMQ 的 MCP Transport"""
def __init__(self, amqp_url: str, queue_name: str = "mcp_messages"):
self.amqp_url = amqp_url
self.queue_name = queue_name
self.connection = None
self.channel = None
self.exchange = None
self._message_callback = None
async def start(self):
# 建立连接
self.connection = await aio_pika.connect_robust(self.amqp_url)
self.channel = await self.connection.channel()
# 声明交换机和队列
self.exchange = await self.channel.declare_exchange(
"mcp_exchange", aio_pika.ExchangeType.DIRECT, durable=True
)
queue = await self.channel.declare_queue(
self.queue_name, durable=True
)
# 绑定队列到交换机
await queue.bind(self.exchange, routing_key=self.queue_name)
# 开始消费消息
await queue.consume(self._on_message)
print(f"RabbitMQ Transport 已连接，监听队列: {self.queue_name}")
async def _on_message(self, message: aio_pika.IncomingMessage):
async with message.process():
payload = json.loads(message.body.decode('utf-8'))
rpc_message = JSONRPCMessage(**payload)
if self._message_callback:
await self._message_callback(rpc_message)
async def send(self, message: JSONRPCMessage):
"""发送消息到指定队列"""
await self.exchange.publish(
aio_pika.Message(
body=json.dumps(message.dict()).encode('utf-8'),
delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
),
routing_key=self.queue_name,
)
async def close(self):
if self.connection:
await self.connection.close()
第四章 MCP 安全加固
4.1 认证与授权机制
OAuth 2.0 集成：
# MCP Server OAuth 2.0 认证中间件
import jwt
import httpx
from functools import wraps
from datetime import datetime, timedelta
class MCPOAuth2Authenticator:
"""MCP Server OAuth 2.0 认证器"""
def __init__(
self,
issuer_url: str,
audience: str,
jwks_cache_ttl: int = 3600
):
self.issuer_url = issuer_url
self.audience = audience
self.jwks_cache_ttl = jwks_cache_ttl
self._jwks_cache = None
self._jwks_cache_time = None
async def get_jwks(self) -> dict:
"""获取并缓存 JWKS"""
now = datetime.utcnow()
if (self._jwks_cache and self._jwks_cache_time and
(now - self._jwks_cache_time).seconds < self.jwks_cache_ttl):
return self._jwks_cache
async with httpx.AsyncClient() as client:
resp = await client.get(f"{self.issuer_url}/.well-known/jwks.json")
self._jwks_cache = resp.json()
self._jwks_cache_time = now
return self._jwks_cache
async def validate_token(self, token: str) -> dict:
"""验证 JWT Token"""
jwks = await self.get_jwks()
# 获取 token 头部的 kid
unverified_header = jwt.get_unverified_header(token)
kid = unverified_header.get("kid")
# 查找对应的公钥
rsa_key = None
for key in jwks.get("keys", []):
if key["kid"] == kid:
rsa_key = jwt.algorithms.RSAAlgorithm.from_jwk(key)
break
if not rsa_key:
raise ValueError(f"未找到 kid={kid} 对应的公钥")
# 验证 token
payload = jwt.decode(
token,
rsa_key,
algorithms=["RS256"],
audience=self.audience,
issuer=self.issuer_url,
)
return payload
class MCPPermissionManager:
"""基于 RBAC 的 MCP 权限管理器"""
# 工具访问权限矩阵
TOOL_PERMISSIONS = {
"admin": ["*"], # 管理员：所有工具
"developer": [
"file_read", "file_write", "code_*", # 文件和代码工具
"db_query", "api_call", # 数据库和 API
],
"analyst": [
"file_read", "db_query", # 只读访问
"report_*", # 报表工具
],
"viewer": ["file_read"], # 最小权限
}
def check_tool_access(self, role: str, tool_name: str) -> bool:
"""检查角色是否有权限访问指定工具"""
permissions = self.TOOL_PERMISSIONS.get(role, [])
for perm in permissions:
if perm == "*":
return True
if perm.endswith("*") and tool_name.startswith(perm[:-1]):
return True
if perm == tool_name:
return True
return False
def filter_tools(self, role: str, tools: list) -> list:
"""过滤出角色可访问的工具列表"""
return [t for t in tools if self.check_tool_access(role, t.name)]
4.2 输入验证与沙箱隔离
# MCP 工具输入验证框架
import os
import re
from pathlib import Path
from typing import Any
from pydantic import BaseModel, validator, field_validator
class FilePathValidator(BaseModel):
"""文件路径安全验证"""
path: str
@field_validator("path")
@classmethod
def validate_path(cls, v):
# 1. 检查路径遍历攻击
if ".." in v:
raise ValueError("路径中不允许包含 '..'")
# 2. 规范化路径
normalized = os.path.normpath(v)
# 3. 检查绝对路径（如不允许）
if os.path.isabs(normalized):
allowed_roots = ["/data/workspace", "/tmp/mcp"]
if not any(normalized.startswith(root) for root in allowed_roots):
raise ValueError(f"绝对路径不允许: {normalized}")
# 4. 检查危险字符
dangerous_patterns = [
r'[;|&`$]', # Shell 注入
r'\x00', # 空字节
r'[\x01-\x1f]', # 控制字符
]
for pattern in dangerous_patterns:
if re.search(pattern, v):
raise ValueError(f"路径包含非法字符")
return normalized
class SQLQueryValidator(BaseModel):
"""SQL 查询安全验证"""
query: str
@field_validator("query")
@classmethod
def validate_sql(cls, v):
# 1. 只允许 SELECT 语句（除非明确授权）
stripped = v.strip().upper()
if not stripped.startswith("SELECT"):
raise ValueError("只允许 SELECT 查询")
# 2. 禁止危险关键字
forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "TRUNCATE", "EXEC"]
for keyword in forbidden:
if re.search(rf'\b{keyword}\b', v, re.IGNORECASE):
raise ValueError(f"禁止使用 {keyword} 语句")
# 3. 检查注释（可能隐藏恶意代码）
if "--" in v or "/*" in v:
raise ValueError("查询中不允许包含注释")
return v
# 沙箱执行器
import subprocess
import tempfile
import resource
class SandboxExecutor:
"""安全沙箱执行器"""
def __init__(
self,
max_memory_mb: int = 256,
max_cpu_seconds: int = 30,
allowed_syscalls: list[str] = None,
network_access: bool = False,
):
self.max_memory_mb = max_memory_mb
self.max_cpu_seconds = max_cpu_seconds
self.network_access = network_access
def execute(self, command: str, stdin_data: str = "") -> dict:
"""在沙箱中执行命令"""
with tempfile.TemporaryDirectory() as tmpdir:
# 构建安全的执行命令
wrapper_script = f"""#!/bin/bash
set -euo pipefail
# 资源限制
ulimit -v {self.max_memory_mb * 1024}
ulimit -t {self.max_cpu_seconds}
# 网络限制
{"# 网络访问已禁用" if not self.network_access else ""}
# 执行命令
cd {tmpdir}
{command}
"""
script_path = os.path.join(tmpdir, "exec.sh")
with open(script_path, "w") as f:
f.write(wrapper_script)
os.chmod(script_path, 0o700)
try:
result = subprocess.run(
["bash", script_path],
input=stdin_data,
capture_output=True,
text=True,
timeout=self.max_cpu_seconds + 5,
cwd=tmpdir,
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"exit_code": result.returncode,
"success": result.returncode == 0,
}
except subprocess.TimeoutExpired:
return {
"stdout": "",
"stderr": "执行超时",
"exit_code": -1,
"success": False,
}
4.3 审计日志与监控
# MCP 审计日志系统
import json
import time
import logging
from enum import Enum
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
class AuditEventType(Enum):
TOOL_CALL = "tool_call"
RESOURCE_ACCESS = "resource_access"
AUTH_SUCCESS = "auth_success"
AUTH_FAILURE = "auth_failure"
PERMISSION_DENIED = "permission_denied"
ERROR = "error"
SESSION_START = "session_start"
SESSION_END = "session_end"
@dataclass
class AuditEvent:
timestamp: float
event_type: AuditEventType
session_id: str
user_id: Optional[str]
tool_name: Optional[str]
resource_uri: Optional[str]
details: dict
ip_address: Optional[str] = None
duration_ms: Optional[float] = None
success: bool = True
error_message: Optional[str] = None
class MCPAuditLogger:
"""MCP 审计日志记录器"""
def __init__(self, log_file: str = "/var/log/mcp/audit.jsonl"):
self.logger = logging.getLogger("mcp.audit")
self.log_file = log_file
# 文件处理器
handler = logging.FileHandler(log_file)
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
# 实时指标
self._metrics = {
"total_calls": 0,
"errors": 0,
"denied": 0,
"avg_duration_ms": 0,
}
def log_event(self, event: AuditEvent):
"""记录审计事件"""
self._metrics["total_calls"] += 1
if not event.success:
self._metrics["errors"] += 1
if event.event_type == AuditEventType.PERMISSION_DENIED:
self._metrics["denied"] += 1
# 写入 JSONL 文件
self.logger.info(json.dumps(asdict(event), default=str))
def log_tool_call(
self,
session_id: str,
user_id: str,
tool_name: str,
arguments: dict,
result: Any,
duration_ms: float,
success: bool = True,
error: str = None,
):
"""记录工具调用"""
event = AuditEvent(
timestamp=time.time(),
event_type=AuditEventType.TOOL_CALL,
session_id=session_id,
user_id=user_id,
tool_name=tool_name,
resource_uri=None,
details={
"arguments": self._sanitize_args(arguments),
"result_size": len(str(result)) if result else 0,
},
duration_ms=duration_ms,
success=success,
error_message=error,
)
self.log_event(event)
def _sanitize_args(self, args: dict) -> dict:
"""脱敏处理参数中的敏感字段"""
sensitive_keys = {"password", "token", "secret", "api_key", "authorization"}
return {
k: "***REDACTED***" if k.lower() in sensitive_keys else v
for k, v in args.items()
}
def get_metrics(self) -> dict:
"""获取实时指标"""
return self._metrics.copy()
第五章 生产级 Server 架构
5.1 高可用架构设计
┌─────────────────┐
│ Load Balancer │
│ (Nginx/HAProxy)│
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌─────┴─────┐ ┌─────┴─────┐ ┌─────┴─────┐
│ MCP Node │ │ MCP Node │ │ MCP Node │
│ #1 │ │ #2 │ │ #3 │
└─────┬─────┘ └─────┬─────┘ └─────┬─────┘
│ │ │
└──────────────┼──────────────┘
│
┌────────┴────────┐
│ 共享状态层 │
│ (Redis Cluster) │
└─────────────────┘
5.2 水平扩展与负载均衡
// MCP 负载均衡器
import { createServer, IncomingMessage, ServerResponse } from "http";
import { WebSocket } from "ws";
interface MCPServerNode {
id: string;
host: string;
port: number;
healthy: boolean;
connections: number;
weight: number;
lastHealthCheck: Date;
}
class MCPLoadBalancer {
private nodes: MCPServerNode[] = [];
private healthCheckInterval: NodeJS.Timeout | null = null;
constructor(private config: {
healthCheckIntervalMs: number;
maxConnectionsPerNode: number;
strategy: "round-robin" | "least-connections" | "weighted";
}) {}
addNode(node: MCPServerNode): void {
this.nodes.push(node);
console.log(`节点已添加: ${node.id} (${node.host}:${node.port})`);
}
removeNode(nodeId: string): void {
this.nodes = this.nodes.filter(n => n.id !== nodeId);
}
// 选择目标节点
selectNode(): MCPServerNode | null {
const healthyNodes = this.nodes.filter(
n => n.healthy && n.connections < this.config.maxConnectionsPerNode
);
if (healthyNodes.length === 0) return null;
switch (this.config.strategy) {
case "round-robin":
return this.roundRobin(healthyNodes);
case "least-connections":
return this.leastConnections(healthyNodes);
case "weighted":
return this.weightedSelect(healthyNodes);
}
}
private roundRobin(nodes: MCPServerNode[]): MCPServerNode {
// 简单轮询
const now = Date.now();
return nodes[now % nodes.length];
}
private leastConnections(nodes: MCPServerNode[]): MCPServerNode {
return nodes.reduce((min, n) =>
n.connections < min.connections ? n : min
);
}
private weightedSelect(nodes: MCPServerNode[]): MCPServerNode {
const totalWeight = nodes.reduce((sum, n) => sum + n.weight, 0);
let random = Math.random() * totalWeight;
for (const node of nodes) {
random -= node.weight;
if (random <= 0) return node;
}
return nodes[0];
}
// 健康检查
startHealthChecks(): void {
this.healthCheckInterval = setInterval(async () => {
await Promise.all(this.nodes.map(node => this.checkNode(node)));
}, this.config.healthCheckIntervalMs);
}
private async checkNode(node: MCPServerNode): Promise<void> {
try {
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 5000);
const resp = await fetch(`http://${node.host}:${node.port}/health`, {
signal: controller.signal,
});
clearTimeout(timeout);
node.healthy = resp.ok;
node.lastHealthCheck = new Date();
} catch {
node.healthy = false;
console.warn(`节点健康检查失败: ${node.id}`);
}
}
stop(): void {
if (this.healthCheckInterval) {
clearInterval(this.healthCheckInterval);
}
}
}
5.3 健康检查与优雅停机
# 优雅停机管理器
import asyncio
import signal
import logging
from typing import Callable, Awaitable
logger = logging.getLogger("mcp.graceful_shutdown")
class GracefulShutdownManager:
"""优雅停机管理器"""
def __init__(self, shutdown_timeout: float = 30.0):
self.shutdown_timeout = shutdown_timeout
self._handlers: list[Callable[[], Awaitable[None]]] = []
self._shutting_down = False
self._active_connections = 0
def register_handler(self, handler: Callable[[], Awaitable[None]]):
"""注册停机处理函数"""
self._handlers.append(handler)
def track_connection(self):
"""跟踪活跃连接"""
self._active_connections += 1
def release_connection(self):
"""释放连接"""
self._active_connections = max(0, self._active_connections - 1)
def setup_signal_handlers(self, loop: asyncio.AbstractEventLoop):
"""设置信号处理器"""
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, lambda s=sig: asyncio.create_task(
self.shutdown(s)
))
async def shutdown(self, sig: signal.Signals = None):
"""执行优雅停机"""
if self._shutting_down:
return
self._shutting_down = True
if sig:
logger.info(f"收到信号 {sig.name}，开始优雅停机...")
# 1. 停止接受新连接
logger.info("停止接受新连接...")
# 2. 等待现有连接完成（带超时）
logger.info(f"等待 {self._active_connections} 个活跃连接完成...")
wait_start = asyncio.get_event_loop().time()
while (self._active_connections > 0 and
asyncio.get_event_loop().time() - wait_start < self.shutdown_timeout):
await asyncio.sleep(0.5)
if self._active_connections > 0:
logger.warning(f"超时！仍有 {self._active_connections} 个连接未完成")
# 3. 执行注册的清理处理器
for handler in self._handlers:
try:
await asyncio.wait_for(handler(), timeout=10)
except asyncio.TimeoutError:
logger.warning(f"处理器 {handler.__name__} 执行超时")
except Exception as e:
logger.error(f"处理器 {handler.__name__} 执行失败: {e}")
logger.info("优雅停机完成")
第六章 主流 MCP 工具生态全景
6.1 文件系统类 Server
官方 Filesystem Server 是最常用的 MCP Server 之一：
// 配置示例
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/home/user/projects",
"/home/user/documents"
]
}
}
}
自定义增强文件系统 Server：
// 增强版文件系统 Server - 支持搜索、监控、版本控制
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import * as fs from "fs/promises";
import * as path from "path";
import { glob } from "glob";
import * as chokidar from "chokidar";
const server = new McpServer({ name: "EnhancedFilesystem", version: "1.0.0" });
// 文件搜索工具
server.tool(
"search_files",
"在指定目录中搜索文件，支持 Glob 模式和正则表达式",
{
directory: z.string().describe("搜索根目录"),
pattern: z.string().describe("Glob 模式或正则表达式"),
search_type: z.enum(["glob", "regex"]).default("glob"),
max_results: z.number().int().min(1).max(1000).default(100),
include_content: z.boolean().default(false).describe("是否搜索文件内容"),
content_pattern: z.string().optional().describe("内容搜索正则"),
},
async ({ directory, pattern, search_type, max_results, include_content, content_pattern }) => {
const results: Array<{ path: string; matches?: string[] }> = [];
// 文件名搜索
if (search_type === "glob") {
const files = await glob(pattern, { cwd: directory, absolute: true });
results.push(...files.slice(0, max_results).map(p => ({ path: p })));
}
// 内容搜索
if (include_content && content_pattern) {
const regex = new RegExp(content_pattern, "gm");
for (const file of results) {
try {
const content = await fs.readFile(file.path, "utf-8");
const matches = content.match(regex);
if (matches) {
file.matches = matches.slice(0, 10);
}
} catch {
// 跳过不可读文件
}
}
}
return {
content: [{ type: "text", text: JSON.stringify({ total: results.length, files: results }, null, 2) }]
};
}
);
// 文件监控资源
server.resource(
"file-watcher",
"watcher://active",
async (uri) => {
// 返回当前活跃的文件监控列表
const watchers = getActiveWatchers();
return {
contents: [{
uri: uri.href,
mimeType: "application/json",
text: JSON.stringify(watchers, null, 2),
}]
};
}
);
6.2 数据库类 Server
多数据库统一访问 Server：
# 多数据库 MCP Server
import asyncpg
import aiomysql
import motor.motor_asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
server = Server("MultiDatabaseServer")
# 数据库连接池
_pools = {}
async def init_pools(config: dict):
"""初始化所有数据库连接池"""
if "postgres" in config:
_pools["postgres"] = await asyncpg.create_pool(
config["postgres"]["url"],
min_size=2,
max_size=20,
)
if "mysql" in config:
_pools["mysql"] = await aiomysql.create_pool(
host=config["mysql"]["host"],
port=config["mysql"]["port"],
user=config["mysql"]["user"],
password=config["mysql"]["password"],
db=config["mysql"]["database"],
minsize=2,
maxsize=20,
)
if "mongodb" in config:
client = motor.motor_asyncio.AsyncIOMotorClient(config["mongodb"]["url"])
_pools["mongodb"] = client[config["mongodb"]["database"]]
@server.list_tools()
async def list_tools():
return [
Tool(
name="db_query",
description="在指定数据库上执行查询",
inputSchema={
"type": "object",
"properties": {
"database": {
"type": "string",
"enum": list(_pools.keys()),
"description": "目标数据库"
},
"query": {
"type": "string",
"description": "SQL/MQL 查询语句"
},
"params": {
"type": "array",
"description": "查询参数（用于参数化查询）",
"items": {}
}
},
"required": ["database", "query"]
}
),
Tool(
name="db_schema",
description="获取数据库 Schema 信息",
inputSchema={
"type": "object",
"properties": {
"database": { "type": "string" },
"table": { "type": "string", "description": "可选，指定表名" }
},
"required": ["database"]
}
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "db_query":
return await execute_query(arguments)
elif name == "db_schema":
return await get_schema(arguments)
else:
raise ValueError(f"未知工具: {name}")
async def execute_query(args: dict) -> list:
db_type = args["database"]
query = args["query"]
params = args.get("params", [])
# 安全检查
if not query.strip().upper().startswith("SELECT"):
# 只读模式：非 SELECT 查询需要 admin 角色
raise PermissionError("只读模式下仅允许 SELECT 查询")
pool = _pools.get(db_type)
if not pool:
raise ValueError(f"数据库 {db_type} 未配置")
try:
if db_type == "postgres":
async with pool.acquire() as conn:
rows = await conn.fetch(query, *params)
columns = list(rows[0].keys()) if rows else []
data = [dict(row) for row in rows]
elif db_type == "mysql":
async with pool.acquire() as conn:
async with conn.cursor(aiomysql.DictCursor) as cur:
await cur.execute(query, params)
data = await cur.fetchall()
columns = list(data[0].keys()) if data else []
elif db_type == "mongodb":
# MongoDB 查询需要特殊处理
collection = pool[query] # query 作为集合名
data = await collection.find(params[0] if params else {}).to_list(1000)
columns = list(data[0].keys()) if data else []
return [TextContent(type="text", text=json.dumps({
"columns": columns,
"rows": data,
"count": len(data),
}, default=str, indent=2))]
except Exception as e:
return [TextContent(type="text", text=f"查询执行失败: {str(e)}")]
6.3 API 集成类 Server
通用 REST API 代理 Server：
// 通用 REST API MCP Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
interface APIEndpoint {
name: string;
baseUrl: string;
auth: {
type: "bearer" | "api_key" | "basic";
credentials: Record<string, string>;
};
endpoints: Array<{
path: string;
method: "GET" | "POST" | "PUT" | "DELETE" | "PATCH";
description: string;
parameters?: Record<string, any>;
}>;
}
class APIProxyServer {
private server: McpServer;
private apis: Map<string, APIEndpoint> = new Map();
constructor() {
this.server = new McpServer({ name: "APIProxy", version: "1.0.0" });
}
registerAPI(config: APIEndpoint): void {
this.apis.set(config.name, config);
// 为每个 API 端点注册 MCP 工具
for (const endpoint of config.endpoints) {
const toolName = `${config.name}_${endpoint.path.replace(/\//g, "_")}`;
this.server.tool(
toolName,
endpoint.description,
endpoint.parameters ?? {},
async (params) => {
return await this.callAPI(config.name, endpoint, params);
}
);
}
}
private async callAPI(
apiName: string,
endpoint: APIEndpoint["endpoints"][0],
params: Record<string, any>
) {
const api = this.apis.get(apiName)!;
const url = new URL(endpoint.path, api.baseUrl);
// 构建请求头
const headers: Record<string, string> = {
"Content-Type": "application/json",
};
// 认证
switch (api.auth.type) {
case "bearer":
headers["Authorization"] = `Bearer ${api.auth.credentials.token}`;
break;
case "api_key":
headers["X-API-Key"] = api.auth.credentials.apiKey;
break;
case "basic":
const encoded = Buffer.from(
`${api.auth.credentials.username}:${api.auth.credentials.password}`
).toString("base64");
headers["Authorization"] = `Basic ${encoded}`;
break;
}
// 发送请求
const response = await fetch(url.toString(), {
method: endpoint.method,
headers,
body: endpoint.method !== "GET" ? JSON.stringify(params) : undefined,
});
const data = await response.json();
return {
content: [{
type: "text" as const,
text: JSON.stringify({
status: response.status,
data,
}, null, 2),
}],
};
}
}
6.4 浏览器与搜索类 Server
Playwright 浏览器自动化 Server：
// 浏览器自动化 MCP Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { chromium, Browser, Page } from "playwright";
const server = new McpServer({ name: "BrowserAutomation", version: "1.0.0" });
let browser: Browser | null = null;
let currentPage: Page | null = null;
// 初始化浏览器
async function ensureBrowser(): Promise<Page> {
if (!browser) {
browser = await chromium.launch({
headless: true,
args: ["--no-sandbox", "--disable-setuid-sandbox"],
});
}
if (!currentPage) {
const context = await browser.newContext({
userAgent: "Mozilla/5.0 (compatible; MCPBot/1.0)",
viewport: { width: 1280, height: 720 },
});
currentPage = await context.newPage();
}
return currentPage;
}
// 导航工具
server.tool(
"browser_navigate",
"导航到指定 URL",
{ url: z.string().url() },
async ({ url }) => {
const page = await ensureBrowser();
await page.goto(url, { waitUntil: "domcontentloaded", timeout: 30000 });
const title = await page.title();
return {
content: [{ type: "text", text: `已导航到: ${url}\n页面标题: ${title}` }],
};
}
);
// 截图工具
server.tool(
"browser_screenshot",
"对当前页面截图",
{
full_page: z.boolean().default(false),
selector: z.string().optional().describe("可选，指定元素选择器"),
},
async ({ full_page, selector }) => {
const page = await ensureBrowser();
let screenshot: Buffer;
if (selector) {
const element = await page.$(selector);
if (!element) throw new Error(`未找到元素: ${selector}`);
screenshot = await element.screenshot({ type: "png" });
} else {
screenshot = await page.screenshot({ type: "png", fullPage: full_page });
}
return {
content: [{
type: "image" as const,
data: screenshot.toString("base64"),
mimeType: "image/png",
}],
};
}
);
// 页面内容提取
server.tool(
"browser_extract",
"提取当前页面的文本内容或结构化数据",
{
selector: z.string().default("body"),
extract_type: z.enum(["text", "html", "links", "table"]).default("text"),
},
async ({ selector, extract_type }) => {
const page = await ensureBrowser();
let result: any;
switch (extract_type) {
case "text":
result = await page.$eval(selector, el => el.textContent?.trim());
break;
case "html":
result = await page.$eval(selector, el => el.innerHTML);
break;
case "links":
result = await page.$$eval(`${selector} a`, links =>
links.map(a => ({ text: a.textContent?.trim(), href: a.href }))
);
break;
case "table":
result = await page.$$eval(`${selector} table tr`, rows =>
rows.map(row => Array.from(row.cells).map(cell => cell.textContent?.trim()))
);
break;
}
return {
content: [{ type: "text", text: JSON.stringify(result, null, 2) }],
};
}
);
第七章 MCP 与 Agent 框架集成
7.1 与 LangChain 集成
# MCP Client 适配器 - 集成到 LangChain
from langchain.tools import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
from typing import Optional, Type
from pydantic import BaseModel, Field
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPToolInput(BaseModel):
"""动态生成的 MCP 工具输入模型"""
arguments: dict = Field(description="工具参数")
class MCPToolAdapter(BaseTool):
"""MCP 工具到 LangChain Tool 的适配器"""
name: str
description: str
mcp_tool_name: str
mcp_session: ClientSession
args_schema: Type[BaseModel] = MCPToolInput
def _run(
self,
arguments: dict,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""同步执行 MCP 工具"""
import asyncio
return asyncio.get_event_loop().run_until_complete(
self._arun(arguments, run_manager)
)
async def _arun(
self,
arguments: dict,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> str:
"""异步执行 MCP 工具"""
result = await self.mcp_session.call_tool(
self.mcp_tool_name, arguments
)
# 提取文本内容
texts = [c.text for c in result.content if hasattr(c, 'text')]
return "\n".join(texts)
class MCPLangChainBridge:
"""MCP 与 LangChain 的桥接器"""
def __init__(self):
self._sessions: dict[str, ClientSession] = {}
self._tools: list[MCPToolAdapter] = []
async def connect_server(
self,
server_name: str,
command: str,
args: list[str],
env: dict[str, str] = None,
):
"""连接 MCP Server 并注册工具"""
server_params = StdioServerParameters(
command=command,
args=args,
env=env,
)
# 建立连接
stdio_transport = await stdio_client(server_params).__aenter__()
read_stream, write_stream = stdio_transport
session = ClientSession(read_stream, write_stream)
await session.__aenter__()
await session.initialize()
self._sessions[server_name] = session
# 获取工具列表并创建适配器
tools_response = await session.list_tools()
for tool in tools_response.tools:
adapter = MCPToolAdapter(
name=f"{server_name}_{tool.name}",
description=f"[{server_name}] {tool.description}",
mcp_tool_name=tool.name,
mcp_session=session,
)
self._tools.append(adapter)
print(f"已连接 MCP Server '{server_name}'，注册 {len(tools_response.tools)} 个工具")
def get_tools(self) -> list[MCPToolAdapter]:
"""获取所有已注册的 LangChain 工具"""
return self._tools
async def close(self):
"""关闭所有连接"""
for session in self._sessions.values():
await session.__aexit__(None, None, None)
# 使用示例
async def create_agent_with_mcp():
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
# 1. 建立 MCP 连接
bridge = MCPLangChainBridge()
await bridge.connect_server(
"filesystem",
"npx",
["-y", "@modelcontextprotocol/server-filesystem", "/home/user/data"],
)
await bridge.connect_server(
"database",
"python",
["-m", "mcp_database_server"],
)
# 2. 创建 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个智能助手，可以使用工具访问文件系统和数据库。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
tools = bridge.get_tools()
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 3. 使用
result = await executor.ainvoke({
"input": "读取 /home/user/data/config.json 的内容"
})
return result
7.2 与 AutoGen 集成
# MCP + AutoGen 多智能体集成
import asyncio
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
class AutoGenMCPBridge:
"""AutoGen 与 MCP 的集成桥"""
def __init__(self):
self.mcp_sessions: dict[str, ClientSession] = {}
self.tool_descriptions: str = ""
async def connect_server(self, name: str, command: str, args: list[str]):
"""连接 MCP Server"""
params = StdioServerParameters(command=command, args=args)
transport = await stdio_client(params).__aenter__()
read_stream, write_stream = transport
session = ClientSession(read_stream, write_stream)
await session.__aenter__()
await session.initialize()
self.mcp_sessions[name] = session
# 生成工具描述供 Agent 使用
tools = await session.list_tools()
for tool in tools.tools:
self.tool_descriptions += f"\n- {tool.name}: {tool.description}"
def create_function_map(self) -> dict:
"""创建 AutoGen 可用的函数映射"""
def make_caller(server_name: str, tool_name: str):
async def caller(**kwargs):
session = self.mcp_sessions[server_name]
result = await session.call_tool(tool_name, kwargs)
return "\n".join(c.text for c in result.content if hasattr(c, 'text'))
return caller
func_map = {}
# ... 遍历所有工具创建映射
return func_map
async def setup_multi_agent_system():
bridge = AutoGenMCPBridge()
await bridge.connect_server("fs", "npx", ["-y", "@modelcontextprotocol/server-filesystem", "/data"])
# 创建多智能体系统
planner = AssistantAgent(
name="Planner",
system_message=f"""你是任务规划者。将复杂任务分解为子步骤。
可用工具：{bridge.tool_descriptions}
将任务分配给合适的执行者。""",
llm_config={"model": "gpt-4o"},
)
executor = AssistantAgent(
name="Executor",
system_message="你是任务执行者。使用提供的工具完成分配给你的子任务。",
llm_config={"model": "gpt-4o"},
function_map=bridge.create_function_map(),
)
reviewer = AssistantAgent(
name="Reviewer",
system_message="你是质量审查者。检查执行结果是否满足要求，指出问题并建议改进。",
llm_config={"model": "gpt-4o"},
)
user_proxy = UserProxyAgent(
name="User",
human_input_mode="TERMINATE",
max_consecutive_auto_reply=10,
)
# 创建群聊
group_chat = GroupChat(
agents=[user_proxy, planner, executor, reviewer],
messages=[],
max_round=20,
)
manager = GroupChatManager(groupchat=group_chat, llm_config={"model": "gpt-4o"})
return user_proxy, manager
7.3 与自研 Agent 框架集成
# 自研 Agent 框架 MCP 集成层
from abc import ABC, abstractmethod
from typing import Any
from dataclasses import dataclass
@dataclass
class ToolCall:
name: str
arguments: dict
id: str
@dataclass
class ToolResult:
tool_call_id: str
content: Any
is_error: bool = False
class MCPAgentIntegration:
"""自研 Agent 框架的 MCP 集成层"""
def __init__(self):
self._servers: dict[str, MCPServerConnection] = {}
self._tool_registry: dict[str, str] = {} # tool_name -> server_name
async def add_server(
self,
name: str,
transport_type: str,
config: dict,
):
"""动态添加 MCP Server"""
if transport_type == "stdio":
conn = StdioConnection(config["command"], config["args"])
elif transport_type == "sse":
conn = SSEConnection(config["url"])
elif transport_type == "websocket":
conn = WebSocketConnection(config["url"])
else:
raise ValueError(f"不支持的传输类型: {transport_type}")
await conn.connect()
self._servers[name] = conn
# 注册工具
tools = await conn.session.list_tools()
for tool in tools.tools:
tool_name = f"{name}.{tool.name}"
self._tool_registry[tool_name] = name
return len(tools.tools)
async def execute_tool(self, call: ToolCall) -> ToolResult:
"""执行工具调用"""
server_name = self._tool_registry.get(call.name)
if not server_name:
return ToolResult(
tool_call_id=call.id,
content=f"未找到工具: {call.name}",
is_error=True,
)
server = self._servers[server_name]
tool_short_name = call.name.split(".", 1)[1]
try:
result = await server.session.call_tool(tool_short_name, call.arguments)
content = "\n".join(
c.text for c in result.content if hasattr(c, 'text')
)
return ToolResult(tool_call_id=call.id, content=content)
except Exception as e:
return ToolResult(tool_call_id=call.id, content=str(e), is_error=True)
def get_tools_description(self) -> str:
"""生成所有工具的描述，供 LLM 使用"""
descriptions = []
for tool_name, server_name in self._tool_registry.items():
server = self._servers[server_name]
# 获取工具详情
tool_info = server.get_tool_info(tool_name.split(".", 1)[1])
if tool_info:
descriptions.append({
"name": tool_name,
"description": tool_info.description,
"parameters": tool_info.inputSchema,
})
return descriptions
第八章 性能优化
8.1 连接池与复用
# MCP 连接池管理器
import asyncio
import time
from typing import Optional
from dataclasses import dataclass, field
@dataclass
class PooledConnection:
session: ClientSession
created_at: float
last_used: float
use_count: int = 0
healthy: bool = True
class MCPConnectionPool:
"""MCP 连接池"""
def __init__(
self,
server_config: dict,
min_size: int = 2,
max_size: int = 10,
max_idle_time: float = 300, # 5 分钟
max_lifetime: float = 3600, # 1 小时
health_check_interval: float = 60,
):
self.config = server_config
self.min_size = min_size
self.max_size = max_size
self.max_idle_time = max_idle_time
self.max_lifetime = max_lifetime
self._pool: list[PooledConnection] = []
self._semaphore = asyncio.Semaphore(max_size)
self._lock = asyncio.Lock()
self._health_task: Optional[asyncio.Task] = None
async def initialize(self):
"""初始化连接池"""
for _ in range(self.min_size):
conn = await self._create_connection()
self._pool.append(conn)
# 启动健康检查
self._health_task = asyncio.create_task(self._health_check_loop())
print(f"连接池已初始化，最小连接数: {self.min_size}")
async def _create_connection(self) -> PooledConnection:
"""创建新连接"""
transport = await stdio_client(
StdioServerParameters(**self.config)
).__aenter__()
read_stream, write_stream = transport
session = ClientSession(read_stream, write_stream)
await session.initialize()
now = time.time()
return PooledConnection(
session=session,
created_at=now,
last_used=now,
)
async def acquire(self) -> PooledConnection:
"""获取连接"""
await self._semaphore.acquire()
async with self._lock:
# 清理过期连接
self._cleanup_idle()
# 查找可用连接
for conn in self._pool:
if conn.healthy:
conn.last_used = time.time()
conn.use_count += 1
return conn
# 创建新连接
conn = await self._create_connection()
self._pool.append(conn)
return conn
async def release(self, conn: PooledConnection):
"""释放连接"""
self._semaphore.release()
def _cleanup_idle(self):
"""清理空闲连接"""
now = time.time()
self._pool = [
c for c in self._pool
if (now - c.last_used < self.max_idle_time and
now - c.created_at < self.max_lifetime and
len(self._pool) > self.min_size)
]
async def _health_check_loop(self):
"""健康检查循环"""
while True:
await asyncio.sleep(60)
async with self._lock:
for conn in self._pool:
try:
# 发送简单的 ping 检测
await asyncio.wait_for(
conn.session.list_tools(),
timeout=5,
)
conn.healthy = True
except:
conn.healthy = False
print(f"连接健康检查失败，标记为不健康")
async def close(self):
"""关闭连接池"""
if self._health_task:
self._health_task.cancel()
for conn in self._pool:
await conn.session.__aexit__(None, None, None)
self._pool.clear()
8.2 缓存策略
# MCP 工具结果缓存
import hashlib
import json
import time
from typing import Any, Optional
from functools import lru_cache
class MCPResultCache:
"""MCP 工具调用结果缓存"""
def __init__(
self,
max_size: int = 1000,
default_ttl: float = 300, # 5 分钟
eviction_policy: str = "lru", # lru, lfu, ttl
):
self.max_size = max_size
self.default_ttl = default_ttl
self.eviction_policy = eviction_policy
self._cache: dict[str, dict] = {}
self._access_order: list[str] = [] # LRU 队列
self._access_count: dict[str, int] = {} # LFU 计数
def _make_key(self, tool_name: str, arguments: dict) -> str:
"""生成缓存键"""
raw = json.dumps({"tool": tool_name, "args": arguments}, sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()
def get(self, tool_name: str, arguments: dict) -> Optional[Any]:
"""获取缓存"""
key = self._make_key(tool_name, arguments)
if key not in self._cache:
return None
entry = self._cache[key]
# 检查 TTL
if time.time() - entry["created_at"] > entry["ttl"]:
del self._cache[key]
return None
# 更新访问信息
entry["last_accessed"] = time.time()
self._access_count[key] = self._access_count.get(key, 0) + 1
# 更新 LRU 顺序
if key in self._access_order:
self._access_order.remove(key)
self._access_order.append(key)
return entry["value"]
def set(self, tool_name: str, arguments: dict, value: Any, ttl: float = None):
"""设置缓存"""
key = self._make_key(tool_name, arguments)
# 检查是否需要淘汰
if len(self._cache) >= self.max_size and key not in self._cache:
self._evict()
self._cache[key] = {
"value": value,
"created_at": time.time(),
"last_accessed": time.time(),
"ttl": ttl or self.default_ttl,
}
self._access_order.append(key)
self._access_count[key] = 1
def _evict(self):
"""根据策略淘汰缓存"""
if self.eviction_policy == "lru":
oldest_key = self._access_order.pop(0)
del self._cache[oldest_key]
elif self.eviction_policy == "lfu":
lfu_key = min(self._access_count, key=self._access_count.get)
del self._cache[lfu_key]
del self._access_count[lfu_key]
elif self.eviction_policy == "ttl":
oldest_key = min(self._cache, key=lambda k: self._cache[k]["created_at"])
del self._cache[oldest_key]
def invalidate(self, tool_name: str = None, pattern: str = None):
"""使缓存失效"""
if tool_name:
keys_to_remove = [
k for k, v in self._cache.items()
if v.get("tool") == tool_name
]
elif pattern:
keys_to_remove = [k for k in self._cache if pattern in k]
else:
keys_to_remove = list(self._cache.keys())
for key in keys_to_remove:
del self._cache[key]
# 使用装饰器自动缓存
def cached(ttl: float = 300, cache_instance: MCPResultCache = None):
"""工具缓存装饰器"""
def decorator(func):
async def wrapper(tool_name: str, arguments: dict, *args, **kwargs):
cache = cache_instance or get_global_cache()
# 尝试从缓存获取
cached_result = cache.get(tool_name, arguments)
if cached_result is not None:
return cached_result
# 执行并缓存
result = await func(tool_name, arguments, *args, **kwargs)
cache.set(tool_name, arguments, result, ttl)
return result
return wrapper
return decorator
8.3 异步与并发处理
# MCP 批量工具调用优化
import asyncio
from typing import Any
from dataclasses import dataclass
@dataclass
class BatchToolCall:
tool_name: str
arguments: dict
call_id: str
@dataclass
class BatchToolResult:
call_id: str
result: Any
error: Optional[str] = None
duration_ms: float = 0
class MCPBatchExecutor:
"""MCP 批量工具执行器"""
def __init__(self, max_concurrency: int = 10, timeout: float = 60):
self.max_concurrency = max_concurrency
self.timeout = timeout
self._semaphore = asyncio.Semaphore(max_concurrency)
async def execute_batch(
self,
session: ClientSession,
calls: list[BatchToolCall],
) -> list[BatchToolResult]:
"""并行执行多个工具调用"""
async def execute_one(call: BatchToolCall) -> BatchToolResult:
async with self._semaphore:
start = time.time()
try:
result = await asyncio.wait_for(
session.call_tool(call.tool_name, call.arguments),
timeout=self.timeout,
)
content = "\n".join(
c.text for c in result.content if hasattr(c, 'text')
)
return BatchToolResult(
call_id=call.call_id,
result=content,
duration_ms=(time.time() - start) * 1000,
)
except asyncio.TimeoutError:
return BatchToolResult(
call_id=call.call_id,
result=None,
error=f"执行超时 ({self.timeout}s)",
duration_ms=(time.time() - start) * 1000,
)
except Exception as e:
return BatchToolResult(
call_id=call.call_id,
result=None,
error=str(e),
duration_ms=(time.time() - start) * 1000,
)
# 并行执行所有调用
results = await asyncio.gather(
*[execute_one(call) for call in calls],
return_exceptions=True,
)
return [r for r in results if isinstance(r, BatchToolResult)]
# 流式工具结果处理
class MCPStreamProcessor:
"""流式处理 MCP 工具结果"""
def __init__(self, chunk_size: int = 4096):
self.chunk_size = chunk_size
async def stream_tool_result(
self,
session: ClientSession,
tool_name: str,
arguments: dict,
callback,
):
"""流式处理大型工具结果"""
result = await session.call_tool(tool_name, arguments)
for content in result.content:
if hasattr(content, 'text') and content.text:
text = content.text
# 分块处理
for i in range(0, len(text), self.chunk_size):
chunk = text[i:i + self.chunk_size]
await callback(chunk)
await asyncio.sleep(0) # 让出执行权
第九章 MCP 网关与代理
9.1 网关架构设计
客户端（Agent/LLM）
│
▼
┌──────────────────────────────────────┐
│ MCP Gateway │
│ ┌──────────────────────────────┐ │
│ │ 认证层 (Auth Middleware) │ │
│ ├──────────────────────────────┤ │
│ │ 路由层 (Router) │ │
│ ├──────────────────────────────┤ │
│ │ 限流层 (Rate Limiter) │ │
│ ├──────────────────────────────┤ │
│ │ 缓存层 (Cache) │ │
│ ├──────────────────────────────┤ │
│ │ 协议转换层 (Protocol) │ │
│ └──────────────────────────────┘ │
└──────────────┬───────────────────────┘
│
┌──────────┼──────────┐
▼ ▼ ▼
┌────────┐ ┌────────┐ ┌────────┐
│MCP Srv1│ │MCP Srv2│ │MCP Srv3│
│ 文件系统│ │ 数据库 │ │ API代理 │
└────────┘ └────────┘ └────────┘
// MCP 网关核心实现
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
interface GatewayRoute {
pattern: string | RegExp;
targetServer: string;
transform?: (request: any) => any;
}
interface GatewayConfig {
routes: GatewayRoute[];
rateLimit: {
windowMs: number;
maxRequests: number;
};
auth: {
type: "jwt" | "api_key";
config: Record<string, any>;
};
}
class MCPGateway {
private clients: Map<string, Client> = new Map();
private routes: GatewayRoute[] = [];
private rateLimiter: RateLimiter;
private cache: MCPResultCache;
private auditor: MCPAuditLogger;
constructor(private config: GatewayConfig) {
this.rateLimiter = new RateLimiter(config.rateLimit);
this.cache = new MCPResultCache({ maxSize: 5000, defaultTTL: 120 });
this.auditor = new MCPAuditLogger();
}
async initialize(): Promise<void> {
// 连接所有后端 Server
for (const route of this.config.routes) {
if (!this.clients.has(route.targetServer)) {
const client = await this.connectServer(route.targetServer);
this.clients.set(route.targetServer, client);
}
}
this.routes = this.config.routes;
console.log(`网关已初始化，路由数: ${this.routes.length}`);
}
private async connectServer(serverName: string): Promise<Client> {
// 根据配置连接后端 MCP Server
const serverConfig = await this.getServerConfig(serverName);
const transport = await this.createTransport(serverConfig);
const client = new Client({ name: "MCPGateway", version: "1.0.0" });
await client.connect(transport);
return client;
}
async handleRequest(
sessionId: string,
userId: string,
toolName: string,
arguments_: Record<string, any>,
): Promise<any> {
const startTime = Date.now();
// 1. 认证检查
if (!await this.authenticate(userId)) {
this.auditor.logEvent({
event_type: AuditEventType.AUTH_FAILURE,
session_id: sessionId,
user_id: userId,
tool_name: toolName,
});
throw new Error("认证失败");
}
// 2. 限流检查
if (!this.rateLimiter.allow(userId)) {
throw new Error("请求过于频繁，请稍后重试");
}
// 3. 路由匹配
const route = this.matchRoute(toolName);
if (!route) {
throw new Error(`未找到匹配路由: ${toolName}`);
}
// 4. 缓存检查
const cached = this.cache.get(toolName, arguments_);
if (cached) {
return cached;
}
// 5. 请求转换
const transformedArgs = route.transform
? route.transform(arguments_)
: arguments_;
// 6. 转发到后端
const client = this.clients.get(route.targetServer)!;
const result = await client.callTool(toolName, transformedArgs);
// 7. 缓存结果
this.cache.set(toolName, arguments_, result);
// 8. 审计日志
this.auditor.logEvent({
event_type: AuditEventType.TOOL_CALL,
session_id: sessionId,
user_id: userId,
tool_name: toolName,
duration_ms: Date.now() - startTime,
});
return result;
}
private matchRoute(toolName: string): GatewayRoute | undefined {
return this.routes.find(route => {
if (typeof route.pattern === "string") {
return toolName === route.pattern || route.pattern === "*";
}
return route.pattern.test(toolName);
});
}
}
9.2 协议转换与聚合
# MCP 协议聚合器 - 将多个 Server 的工具聚合为统一视图
import asyncio
from typing import Any
from mcp import ClientSession
class MCPAggregator:
"""MCP 多 Server 聚合器"""
def __init__(self):
self._sessions: dict[str, ClientSession] = {}
self._tool_map: dict[str, tuple[str, str]] = {} # prefixed_name -> (server, original_name)
async def add_server(self, prefix: str, session: ClientSession):
"""添加 Server 并注册工具（带前缀避免冲突）"""
self._sessions[prefix] = session
tools = await session.list_tools()
for tool in tools.tools:
prefixed_name = f"{prefix}.{tool.name}"
self._tool_map[prefixed_name] = (prefix, tool.name)
async def list_all_tools(self) -> list[dict]:
"""列出所有 Server 的工具"""
all_tools = []
for prefix, session in self._sessions.items():
tools = await session.list_tools()
for tool in tools.tools:
all_tools.append({
"name": f"{prefix}.{tool.name}",
"original_name": tool.name,
"server": prefix,
"description": f"[{prefix}] {tool.description}",
"inputSchema": tool.inputSchema,
})
return all_tools
async def call_tool(self, prefixed_name: str, arguments: dict) -> Any:
"""统一的工具调用接口"""
if prefixed_name not in self._tool_map:
raise ValueError(f"未知工具: {prefixed_name}")
server_prefix, original_name = self._tool_map[prefixed_name]
session = self._sessions[server_prefix]
return await session.call_tool(original_name, arguments)
async def aggregate_resources(self) -> list[dict]:
"""聚合所有 Server 的资源"""
all_resources = []
for prefix, session in self._sessions.items():
resources = await session.list_resources()
for resource in resources.resources:
all_resources.append({
"uri": resource.uri,
"name": f"[{prefix}] {resource.name}",
"description": resource.description,
"mimeType": resource.mimeType,
})
return all_resources
9.3 流量控制与熔断
# MCP 熔断器实现
import time
import asyncio
from enum import Enum
from dataclasses import dataclass, field
class CircuitState(Enum):
CLOSED = "closed" # 正常状态
OPEN = "open" # 熔断状态
HALF_OPEN = "half_open" # 半开状态
@dataclass
class CircuitBreaker:
"""熔断器"""
failure_threshold: int = 5 # 失败阈值
recovery_timeout: float = 30.0 # 恢复超时（秒）
half_open_max_calls: int = 3 # 半开状态最大尝试次数
state: CircuitState = CircuitState.CLOSED
failure_count: int = 0
success_count: int = 0
last_failure_time: float = 0
half_open_calls: int = 0
def allow_request(self) -> bool:
"""判断是否允许请求"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# 检查是否可以进入半开状态
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.half_open_max_calls
return False
def record_success(self):
"""记录成功"""
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
self.half_open_calls += 1
if self.success_count >= self.half_open_max_calls:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
else:
self.failure_count = max(0, self.failure_count - 1)
def record_failure(self):
"""记录失败"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
class MCPResilientClient:
"""带熔断和重试的 MCP 客户端"""
def __init__(
self,
session: ClientSession,
max_retries: int = 3,
retry_delay: float = 1.0,
circuit_breaker: CircuitBreaker = None,
):
self.session = session
self.max_retries = max_retries
self.retry_delay = retry_delay
self.circuit_breaker = circuit_breaker or CircuitBreaker()
async def call_tool(self, name: str, arguments: dict) -> Any:
"""带熔断和重试的工具调用"""
if not self.circuit_breaker.allow_request():
raise CircuitOpenError(f"熔断器已打开，拒绝调用: {name}")
last_error = None
for attempt in range(self.max_retries):
try:
result = await self.session.call_tool(name, arguments)
self.circuit_breaker.record_success()
return result
except Exception as e:
last_error = e
self.circuit_breaker.record_failure()
if attempt < self.max_retries - 1:
delay = self.retry_delay * (2 ** attempt) # 指数退避
await asyncio.sleep(delay)
raise last_error
第十章 实战项目：企业级 MCP Server 集群
10.1 项目架构设计
我们将构建一个完整的企业级 MCP Server 集群，包含以下组件：
项目结构：
mcp-enterprise-cluster/
├── gateway/ # MCP 网关
│ ├── src/
│ │ ├── index.ts
│ │ ├── router.ts
│ │ ├── auth.ts
│ │ ├── rate_limiter.ts
│ │ └── cache.ts
│ └── package.json
├── servers/ # MCP Server 集合
│ ├── filesystem/
│ │ ├── index.ts
│ │ └── package.json
│ ├── database/
│ │ ├── index.py
│ │ └── requirements.txt
│ └── api-proxy/
│ ├── index.ts
│ └── package.json
├── shared/ # 共享库
│ ├── types/
│ ├── utils/
│ └── middleware/
├── deploy/ # 部署配置
│ ├── docker-compose.yml
│ ├── k8s/
│ └── nginx.conf
├── monitoring/ # 监控配置
│ ├── prometheus.yml
│ └── grafana/
└── tests/ # 测试
├── integration/
└── load/
10.2 核心模块实现
网关入口（TypeScript）：
// gateway/src/index.ts
import express from "express";
import { WebSocketServer } from "ws";
import { MCPRouter } from "./router";
import { JWTAuthenticator } from "./auth";
import { RateLimiter } from "./rate_limiter";
import { MCPResultCache } from "./cache";
import { MCPAuditLogger } from "@shared/audit";
async function main() {
const app = express();
app.use(express.json());
// 初始化组件
const auth = new JWTAuthenticator({
issuer: process.env.JWT_ISSUER!,
audience: process.env.JWT_AUDIENCE!,
jwksUrl: process.env.JWKS_URL!,
});
const rateLimiter = new RateLimiter({
windowMs: 60_000,
maxRequests: parseInt(process.env.RATE_LIMIT ?? "100"),
});
const cache = new MCPResultCache({
maxSize: 5000,
defaultTTL: 120,
});
const auditLogger = new MCPAuditLogger({
logFile: process.env.AUDIT_LOG ?? "/var/log/mcp/audit.jsonl",
});
const router = new MCPRouter({
auth,
rateLimiter,
cache,
auditLogger,
});
// 注册后端 Server
await router.addServer("filesystem", {
type: "stdio",
command: "node",
args: ["./servers/filesystem/index.js"],
});
await router.addServer("database", {
type: "stdio",
command: "python",
args: ["-m", "servers.database.index"],
});
await router.addServer("api-proxy", {
type: "sse",
url: process.env.API_PROXY_URL ?? "http://localhost:3001",
});
// HTTP 端点（SSE 传输）
app.post("/mcp/message", auth.middleware(), async (req, res) => {
const userId = req.auth!.userId;
const sessionId = req.headers["x-session-id"] as string;
try {
const result = await router.handleMessage(sessionId, userId, req.body);
res.json(result);
} catch (err: any) {
res.status(err.status ?? 500).json({ error: err.message });
}
});
// WebSocket 端点
const wss = new WebSocketServer({ port: 8080 });
wss.on("connection", async (ws, req) => {
// WebSocket 认证
const token = new URL(req.url!, `http://${req.headers.host}`).searchParams.get("token");
if (!token) {
ws.close(4001, "Missing token");
return;
}
const userId = await auth.validateToken(token);
if (!userId) {
ws.close(4001, "Invalid token");
return;
}
const sessionId = crypto.randomUUID();
console.log(`WebSocket 连接建立: session=${sessionId}, user=${userId}`);
ws.on("message", async (data) => {
try {
const message = JSON.parse(data.toString());
const result = await router.handleMessage(sessionId, userId, message);
ws.send(JSON.stringify(result));
} catch (err: any) {
ws.send(JSON.stringify({ error: err.message }));
}
});
ws.on("close", () => {
console.log(`WebSocket 连接关闭: session=${sessionId}`);
});
});
// 健康检查
app.get("/health", (req, res) => {
const status = router.getHealthStatus();
res.json(status);
});
// 指标端点
app.get("/metrics", async (req, res) => {
const metrics = await router.getMetrics();
res.json(metrics);
});
const port = parseInt(process.env.PORT ?? "3000");
app.listen(port, () => {
console.log(`MCP Gateway 启动于端口 ${port}`);
});
}
main().catch(console.error);
数据库 MCP Server（Python）：
# servers/database/index.py
import asyncio
import json
import os
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import asyncpg
import aiomysql
server = Server("EnterpriseDatabaseServer")
# 连接池管理
_pools = {}
async def init():
"""初始化数据库连接池"""
# PostgreSQL
pg_url = os.getenv("POSTGRES_URL")
if pg_url:
_pools["postgres"] = await asyncpg.create_pool(pg_url, min_size=5, max_size=20)
# MySQL
mysql_config = os.getenv("MYSQL_URL")
if mysql_config:
config = json.loads(mysql_config)
_pools["mysql"] = await aiomysql.create_pool(
host=config["host"],
port=config.get("port", 3306),
user=config["user"],
password=config["password"],
db=config["database"],
minsize=5,
maxsize=20,
)
print(f"数据库 Server 已初始化，连接池: {list(_pools.keys())}")
@server.list_tools()
async def list_tools():
return [
Tool(
name="query",
description="执行安全的只读数据库查询",
inputSchema={
"type": "object",
"properties": {
"database": {
"type": "string",
"enum": list(_pools.keys()),
},
"sql": {
"type": "string",
"description": "SQL 查询语句（仅 SELECT）",
},
"params": {
"type": "array",
"items": {},
"description": "参数化查询的参数",
},
"limit": {
"type": "integer",
"default": 100,
"description": "最大返回行数",
},
},
"required": ["database", "sql"],
},
),
Tool(
name="schema",
description="获取数据库表结构信息",
inputSchema={
"type": "object",
"properties": {
"database": { "type": "string" },
"table": { "type": "string" },
},
"required": ["database"],
},
),
Tool(
name="explain",
description="获取查询执行计划",
inputSchema={
"type": "object",
"properties": {
"database": { "type": "string" },
"sql": { "type": "string" },
},
"required": ["database", "sql"],
},
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "query":
return await handle_query(arguments)
elif name == "schema":
return await handle_schema(arguments)
elif name == "explain":
return await handle_explain(arguments)
raise ValueError(f"Unknown tool: {name}")
async def handle_query(args: dict) -> list:
db_type = args["database"]
sql = args["sql"]
params = args.get("params", [])
limit = args.get("limit", 100)
# 安全校验
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
raise ValueError("仅允许 SELECT 查询")
forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "EXEC"]
for kw in forbidden:
if f" {kw} " in f" {sql_upper} ":
raise ValueError(f"禁止使用 {kw} 语句")
# 添加 LIMIT 保护
if "LIMIT" not in sql_upper:
sql = f"{sql.rstrip(';')} LIMIT {limit}"
pool = _pools.get(db_type)
if not pool:
raise ValueError(f"数据库 {db_type} 未配置")
try:
if db_type == "postgres":
async with pool.acquire() as conn:
rows = await conn.fetch(sql, *params)
data = {
"columns": list(rows[0].keys()) if rows else [],
"rows": [dict(r) for r in rows],
"count": len(rows),
}
elif db_type == "mysql":
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute(sql, params)
rows = await cur.fetchall()
columns = [d[0] for d in cur.description] if cur.description else []
data = {
"columns": columns,
"rows": [dict(zip(columns, r)) for r in rows],
"count": len(rows),
}
return [TextContent(type="text", text=json.dumps(data, default=str, indent=2))]
except Exception as e:
return [TextContent(type="text", text=f"查询失败: {str(e)}")]
async def handle_schema(args: dict) -> list:
db_type = args["database"]
table = args.get("table")
pool = _pools.get(db_type)
if db_type == "postgres":
async with pool.acquire() as conn:
if table:
rows = await conn.fetch("""
SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_name = $1
ORDER BY ordinal_position
""", table)
else:
rows = await conn.fetch("""
SELECT table_name, column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position
""")
data = [dict(r) for r in rows]
return [TextContent(type="text", text=json.dumps(data, indent=2))]
async def main():
await init()
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
10.3 部署与运维
Docker Compose 部署：
# deploy/docker-compose.yml
version: "3.8"
services:
# MCP 网关
gateway:
build:
context: ..
dockerfile: gateway/Dockerfile
ports:
- "3000:3000" # HTTP
- "8080:8080" # WebSocket
environment:
- JWT_ISSUER=https://auth.example.com
- JWT_AUDIENCE=mcp-gateway
- JWKS_URL=https://auth.example.com/.well-known/jwks.json
- RATE_LIMIT=100
- AUDIT_LOG=/var/log/mcp/audit.jsonl
- POSTGRES_URL=postgresql://mcp:secret@postgres:5432/mcp
- API_PROXY_URL=http://api-proxy:3001
volumes:
- audit-logs:/var/log/mcp
depends_on:
- postgres
- redis
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:3000/health"]
interval: 30s
timeout: 10s
retries: 3
# PostgreSQL
postgres:
image: postgres:16-alpine
environment:
POSTGRES_DB: mcp
POSTGRES_USER: mcp
POSTGRES_PASSWORD: secret
volumes:
- pg-data:/var/lib/postgresql/data
restart: unless-stopped
# Redis（缓存和会话）
redis:
image: redis:7-alpine
command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
volumes:
- redis-data:/data
restart: unless-stopped
# API 代理 Server
api-proxy:
build:
context: ..
dockerfile: servers/api-proxy/Dockerfile
environment:
- API_CONFIG_PATH=/config/apis.json
volumes:
- ./config:/config
restart: unless-stopped
# 监控
prometheus:
image: prom/prometheus
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
restart: unless-stopped
grafana:
image: grafana/grafana
ports:
- "3001:3000"
volumes:
- grafana-data:/var/lib/grafana
restart: unless-stopped
volumes:
pg-data:
redis-data:
audit-logs:
grafana-data:
Kubernetes 部署：
# deploy/k8s/gateway-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-gateway
labels:
app: mcp-gateway
spec:
replicas: 3
selector:
matchLabels:
app: mcp-gateway
template:
metadata:
labels:
app: mcp-gateway
spec:
containers:
- name: gateway
image: mcp-enterprise/gateway:latest
ports:
- containerPort: 3000
- containerPort: 8080
env:
- name: JWT_ISSUER
valueFrom:
secretKeyRef:
name: mcp-secrets
key: jwt-issuer
- name: REDIS_URL
value: "redis://mcp-redis:6379"
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "2000m"
memory: "2Gi"
livenessProbe:
httpGet:
path: /health
port: 3000
initialDelaySeconds: 15
periodSeconds: 20
readinessProbe:
httpGet:
path: /health
port: 3000
initialDelaySeconds: 5
periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: mcp-gateway
spec:
selector:
app: mcp-gateway
ports:
- name: http
port: 3000
targetPort: 3000
- name: websocket
port: 8080
targetPort: 8080
type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: mcp-gateway-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: mcp-gateway
minReplicas: 3
maxReplicas: 20
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
负载测试脚本：
# tests/load/load_test.py
import asyncio
import time
import statistics
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
async def load_test():
"""MCP Server 负载测试"""
num_concurrent = 50
num_requests = 1000
results = []
errors = 0
# 建立连接池
sessions = []
for i in range(10):
transport = await stdio_client(
StdioServerParameters(
command="node",
args=["./gateway/dist/index.js"],
)
).__aenter__()
read_stream, write_stream = transport
session = ClientSession(read_stream, write_stream)
await session.initialize()
sessions.append(session)
print(f"已建立 {len(sessions)} 个连接")
async def make_request(session_id: int):
nonlocal errors
session = sessions[session_id % len(sessions)]
start = time.time()
try:
result = await session.call_tool("query", {
"database": "postgres",
"sql": "SELECT * FROM users LIMIT 10",
})
duration = (time.time() - start) * 1000
results.append(duration)
except Exception as e:
errors += 1
# 并发执行
semaphore = asyncio.Semaphore(num_concurrent)
async def bounded_request(i):
async with semaphore:
await make_request(i)
start_time = time.time()
await asyncio.gather(*[bounded_request(i) for i in range(num_requests)])
total_time = time.time() - start_time
# 统计结果
print(f"\n=== 负载测试结果 ===")
print(f"总请求数: {num_requests}")
print(f"成功: {num_requests - errors}")
print(f"失败: {errors}")
print(f"总耗时: {total_time:.2f}s")
print(f"QPS: {num_requests / total_time:.2f}")
print(f"延迟 P50: {statistics.median(results):.2f}ms")
print(f"延迟 P95: {sorted(results)[int(len(results) * 0.95)]:.2f}ms")
print(f"延迟 P99: {sorted(results)[int(len(results) * 0.99)]:.2f}ms")
print(f"平均延迟: {statistics.mean(results):.2f}ms")
if __name__ == "__main__":
asyncio.run(load_test())
第十一章 常见问题与解决方案
Q1: MCP Server 启动后客户端无法连接
症状：Server 进程启动但客户端报连接错误。
排查步骤：
# 1. 检查进程是否存活
ps aux | grep mcp
# 2. 检查端口是否监听（SSE/WS 模式）
netstat -tlnp | grep 3000
# 3. 检查日志输出
# stdio 模式下，日志必须输出到 stderr，stdout 保留给 JSON-RPC
# 常见错误：console.log 打印到 stdout 导致协议解析失败
# 4. 验证 JSON-RPC 握手
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}' | node ./server.js
常见原因与修复：
// ❌ 错误：日志输出到 stdout
console.log("Server started"); // 会污染 JSON-RPC 通道
// ✅ 正确：日志输出到 stderr
console.error("Server started");
// 或使用 MCP 内置日志
server.sendLoggingMessage({
level: "info",
data: "Server started",
});
Q2: 工具调用超时
症状：长时间运行的工具调用频繁超时。
解决方案：
// 方案1：使用进度通知保持连接活跃
server.tool("long_task", {}, async (args) => {
const totalSteps = 100;
for (let i = 0; i < totalSteps; i++) {
await doWork(i);
// 定期发送进度通知，防止客户端超时
if (i % 10 === 0) {
await server.sendNotification({
method: "notifications/progress",
params: {
progress: i,
total: totalSteps,
message: `处理进度: ${i}/${totalSteps}`,
},
});
}
}
return { content: [{ type: "text", text: "任务完成" }] };
});
// 方案2：拆分为异步任务
server.tool("submit_task", {}, async (args) => {
const taskId = await submitAsyncTask(args);
return {
content: [{
type: "text",
text: `任务已提交，ID: ${taskId}。使用 check_task 工具查看状态。`,
}],
};
});
server.tool("check_task", {}, async ({ taskId }) => {
const status = await getTaskStatus(taskId);
return {
content: [{ type: "text", text: JSON.stringify(status) }],
};
});
Q3: 多客户端并发冲突
症状：多个客户端同时调用同一 Server 的写操作导致数据不一致。
解决方案：
# 使用分布式锁保护写操作
import asyncio
from contextlib import asynccontextmanager
class MCPWriteLock:
"""MCP 写操作分布式锁"""
def __init__(self, redis_client):
self.redis = redis_client
@asynccontextmanager
async def acquire(self, resource: str, timeout: float = 30):
lock_key = f"mcp:lock:{resource}"
lock_id = str(uuid.uuid4())
acquired = False
try:
# 尝试获取锁
while not acquired:
acquired = await self.redis.set(
lock_key, lock_id, nx=True, ex=int(timeout)
)
if not acquired:
await asyncio.sleep(0.1)
yield
finally:
# 释放锁（仅释放自己持有的锁）
if acquired:
current = await self.redis.get(lock_key)
if current == lock_id:
await self.redis.delete(lock_key)
# 在工具中使用
write_lock = MCPWriteLock(redis_client)
@server.call_tool()
async def call_tool(name: str, arguments: dict):
if name in WRITE_TOOLS:
async with write_lock.acquire(f"tool:{name}"):
return await execute_write_tool(name, arguments)
return await execute_read_tool(name, arguments)
Q4: 内存泄漏问题
症状：Server 运行一段时间后内存持续增长。
排查与修复：
// 内存监控工具
class MemoryMonitor {
private checkInterval: NodeJS.Timeout;
start(intervalMs: number = 30000) {
this.checkInterval = setInterval(() => {
const usage = process.memoryUsage();
const mb = (bytes: number) => (bytes / 1024 / 1024).toFixed(2);
console.error(`[Memory] RSS: ${mb(usage.rss)}MB, Heap: ${mb(usage.heapUsed)}/${mb(usage.heapTotal)}MB, External: ${mb(usage.external)}MB`);
// 堆内存超过阈值时触发 GC
if (usage.heapUsed > 500 * 1024 * 1024) { // 500MB
console.warn("[Memory] 堆内存超过 500MB，触发强制 GC");
global.gc?.();
}
}, intervalMs);
}
stop() {
clearInterval(this.checkInterval);
}
}
// 常见内存泄漏：未清理的事件监听器和定时器
// ❌ 错误示例
server.tool("leaky_tool", {}, async () => {
const emitter = new EventEmitter();
emitter.on("data", () => {}); // 永远不会被清理
// ...
});
// ✅ 正确示例
server.tool("safe_tool", {}, async () => {
const emitter = new EventEmitter();
const handler = () => {};
emitter.on("data", handler);
try {
// ... 处理逻辑
} finally {
emitter.removeListener("data", handler);
emitter.removeAllListeners();
}
});
Q5: SSE 连接不稳定
症状：SSE 传输模式下连接频繁断开重连。
解决方案：
// SSE 重连与心跳机制
class RobustSSEClient {
private eventSource: EventSource | null = null;
private reconnectAttempts = 0;
private maxReconnectAttempts = 10;
private heartbeatTimeout: NodeJS.Timeout | null = null;
connect(url: string) {
this.eventSource = new EventSource(url);
this.eventSource.onopen = () => {
console.log("SSE 连接已建立");
this.reconnectAttempts = 0;
this.startHeartbeatCheck();
};
this.eventSource.onmessage = (event) => {
this.resetHeartbeatCheck();
const message = JSON.parse(event.data);
this.handleMessage(message);
};
this.eventSource.onerror = () => {
console.warn("SSE 连接错误");
this.stopHeartbeatCheck();
this.attemptReconnect(url);
};
}
private startHeartbeatCheck() {
this.heartbeatTimeout = setTimeout(() => {
console.warn("心跳超时，主动重连");
this.eventSource?.close();
this.attemptReconnect(this.eventSource!.url);
}, 60000); // 60秒无消息则重连
}
private resetHeartbeatCheck() {
if (this.heartbeatTimeout) {
clearTimeout(this.heartbeatTimeout);
}
this.startHeartbeatCheck();
}
private attemptReconnect(url: string) {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error("达到最大重连次数，停止重连");
return;
}
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
console.log(`${delay}ms 后尝试重连 (${this.reconnectAttempts + 1}/${this.maxReconnectAttempts})`);
setTimeout(() => {
this.reconnectAttempts++;
this.connect(url);
}, delay);
}
}
总结与展望
本教程系统性地讲解了 MCP 工具生态的高级开发技能，从协议底层到生产架构，从安全加固到性能优化，覆盖了企业级 MCP 开发的方方面面。
核心要点回顾
| 领域 | 关键技能 |
|---|---|
| 协议理解 | JSON-RPC 2.0 扩展、能力协商、版本兼容 |
| Server 开发 | 动态资源、组合工具、采样机制 |
| Transport | WebSocket、gRPC、消息队列自定义传输 |
| 安全加固 | OAuth 2.0、RBAC、输入验证、沙箱隔离 |
| 生产架构 | 高可用、水平扩展、优雅停机 |
| 工具生态 | 文件系统、数据库、API、浏览器 Server |
| Agent 集成 | LangChain、AutoGen、自研框架集成 |
| 性能优化 | 连接池、缓存、批量并发 |
| 网关代理 | 路由、聚合、熔断、限流 |
MCP 生态发展趋势
- 标准化推进：MCP 正在成为 AI 工具调用的事实标准，预计会有更多框架原生支持
- 安全增强：OAuth 2.1、mTLS 等企业级安全特性将逐步纳入协议规范
- 性能优化：HTTP/2、Protocol Buffers 等高性能传输方案将成为可选配置
- 生态繁荣：MCP Server 注册中心、版本管理、依赖解析等基础设施将逐步完善
- 跨平台互通：MCP 网关将支持与其他 AI 工具协议（如 OpenAPI、GraphQL）的互通
下一步学习建议
- 参与 MCP 官方仓库 的 Issue 和 PR
- 构建自己的 MCP Server 并发布到社区
- 关注 MCP 协议规范的更新动态
- 在生产环境中实践本教程的架构模式
版权声明：本教程内容基于 MCP 协议公开规范撰写，所有代码示例为原创编写，可自由学习和使用。
