前九讲我们完成了 MCP Server 的开发、集成和编排。但所有这些都在本地开发环境中运行。这一讲我们要把 MCP Server部署到生产环境——让它稳定、安全、可监控地对外提供服务。一、生产环境的挑战维度开发环境生产环境传输方式stdio本地进程SSE网络服务认证无API Key / OAuth并发单用户多用户并发可靠性挂了重启高可用、自动恢复监控print 调试日志、指标、告警部署命令行启动Docker / K8s二、从 stdio 迁移到 SSEstdio 模式只适合本地开发。生产环境需要 SSEServer-Sent Events模式让 MCP Server 作为一个独立的 HTTP 服务运行。2.1 SSE Server 实现# sse_mcp_server.py import asyncio import json import uuid from datetime import datetime from mcp.server import Server from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route, Mount from starlette.middleware import Middleware from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import JSONResponse import uvicorn # 导入之前的 Server 逻辑 from db_mcp_server import server as db_server class AuthMiddleware(BaseHTTPMiddleware): 认证中间件 def __init__(self, app, api_keys: list[str]): super().__init__(app) self.api_keys api_keys async def dispatch(self, request, call_next): # SSE 连接端点不需要认证 if request.url.path /sse: return await call_next(request) # 其他端点需要 API Key api_key request.headers.get(X-API-Key, ) if api_key not in self.api_keys: return JSONResponse( {error: Unauthorized}, status_code401 ) return await call_next(request) class RateLimitMiddleware(BaseHTTPMiddleware): 限流中间件 def __init__(self, app, max_requests: int 100, window_seconds: int 60): super().__init__(app) self.max_requests max_requests self.window_seconds window_seconds self.requests {} async def dispatch(self, request, call_next): client_ip request.client.host now datetime.now().timestamp() # 清理过期记录 if client_ip in self.requests: self.requests[client_ip] [ t for t in self.requests[client_ip] if now - t self.window_seconds ] # 检查限流 if client_ip in self.requests and len(self.requests[client_ip]) self.max_requests: return JSONResponse( {error: Rate limit exceeded}, status_code429, headers{Retry-After: str(self.window_seconds)} ) # 记录请求 if client_ip not in self.requests: self.requests[client_ip] [] self.requests[client_ip].append(now) return await call_next(request) # 创建 SSE 传输 sse_transport SseServerTransport(/messages) # 创建 Starlette 应用 async def handle_sse(request): async with sse_transport.connect_sse( request.scope, request.receive, request._send ) as sessions: await db_server.run( sessions, db_server.create_initialization_options() ) async def handle_messages(request): await sse_transport.handle_post_message( request.scope, request.receive, request._send ) # 健康检查端点 async def health_check(request): return JSONResponse({ status: healthy, timestamp: datetime.now().isoformat(), version: 1.0.0, uptime: ... # 可以从启动时间计算 }) # 指标端点 async def metrics(request): return JSONResponse({ connections: len(sse_transport._sessions) if hasattr(sse_transport, _sessions) else 0, tools_count: len(db_server._tools) if hasattr(db_server, _tools) else 0 }) # 路由配置 routes [ Route(/sse, endpointhandle_sse), Route(/messages, endpointhandle_messages, methods[POST]), Route(/health, endpointhealth_check), Route(/metrics, endpointmetrics), ] # 中间件配置 middleware [ Middleware(AuthMiddleware, api_keys[sk-prod-xxx, sk-staging-xxx]), Middleware(RateLimitMiddleware, max_requests100, window_seconds60), ] app Starlette( routesroutes, middlewaremiddleware, on_startup[lambda: print( MCP Server 已启动)], on_shutdown[lambda: print( MCP Server 已关闭)] ) if __name__ __main__: uvicorn.run( app, host0.0.0.0, port8000, workers4, # 多 worker 提高并发 log_levelinfo )2.2 SSE Client 连接# sse_client.py import asyncio from mcp import ClientSession from mcp.client.sse import sse_client async def connect_to_production(): 连接到生产环境的 MCP Server server_url http://localhost:8000/sse headers {X-API-Key: sk-prod-xxx} async with sse_client(server_url, headersheaders) as streams: async with ClientSession(streams[0], streams[1]) as session: await session.initialize() # 获取工具列表 tools await session.list_tools() print(f已连接发现 {len(tools.tools)} 个工具) # 调用工具 result await session.call_tool(query_database, { sql: SELECT * FROM employees LIMIT 5 }) print(result.content[0].text) if __name__ __main__: asyncio.run(connect_to_production())三、Docker 容器化部署3.1 Dockerfile# Dockerfile FROM python:3.11-slim LABEL maintaineryour-teamcompany.com LABEL descriptionMCP Database Server - Production # 设置环境变量 ENV PYTHONUNBUFFERED1 \ PYTHONDONTWRITEBYTECODE1 \ PIP_NO_CACHE_DIR1 # 安装系统依赖 RUN apt-get update apt-get install -y --no-install-recommends \ gcc \ rm -rf /var/lib/apt/lists/* # 创建工作目录 WORKDIR /app # 复制依赖文件 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非 root 用户 RUN useradd -m -u 1000 mcp chown -R mcp:mcp /app USER mcp # 暴露端口 EXPOSE 8000 # 健康检查 HEALTHCHECK --interval30s --timeout10s --start-period5s --retries3 \ CMD curl -f http://localhost:8000/health || exit 1 # 启动命令 CMD [uvicorn, sse_mcp_server:app, --host, 0.0.0.0, --port, 8000, --workers, 4]3.2 docker-compose.yml# docker-compose.yml version: 3.8 services: mcp-db-server: build: context: . dockerfile: Dockerfile container_name: mcp-db-server ports: - 8000:8000 environment: - DB_TYPE${DB_TYPE:-sqlite} - DB_DATABASE/data/database.db - API_KEYS${API_KEYS:-sk-default} - LOG_LEVELINFO - MAX_REQUESTS_PER_MINUTE100 volumes: - ./data:/data - ./logs:/app/logs restart: unless-stopped networks: - mcp-network deploy: resources: limits: cpus: 1 memory: 512M reservations: cpus: 0.5 memory: 256M mcp-fs-server: build: context: . dockerfile: Dockerfile.fs container_name: mcp-fs-server ports: - 8001:8000 environment: - FS_ROOT/data/sandbox - API_KEYS${API_KEYS:-sk-default} volumes: - ./sandbox:/data/sandbox restart: unless-stopped networks: - mcp-network nginx: image: nginx:alpine container_name: mcp-gateway ports: - 443:443 volumes: - ./nginx.conf:/etc/nginx/nginx.conf - ./ssl:/etc/nginx/ssl depends_on: - mcp-db-server - mcp-fs-server networks: - mcp-network networks: mcp-network: driver: bridge3.3 Nginx 反向代理配置# nginx.conf events { worker_connections 1024; } http { upstream mcp_db_server { server mcp-db-server:8000; } upstream mcp_fs_server { server mcp-fs-server:8000; } server { listen 443 ssl; server_name mcp.example.com; ssl_certificate /etc/nginx/ssl/cert.pem; ssl_certificate_key /etc/nginx/ssl/key.pem; # 数据库 Server location /db/ { proxy_pass http://mcp_db_server/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_buffering off; proxy_cache off; # SSE 支持 proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; } # 文件系统 Server location /fs/ { proxy_pass http://mcp_fs_server/; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_buffering off; proxy_cache off; proxy_http_version 1.1; chunked_transfer_encoding off; } # 健康检查 location /health { proxy_pass http://mcp_db_server/health; } } }四、日志与监控4.1 结构化日志# logging_config.py import json import logging import sys from datetime import datetime class StructuredFormatter(logging.Formatter): 结构化日志格式化器 def format(self, record): log_entry { timestamp: datetime.utcnow().isoformat() Z, level: record.levelname, logger: record.name, message: record.getMessage(), } # 添加额外字段 if hasattr(record, session_id): log_entry[session_id] record.session_id if hasattr(record, tool_name): log_entry[tool] record.tool_name if hasattr(record, duration_ms): log_entry[duration_ms] record.duration_ms # 添加异常信息 if record.exc_info: log_entry[exception] self.formatException(record.exc_info) return json.dumps(log_entry, ensure_asciiFalse) def setup_logging(): 配置日志 handler logging.StreamHandler(sys.stdout) handler.setFormatter(StructuredFormatter()) root_logger logging.getLogger() root_logger.addHandler(handler) root_logger.setLevel(logging.INFO) # 文件日志 file_handler logging.FileHandler(/app/logs/mcp_server.log) file_handler.setFormatter(StructuredFormatter()) root_logger.addHandler(file_handler) return root_logger4.2 Prometheus 指标# metrics.py from prometheus_client import Counter, Histogram, Gauge, generate_latest from starlette.responses import Response import time # 定义指标 TOOL_CALLS_TOTAL Counter( mcp_tool_calls_total, Total number of tool calls, [server, tool, status] ) TOOL_CALL_DURATION Histogram( mcp_tool_call_duration_seconds, Duration of tool calls, [server, tool], buckets(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0) ) ACTIVE_CONNECTIONS Gauge( mcp_active_connections, Number of active MCP connections ) TOOL_ERRORS_TOTAL Counter( mcp_tool_errors_total, Total number of tool errors, [server, tool, error_type] ) class MetricsMiddleware: 指标采集中间件 async def __call__(self, request, call_next): start time.time() response await call_next(request) duration time.time() - start # 记录指标 if request.url.path /sse: ACTIVE_CONNECTIONS.inc() return response # 指标端点 async def metrics_endpoint(request): return Response( contentgenerate_latest(), media_typetext/plain )4.3 告警配置# alerts.yml (Prometheus AlertManager 配置) groups: - name: mcp_alerts rules: - alert: HighErrorRate expr: rate(mcp_tool_errors_total[5m]) 0.1 for: 5m labels: severity: critical annotations: summary: MCP Server 错误率过高 description: 过去5分钟错误率超过10% - alert: HighLatency expr: histogram_quantile(0.95, rate(mcp_tool_call_duration_seconds_bucket[5m])) 5 for: 5m labels: severity: warning annotations: summary: MCP Server 响应缓慢 description: P95 延迟超过5秒 - alert: LowConnections expr: mcp_active_connections 1 for: 1m labels: severity: critical annotations: summary: MCP Server 无活跃连接 description: 没有客户端连接到 MCP Server五、生产部署 Checklist5.1 安全加固# security.py import os from typing import Optional class SecurityConfig: 安全配置 # API Key 管理 API_KEYS os.environ.get(API_KEYS, ).split(,) # TLS/SSL SSL_CERT os.environ.get(SSL_CERT, ) SSL_KEY os.environ.get(SSL_KEY, ) # IP 白名单 ALLOWED_IPS os.environ.get(ALLOWED_IPS, 0.0.0.0/0).split(,) # 请求大小限制 MAX_REQUEST_SIZE 10 * 1024 * 1024 # 10MB # 超时设置 TOOL_TIMEOUT 30 # 秒 CONNECTION_TIMEOUT 60 # 秒 def validate_api_key(api_key: str) - bool: 验证 API Key return api_key in SecurityConfig.API_KEYS def validate_ip(client_ip: str) - bool: 验证客户端 IP import ipaddress for allowed in SecurityConfig.ALLOWED_IPS: if client_ip in ipaddress.ip_network(allowed): return True return False5.2 部署前检查清单## 部署前检查清单 ### 安全 - [ ] API Key 已设置且强度足够 - [ ] TLS/SSL 证书已配置 - [ ] IP 白名单已配置如需要 - [ ] 数据库使用只读账号 - [ ] 文件系统沙箱目录权限正确 ### 可靠性 - [ ] 健康检查端点正常工作 - [ ] 超时设置合理 - [ ] 重试机制已实现 - [ ] 资源限制已配置CPU/内存 ### 可观测性 - [ ] 结构化日志已配置 - [ ] Prometheus 指标已暴露 - [ ] 告警规则已配置 - [ ] 日志持久化存储已配置 ### 部署 - [ ] Docker 镜像构建成功 - [ ] docker-compose 配置正确 - [ ] Nginx 反向代理配置正确 - [ ] 端口映射正确 - [ ] 数据卷挂载正确 ### 测试 - [ ] SSE 连接测试通过 - [ ] 工具调用测试通过 - [ ] 认证测试通过 - [ ] 限流测试通过 - [ ] 压力测试通过六、生产环境运维6.1 启动脚本#!/bin/bash # deploy.sh - 生产部署脚本 set -e echo 开始部署 MCP Server... # 1. 构建 Docker 镜像 echo 构建 Docker 镜像... docker build -t mcp-server:latest . # 2. 备份旧数据 echo 备份数据... BACKUP_DIR./backups/$(date %Y%m%d_%H%M%S) mkdir -p $BACKUP_DIR cp -r ./data/* $BACKUP_DIR/ 2/dev/null || true # 3. 启动服务 echo ▶️ 启动服务... docker-compose up -d # 4. 等待服务就绪 echo ⏳ 等待服务就绪... for i in {1..30}; do if curl -sf http://localhost:8000/health /dev/null 21; then echo ✅ 服务已就绪 break fi sleep 1 done # 5. 运行健康检查 echo 运行健康检查... python scripts/health_check.py echo ✅ 部署完成6.2 滚动更新# docker-compose.prod.yml version: 3.8 services: mcp-server: image: registry.example.com/mcp-server:${VERSION} deploy: replicas: 3 update_config: parallelism: 1 delay: 10s order: start-first restart_policy: condition: any delay: 5s max_attempts: 3七、性能优化7.1 连接池优化# connection_pool_prod.py from concurrent.futures import ThreadPoolExecutor import asyncio class ProductionConnectionPool: 生产环境连接池 def __init__(self, min_size5, max_size50): self.min_size min_size self.max_size max_size self._pool ThreadPoolExecutor(max_workersmax_size) self._connections [] async def get_connection(self): 获取连接异步 loop asyncio.get_event_loop() return await loop.run_in_executor( self._pool, self._create_connection ) def _create_connection(self): 创建数据库连接同步 import sqlite3 conn sqlite3.connect(production.db) conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA synchronousNORMAL) conn.execute(PRAGMA cache_size-64000) # 64MB 缓存 return conn7.2 查询缓存# cache.py import hashlib import json from datetime import datetime, timedelta import redis class QueryCache: 查询结果缓存 def __init__(self, redis_urlredis://localhost:6379/0): self.redis redis.from_url(redis_url) self.ttl 300 # 5 分钟 def _make_key(self, sql: str) - str: 生成缓存键 return fquery:{hashlib.md5(sql.encode()).hexdigest()} def get(self, sql: str): 获取缓存 key self._make_key(sql) data self.redis.get(key) if data: return json.loads(data) return None def set(self, sql: str, result: str): 设置缓存 key self._make_key(sql) self.redis.setex(key, self.ttl, result) def invalidate(self, pattern: str *): 清除缓存 for key in self.redis.scan_iter(fquery:{pattern}): self.redis.delete(key)八、课后作业实现灰度发布让新版 MCP Server 只对部分用户生效逐步放量。添加请求追踪为每个请求生成 Trace ID贯穿所有日志和指标方便排查问题。挑战题实现一个 MCP Server 的“控制面板”——Web 页面展示实时连接数、QPS、错误率、延迟分布并支持动态调整限流阈值。九、专栏总结10 讲回顾讲次核心内容产出第1讲MCP 协议基础理解 MCP 的价值第2讲协议核心原理Transport/Tool/Resource第3讲手写 MCP Server第一个可运行的 Server第4讲MCP Client 开发连接、发现、调用第5讲数据库 MCP Server安全查询数据库第6讲文件系统 MCP Server安全读写文件第7讲API 网关 MCP Server调用外部 API第8讲MCP Agent 集成Agent 自主使用工具第9讲多 Server 编排协同完成复杂任务第10讲生产化部署安全、稳定、可监控你现在掌握的技能✅ 理解 MCP 协议的设计哲学和工作原理✅ 能手写 MCP Server 暴露任意工具✅ 能开发 MCP Client 连接和调用工具✅ 能对接数据库、文件系统、外部 API✅ 能将 MCP 集成到 Agent 中✅ 能编排多个 MCP Server 协同工作✅ 能安全地将 MCP Server 部署到生产环境下一步可以做什么贡献社区把你写的 MCP Server 开源到 GitHub深入框架学习 LangChain MCP、AutoGen MCP 等集成方案垂直场景针对客服、运维、数据分析等场景定制 MCP Server协议演进关注 MCP 协议的更新和新特性 开发之余处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top子页 PDF 大师PDF 大师 - zz365工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。