MCP 实战指南从本地 AI 工具到 Web 项目的完整接入方案阅读时间约35分钟 |你将学到MCP 的真实调用原理、3种本地接入方式、2种 Web 接入架构、完整可运行代码核心观点MCP 不只是一个协议规范它是一套工具发现 工具调用 结果返回的完整通信机制。本地用 stdioWeb 用 HTTP/SSE原理相同传输方式不同。一、MCP 到底是什么3分钟真正搞懂1.1 一句话说清楚MCPModel Context Protocol AI 应用调用外部工具的标准协议。就像 USB 是硬件接口标准MCP 是 AI 调用工具的接口标准。没有 MCP 的世界 有 MCP 的世界 ┌──────────┐ ┌──────────┐ │ AI 应用 │ │ AI 应用 │ │ │ │ │ │ 调GitHub → 写一套适配代码 │ 统一调用 │────→ MCP Server AGitHub │ 调数据库 → 又写一套适配代码 │ MCP 协议 │────→ MCP Server B数据库 │ 调搜索引擎 → 再写一套适配代码 │ │────→ MCP Server C搜索 └──────────┘ └──────────┘ 每个工具一套接口维护成本爆炸 一套协议所有工具通用1.2 MCP 的三个核心能力能力说明类比ToolsAI 可以调用的函数你给AI的工具箱ResourcesAI 可以读取的数据你给AI的资料库Prompts预设的提示词模板你给AI的话术本实际项目中最常用的是Tools占 95% 的使用场景。1.3 通信协议JSON-RPC 2.0MCP 用 JSON-RPC 2.0 通信。就两种消息// 请求AI 应用 → MCP Server { jsonrpc: 2.0, id: 1, method: tools/call, params: { name: search_github, arguments: {query: mcp server} } } // 响应MCP Server → AI 应用 { jsonrpc: 2.0, id: 1, result: { content: [ {type: text, text: 找到 1234 个仓库...} ] } }就这么简单。请求带着方法名和参数响应带着结果。二、本地 AI 工具中使用 MCP2.1 本地 MCP 的通信方式stdio本地 AI 工具OpenClaw、Claude Desktop、Claude Code用的是stdio标准输入输出通信。┌─────────────────────┐ stdin/stdout ┌─────────────────────┐ │ AI 应用进程 │ ←──────────────────→ │ MCP Server 进程 │ │ (Claude Desktop) │ JSON-RPC 2.0 │ (子进程) │ └─────────────────────┘ └─────────────────────┘ │ │ │ 不需要网络 │ 调用实际资源 │ 同一台机器 │ ▼ ▼ 用户交互 文件系统/数据库/API关键点MCP Server 是 AI 应用拉起来的子进程通过管道pipe通信不需要网络。2.2 实战一Claude Desktop 接入 MCP Server第 1 步安装 MCP Server# 安装文件系统 MCP Server npm install -g modelcontextprotocol/server-filesystem # 安装 GitHub MCP Server npm install -g modelcontextprotocol/server-github # 安装 SQLite MCP Server npm install -g modelcontextprotocol/server-sqlite第 2 步配置 Claude Desktop编辑~/Library/Application Support/Claude/claude_desktop_config.jsonmacOS或%APPDATA%\Claude\claude_desktop_config.jsonWindows{ mcpServers: { filesystem: { command: npx, args: [ -y, modelcontextprotocol/server-filesystem, /Users/xiaoyuer/projects, /Users/xiaoyuer/Documents ] }, github: { command: npx, args: [-y, modelcontextprotocol/server-github], env: { GITHUB_PERSONAL_ACCESS_TOKEN: ghp_your_token_here } }, sqlite: { command: npx, args: [ -y, modelcontextprotocol/server-sqlite, /path/to/your/database.db ] } } }第 3 步重启 Claude Desktop开始使用重启后Claude 会自动拉起配置中的 MCP Server 进程。你可以在对话中直接说读一下 /Users/xiaoyuer/projects/main.py搜索 GitHub 上 star 最多的 MCP 项目查一下数据库里最近的订单Claude 会自动调用对应的 MCP Server。2.3 实战二OpenClaw 接入 MCP ServerOpenClaw 的 MCP 配置在~/.openclaw/config.yaml中# ~/.openclaw/config.yaml mcp: servers: # 文件系统 MCP Server filesystem: command: npx args: - -y - modelcontextprotocol/server-filesystem - /Users/xiaoyuer/projects # 自定义 MCP ServerPython 写的 my-tools: command: python args: - /Users/xiaoyuer/mcp-servers/my_tools_server.py # 远程 MCP ServerHTTP/SSE remote-search: url: https://mcp.example.com/sseOpenClaw 的特殊能力同时支持 stdio 和 HTTP/SSE 两种方式连接 MCP Server。2.4 实战三Claude Code 接入 MCP ServerClaude Code 通过命令行配置# 添加 MCP Server claude mcp add filesystem npx -y modelcontextprotocol/server-filesystem /Users/xiaoyuer/projects # 添加带环境变量的 MCP Server claude mcp add github -e GITHUB_PERSONAL_ACCESS_TOKENghp_xxx -- npx -y modelcontextprotocol/server-github # 查看已配置的 MCP Server claude mcp list # 删除 MCP Server claude mcp remove filesystem2.5 自己写一个 MCP ServerPython这是最重要的实战能力。项目里经常需要自定义工具。 自定义 MCP Server - 项目管理工具 功能查询项目状态、搜索代码、管理任务 import json import sys import asyncio from typing import Any # 安装依赖pip install mcp from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Tool, TextContent, ImageContent, EmbeddedResource, ) # 创建 Server 实例 app Server(project-manager) # 1. 注册工具列表 app.list_tools() async def list_tools() - list[Tool]: 告诉 AI 应用我提供哪些工具 return [ Tool( nameget_project_status, description获取项目状态包括进度、待办事项、最近变更, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 } }, required: [project_name] } ), Tool( namesearch_code, description在项目代码中搜索关键词, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 }, keyword: { type: string, description: 搜索关键词 }, file_type: { type: string, description: 文件类型过滤如 py/js/ts, default: None } }, required: [project_name, keyword] } ), Tool( namecreate_task, description创建项目管理任务, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 }, title: { type: string, description: 任务标题 }, priority: { type: string, enum: [high, medium, low], description: 优先级, default: medium }, assignee: { type: string, description: 负责人, default: None } }, required: [project_name, title] } ), ] # 2. 实现工具逻辑 app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) - list[TextContent | ImageContent | EmbeddedResource]: 执行工具调用 if name get_project_status: return await handle_get_project_status(arguments) elif name search_code: return await handle_search_code(arguments) elif name create_task: return await handle_create_task(arguments) else: return [TextContent(typetext, textf未知工具{name})] async def handle_get_project_status(args: dict) - list[TextContent]: 获取项目状态 project_name args[project_name] # 这里替换成你的实际逻辑 # 比如查数据库、调API、读文件等 status { project: project_name, progress: 68%, open_issues: 12, recent_commits: [ {hash: a1b2c3d, message: feat: 添加用户认证, author: zhangsan}, {hash: e4f5g6h, message: fix: 修复分页bug, author: lisi}, ], pending_tasks: [ {title: 接口压测, priority: high}, {title: 文档更新, priority: medium}, ] } return [TextContent( typetext, textjson.dumps(status, ensure_asciiFalse, indent2) )] async def handle_search_code(args: dict) - list[TextContent]: 搜索代码 project_name args[project_name] keyword args[keyword] file_type args.get(file_type) # 这里替换成你的实际逻辑 # 比如用 grep/ripgrep 搜索、调 GitHub API 等 results [ { file: fsrc/api/{project_name}/routes.py, line: 42, content: fdef search_{keyword}(query: str): }, { file: fsrc/services/{project_name}/service.py, line: 88, content: fclass {keyword.capitalize()}Service: }, ] return [TextContent( typetext, textjson.dumps(results, ensure_asciiFalse, indent2) )] async def handle_create_task(args: dict) - list[TextContent]: 创建任务 project_name args[project_name] title args[title] priority args.get(priority, medium) assignee args.get(assignee) # 这里替换成你的实际逻辑 # 比如调 Jira API、写数据库等 task { id: TASK-2026-001, project: project_name, title: title, priority: priority, assignee: assignee or 未分配, status: created } return [TextContent( typetext, textjson.dumps(task, ensure_asciiFalse, indent2) )] # 3. 启动 Server async def main(): 启动 MCP Serverstdio 模式 async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ __main__: asyncio.run(main())配置到 Claude Desktop{ mcpServers: { project-manager: { command: python, args: [/Users/xiaoyuer/mcp-servers/my_tools_server.py] } } }重启 Claude Desktop 后你就能直接说查一下 smart-trip-planner 项目状态或在项目里搜一下 auth 相关代码Claude 会自动调用你的 MCP Server。三、Web 项目中使用 MCP3.1 Web 项目和本地 AI 工具的核心区别本地 AI 工具Claude Desktop / OpenClaw ┌─────────────────┐ stdio管道 ┌─────────────────┐ │ AI 应用进程 │ ←────────────→ │ MCP Server 进程 │ │ 同一台机器 │ │ 同一台机器 │ └─────────────────┘ └─────────────────┘ 特点不需要网络进程间直接通信 Web 项目 ┌─────────────────┐ HTTP/SSE网络 ┌─────────────────┐ │ Web 后端 │ ←───────────────→ │ MCP Server 服务 │ │ 任意机器 │ │ 可能是另一台机器 │ └─────────────────┘ └─────────────────┘ 特点需要网络跨机器通信需要考虑认证、超时、负载均衡核心区别维度本地 AI 工具Web 项目通信方式stdio管道HTTP/SSE网络MCP Server 生命周期随 AI 应用启动/关闭独立部署长期运行认证不需要需要考虑API Key / OAuth并发单用户多用户并发部署本地进程云服务器 / Docker延迟极低进程间较高网络传输3.2 Web 项目的两种 MCP 架构架构一Web 后端 MCP Client调用外部 MCP Server┌──────────────┐ ┌──────────────────┐ HTTP/SSE ┌──────────────────┐ │ 前端 │ ──→ │ Web 后端 │ ──────────────→ │ MCP Server │ │ (React/Vue) │ │ (FastAPI/Next) │ ←────────────── │ (远程服务) │ └──────────────┘ │ │ │ │ │ MCP Client 逻辑 │ │ 工具实现 │ └──────────────────┘ └──────────────────┘适用场景你的 Web 项目需要调用现成的 MCP Server 提供的工具。架构二Web 后端 MCP Server给 AI 应用提供工具┌──────────────────┐ HTTP/SSE ┌──────────────────┐ ┌──────────────┐ │ AI 应用 │ ──────────────→ │ Web 后端 │ ──→ │ 数据库/API │ │ (Claude/OpenClaw)│ ←────────────── │ (FastAPI/Next) │ ←── │ │ │ │ │ MCP Server 逻辑 │ └──────────────┘ └──────────────────┘ └──────────────────┘适用场景你想让 AI 应用直接调用你 Web 项目的功能。3.3 实战Web 后端作为 MCP Client Web 后端 - MCP Client FastAPI 项目调用远程 MCP Server 的工具 import httpx import json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(titleMCP Client Demo) class MCPClient: MCP 客户端HTTP 模式 def __init__(self, server_url: str, api_key: Optional[str] None): self.server_url server_url.rstrip(/) self.headers {} if api_key: self.headers[Authorization] fBearer {api_key} self._request_id 0 self._client None async def _get_client(self) - httpx.AsyncClient: if self._client is None or self._client.is_closed: self._client httpx.AsyncClient( timeout30.0, headersself.headers, ) return self._client async def _send_request(self, method: str, params: dict None) - dict: 发送 JSON-RPC 请求 self._request_id 1 payload { jsonrpc: 2.0, id: self._request_id, method: method, } if params: payload[params] params client await self._get_client() response await client.post( f{self.server_url}/mcp, jsonpayload, ) response.raise_for_status() data response.json() if error in data: raise Exception(fMCP Error: {data[error]}) return data.get(result, {}) async def initialize(self) - dict: 初始化连接 return await self._send_request(initialize, { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: { name: web-backend, version: 1.0.0 } }) async def list_tools(self) - list: 获取工具列表 result await self._send_request(tools/list) return result.get(tools, []) async def call_tool(self, name: str, arguments: dict) - dict: 调用工具 return await self._send_request(tools/call, { name: name, arguments: arguments }) async def close(self): if self._client and not self._client.is_closed: await self._client.aclose() # 全局 MCP Client 实例 mcp_client MCPClient( server_urlhttps://mcp.example.com, api_keyyour-api-key # 生产环境从环境变量读取 ) app.on_event(startup) async def startup(): 启动时初始化 MCP 连接 await mcp_client.initialize() tools await mcp_client.list_tools() print(f已连接 MCP Server可用工具{[t[name] for t in tools]}) app.on_event(shutdown) async def shutdown(): 关闭时断开连接 await mcp_client.close() # 业务接口 class SearchRequest(BaseModel): query: str project: Optional[str] None class TaskRequest(BaseModel): project_name: str title: str priority: str medium app.post(/api/search) async def search(req: SearchRequest): 搜索接口 - 调用 MCP Server 的搜索工具 result await mcp_client.call_tool(search_code, { project_name: req.project or default, keyword: req.query, }) return {data: result} app.post(/api/tasks) async def create_task(req: TaskRequest): 创建任务接口 - 调用 MCP Server 的任务工具 result await mcp_client.call_tool(create_task, { project_name: req.project_name, title: req.title, priority: req.priority, }) return {data: result} app.get(/api/tools) async def get_available_tools(): 查看当前可用的 MCP 工具 tools await mcp_client.list_tools() return {tools: tools} # 运行uvicorn main:app --host 0.0.0.0 --port 80003.4 实战Web 后端作为 MCP Server Web 后端 - MCP Server 把 Web 项目的功能暴露为 MCP 工具给 AI 应用调用 import json from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from typing import Any app FastAPI(titleMCP Server Demo) # 你的业务逻辑 class BusinessLogic: 业务逻辑层和 MCP 无关 staticmethod async def query_orders(user_id: str, status: str None) - list: 查询订单 # 实际项目中查数据库 return [ {id: ORD-001, user: user_id, amount: 299.0, status: paid}, {id: ORD-002, user: user_id, amount: 158.0, status: pending}, ] staticmethod async def get_user_info(user_id: str) - dict: 查询用户信息 # 实际项目中查数据库 return { id: user_id, name: 张三, level: VIP, orders_count: 42, } staticmethod async def cancel_order(order_id: str, reason: str) - dict: 取消订单 # 实际项目中写数据库 return { order_id: order_id, status: cancelled, reason: reason, refund_amount: 299.0, } biz BusinessLogic() # MCP 协议层 def mcp_response(req_id: int, result: Any) - dict: return {jsonrpc: 2.0, id: req_id, result: result} def mcp_error(req_id: int, code: int, message: str) - dict: return {jsonrpc: 2.0, id: req_id, error: {code: code, message: message}} # 工具注册表 TOOLS [ { name: query_orders, description: 查询用户订单, inputSchema: { type: object, properties: { user_id: {type: string, description: 用户ID}, status: {type: string, description: 订单状态过滤, default: None} }, required: [user_id] } }, { name: get_user_info, description: 查询用户信息, inputSchema: { type: object, properties: { user_id: {type: string, description: 用户ID} }, required: [user_id] } }, { name: cancel_order, description: 取消订单危险操作需要确认, inputSchema: { type: object, properties: { order_id: {type: string, description: 订单ID}, reason: {type: string, description: 取消原因} }, required: [order_id, reason] } } ] # 工具处理函数映射 TOOL_HANDLERS { query_orders: lambda args: biz.query_orders(args[user_id], args.get(status)), get_user_info: lambda args: biz.get_user_info(args[user_id]), cancel_order: lambda args: biz.cancel_order(args[order_id], args[reason]), } app.post(/mcp) async def handle_mcp(request: Request): 处理 MCP 请求 body await request.json() method body.get(method) params body.get(params, {}) req_id body.get(id) # 1. 初始化握手 if method initialize: return mcp_response(req_id, { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: { name: order-management-mcp, version: 1.0.0 } }) # 2. 初始化确认 if method notifications/initialized: return JSONResponse(content{}) # 3. 工具列表 if method tools/list: return mcp_response(req_id, {tools: TOOLS}) # 4. 工具调用 if method tools/call: tool_name params.get(name) arguments params.get(arguments, {}) handler TOOL_HANDLERS.get(tool_name) if not handler: return mcp_error(req_id, -32601, f未知工具{tool_name}) try: result await handler(arguments) return mcp_response(req_id, { content: [ {type: text, text: json.dumps(result, ensure_asciiFalse)} ] }) except Exception as e: return mcp_error(req_id, -32603, f工具执行失败{str(e)}) return mcp_error(req_id, -32601, f未知方法{method}) # 健康检查 app.get(/health) async def health(): return {status: ok, tools_count: len(TOOLS)} # 运行uvicorn main:app --host 0.0.0.0 --port 8080Claude Desktop 配置远程 MCP Server{ mcpServers: { order-management: { url: https://your-api.com/mcp } } }这样 Claude Desktop 就能直接查询订单、取消订单了。四、原理对比本地 vs Web4.1 通信方式对比本地模式stdio AI 应用进程 MCP Server 进程 ┌──────────┐ stdin写 ┌──────────┐ │ │ ──────────────→ │ │ │ stdout │ ←────────────── │ stdout │ │ 读 │ stdout写 │ 输出 │ └──────────┘ └──────────┘ 数据流AI 应用 → pipe → MCP Server → pipe → AI 应用 特点操作系统级管道零网络延迟进程同生命周期 Web 模式HTTP/SSE Web 后端 MCP Server ┌──────────┐ HTTP POST ┌──────────┐ │ │ ──────────────→ │ │ │ │ ←────────────── │ │ │ │ HTTP Response │ │ └──────────┘ └──────────┘ 数据流Web 后端 → 网络请求 → MCP Server → 网络响应 → Web 后端 特点需要网络独立部署需要认证可跨机器4.2 生命周期对比 本地模式的 MCP Server 生命周期 # 1. AI 应用启动时自动拉起 MCP Server import subprocess mcp_process subprocess.Popen( [npx, -y, modelcontextprotocol/server-filesystem, /tmp], stdinsubprocess.PIPE, # AI 应用写请求 stdoutsubprocess.PIPE, # AI 应用读响应 stderrsubprocess.PIPE, ) # 2. 通过管道通信 request json.dumps({jsonrpc: 2.0, id: 1, method: tools/list, params: {}}) mcp_process.stdin.write(f{request}\n.encode()) mcp_process.stdin.flush() response mcp_process.stdout.readline().decode() print(json.loads(response)) # 3. AI 应用关闭时MCP Server 也关闭 mcp_process.terminate() Web 模式的 MCP Server 生命周期 # 1. MCP Server 独立部署长期运行 # uvicorn mcp_server:app --host 0.0.0.0 --port 8080 # 2. Web 后端按需连接 import httpx async with httpx.AsyncClient() as client: response await client.post(http://mcp-server:8080/mcp, json{ jsonrpc: 2.0, id: 1, method: tools/list, params: {} }) print(response.json()) # 3. Web 后端关闭不影响 MCP Server # MCP Server 独立运行可以被多个 Web 后端调用4.3 完整对比表维度本地模式stdioWeb 模式HTTP/SSE通信操作系统管道HTTP 网络请求延迟微秒级毫秒~秒级认证不需要API Key / OAuth并发单用户多用户部署随 AI 应用启停独立部署长期运行发现配置文件声明URL 自动发现适用个人工具团队/生产环境错误处理进程崩溃 通信断需要重试/超时/熔断监控日志文件APM 健康检查多实例不支持负载均衡五、SSE 模式深入SSEServer-Sent Events是 MCP 的流式通信模式适合长时间运行的工具。5.1 为什么需要 SSE普通 HTTP请求 → 等待 → 一次性返回 问题工具执行要 30 秒 → HTTP 超时 SSE请求 → 持续推送进度 → 最终返回结果 优势不会超时用户能看到进度5.2 SSE MCP Server 实现 SSE 模式的 MCP Server 适合执行时间长的工具数据分析、批量处理 import json import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.get(/sse) async def sse_endpoint(): SSE 端点 async def event_stream(): # 1. 发送初始化信息 yield fdata: {json.dumps({type: endpoint, endpoint: /mcp})}\n\n # 保持连接 while True: await asyncio.sleep(30) yield f: keepalive\n\n return StreamingResponse( event_stream(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } ) app.post(/mcp) async def handle_mcp_sse(request): 处理 MCP 请求支持流式响应 body await request.json() method body.get(method) params body.get(params, {}) if method tools/call: tool_name params.get(name) if tool_name long_running_analysis: # 流式返回进度 async def stream_result(): for i in range(5): progress { type: progress, message: f分析中... {i1}/5, percentage: (i 1) * 20 } yield fdata: {json.dumps(progress)}\n\n await asyncio.sleep(2) final { type: result, content: [{type: text, text: 分析完成共发现 3 个异常模式}] } yield fdata: {json.dumps(final)}\n\n return StreamingResponse( stream_result(), media_typetext/event-stream )六、生产环境部署6.1 Docker 部署 MCP Server# Dockerfile FROM python:3.10-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 健康检查 HEALTHCHECK --interval30s --timeout5s \ CMD curl -f http://localhost:8080/health || exit 1 # 启动 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8080]# docker-compose.yml version: 3.8 services: mcp-server: build: . ports: - 8080:8080 environment: - DATABASE_URLpostgresql://user:passdb:5432/mydb - API_KEY${MCP_API_KEY} depends_on: - db restart: unless-stopped deploy: resources: limits: memory: 512M cpus: 0.5 db: image: postgres:15 environment: POSTGRES_DB: mydb POSTGRES_USER: user POSTGRES_PASSWORD: pass volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:6.2 认证中间件 MCP Server 认证中间件 生产环境必须加否则任何人都能调用你的工具 from fastapi import FastAPI, Request, HTTPException from fastapi.middleware import Middleware from fastapi.middleware.cors import CORSMiddleware import os import time import hashlib import hmac API_KEY os.environ.get(MCP_API_KEY, ) ALLOWED_ORIGINS os.environ.get(MCP_ALLOWED_ORIGINS, *).split(,) # API Key 认证 async def verify_api_key(request: Request, call_next): path request.url.path # 健康检查不需要认证 if path /health: return await call_next(request) # 检查 API Key auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detailMissing API Key) key auth_header.replace(Bearer , ) if key ! API_KEY: raise HTTPException(status_code403, detailInvalid API Key) return await call_next(request) # 速率限制简单版 request_counts {} async def rate_limit(request: Request, call_next): client_ip request.client.host now int(time.time() / 60) # 按分钟计数 key f{client_ip}:{now} request_counts[key] request_counts.get(key, 0) 1 if request_counts[key] 60: # 每分钟最多 60 次 raise HTTPException(status_code429, detailToo many requests) # 清理旧记录 old_keys [k for k in request_counts if int(k.split(:)[1]) now - 5] for k in old_keys: del request_counts[k] return await call_next(request) # 创建带中间件的 App app FastAPI( middleware[ Middleware(verify_api_key), Middleware(rate_limit), ] ) app.add_middleware( CORSMiddleware, allow_originsALLOWED_ORIGINS, allow_methods[POST, GET], allow_headers[*], )七、踩坑记录坑症状解决方法stdio 模式卡死MCP Server 不响应检查 stdout 是否被缓冲加sys.stdout.flush()HTTP 超时长时间工具执行失败用 SSE 模式或加大超时时间工具列表为空AI 应用看不到工具检查list_tools()返回格式是否正确编码问题中文返回乱码确保 UTF-8 编码JSON 加ensure_asciiFalse认证失败API Key 传了但还是 401检查 Bearer 前缀、环境变量是否生效并发冲突多用户同时调用同一工具加锁或用队列注意共享状态MCP Server 崩溃AI 应用报错连接断开加进程守护systemd / supervisor本地模式加自动重启JSON-RPC id 不一致请求和响应对不上严格用递增 id不要随机生成八、决策树什么时候用什么模式你要做什么 │ ├── 个人用本地 AI 工具Claude Desktop / OpenClaw │ └── 用 stdio 模式 │ ├── 现成的 MCP Server → npm install 配置 │ └── 自定义工具 → Python/TS 写 Server 配置 │ ├── Web 项目调用外部 MCP Server 的工具 │ └── 用 HTTP 模式Client │ ├── 工具执行快5s→ 普通 HTTP POST │ └── 工具执行慢5s→ SSE 流式 │ ├── 把 Web 项目的功能暴露给 AI 应用 │ └── 用 HTTP 模式Server │ ├── 内网使用 → 简单 API Key 认证 │ └── 公网使用 → API Key 速率限制 HTTPS │ └── 既要本地用又要 Web 用 └── 写一个 MCP Server同时支持 stdio 和 HTTP ├── stdio 入口if __name__ __main__: stdio_server() └── HTTP 入口uvicorn FastAPI 路由总结MCP 本质是 JSON-RPC 2.0 协议不管本地还是 Web通信内容一样本地用 stdioAI 应用自动拉起 MCP Server 进程零网络延迟Web 用 HTTP/SSEMCP Server 独立部署需要认证和监控90% 的场景用现成的 MCP Server 就够了自定义 Server 只在需要时写生产环境必须加认证API Key 是最低要求长耗时工具用 SSE避免 HTTP 超时MCP 的设计很克制只解决AI 怎么调用工具这一个问题。也正因为克制它才可能成为标准。
MCP实战指南：本地AI工具与Web项目的完整接入方案
发布时间：2026/5/25 8:50:19
MCP 实战指南从本地 AI 工具到 Web 项目的完整接入方案阅读时间约35分钟 |你将学到MCP 的真实调用原理、3种本地接入方式、2种 Web 接入架构、完整可运行代码核心观点MCP 不只是一个协议规范它是一套工具发现 工具调用 结果返回的完整通信机制。本地用 stdioWeb 用 HTTP/SSE原理相同传输方式不同。一、MCP 到底是什么3分钟真正搞懂1.1 一句话说清楚MCPModel Context Protocol AI 应用调用外部工具的标准协议。就像 USB 是硬件接口标准MCP 是 AI 调用工具的接口标准。没有 MCP 的世界 有 MCP 的世界 ┌──────────┐ ┌──────────┐ │ AI 应用 │ │ AI 应用 │ │ │ │ │ │ 调GitHub → 写一套适配代码 │ 统一调用 │────→ MCP Server AGitHub │ 调数据库 → 又写一套适配代码 │ MCP 协议 │────→ MCP Server B数据库 │ 调搜索引擎 → 再写一套适配代码 │ │────→ MCP Server C搜索 └──────────┘ └──────────┘ 每个工具一套接口维护成本爆炸 一套协议所有工具通用1.2 MCP 的三个核心能力能力说明类比ToolsAI 可以调用的函数你给AI的工具箱ResourcesAI 可以读取的数据你给AI的资料库Prompts预设的提示词模板你给AI的话术本实际项目中最常用的是Tools占 95% 的使用场景。1.3 通信协议JSON-RPC 2.0MCP 用 JSON-RPC 2.0 通信。就两种消息// 请求AI 应用 → MCP Server { jsonrpc: 2.0, id: 1, method: tools/call, params: { name: search_github, arguments: {query: mcp server} } } // 响应MCP Server → AI 应用 { jsonrpc: 2.0, id: 1, result: { content: [ {type: text, text: 找到 1234 个仓库...} ] } }就这么简单。请求带着方法名和参数响应带着结果。二、本地 AI 工具中使用 MCP2.1 本地 MCP 的通信方式stdio本地 AI 工具OpenClaw、Claude Desktop、Claude Code用的是stdio标准输入输出通信。┌─────────────────────┐ stdin/stdout ┌─────────────────────┐ │ AI 应用进程 │ ←──────────────────→ │ MCP Server 进程 │ │ (Claude Desktop) │ JSON-RPC 2.0 │ (子进程) │ └─────────────────────┘ └─────────────────────┘ │ │ │ 不需要网络 │ 调用实际资源 │ 同一台机器 │ ▼ ▼ 用户交互 文件系统/数据库/API关键点MCP Server 是 AI 应用拉起来的子进程通过管道pipe通信不需要网络。2.2 实战一Claude Desktop 接入 MCP Server第 1 步安装 MCP Server# 安装文件系统 MCP Server npm install -g modelcontextprotocol/server-filesystem # 安装 GitHub MCP Server npm install -g modelcontextprotocol/server-github # 安装 SQLite MCP Server npm install -g modelcontextprotocol/server-sqlite第 2 步配置 Claude Desktop编辑~/Library/Application Support/Claude/claude_desktop_config.jsonmacOS或%APPDATA%\Claude\claude_desktop_config.jsonWindows{ mcpServers: { filesystem: { command: npx, args: [ -y, modelcontextprotocol/server-filesystem, /Users/xiaoyuer/projects, /Users/xiaoyuer/Documents ] }, github: { command: npx, args: [-y, modelcontextprotocol/server-github], env: { GITHUB_PERSONAL_ACCESS_TOKEN: ghp_your_token_here } }, sqlite: { command: npx, args: [ -y, modelcontextprotocol/server-sqlite, /path/to/your/database.db ] } } }第 3 步重启 Claude Desktop开始使用重启后Claude 会自动拉起配置中的 MCP Server 进程。你可以在对话中直接说读一下 /Users/xiaoyuer/projects/main.py搜索 GitHub 上 star 最多的 MCP 项目查一下数据库里最近的订单Claude 会自动调用对应的 MCP Server。2.3 实战二OpenClaw 接入 MCP ServerOpenClaw 的 MCP 配置在~/.openclaw/config.yaml中# ~/.openclaw/config.yaml mcp: servers: # 文件系统 MCP Server filesystem: command: npx args: - -y - modelcontextprotocol/server-filesystem - /Users/xiaoyuer/projects # 自定义 MCP ServerPython 写的 my-tools: command: python args: - /Users/xiaoyuer/mcp-servers/my_tools_server.py # 远程 MCP ServerHTTP/SSE remote-search: url: https://mcp.example.com/sseOpenClaw 的特殊能力同时支持 stdio 和 HTTP/SSE 两种方式连接 MCP Server。2.4 实战三Claude Code 接入 MCP ServerClaude Code 通过命令行配置# 添加 MCP Server claude mcp add filesystem npx -y modelcontextprotocol/server-filesystem /Users/xiaoyuer/projects # 添加带环境变量的 MCP Server claude mcp add github -e GITHUB_PERSONAL_ACCESS_TOKENghp_xxx -- npx -y modelcontextprotocol/server-github # 查看已配置的 MCP Server claude mcp list # 删除 MCP Server claude mcp remove filesystem2.5 自己写一个 MCP ServerPython这是最重要的实战能力。项目里经常需要自定义工具。 自定义 MCP Server - 项目管理工具 功能查询项目状态、搜索代码、管理任务 import json import sys import asyncio from typing import Any # 安装依赖pip install mcp from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Tool, TextContent, ImageContent, EmbeddedResource, ) # 创建 Server 实例 app Server(project-manager) # 1. 注册工具列表 app.list_tools() async def list_tools() - list[Tool]: 告诉 AI 应用我提供哪些工具 return [ Tool( nameget_project_status, description获取项目状态包括进度、待办事项、最近变更, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 } }, required: [project_name] } ), Tool( namesearch_code, description在项目代码中搜索关键词, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 }, keyword: { type: string, description: 搜索关键词 }, file_type: { type: string, description: 文件类型过滤如 py/js/ts, default: None } }, required: [project_name, keyword] } ), Tool( namecreate_task, description创建项目管理任务, inputSchema{ type: object, properties: { project_name: { type: string, description: 项目名称 }, title: { type: string, description: 任务标题 }, priority: { type: string, enum: [high, medium, low], description: 优先级, default: medium }, assignee: { type: string, description: 负责人, default: None } }, required: [project_name, title] } ), ] # 2. 实现工具逻辑 app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) - list[TextContent | ImageContent | EmbeddedResource]: 执行工具调用 if name get_project_status: return await handle_get_project_status(arguments) elif name search_code: return await handle_search_code(arguments) elif name create_task: return await handle_create_task(arguments) else: return [TextContent(typetext, textf未知工具{name})] async def handle_get_project_status(args: dict) - list[TextContent]: 获取项目状态 project_name args[project_name] # 这里替换成你的实际逻辑 # 比如查数据库、调API、读文件等 status { project: project_name, progress: 68%, open_issues: 12, recent_commits: [ {hash: a1b2c3d, message: feat: 添加用户认证, author: zhangsan}, {hash: e4f5g6h, message: fix: 修复分页bug, author: lisi}, ], pending_tasks: [ {title: 接口压测, priority: high}, {title: 文档更新, priority: medium}, ] } return [TextContent( typetext, textjson.dumps(status, ensure_asciiFalse, indent2) )] async def handle_search_code(args: dict) - list[TextContent]: 搜索代码 project_name args[project_name] keyword args[keyword] file_type args.get(file_type) # 这里替换成你的实际逻辑 # 比如用 grep/ripgrep 搜索、调 GitHub API 等 results [ { file: fsrc/api/{project_name}/routes.py, line: 42, content: fdef search_{keyword}(query: str): }, { file: fsrc/services/{project_name}/service.py, line: 88, content: fclass {keyword.capitalize()}Service: }, ] return [TextContent( typetext, textjson.dumps(results, ensure_asciiFalse, indent2) )] async def handle_create_task(args: dict) - list[TextContent]: 创建任务 project_name args[project_name] title args[title] priority args.get(priority, medium) assignee args.get(assignee) # 这里替换成你的实际逻辑 # 比如调 Jira API、写数据库等 task { id: TASK-2026-001, project: project_name, title: title, priority: priority, assignee: assignee or 未分配, status: created } return [TextContent( typetext, textjson.dumps(task, ensure_asciiFalse, indent2) )] # 3. 启动 Server async def main(): 启动 MCP Serverstdio 模式 async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ __main__: asyncio.run(main())配置到 Claude Desktop{ mcpServers: { project-manager: { command: python, args: [/Users/xiaoyuer/mcp-servers/my_tools_server.py] } } }重启 Claude Desktop 后你就能直接说查一下 smart-trip-planner 项目状态或在项目里搜一下 auth 相关代码Claude 会自动调用你的 MCP Server。三、Web 项目中使用 MCP3.1 Web 项目和本地 AI 工具的核心区别本地 AI 工具Claude Desktop / OpenClaw ┌─────────────────┐ stdio管道 ┌─────────────────┐ │ AI 应用进程 │ ←────────────→ │ MCP Server 进程 │ │ 同一台机器 │ │ 同一台机器 │ └─────────────────┘ └─────────────────┘ 特点不需要网络进程间直接通信 Web 项目 ┌─────────────────┐ HTTP/SSE网络 ┌─────────────────┐ │ Web 后端 │ ←───────────────→ │ MCP Server 服务 │ │ 任意机器 │ │ 可能是另一台机器 │ └─────────────────┘ └─────────────────┘ 特点需要网络跨机器通信需要考虑认证、超时、负载均衡核心区别维度本地 AI 工具Web 项目通信方式stdio管道HTTP/SSE网络MCP Server 生命周期随 AI 应用启动/关闭独立部署长期运行认证不需要需要考虑API Key / OAuth并发单用户多用户并发部署本地进程云服务器 / Docker延迟极低进程间较高网络传输3.2 Web 项目的两种 MCP 架构架构一Web 后端 MCP Client调用外部 MCP Server┌──────────────┐ ┌──────────────────┐ HTTP/SSE ┌──────────────────┐ │ 前端 │ ──→ │ Web 后端 │ ──────────────→ │ MCP Server │ │ (React/Vue) │ │ (FastAPI/Next) │ ←────────────── │ (远程服务) │ └──────────────┘ │ │ │ │ │ MCP Client 逻辑 │ │ 工具实现 │ └──────────────────┘ └──────────────────┘适用场景你的 Web 项目需要调用现成的 MCP Server 提供的工具。架构二Web 后端 MCP Server给 AI 应用提供工具┌──────────────────┐ HTTP/SSE ┌──────────────────┐ ┌──────────────┐ │ AI 应用 │ ──────────────→ │ Web 后端 │ ──→ │ 数据库/API │ │ (Claude/OpenClaw)│ ←────────────── │ (FastAPI/Next) │ ←── │ │ │ │ │ MCP Server 逻辑 │ └──────────────┘ └──────────────────┘ └──────────────────┘适用场景你想让 AI 应用直接调用你 Web 项目的功能。3.3 实战Web 后端作为 MCP Client Web 后端 - MCP Client FastAPI 项目调用远程 MCP Server 的工具 import httpx import json from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(titleMCP Client Demo) class MCPClient: MCP 客户端HTTP 模式 def __init__(self, server_url: str, api_key: Optional[str] None): self.server_url server_url.rstrip(/) self.headers {} if api_key: self.headers[Authorization] fBearer {api_key} self._request_id 0 self._client None async def _get_client(self) - httpx.AsyncClient: if self._client is None or self._client.is_closed: self._client httpx.AsyncClient( timeout30.0, headersself.headers, ) return self._client async def _send_request(self, method: str, params: dict None) - dict: 发送 JSON-RPC 请求 self._request_id 1 payload { jsonrpc: 2.0, id: self._request_id, method: method, } if params: payload[params] params client await self._get_client() response await client.post( f{self.server_url}/mcp, jsonpayload, ) response.raise_for_status() data response.json() if error in data: raise Exception(fMCP Error: {data[error]}) return data.get(result, {}) async def initialize(self) - dict: 初始化连接 return await self._send_request(initialize, { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: { name: web-backend, version: 1.0.0 } }) async def list_tools(self) - list: 获取工具列表 result await self._send_request(tools/list) return result.get(tools, []) async def call_tool(self, name: str, arguments: dict) - dict: 调用工具 return await self._send_request(tools/call, { name: name, arguments: arguments }) async def close(self): if self._client and not self._client.is_closed: await self._client.aclose() # 全局 MCP Client 实例 mcp_client MCPClient( server_urlhttps://mcp.example.com, api_keyyour-api-key # 生产环境从环境变量读取 ) app.on_event(startup) async def startup(): 启动时初始化 MCP 连接 await mcp_client.initialize() tools await mcp_client.list_tools() print(f已连接 MCP Server可用工具{[t[name] for t in tools]}) app.on_event(shutdown) async def shutdown(): 关闭时断开连接 await mcp_client.close() # 业务接口 class SearchRequest(BaseModel): query: str project: Optional[str] None class TaskRequest(BaseModel): project_name: str title: str priority: str medium app.post(/api/search) async def search(req: SearchRequest): 搜索接口 - 调用 MCP Server 的搜索工具 result await mcp_client.call_tool(search_code, { project_name: req.project or default, keyword: req.query, }) return {data: result} app.post(/api/tasks) async def create_task(req: TaskRequest): 创建任务接口 - 调用 MCP Server 的任务工具 result await mcp_client.call_tool(create_task, { project_name: req.project_name, title: req.title, priority: req.priority, }) return {data: result} app.get(/api/tools) async def get_available_tools(): 查看当前可用的 MCP 工具 tools await mcp_client.list_tools() return {tools: tools} # 运行uvicorn main:app --host 0.0.0.0 --port 80003.4 实战Web 后端作为 MCP Server Web 后端 - MCP Server 把 Web 项目的功能暴露为 MCP 工具给 AI 应用调用 import json from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from typing import Any app FastAPI(titleMCP Server Demo) # 你的业务逻辑 class BusinessLogic: 业务逻辑层和 MCP 无关 staticmethod async def query_orders(user_id: str, status: str None) - list: 查询订单 # 实际项目中查数据库 return [ {id: ORD-001, user: user_id, amount: 299.0, status: paid}, {id: ORD-002, user: user_id, amount: 158.0, status: pending}, ] staticmethod async def get_user_info(user_id: str) - dict: 查询用户信息 # 实际项目中查数据库 return { id: user_id, name: 张三, level: VIP, orders_count: 42, } staticmethod async def cancel_order(order_id: str, reason: str) - dict: 取消订单 # 实际项目中写数据库 return { order_id: order_id, status: cancelled, reason: reason, refund_amount: 299.0, } biz BusinessLogic() # MCP 协议层 def mcp_response(req_id: int, result: Any) - dict: return {jsonrpc: 2.0, id: req_id, result: result} def mcp_error(req_id: int, code: int, message: str) - dict: return {jsonrpc: 2.0, id: req_id, error: {code: code, message: message}} # 工具注册表 TOOLS [ { name: query_orders, description: 查询用户订单, inputSchema: { type: object, properties: { user_id: {type: string, description: 用户ID}, status: {type: string, description: 订单状态过滤, default: None} }, required: [user_id] } }, { name: get_user_info, description: 查询用户信息, inputSchema: { type: object, properties: { user_id: {type: string, description: 用户ID} }, required: [user_id] } }, { name: cancel_order, description: 取消订单危险操作需要确认, inputSchema: { type: object, properties: { order_id: {type: string, description: 订单ID}, reason: {type: string, description: 取消原因} }, required: [order_id, reason] } } ] # 工具处理函数映射 TOOL_HANDLERS { query_orders: lambda args: biz.query_orders(args[user_id], args.get(status)), get_user_info: lambda args: biz.get_user_info(args[user_id]), cancel_order: lambda args: biz.cancel_order(args[order_id], args[reason]), } app.post(/mcp) async def handle_mcp(request: Request): 处理 MCP 请求 body await request.json() method body.get(method) params body.get(params, {}) req_id body.get(id) # 1. 初始化握手 if method initialize: return mcp_response(req_id, { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: { name: order-management-mcp, version: 1.0.0 } }) # 2. 初始化确认 if method notifications/initialized: return JSONResponse(content{}) # 3. 工具列表 if method tools/list: return mcp_response(req_id, {tools: TOOLS}) # 4. 工具调用 if method tools/call: tool_name params.get(name) arguments params.get(arguments, {}) handler TOOL_HANDLERS.get(tool_name) if not handler: return mcp_error(req_id, -32601, f未知工具{tool_name}) try: result await handler(arguments) return mcp_response(req_id, { content: [ {type: text, text: json.dumps(result, ensure_asciiFalse)} ] }) except Exception as e: return mcp_error(req_id, -32603, f工具执行失败{str(e)}) return mcp_error(req_id, -32601, f未知方法{method}) # 健康检查 app.get(/health) async def health(): return {status: ok, tools_count: len(TOOLS)} # 运行uvicorn main:app --host 0.0.0.0 --port 8080Claude Desktop 配置远程 MCP Server{ mcpServers: { order-management: { url: https://your-api.com/mcp } } }这样 Claude Desktop 就能直接查询订单、取消订单了。四、原理对比本地 vs Web4.1 通信方式对比本地模式stdio AI 应用进程 MCP Server 进程 ┌──────────┐ stdin写 ┌──────────┐ │ │ ──────────────→ │ │ │ stdout │ ←────────────── │ stdout │ │ 读 │ stdout写 │ 输出 │ └──────────┘ └──────────┘ 数据流AI 应用 → pipe → MCP Server → pipe → AI 应用 特点操作系统级管道零网络延迟进程同生命周期 Web 模式HTTP/SSE Web 后端 MCP Server ┌──────────┐ HTTP POST ┌──────────┐ │ │ ──────────────→ │ │ │ │ ←────────────── │ │ │ │ HTTP Response │ │ └──────────┘ └──────────┘ 数据流Web 后端 → 网络请求 → MCP Server → 网络响应 → Web 后端 特点需要网络独立部署需要认证可跨机器4.2 生命周期对比 本地模式的 MCP Server 生命周期 # 1. AI 应用启动时自动拉起 MCP Server import subprocess mcp_process subprocess.Popen( [npx, -y, modelcontextprotocol/server-filesystem, /tmp], stdinsubprocess.PIPE, # AI 应用写请求 stdoutsubprocess.PIPE, # AI 应用读响应 stderrsubprocess.PIPE, ) # 2. 通过管道通信 request json.dumps({jsonrpc: 2.0, id: 1, method: tools/list, params: {}}) mcp_process.stdin.write(f{request}\n.encode()) mcp_process.stdin.flush() response mcp_process.stdout.readline().decode() print(json.loads(response)) # 3. AI 应用关闭时MCP Server 也关闭 mcp_process.terminate() Web 模式的 MCP Server 生命周期 # 1. MCP Server 独立部署长期运行 # uvicorn mcp_server:app --host 0.0.0.0 --port 8080 # 2. Web 后端按需连接 import httpx async with httpx.AsyncClient() as client: response await client.post(http://mcp-server:8080/mcp, json{ jsonrpc: 2.0, id: 1, method: tools/list, params: {} }) print(response.json()) # 3. Web 后端关闭不影响 MCP Server # MCP Server 独立运行可以被多个 Web 后端调用4.3 完整对比表维度本地模式stdioWeb 模式HTTP/SSE通信操作系统管道HTTP 网络请求延迟微秒级毫秒~秒级认证不需要API Key / OAuth并发单用户多用户部署随 AI 应用启停独立部署长期运行发现配置文件声明URL 自动发现适用个人工具团队/生产环境错误处理进程崩溃 通信断需要重试/超时/熔断监控日志文件APM 健康检查多实例不支持负载均衡五、SSE 模式深入SSEServer-Sent Events是 MCP 的流式通信模式适合长时间运行的工具。5.1 为什么需要 SSE普通 HTTP请求 → 等待 → 一次性返回 问题工具执行要 30 秒 → HTTP 超时 SSE请求 → 持续推送进度 → 最终返回结果 优势不会超时用户能看到进度5.2 SSE MCP Server 实现 SSE 模式的 MCP Server 适合执行时间长的工具数据分析、批量处理 import json import asyncio from fastapi import FastAPI from fastapi.responses import StreamingResponse app FastAPI() app.get(/sse) async def sse_endpoint(): SSE 端点 async def event_stream(): # 1. 发送初始化信息 yield fdata: {json.dumps({type: endpoint, endpoint: /mcp})}\n\n # 保持连接 while True: await asyncio.sleep(30) yield f: keepalive\n\n return StreamingResponse( event_stream(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } ) app.post(/mcp) async def handle_mcp_sse(request): 处理 MCP 请求支持流式响应 body await request.json() method body.get(method) params body.get(params, {}) if method tools/call: tool_name params.get(name) if tool_name long_running_analysis: # 流式返回进度 async def stream_result(): for i in range(5): progress { type: progress, message: f分析中... {i1}/5, percentage: (i 1) * 20 } yield fdata: {json.dumps(progress)}\n\n await asyncio.sleep(2) final { type: result, content: [{type: text, text: 分析完成共发现 3 个异常模式}] } yield fdata: {json.dumps(final)}\n\n return StreamingResponse( stream_result(), media_typetext/event-stream )六、生产环境部署6.1 Docker 部署 MCP Server# Dockerfile FROM python:3.10-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 健康检查 HEALTHCHECK --interval30s --timeout5s \ CMD curl -f http://localhost:8080/health || exit 1 # 启动 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8080]# docker-compose.yml version: 3.8 services: mcp-server: build: . ports: - 8080:8080 environment: - DATABASE_URLpostgresql://user:passdb:5432/mydb - API_KEY${MCP_API_KEY} depends_on: - db restart: unless-stopped deploy: resources: limits: memory: 512M cpus: 0.5 db: image: postgres:15 environment: POSTGRES_DB: mydb POSTGRES_USER: user POSTGRES_PASSWORD: pass volumes: - pgdata:/var/lib/postgresql/data volumes: pgdata:6.2 认证中间件 MCP Server 认证中间件 生产环境必须加否则任何人都能调用你的工具 from fastapi import FastAPI, Request, HTTPException from fastapi.middleware import Middleware from fastapi.middleware.cors import CORSMiddleware import os import time import hashlib import hmac API_KEY os.environ.get(MCP_API_KEY, ) ALLOWED_ORIGINS os.environ.get(MCP_ALLOWED_ORIGINS, *).split(,) # API Key 认证 async def verify_api_key(request: Request, call_next): path request.url.path # 健康检查不需要认证 if path /health: return await call_next(request) # 检查 API Key auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detailMissing API Key) key auth_header.replace(Bearer , ) if key ! API_KEY: raise HTTPException(status_code403, detailInvalid API Key) return await call_next(request) # 速率限制简单版 request_counts {} async def rate_limit(request: Request, call_next): client_ip request.client.host now int(time.time() / 60) # 按分钟计数 key f{client_ip}:{now} request_counts[key] request_counts.get(key, 0) 1 if request_counts[key] 60: # 每分钟最多 60 次 raise HTTPException(status_code429, detailToo many requests) # 清理旧记录 old_keys [k for k in request_counts if int(k.split(:)[1]) now - 5] for k in old_keys: del request_counts[k] return await call_next(request) # 创建带中间件的 App app FastAPI( middleware[ Middleware(verify_api_key), Middleware(rate_limit), ] ) app.add_middleware( CORSMiddleware, allow_originsALLOWED_ORIGINS, allow_methods[POST, GET], allow_headers[*], )七、踩坑记录坑症状解决方法stdio 模式卡死MCP Server 不响应检查 stdout 是否被缓冲加sys.stdout.flush()HTTP 超时长时间工具执行失败用 SSE 模式或加大超时时间工具列表为空AI 应用看不到工具检查list_tools()返回格式是否正确编码问题中文返回乱码确保 UTF-8 编码JSON 加ensure_asciiFalse认证失败API Key 传了但还是 401检查 Bearer 前缀、环境变量是否生效并发冲突多用户同时调用同一工具加锁或用队列注意共享状态MCP Server 崩溃AI 应用报错连接断开加进程守护systemd / supervisor本地模式加自动重启JSON-RPC id 不一致请求和响应对不上严格用递增 id不要随机生成八、决策树什么时候用什么模式你要做什么 │ ├── 个人用本地 AI 工具Claude Desktop / OpenClaw │ └── 用 stdio 模式 │ ├── 现成的 MCP Server → npm install 配置 │ └── 自定义工具 → Python/TS 写 Server 配置 │ ├── Web 项目调用外部 MCP Server 的工具 │ └── 用 HTTP 模式Client │ ├── 工具执行快5s→ 普通 HTTP POST │ └── 工具执行慢5s→ SSE 流式 │ ├── 把 Web 项目的功能暴露给 AI 应用 │ └── 用 HTTP 模式Server │ ├── 内网使用 → 简单 API Key 认证 │ └── 公网使用 → API Key 速率限制 HTTPS │ └── 既要本地用又要 Web 用 └── 写一个 MCP Server同时支持 stdio 和 HTTP ├── stdio 入口if __name__ __main__: stdio_server() └── HTTP 入口uvicorn FastAPI 路由总结MCP 本质是 JSON-RPC 2.0 协议不管本地还是 Web通信内容一样本地用 stdioAI 应用自动拉起 MCP Server 进程零网络延迟Web 用 HTTP/SSEMCP Server 独立部署需要认证和监控90% 的场景用现成的 MCP Server 就够了自定义 Server 只在需要时写生产环境必须加认证API Key 是最低要求长耗时工具用 SSE避免 HTTP 超时MCP 的设计很克制只解决AI 怎么调用工具这一个问题。也正因为克制它才可能成为标准。
