MCP（Model Context Protocol）是Anthropic在2024年末推出的开放协议，旨在解决AI Agent与外部工具/数据源集成的碎片化问题。两年后的今天，MCP已经成为AI工具生态的事实标准，支持MCP的工具超过3000个。本文深入探讨MCP的工程实践，从协议理解到生产部署。

MCP解决了什么问题### 没有MCP之前的混乱在MCP出现之前，每个AI工具都有自己的集成方式：Claude插件：一套APIChatGPT插件：另一套APILangChain工具：又一套抽象AutoGEN工具：再一套定义...结果：- 同一个工具需要为N个AI平台写N份集成代码- 开发者疲于维护多套适配器- 用户体验碎片化### MCP的标准化方案 MCP标准协议 ┌──────────────────────┐ │ Claude / 其他支持MCP │ │ 的AI应用 │ └──────────────────────┘ ↕ MCP ┌──────────────────────┐ │ MCP Server │ │ （数据库/文件/API等） │ └──────────────────────┘一次实现 → 所有支持MCP的AI应用均可使用## MCP核心概念### 三种资源类型Resources（资源）：AI可以读取的数据json{ "name": "database://users", "description": "用户数据表", "mimeType": "application/json"}Tools（工具）：AI可以调用的操作json{ "name": "execute_sql", "description": "在数据库中执行SQL查询", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "SQL语句"}, "limit": {"type": "integer", "default": 100} }, "required": ["query"] }}Prompts（提示词）：AI可调用的预定义Prompt模板json{ "name": "analyze_data", "description": "分析数据集并生成报告", "arguments": [ {"name": "dataset_name", "required": true}, {"name": "analysis_type", "required": false} ]}### 传输层MCP支持两种传输方式：1. stdio（标准输入/输出）：适合本地工具 AI应用 ←→ 进程stdin/stdout ←→ MCP Server2. HTTP SSE（服务器推送事件）：适合远程服务 AI应用 ←→ HTTP连接 ←→ MCP Server## 构建第一个MCP Server### 使用Python MCP SDKpythonfrom mcp.server import Server, NotificationOptionsfrom mcp.server.models import InitializationOptionsimport mcp.server.stdioimport mcp.types as typesimport asyncio# 初始化MCP Serverapp = Server("my-database-server")# ─── 定义工具 ───────────────────────────────────────────@app.list_tools()async def handle_list_tools() -> list[types.Tool]: """列出所有可用工具""" return [ types.Tool( name="query_users", description="查询用户数据库，支持按条件筛选", inputSchema={ "type": "object", "properties": { "filter": { "type": "object", "description": "查询条件，如 {'age': {'$gt': 18}}", }, "limit": { "type": "integer", "description": "返回记录数上限", "default": 10 } } } ), types.Tool( name="create_report", description="基于查询结果生成CSV格式报告", inputSchema={ "type": "object", "properties": { "data": {"type": "array", "description": "数据数组"}, "title": {"type": "string", "description": "报告标题"} }, "required": ["data", "title"] } ) ]@app.call_tool()async def handle_call_tool( name: str, arguments: dict) -> list[types.TextContent | types.ImageContent]: """处理工具调用""" if name == "query_users": filter_cond = arguments.get("filter", {}) limit = arguments.get("limit", 10) # 实际数据库查询 try: results = await db.users.find(filter_cond).limit(limit).to_list() return [types.TextContent( type="text", text=json.dumps(results, ensure_ascii=False, indent=2) )] except Exception as e: return [types.TextContent( type="text", text=f"查询错误：{str(e)}" )] elif name == "create_report": data = arguments["data"] title = arguments["title"] csv_content = f"# {title}\n" if data: headers = list(data[0].keys()) csv_content += ",".join(headers) + "\n" for row in data: csv_content += ",".join(str(row.get(h, '')) for h in headers) + "\n" return [types.TextContent(type="text", text=csv_content)] raise ValueError(f"未知工具：{name}")# ─── 定义资源 ───────────────────────────────────────────@app.list_resources()async def handle_list_resources() -> list[types.Resource]: """列出可用数据资源""" return [ types.Resource( uri="database://schema", name="数据库Schema", description="数据库表结构定义", mimeType="application/json" ), types.Resource( uri="database://stats", name="数据库统计", description="实时统计信息（记录数、最后更新时间等）", mimeType="application/json" ) ]@app.read_resource()async def handle_read_resource(uri: str) -> str: """读取资源内容""" if str(uri) == "database://schema": schema = await db.command("dbStats") return json.dumps(schema, ensure_ascii=False, indent=2) elif str(uri) == "database://stats": stats = { "total_users": await db.users.count_documents({}), "last_updated": datetime.now().isoformat() } return json.dumps(stats, ensure_ascii=False) raise ValueError(f"未知资源：{uri}")# ─── 启动Server ───────────────────────────────────────────async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, InitializationOptions( server_name="my-database-server", server_version="1.0.0", capabilities=app.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={} ) ) )if __name__ == "__main__": asyncio.run(main())### 配置到Claude Desktopjson// ~/.config/claude/claude_desktop_config.json（macOS）// %APPDATA%\Claude\claude_desktop_config.json（Windows）{ "mcpServers": { "my-database": { "command": "python", "args": ["/path/to/mcp_server.py"], "env": { "DATABASE_URL": "mongodb://localhost:27017/mydb", "LOG_LEVEL": "INFO" } } }}## 构建HTTP MCP Server（用于远程部署）pythonfrom fastapi import FastAPIfrom fastapi.responses import StreamingResponseimport asyncioimport jsonapp_http = FastAPI(title="MCP Database Server")# SSE连接管理connections: dict[str, asyncio.Queue] = {}@app_http.get("/sse")async def sse_endpoint(): """SSE连接入口""" conn_id = str(uuid.uuid4()) connections[conn_id] = asyncio.Queue() async def event_generator(): try: while True: message = await connections[conn_id].get() yield f"data: {json.dumps(message)}\n\n" finally: del connections[conn_id] return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no" } )@app_http.post("/message")async def handle_message(request: Request): """处理来自AI客户端的消息""" body = await request.json() method = body.get("method") if method == "tools/list": return await list_tools_handler() elif method == "tools/call": return await call_tool_handler(body["params"]) elif method == "resources/list": return await list_resources_handler() return {"error": f"Unknown method: {method}"}## MCP安全最佳实践### 工具权限控制pythonfrom enum import Enumfrom functools import wrapsclass PermissionLevel(Enum): READ_ONLY = "read_only" READ_WRITE = "read_write" ADMIN = "admin"# 工具权限声明TOOL_PERMISSIONS = { "query_users": PermissionLevel.READ_ONLY, "update_user": PermissionLevel.READ_WRITE, "delete_user": PermissionLevel.ADMIN, "create_report": PermissionLevel.READ_ONLY,}def require_permission(required_level: PermissionLevel): """权限检查装饰器""" def decorator(func): @wraps(func) async def wrapper(tool_name: str, arguments: dict, context: dict): user_level = context.get("permission_level", PermissionLevel.READ_ONLY) tool_level = TOOL_PERMISSIONS.get(tool_name) if not has_permission(user_level, tool_level): raise PermissionError( f"工具 '{tool_name}' 需要 {tool_level.value} 权限，" f"当前权限：{user_level.value}" ) return await func(tool_name, arguments, context) return wrapper return decorator### 输入验证与沙箱pythonimport astimport reclass MCPInputSanitizer: """MCP工具输入安全检查""" # SQL注入检测 SQL_INJECTION_PATTERNS = [ r";\s*DROP\s+TABLE", r";\s*DELETE\s+FROM", r"UNION\s+SELECT", r"1\s*=\s*1", ] # 路径遍历攻击检测 PATH_TRAVERSAL_PATTERNS = [ r"\.\./", r"\.\.\\", r"%2e%2e", ] def sanitize_sql_query(self, query: str) -> str: """SQL查询安全检查""" query_upper = query.upper() for pattern in self.SQL_INJECTION_PATTERNS: if re.search(pattern, query_upper): raise ValueError(f"危险的SQL模式：{pattern}") # 只允许SELECT语句 if not query_upper.strip().startswith("SELECT"): raise ValueError("只允许SELECT查询") return query def sanitize_file_path(self, path: str) -> str: """文件路径安全检查""" for pattern in self.PATH_TRAVERSAL_PATTERNS: if re.search(pattern, path.lower()): raise ValueError("危险的文件路径") # 限制在允许目录内 import os allowed_base = "/app/data" full_path = os.path.normpath(os.path.join(allowed_base, path)) if not full_path.startswith(allowed_base): raise ValueError("文件路径越界") return full_path## 实用MCP Server推荐（2026年）yaml# 常用的开源MCP Server数据库类： - mcp-server-postgres：PostgreSQL读写 - mcp-server-sqlite：SQLite操作 - mcp-server-mongodb：MongoDB查询 文件系统类： - mcp-server-filesystem：本地文件读写 - mcp-server-s3：AWS S3操作 开发工具类： - mcp-server-github：GitHub Issues/PR/代码操作 - mcp-server-docker：容器管理 - mcp-server-kubernetes：K8s资源管理 搜索类： - mcp-server-brave-search：Brave搜索引擎 - mcp-server-fetch：网页内容获取 生产力类： - mcp-server-slack：Slack消息发送 - mcp-server-notion：Notion笔记操作 - mcp-server-google-maps：地图和位置服务## MCP vs Function Calling：怎么选| 维度 | MCP | Function Calling ||------|-----|-----------------|| 标准化 | 跨AI平台标准 | 各平台不同实现 || 部署方式 | 独立进程/服务 | 代码内定义 || 适用场景 | 可复用工具、数据源 | 特定应用内的工具 || 安全隔离 | 独立进程，隔离好 | 同进程，隔离弱 || 开发成本 | 较高（需要独立服务） | 较低（直接写函数） || 生态 | 快速增长（3000+） | 需要自行维护 |决策规则：- 工具需要被多个AI应用/用户共享 → MCP- 工具只用于特定应用的内部逻辑 → Function Calling- 工具需要持久运行（如数据库连接池） → MCP- 工具是一次性/轻量 → Function Calling## 总结MCP代表了AI工具集成从"各自为战"到"标准化生态"的范式转变：1. 协议简单清晰：三种资源类型（Tools/Resources/Prompts），学习成本低2. 生态快速成长：3000+工具，基本覆盖所有常见集成需求3. 安全设计先行：进程隔离、权限控制、输入验证要从第一天做起4. HTTP SSE支持远程部署：不局限于本地工具，可构建共享服务5. 选型要理智：不是所有工具都需要MCP，简单的Function Calling有时更合适掌握MCP，意味着你构建的工具可以被整个AI生态系统复用，而不是局限于单一平台。这是2026年AI工程师的核心竞争力之一。