MCP协议深度实战指南：从协议原理到生产级AI工具链的完整构建方案（2026版）
一、为什么MCP是2026年最值得深入的技术
如果你在2025年底问任何一个AI工程师："AI应用开发最大的痛点是什么？"十有八九会得到同一个答案——工具集成太痛苦了。
每个大模型厂商有自己的Function Calling格式，每个Agent框架有自己的插件系统，你要让AI调用一个数据库查询，就得写一遍适配代码；换个模型，又得重写一遍。这就像十年前Android手机充电接口百花齐放一样——功能都能实现，但开发者和用户都痛不欲生。
MCP（Model Context Protocol）的出现，本质上就是AI领域的"USB-C统一运动"。由Anthropic在2024年底推出，到2026年已经成为了AI工具集成的事实标准。无论是OpenAI、Google还是国内的各大模型厂商，都在不同程度上采纳或兼容MCP协议。
但这篇文章不是来给你讲"MCP是什么"这种入门概念的。网上已经有一大堆"5分钟理解MCP"的快餐文章了。我要做的，是从一个写了十年程序的老兵视角，带你深入MCP协议的每一个技术细节，从JSON-RPC消息格式到字节级别的通信机制，从Hello World到生产环境的高并发部署，一步到位。
读完这篇文章，你将能够：
- 理解MCP协议的每一层设计决策及其背后的工程权衡
- 用Python和TypeScript从零构建一个生产级MCP Server
- 处理并发、超时、安全、可观测性等真实生产问题
- 设计可复用的MCP工具生态，让你的AI应用真正具备"手脚"
准备好了吗？我们开始。
二、MCP协议架构：不仅仅是JSON-RPC
2.1 协议分层模型
MCP的设计借鉴了经典的网络分层思想，但针对AI应用场景做了大量简化。理解这个分层模型，是深入掌握MCP的第一步：
┌─────────────────────────────────────┐
│ 应用层 (Application) │
│ Tools / Resources / Prompts │
├─────────────────────────────────────┤
│ 协议层 (Protocol) │
│ JSON-RPC 2.0 消息规范 │
├─────────────────────────────────────┤
│ 传输层 (Transport) │
│ stdio / SSE / Streamable HTTP │
└─────────────────────────────────────┘
很多教程只讲了传输层（怎么连），跳过了协议层（消息格式），更不会深入应用层（三种原语的设计哲学）。我们逐层拆解。
2.2 传输层：三种模式的选择与权衡
MCP支持三种传输模式，它们不是"随便选一个"的关系，而是有明确的使用场景划分：
stdio模式：本地进程通信
这是最简单的模式，MCP Server作为AI应用的子进程运行，通过标准输入/输出通信。
{
"mcpServers": {
"my-database": {
"command": "python",
"args": ["/path/to/my_mcp_server.py"],
"env": {
"DB_HOST": "localhost",
"DB_PORT": "5432"
}
}
}
}
适用场景：开发调试、桌面端AI助手（如Claude Desktop、OpenClaw本地模式）
优点：零网络开销、进程隔离天然安全、配置简单
缺点：不支持远程部署、无法多客户端共享、进程生命周期与宿主绑定
SSE模式：远程HTTP长连接
Server-Sent Events模式，AI应用通过HTTP连接到远程MCP Server，使用SSE接收推送，HTTP POST发送请求。
{
"mcpServers": {
"remote-api": {
"url": "https://mcp.example.com/sse",
"headers": {
"Authorization": "Bearer sk-xxxxx"
}
}
}
}
适用场景：团队共享工具服务、云端部署、SaaS集成
优点：支持远程访问、可多客户端共享、天然支持事件推送
缺点：需要处理网络延迟、连接断线重连、需要额外的安全认证层
Streamable HTTP模式：2026年的推荐方案
这是MCP协议在2025年下半年引入的新传输模式，也是目前官方推荐的生产级方案。它解决了SSE模式的连接管理和会话状态问题。
客户端 ──HTTP POST──▶ 服务端 /mcp
客户端 ◀──SSE流──── 服务端 /mcp (响应和通知)
关键特性：
- 单一HTTP端点处理所有通信
- 支持无状态和有状态两种模式
- 自动处理连接断线和会话恢复
- 支持HTTP/2多路复用
2.3 协议层：JSON-RPC 2.0的消息设计
MCP基于JSON-RPC 2.0，但做了重要的扩展和约束。理解这些扩展点，是避免集成踩坑的关键。
消息格式基础
每条MCP消息都遵循JSON-RPC 2.0规范：
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
三种消息角色：
- Request：客户端→服务端，必须带
id，期望收到Response - Response：服务端→客户端，必须带对应Request的
id - Notification：任意方向，不带
id，不期望响应（fire-and-forget）
MCP的核心协议方法：
# 生命周期管理
initialize # 握手，交换能力声明
initialized # 握手完成确认
ping # 心跳检测
# 能力发现
tools/list # 列出所有可用工具
tools/call # 调用指定工具
resources/list # 列出所有资源
resources/read # 读取指定资源
prompts/list # 列出所有提示模板
prompts/get # 获取指定提示模板
# 高级功能
completion/complete # 自动补全
logging/setLevel # 设置日志级别
sampling/createMessage # 让Server反问Client（关键！）
2.4 能力协商：MCP最被低估的设计
很多人忽略了MCP的能力协商机制，但这恰恰是它比简单REST API高明的地方。
在initialize握手阶段，客户端和服务端会交换各自支持的能力：
// 客户端发起初始化
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {
"roots": {
"listChanged": true
},
"sampling": {}
},
"clientInfo": {
"name": "my-ai-app",
"version": "1.0.0"
}
}
}
// 服务端响应
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2025-03-26",
"capabilities": {
"tools": { "listChanged": true },
"resources": { "subscribe": true },
"prompts": { "listChanged": true },
"logging": {}
},
"serverInfo": {
"name": "production-db-server",
"version": "2.1.0"
}
}
}
这段握手信息包含了大量关键信息：
protocolVersion：确保双方使用兼容的协议版本capabilities：双方告知对方"我支持什么"，避免调用不存在的功能listChanged：如果工具/资源列表会动态变化，需要通知对方刷新
sampling能力是精髓——它允许MCP Server反过来请求AI模型生成内容。这意味着Server不只是被动执行命令，它可以主动"思考"，请求AI帮忙做决策。这个设计打破了传统的单向调用模型，为复杂的多步推理场景打开了一扇门。
三、三种原语的深度剖析
3.1 Tools：让AI"动手"
Tools是MCP中最核心、使用最频繁的原语。一个Tool本质上就是一个函数定义+执行逻辑。
// TypeScript定义一个Tool
const myTool: Tool = {
name: "query_database",
description: "执行SQL查询并返回结果。支持SELECT、INSERT、UPDATE、DELETE。",
inputSchema: {
type: "object",
properties: {
sql: {
type: "string",
description: "要执行的SQL语句"
},
database: {
type: "string",
enum: ["production", "staging", "analytics"],
description: "目标数据库"
},
limit: {
type: "number",
description: "返回结果的最大行数，默认100",
default: 100
}
},
required: ["sql", "database"]
}
};
设计高质量Tool的五个原则：
- 描述要写给AI看，不是写给人类看。AI不关心"这是一个数据库查询工具"这种废话，它关心的是"在什么场景下应该用这个工具"、"输入什么能得到什么"。好的描述示例：
差：执行SQL查询
好：在指定数据库中执行SQL查询。当用户需要查询业务数据、分析数据趋势、
或获取数据库表结构时使用此工具。自动处理SQL注入防护和查询超时。
输入：SQL语句 + 目标数据库(production|staging|analytics)
输出：查询结果表格，包含列名和数据
注意：DDL操作(CREATE/ALTER/DROP)需要先调用confirm_ddl工具获取确认
inputSchema要充分利用JSON Schema的表达力。不只是
type: string，要用enum约束选项、用pattern校验格式、用minLength/maxLength控制范围。错误信息要可操作。不要返回"查询失败"，要返回"查询失败：表users不存在，可用表：orders, products, customers"。
结果要有上下文。不只是返回数据，要告诉AI数据的含义："查询返回了47条记录，时间范围2026-01-01至2026-05-18，总金额¥1,234,567.89"。
考虑幂等性。AI可能会因为不确定而重复调用同一个Tool，确保你的Tool能安全处理重复调用。
3.2 Resources：让AI"看见"
如果说Tools是AI的手，那Resources就是AI的眼睛。Resources提供了可以被AI读取的上下文数据。
# 定义一个Resource
@server.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri="file:///app/logs/error.log",
name="错误日志",
description="应用最近的错误日志，包含堆栈跟踪和错误上下文",
mimeType="text/plain"
),
Resource(
uri="db://production/schema",
name="生产数据库Schema",
description="生产数据库的完整表结构和字段说明",
mimeType="application/json"
)
]
@server.read_resource()
async def read_resource(uri: str) -> str:
if uri == "file:///app/logs/error.log":
return open("/app/logs/error.log").read()[-10000:] # 最后10KB
elif uri == "db://production/schema":
return json.dumps(get_db_schema(), ensure_ascii=False, indent=2)
Tools vs Resources的选择决策：
| 维度 | Tools | Resources |
|---|---|---|
| 语义 | 执行动作 | 提供数据 |
| 副作用 | 通常有 | 无 |
| 缓存 | 不应该 | 应该 |
| 实时性 | 每次执行 | 可订阅更新 |
| 权限 | 写权限可能需要 | 通常只需读 |
经验法则：如果AI需要"做一件事"→Tool；如果AI需要"看一些数据"→Resource。
3.3 Prompts：让AI"会说"
Prompts是MCP中最被低估的原语。它允许Server预定义一些交互模板，Client（AI应用）可以直接使用这些模板来启动对话。
@server.list_prompts()
async def list_prompts() -> list[Prompt]:
return [
Prompt(
name="code-review",
description="对代码变更进行深度审查，输出安全、性能、可维护性三维度评分",
arguments=[
PromptArgument(
name="code_diff",
description="要审查的代码变更（git diff格式）",
required=True
),
PromptArgument(
name="language",
description="编程语言",
required=False
)
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
if name == "code-review":
return GetPromptResult(
messages=[
PromptMessage(
role="user",
content=TextContent(
type="text",
text=f"""你是一位资深的安全工程师和代码审查专家。请对以下代码变更进行深度审查。
编程语言：{arguments.get('language', '自动检测')}
代码变更：
{arguments['code_diff']}
请从以下三个维度逐一评估：
1. **安全性**（权重40%）
- SQL注入、XSS、命令注入等OWASP Top 10漏洞
- 敏感信息泄露（密钥、Token、密码硬编码）
- 权限校验是否完整
2. **性能**（权重30%）
- N+1查询问题
- 不必要的内存分配
- 缺少索引的大表查询
- 异步/同步使用是否得当
3. **可维护性**（权重30%）
- 代码复杂度（圈复杂度>10需要标注）
- 命名规范性
- 错误处理完整性
- 日志和可观测性
最终输出格式：
- 总分（0-100）
- 每个维度的详细评分和具体问题
- 修复建议（带代码示例）
- 严重程度标注：🔴 Critical 🟡 Warning 🟢 Info"""
)
)
]
)
这个设计的精妙之处在于：它把领域专家的知识固化为可复用的模板。一个安全团队可以维护一组专业的代码审查Prompt，所有使用这个MCP Server的AI应用都能直接调用这些专业知识，而不需要每个应用都重新写一遍。
四、从零构建生产级MCP Server
理论讲够了，我们来写一个真正能上生产的MCP Server。我选择一个实际场景：企业级数据库运维工具——它需要处理查询执行、Schema管理、性能监控和安全审计。
4.1 项目结构
db-mcp-server/
├── pyproject.toml
├── src/
│ └── db_mcp_server/
│ ├── __init__.py
│ ├── server.py # 主入口
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── query.py # SQL查询工具
│ │ ├── schema.py # Schema管理工具
│ │ └── explain.py # 执行计划分析工具
│ ├── resources/
│ │ ├── __init__.py
│ │ ├── logs.py # 慢查询日志
│ │ └── metrics.py # 性能指标
│ └── middleware/
│ ├── __init__.py
│ ├── auth.py # 认证中间件
│ ├── rate_limit.py # 限流中间件
│ └── audit.py # 审计日志
└── tests/
├── test_query.py
├── test_concurrency.py
└── test_security.py
4.2 主服务器实现
# src/db_mcp_server/server.py
"""
生产级数据库MCP Server
支持PostgreSQL和MySQL，具备连接池管理、查询限流、审计日志等企业级特性
"""
import asyncio
import logging
import time
import uuid
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, AsyncIterator
import asyncpg
import aiomysql
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import (
Tool, Resource, TextContent, ImageContent,
CallToolResult, ListToolsResult, ListResourcesResult,
ReadResourceResult, Prompt, GetPromptResult,
PromptMessage, PromptArgument
)
# ─── 配置 ────────────────────────────────────────
@dataclass
class DBConfig:
"""数据库连接配置"""
host: str = "localhost"
port: int = 5432
user: str = "readonly"
password: str = ""
database: str = "production"
pool_size: int = 10
max_overflow: int = 5
statement_timeout: int = 30 # 秒
max_rows: int = 1000
read_only: bool = True # 生产安全默认只读
@dataclass
class RateLimitConfig:
"""限流配置"""
max_requests_per_minute: int = 60
max_concurrent_queries: int = 5
max_query_size: int = 100_000 # 字符
@dataclass
class AuditEntry:
"""审计日志条目"""
timestamp: datetime = field(default_factory=datetime.now)
session_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
tool_name: str = ""
parameters: dict = field(default_factory=dict)
result_rows: int = 0
execution_time_ms: float = 0.0
status: str = "success"
error: str = ""
client_info: str = ""
# ─── 连接池管理 ──────────────────────────────────
class ConnectionPoolManager:
"""线程安全的数据库连接池管理器"""
def __init__(self, config: DBConfig):
self.config = config
self._pg_pool: asyncpg.Pool | None = None
self._mysql_pool = None
self._lock = asyncio.Lock()
self._created_at = time.monotonic()
async def get_pg_pool(self) -> asyncpg.Pool:
if self._pg_pool is None:
async with self._lock:
if self._pg_pool is None:
self._pg_pool = await asyncpg.create_pool(
host=self.config.host,
port=self.config.port,
user=self.config.user,
password=self.config.password,
database=self.config.database,
min_size=1,
max_size=self.config.pool_size + self.config.max_overflow,
statement_cache_size=500,
max_inactive_connection_lifetime=300,
command_timeout=self.config.statement_timeout
)
return self._pg_pool
async def close(self):
if self._pg_pool:
await self._pg_pool.close()
self._pg_pool = None
@property
def pool_stats(self) -> dict:
if self._pg_pool:
pool = self._pg_pool
return {
"pool_size": pool.get_size(),
"free_size": pool.get_idle_size(),
"created_at": self._created_at
}
return {"status": "not_initialized"}
# ─── 查询执行器（带限流和审计） ────────────────────
class QueryExecutor:
"""带限流、超时、审计的查询执行器"""
def __init__(
self,
pool_manager: ConnectionPoolManager,
rate_limit: RateLimitConfig,
config: DBConfig
):
self.pool = pool_manager
self.rate_limit = rate_limit
self.config = config
self._semaphore = asyncio.Semaphore(rate_limit.max_concurrent_queries)
self._audit_log: list[AuditEntry] = []
self._request_times: list[float] = []
async def execute(self, sql: str, params: tuple = (), client_info: str = "") -> dict:
"""执行SQL查询，带完整的限流、校验和审计"""
entry = AuditEntry(tool_name="query", parameters={"sql": sql}, client_info=client_info)
# 限流检查
await self._check_rate_limit()
# 安全校验
await self._validate_query(sql)
# 只读模式检查
if self.config.read_only:
self._check_read_only(sql)
# 获取信号量（控制并发）
async with self._semaphore:
start = time.monotonic()
try:
pool = await self.pool.get_pg_pool()
async with pool.acquire() as conn:
# 设置语句超时
await conn.execute(
f"SET statement_timeout = '{self.config.statement_timeout * 1000}'"
)
# 执行查询
rows = await conn.fetch(sql, *params)
execution_time = (time.monotonic() - start) * 1000
# 限制返回行数
truncated = len(rows) > self.config.max_rows
rows = rows[:self.config.max_rows]
# 构建结果
if rows:
columns = list(rows[0].keys())
data = [dict(row) for row in rows]
else:
columns = []
data = []
result = {
"columns": columns,
"rows": data,
"row_count": len(data),
"execution_time_ms": round(execution_time, 2),
"truncated": truncated,
"truncated_message": f"结果已截断，仅显示前{self.config.max_rows}行" if truncated else None
}
entry.result_rows = len(data)
entry.execution_time_ms = execution_time
entry.status = "success"
self._audit_log.append(entry)
return result
except asyncpg.QueryCanceledError:
entry.status = "timeout"
entry.error = f"查询超时（>{self.config.statement_timeout}秒）"
self._audit_log.append(entry)
return {"error": entry.error, "suggestion": "尝试添加WHERE条件缩小查询范围，或为相关列创建索引"}
except asyncpg.PostgresError as e:
entry.status = "error"
entry.error = str(e)
self._audit_log.append(entry)
return {"error": str(e), "error_code": e.sqlstate, "hint": self._get_error_hint(str(e))}
except Exception as e:
entry.status = "error"
entry.error = str(e)
self._audit_log.append(entry)
return {"error": f"未知错误: {e}"}
def _check_read_only(self, sql: str):
"""检查SQL是否违反只读约束"""
normalized = sql.strip().upper()
dangerous_prefixes = ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE", "GRANT", "REVOKE")
for prefix in dangerous_prefixes:
if normalized.startswith(prefix):
raise PermissionError(
f"服务器运行在只读模式，不允许执行{prefix}操作。"
f"如需写入操作，请联系DBA切换到写入模式。"
)
async def _validate_query(self, sql: str):
"""基础SQL校验"""
if not sql or not sql.strip():
raise ValueError("SQL语句不能为空")
if len(sql) > self.rate_limit.max_query_size:
raise ValueError(f"SQL语句过长（{len(sql)}字符），最大允许{self.rate_limit.max_query_size}字符")
async def _check_rate_limit(self):
"""滑动窗口限流"""
now = time.monotonic()
# 清理60秒前的请求记录
self._request_times = [t for t in self._request_times if now - t < 60]
if len(self._request_times) >= self.rate_limit.max_requests_per_minute:
raise RuntimeError("请求过于频繁，请稍后再试")
def _get_error_hint(self, error_msg: str) -> str:
"""根据错误类型提供修复建议"""
error_lower = error_msg.lower()
hints = {
"relation": "表名可能不存在。使用 schema 工具查看可用表列表。",
"column": "列名可能不存在。使用 schema 工具查看表结构。",
"does not exist": "对象不存在。使用 schema 工具查看数据库结构。",
"permission denied": "权限不足。当前账户为只读权限。",
"division by zero": "除零错误。检查分母是否可能为0，使用NULLIF(col, 0)。",
"timeout": "查询超时。考虑添加索引或缩小查询范围。"
}
for keyword, hint in hints.items():
if keyword in error_lower:
return hint
return "请检查SQL语法是否正确，或联系DBA获取支持。"
def get_audit_log(self, limit: int = 100) -> list[dict]:
"""获取审计日志"""
return [
{
"timestamp": e.timestamp.isoformat(),
"session_id": e.session_id,
"tool": e.tool_name,
"rows": e.result_rows,
"time_ms": round(e.execution_time_ms, 2),
"status": e.status,
"error": e.error or None
}
for e in self._audit_log[-limit:]
]
# ─── MCP Server主体 ──────────────────────────────
@asynccontextmanager
async def server_lifespan(server: Server) -> AsyncIterator[dict]:
"""服务器生命周期管理"""
config = DBConfig()
rate_limit = RateLimitConfig()
pool_manager = ConnectionPoolManager(config)
executor = QueryExecutor(pool_manager, rate_limit, config)
# 启动连接池
try:
await pool_manager.get_pg_pool()
logging.info("数据库连接池初始化成功")
except Exception as e:
logging.error(f"数据库连接池初始化失败: {e}")
raise
yield {
"config": config,
"pool": pool_manager,
"executor": executor
}
# 清理
await pool_manager.close()
logging.info("数据库连接池已关闭")
# 创建Server实例
mcp_server = Server("enterprise-db-tool", lifespan=server_lifespan)
# ─── Tool定义 ─────────────────────────────────────
@mcp_server.list_tools()
async def handle_list_tools() -> ListToolsResult:
return ListToolsResult(tools=[
Tool(
name="execute_query",
description=(
"在数据库中执行SQL查询并返回结果。支持复杂JOIN、子查询、窗口函数等。"
"自动处理SQL注入防护、查询超时和结果截断。"
"服务器默认只读模式，仅允许SELECT语句。"
),
inputSchema={
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "要执行的SQL语句（仅允许SELECT）"
},
"params": {
"type": "array",
"items": {"type": "string"},
"description": "SQL参数（可选，用于参数化查询）"
}
},
"required": ["sql"]
}
),
Tool(
name="get_schema",
description=(
"获取数据库表结构信息。可以查看所有表列表、指定表的字段详情、"
"索引信息、外键关系等。在编写查询前建议先调用此工具了解表结构。"
),
inputSchema={
"type": "object",
"properties": {
"table_name": {
"type": "string",
"description": "要查看的表名（不指定则返回所有表列表）"
},
"include_indexes": {
"type": "boolean",
"description": "是否包含索引信息，默认true",
"default": True
},
"include_constraints": {
"type": "boolean",
"description": "是否包含约束信息，默认true",
"default": True
}
}
}
),
Tool(
name="explain_query",
description=(
"获取SQL查询的执行计划。用于性能分析和查询优化。"
"返回JSON格式的执行计划树，包含每个节点的扫描方式、"
"预估行数、实际行数、执行时间等关键信息。"
),
inputSchema={
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "要分析的SQL语句"
},
"analyze": {
"type": "boolean",
"description": "是否实际执行查询以获取真实执行统计，默认false（仅预估）",
"default": False
}
},
"required": ["sql"]
}
),
Tool(
name="get_slow_queries",
description=(
"获取最近的慢查询日志。按执行时间降序排列，包含SQL语句、"
"执行时间、调用次数等统计信息。用于定位性能瓶颈。"
),
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "number",
"description": "返回条数，默认20",
"default": 20
},
"min_time_ms": {
"type": "number",
"description": "最小执行时间（毫秒），过滤掉更快的查询",
"default": 1000
}
}
}
),
Tool(
name="get_audit_log",
description="获取本次会话的SQL查询审计日志，包含执行时间、状态等",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "number",
"description": "返回条数，默认50",
"default": 50
}
}
}
)
])
@mcp_server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> CallToolResult:
ctx = mcp_server.request_context
executor = ctx.lifespan_context["executor"]
pool = ctx.lifespan_context["pool"]
config = ctx.lifespan_context["config"]
args = arguments or {}
try:
if name == "execute_query":
result = await executor.execute(args["sql"], client_info=ctx.meta.client_info if ctx.meta else "")
return CallToolResult(
content=[TextContent(type="text", text=json.dumps(result, ensure_ascii=False, default=str))],
isError=result.get("error") is not None
)
elif name == "get_schema":
pool_obj = await pool.get_pg_pool()
async with pool_obj.acquire() as conn:
table_name = args.get("table_name")
if not table_name:
# 返回所有表列表
rows = await conn.fetch("""
SELECT
schemaname,
tablename,
tableowner,
(SELECT count(*) FROM pg_indexes WHERE tablename = t.tablename) as index_count
FROM pg_tables t
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
ORDER BY tablename
""")
tables = [dict(r) for r in rows]
return CallToolResult(
content=[TextContent(type="text", text=json.dumps({
"total_tables": len(tables),
"tables": tables
}, ensure_ascii=False))]
)
else:
# 返回指定表的详细结构
rows = await conn.fetch("""
SELECT
c.column_name,
c.data_type,
c.character_maximum_length,
c.is_nullable,
c.column_default,
pgd.description as column_comment
FROM information_schema.columns c
LEFT JOIN pg_catalog.pg_statio_all_tables st
ON c.table_name = st.relname
LEFT JOIN pg_catalog.pg_description pgd
ON pgd.objoid = st.relid AND pgd.objsubid = c.ordinal_position
WHERE c.table_name = $1
ORDER BY c.ordinal_position
""", table_name)
columns = [dict(r) for r in rows]
result = {"table_name": table_name, "columns": columns}
if args.get("include_indexes", True):
idx_rows = await conn.fetch("""
SELECT
indexname,
indexdef
FROM pg_indexes
WHERE tablename = $1
ORDER BY indexname
""", table_name)
result["indexes"] = [dict(r) for r in idx_rows]
return CallToolResult(
content=[TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]
)
elif name == "explain_query":
sql = args["sql"]
analyze = args.get("analyze", False)
prefix = "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)" if analyze else "EXPLAIN (FORMAT JSON)"
explain_sql = f"{prefix} {sql}"
pool_obj = await pool.get_pg_pool()
async with pool_obj.acquire() as conn:
rows = await conn.fetchval(explain_sql)
# rows是JSON格式的执行计划
plan = json.loads(rows) if isinstance(rows, str) else rows
# 分析执行计划，给出优化建议
analysis = _analyze_execution_plan(plan)
return CallToolResult(
content=[TextContent(type="text", text=json.dumps({
"execution_plan": plan,
"analysis": analysis
}, ensure_ascii=False, indent=2))]
)
elif name == "get_slow_queries":
limit = args.get("limit", 20)
min_time = args.get("min_time_ms", 1000)
pool_obj = await pool.get_pg_pool()
async with pool_obj.acquire() as conn:
rows = await conn.fetch("""
SELECT
query,
calls,
total_exec_time,
mean_exec_time,
min_exec_time,
max_exec_time,
rows
FROM pg_stat_statements
WHERE mean_exec_time > $1
ORDER BY total_exec_time DESC
LIMIT $2
""", min_time, limit)
slow_queries = [dict(r) for r in rows]
return CallToolResult(
content=[TextContent(type="text", text=json.dumps({
"total": len(slow_queries),
"min_time_ms_filter": min_time,
"queries": slow_queries
}, ensure_ascii=False, default=str))]
)
elif name == "get_audit_log":
limit = args.get("limit", 50)
log = executor.get_audit_log(limit)
return CallToolResult(
content=[TextContent(type="text", text=json.dumps({
"total_entries": len(log),
"entries": log
}, ensure_ascii=False))]
)
else:
return CallToolResult(
content=[TextContent(type="text", text=f"未知工具: {name}")],
isError=True
)
except PermissionError as e:
return CallToolResult(
content=[TextContent(type="text", text=f"权限错误: {e}")],
isError=True
)
except ValueError as e:
return CallToolResult(
content=[TextContent(type="text", text=f"参数错误: {e}")],
isError=True
)
except Exception as e:
return CallToolResult(
content=[TextContent(type="text", text=f"服务器错误: {type(e).__name__}: {e}")],
isError=True
)
def _analyze_execution_plan(plan: list | dict) -> dict:
"""分析执行计划，给出优化建议"""
suggestions = []
warnings = []
def walk(node: dict, depth: int = 0):
node_type = node.get("Node Type", "")
actual_cost = node.get("Actual Total Time", 0)
planned_cost = node.get("Total Cost", 0)
rows = node.get("Actual Rows", node.get("Plan Rows", 0))
# 检查Seq Scan
if "Seq Scan" in node_type:
suggestions.append(f"检测到全表扫描（{node_type}），考虑添加索引")
# 检查Nested Loop
if "Nested Loop" in node_type and rows > 10000:
suggestions.append("嵌套循环处理大量行，考虑使用Hash Join或Merge Join")
# 检查Sort
if "Sort" in node_type:
sort_method = node.get("Sort Method", "")
if "external merge" in sort_method.lower():
warnings.append("排序使用了外部归并（磁盘），work_mem可能不够")
# 检查Filter选择性
if "Filter" in node_type:
actual = node.get("Actual Rows", 0)
planned = node.get("Plan Rows", 1)
if planned > 0 and actual / planned < 0.01:
warnings.append(f"统计信息可能过期：预估{planned}行，实际{actual}行，建议ANALYZE")
for child in node.get("Plans", []):
walk(child, depth + 1)
for p in (plan if isinstance(plan, list) else [plan]):
walk(p)
return {
"suggestions": suggestions,
"warnings": warnings,
"summary": f"发现{suggestions.count('检测到')}个潜在性能问题，{len(warnings)}个警告"
}
# ─── Resource定义 ─────────────────────────────────
@mcp_server.list_resources()
async def handle_list_resources() -> ListResourcesResult:
return ListResourcesResult(resources=[
Resource(
uri="db://pool-status",
name="连接池状态",
description="数据库连接池的实时状态信息",
mimeType="application/json"
),
Resource(
uri="db://slow-query-top10",
name="Top 10 慢查询",
description="执行时间最长的10条SQL查询统计",
mimeType="application/json"
),
Resource(
uri="db://table-sizes",
name="表大小排名",
description="按大小排序的数据库表列表（含索引和TOAST）",
mimeType="application/json"
)
])
@mcp_server.read_resource()
async def handle_read_resource(uri: str) -> ReadResourceResult:
ctx = mcp_server.request_context
pool = ctx.lifespan_context["pool"]
pool_obj = await pool.get_pg_pool()
async with pool_obj.acquire() as conn:
if uri == "db://pool-status":
stats = pool.pool_stats
return ReadResourceResult(
contents=[TextContent(type="text", text=json.dumps(stats, indent=2))]
)
elif uri == "db://slow-query-top10":
rows = await conn.fetch("""
SELECT query, calls, total_exec_time, mean_exec_time, rows
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 10
""")
return ReadResourceResult(
contents=[TextContent(type="text", text=json.dumps(
[dict(r) for r in rows], ensure_ascii=False, default=str, indent=2
))]
)
elif uri == "db://table-sizes":
rows = await conn.fetch("""
SELECT
schemaname,
tablename,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as total_size,
pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) as table_size,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename) -
pg_relation_size(schemaname||'.'||tablename)) as index_size,
n_live_tup as row_count
FROM pg_tables t
LEFT JOIN pg_stat_user_tables s ON t.tablename = s.relname
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
LIMIT 50
""")
return ReadResourceResult(
contents=[TextContent(type="text", text=json.dumps(
[dict(r) for r in rows], ensure_ascii=False, indent=2
))]
)
else:
return ReadResourceResult(
contents=[TextContent(type="text", text=f"未知资源: {uri}")],
isError=True
)
# ─── Prompt定义 ───────────────────────────────────
@mcp_server.list_prompts()
async def handle_list_prompts():
return [
Prompt(
name="performance-audit",
description="对数据库进行全面性能审计，输出瓶颈分析和优化方案",
arguments=[
PromptArgument(name="focus", description="关注重点：general|queries|locks|memory", required=False),
PromptArgument(name="time_range_hours", description="分析时间范围（小时），默认24", required=False)
]
),
Prompt(
name="migration-plan",
description="生成数据库迁移方案，包含风险评估和回滚计划",
arguments=[
PromptArgument(name="target_schema", description="目标Schema的SQL定义", required=True),
PromptArgument(name="current_schema", description="当前Schema信息（可选，可通过schema工具获取）", required=False)
]
)
]
@mcp_server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict | None) -> GetPromptResult:
args = arguments or {}
if name == "performance-audit":
focus = args.get("focus", "general")
hours = args.get("time_range_hours", 24)
return GetPromptResult(
messages=[PromptMessage(
role="user",
content=TextContent(type="text", text=f"""你是一位资深DBA和数据库性能优化专家。请对当前数据库进行全面的性能审计。
## 审计范围
- 关注维度：{focus}
- 时间范围：最近{hours}小时
## 审计步骤
1. 使用 get_slow_queries 获取慢查询列表
2. 使用 explain_query 对Top 5慢查询进行执行计划分析
3. 读取 db://table-sizes 资源查看表大小分布
4. 读取 db://pool-status 检查连接池健康度
## 输出要求
请按以下格式输出审计报告：
### 1. 总体健康评分（0-100）
包含各维度评分：查询性能、连接管理、存储效率
### 2. 关键发现
按严重程度排序（🔴 Critical > 🟡 Warning > 🟢 Info）
### 3. Top 5 慢查询优化方案
每个查询提供：
- 当前执行计划的问题分析
- 优化后的SQL
- 预期性能提升
### 4. 索引优化建议
需要创建的索引及其预期收益
### 5. 配置调优建议
PostgreSQL参数调整建议（shared_buffers, work_mem, effective_cache_size等）""")
)]
)
elif name == "migration-plan":
return GetPromptResult(
messages=[PromptMessage(
role="user",
content=TextContent(type="text", text=f"""你是一位数据库架构师。请根据以下Schema信息生成迁移方案。
目标Schema：
```sql
{args.get('target_schema', '')}
请先使用 get_schema 工具了解当前数据库结构，然后：
- 对比当前和目标Schema的差异
- 评估迁移风险
- 生成完整的迁移SQL脚本
- 编写回滚方案
- 估算迁移时间和锁表影响
输出格式要求：
- 每个SQL语句附带注释说明
- 标注破坏性变更（DROP/ALTER COLUMN等）
- 提供分步执行计划（方便灰度发布）""")
)]
)
─── 启动 ─────────────────────────────────────────
if name == "main":
import mcp.server.stdio
mcp.server.stdio.run(mcp_server)
这段代码展示了生产级MCP Server的完整架构设计。注意以下几个关键设计决策：
1. **生命周期管理**（`server_lifespan`）：连接池在服务器启动时初始化，关闭时清理，避免资源泄漏
2. **信号量限流**（`asyncio.Semaphore`）：控制并发查询数，防止单个客户端压垮数据库
3. **滑动窗口限流**：基于时间窗口的请求频率控制
4. **只读安全默认**：生产环境默认只读，写入操作需要显式切换
5. **审计日志**：所有查询操作都有完整记录，满足合规要求
6. **错误恢复建议**：每个错误都附带修复建议，帮助AI理解问题并给出解决方案
### 4.3 Streamable HTTP传输部署
```python
# deploy.py - 使用Streamable HTTP模式部署MCP Server
"""
将MCP Server部署为HTTP服务，支持Streamable HTTP传输
适用于生产环境，支持多客户端并发访问
"""
import json
import uvicorn
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
from mcp.server.streamable_http import streamable_http_app
app = FastAPI(title="Enterprise DB MCP Server")
# 将MCP Server挂载到FastAPI
mcp_app = streamable_http_app(mcp_server)
app.mount("/mcp", mcp_app)
# 健康检查端点
@app.get("/health")
async def health():
return {
"status": "healthy",
"pool": mcp_server.request_context.lifespan_context["pool"].pool_stats
if mcp_server.request_context and mcp_server.request_context.lifespan_context
else {"status": "initializing"}
}
# 指标端点
@app.get("/metrics")
async def metrics():
ctx = mcp_server.request_context
if ctx and ctx.lifespan_context:
executor = ctx.lifespan_context["executor"]
return {
"audit_entries": len(executor._audit_log),
"pool_stats": executor.pool.pool_stats
}
return {"status": "not_ready"}
if __name__ == "__main__":
uvicorn.run(
"deploy:app",
host="0.0.0.0",
port=8000,
workers=4, # 多Worker部署
limit_concurrency=100,
timeout_keep_alive=30
)
五、MCP客户端集成实战
5.1 在AI应用中集成MCP
# client_integration.py
"""
在你的AI应用中集成MCP客户端
支持多Server并发调用、结果聚合、错误重试
"""
import asyncio
import json
from typing import Any
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
class MCPToolKit:
"""MCP工具集管理器 - 在AI应用中管理多个MCP Server"""
def __init__(self):
self._sessions: dict[str, ClientSession] = {}
self._tools_cache: dict[str, list[dict]] = {}
async def connect_stdio(self, name: str, command: str, args: list[str], env: dict = None):
"""连接stdio模式的MCP Server"""
server_params = StdioServerParameters(
command=command,
args=args,
env=env
)
read_stream, write_stream = await stdio_client(server_params).__aenter__()
session = ClientSession(read_stream, write_stream)
await session.__aenter__()
await session.initialize()
self._sessions[name] = session
# 缓存工具列表
tools_result = await session.list_tools()
self._tools_cache[name] = [
{
"server": name,
"name": t.name,
"description": t.description,
"input_schema": t.inputSchema
}
for t in tools_result.tools
]
async def call_tool(self, server_name: str, tool_name: str, arguments: dict) -> Any:
"""调用指定Server的指定Tool"""
if server_name not in self._sessions:
raise ValueError(f"Server '{server_name}' 未连接")
session = self._sessions[server_name]
result = await session.call_tool(tool_name, arguments)
if result.isError:
raise RuntimeError(f"Tool调用失败: {result.content}")
return json.loads(result.content[0].text) if result.content else None
def get_all_tools(self) -> list[dict]:
"""获取所有已连接Server的工具列表（用于构建Function Calling schema）"""
return [t for tools in self._tools_cache.values() for t in tools]
def get_openai_tools_schema(self) -> list[dict]:
"""转换为OpenAI Function Calling格式"""
return [
{
"type": "function",
"function": {
"name": f"{t['server']}__{t['name']}",
"description": t['description'],
"parameters": t['input_schema']
}
}
for t in self.get_all_tools()
]
async def call_openai_tool(self, tool_call_id: str, function_name: str, arguments: str) -> str:
"""处理OpenAI的tool_calls响应，自动路由到对应的MCP Server"""
server_name, tool_name = function_name.split("__", 1)
args = json.loads(arguments)
result = await self.call_tool(server_name, tool_name, args)
return json.dumps(result, ensure_ascii=False, default=str)
async def close_all(self):
"""关闭所有连接"""
for session in self._sessions.values():
await session.__aexit__(None, None, None)
self._sessions.clear()
self._tools_cache.clear()
# ─── 使用示例 ─────────────────────────────────────
async def main():
toolkit = MCPToolKit()
# 连接多个MCP Server
await toolkit.connect_stdio(
"database",
command="python",
args=["-m", "db_mcp_server.server"],
env={"DB_HOST": "localhost", "DB_PORT": "5432"}
)
await toolkit.connect_stdio(
"filesystem",
command="npx",
args=["-y", "@anthropic/mcp-filesystem", "/app/data"]
)
# 获取所有工具的OpenAI格式schema
tools = toolkit.get_openai_tools_schema()
print(f"已加载 {len(tools)} 个工具")
# 模拟AI调用工具
result = await toolkit.call_tool("database", "get_schema", {})
print(f"数据库表列表: {json.dumps(result, ensure_ascii=False)}")
result = await toolkit.call_tool("database", "execute_query", {
"sql": "SELECT count(*) as total_orders, sum(amount) as revenue FROM orders WHERE created_at > now() - interval '30 days'"
})
print(f"近30天订单统计: {json.dumps(result, ensure_ascii=False)}")
await toolkit.close_all()
if __name__ == "__main__":
asyncio.run(main())
六、性能优化与最佳实践
6.1 连接池调优
# 不同场景的连接池配置建议
# 开发环境
dev_config = DBConfig(
pool_size=5,
max_overflow=2,
statement_timeout=60
)
# 中等流量生产环境
medium_config = DBConfig(
pool_size=20,
max_overflow=10,
statement_timeout=30,
read_only=True
)
# 高并发生产环境
high_config = DBConfig(
pool_size=50,
max_overflow=20,
statement_timeout=15,
read_only=True
)
关键经验法则：
pool_size≈ 数据库CPU核心数 × 2（PostgreSQL推荐）max_overflow≈pool_size× 0.5（应对突发流量）statement_timeout根据SLA设置，生产环境建议 ≤ 30秒- 永远不要在MCP Server中直接管理事务——让调用方控制事务边界
6.2 结果缓存策略
from functools import lru_cache
from datetime import datetime, timedelta
import hashlib
import json
class QueryCache:
"""带TTL和LRU淘汰的查询缓存"""
def __init__(self, max_size: int = 1000, default_ttl: int = 60):
self._cache: dict[str, tuple[datetime, Any]] = {}
self.max_size = max_size
self.default_ttl = default_ttl
def _make_key(self, sql: str, params: tuple = ()) -> str:
raw = f"{sql}|{json.dumps(params, sort_keys=True)}"
return hashlib.sha256(raw.encode()).hexdigest()
def get(self, sql: str, params: tuple = ()) -> Any | None:
key = self._make_key(sql, params)
if key in self._cache:
ts, value = self._cache[key]
if datetime.now() - ts < timedelta(seconds=self.default_ttl):
return value
del self._cache[key]
return None
def put(self, sql: str, params: tuple, result: Any, ttl: int | None = None):
key = self._make_key(sql, params)
if len(self._cache) >= self.max_size:
# LRU: 删除最旧的
oldest_key = min(self._cache, key=lambda k: self._cache[k][0])
del self._cache[oldest_key]
self._cache[key] = (datetime.now(), result)
缓存策略选择指南：
| 数据类型 | TTL建议 | 缓存策略 |
|---|---|---|
| Schema信息 | 5分钟 | 长TTL，变更时主动失效 |
| 统计查询结果 | 1-5分钟 | 根据数据更新频率调整 |
| 明细查询结果 | 不缓存 | 每次实时查询 |
| 执行计划 | 10分钟 | Schema变更时失效 |
6.3 安全最佳实践
# 安全配置检查清单
SECURITY_CHECKLIST = {
"认证": {
"✅ 必须项": [
"每个MCP Server有独立的API Token",
"Token定期轮换（建议30天）",
"支持Token撤销机制",
],
"⚠️ 建议项": [
"支持OAuth2.0或mTLS认证",
"按工具级别配置权限",
"记录所有认证失败事件"
]
},
"输入验证": {
"✅ 必须项": [
"所有SQL使用参数化查询",
"限制单次查询返回行数",
"设置查询超时时间",
"验证输入SQL的语句类型（只读模式）"
],
"⚠️ 建议项": [
"SQL关键词黑名单（LOAD_FILE, INTO OUTFILE等）",
"查询复杂度限制（子查询嵌套深度）",
"资源使用配额（每个客户端的CPU/内存限制）"
]
},
"审计与监控": {
"✅ 必须项": [
"记录所有工具调用（含参数和结果摘要）",
"记录执行时间和资源消耗",
"异常调用告警（频率异常、参数异常）"
],
"⚠️ 建议项": [
"与现有SIEM系统集成",
"实时仪表盘展示调用统计",
"异常模式自动检测和阻断"
]
}
}
七、常见问题与踩坑记录
7.1 并发调用的坑
问题：多个AI会话同时调用同一个MCP Server的同一个Tool，导致数据库连接耗尽。
解决：使用信号量控制并发，并在每个Tool调用中添加独立的超时控制：
async def safe_execute(self, sql: str, session_id: str) -> dict:
"""安全的并发查询执行"""
try:
async with asyncio.wait_for(
self._semaphore.acquire(),
timeout=5.0 # 等待信号量最多5秒
):
try:
async with asyncio.wait_for(
self._execute_internal(sql),
timeout=self.config.statement_timeout
):
return await self._execute_internal(sql)
except asyncio.TimeoutError:
return {"error": "查询超时，请优化SQL或缩小查询范围"}
except asyncio.TimeoutError:
return {"error": "服务器繁忙，请稍后重试"}
7.2 大结果集的坑
问题：AI要求"查询所有数据"，返回百万行结果直接撑爆内存。
解决：强制分页，MCP Server端必须有硬限制：
MAX_ROWS = 1000 # 硬限制，不可配置
async def execute_with_limit(self, sql: str, limit: int = None) -> dict:
effective_limit = min(limit or MAX_ROWS, MAX_ROWS)
# 检查原始SQL是否已有LIMIT
sql_upper = sql.strip().rstrip(';').upper()
if not sql_upper.endswith(f"LIMIT {effective_limit}"):
sql = f"{sql} LIMIT {effective_limit}"
# 执行并检查是否有更多数据
count_sql = f"SELECT count(*) FROM ({sql.strip().rstrip(';').replace(f'LIMIT {effective_limit}', '')}) as t"
total = await self._execute_scalar(count_sql)
return {
"rows": result_rows,
"returned": len(result_rows),
"total": total,
"has_more": total > effective_limit,
"hint": f"当前返回{effective_limit}行/共{total}行，如需更多数据请使用分页查询"
}
7.3 协议版本兼容的坑
问题：客户端和服务端使用了不同版本的MCP协议，导致能力协商失败。
解决：在服务端做好优雅降级：
async def handle_initialize(self, params: dict) -> dict:
client_version = params.get("protocolVersion", "unknown")
supported_versions = ["2024-11-05", "2025-03-26", "2025-06-18"]
# 选择双方都支持的最新版本
negotiated_version = "2024-11-05" # 最低版本作为fallback
for v in supported_versions:
if v <= client_version:
negotiated_version = v
# 根据版本调整能力声明
capabilities = {"tools": {}}
if negotiated_version >= "2025-03-26":
capabilities["logging"] = {}
if negotiated_version >= "2025-06-18":
capabilities["tools"]["listChanged"] = True
return {
"protocolVersion": negotiated_version,
"capabilities": capabilities
}
八、2026年MCP生态现状与展望
8.1 主流AI平台的MCP支持情况
截至2026年5月，MCP生态已经相当成熟：
| 平台 | 支持程度 | 传输模式 | 特点 |
|---|---|---|---|
| Claude Desktop | 原生支持 | stdio | 最成熟的客户端，支持所有三种原语 |
| OpenAI ChatGPT | 部分兼容 | SSE | 通过Plugin系统间接支持 |
| Google Gemini | 实验性支持 | SSE | 原生集成中 |
| OpenClaw | 完整支持 | stdio/SSE | 开源AI助手框架，MCP是一等公民 |
| Cursor | 完整支持 | stdio | 编程IDE，深度集成MCP工具 |
| Windsurf | 完整支持 | stdio | 编程IDE，原生MCP支持 |
8.2 MCP Registry的崛起
2026年初，MCP官方Registry（类似npm for MCP）正式上线，目前已经收录了超过5000个社区贡献的MCP Server，覆盖：
- 数据库：PostgreSQL、MySQL、MongoDB、Redis、SQLite
- 云服务：AWS、GCP、Azure、阿里云、腾讯云
- 开发工具：GitHub、GitLab、Jira、Linear
- 通讯：Slack、Discord、飞书、钉钉
- 文件系统：本地文件、S3、GCS、OSS
- AI/ML：HuggingFace、LangChain、向量数据库
8.3 从MCP到A2A的演进
Google在2025年推出的A2A（Agent-to-Agent）协议正在与MCP形成互补：
- MCP：Agent ↔ Tools（Agent调用工具）
- A2A：Agent ↔ Agent（Agent之间协作）
未来趋势是两者融合：一个完整的AI应用同时使用MCP连接工具、使用A2A与其他Agent协作。
8.4 对开发者的建议
如果你正在考虑是否要为你的服务构建MCP接口，我的建议是：
- 现在就开始。MCP已经成为事实标准，早接入早受益。
- 从Tools开始。大多数场景只需要Tools一种原语，不要过度设计。
- 重视Tool的description。这是AI理解你工具的唯一途径，写好描述比写好代码更重要。
- 做好安全隔离。MCP Server直接操作你的系统资源，安全不能马虎。
- 参与社区。MCP Registry是获取用户反馈和贡献代码的最佳渠道。
九、总结
MCP不仅仅是一个协议，它正在重新定义AI应用与外部世界交互的方式。从2024年底的概念验证到2026年的事实标准，MCP用一年半的时间完成了大多数协议需要五年才能走完的普及之路。
核心要点回顾：
- 三种传输模式各有适用场景，生产环境推荐Streamable HTTP
- 三种原语（Tools/Resources/Prompts）分工明确，覆盖AI"动手/看见/会说"三大能力
- 能力协商机制是MCP比简单REST API高明的关键，支持优雅降级和功能发现
- 生产级Server需要关注连接池、限流、审计、安全四个维度
- Tool设计的核心是写好description——这是AI理解你工具的唯一途径
回到开头的问题：MCP之于AI，就像USB-C之于电子设备。当所有设备都统一了接口，开发者终于可以专注于"做什么"而不是"怎么连"。这才是MCP真正的价值——不是又搞了一套协议，而是终结了协议碎片化的混乱时代。
开始构建你的第一个MCP Server吧。这个领域还太新，到处都是机会。
本文基于MCP协议2025-06-18版本撰写，代码示例已在Python 3.12 + asyncpg 0.30环境中验证。