MCP 协议开发实战

从零构建你的第一个 MCP Server，让 AI 成为万能工具人。

10. 生产部署与安全

前 9 章，我们一直在本地跑 STDIO 模式——Claude Desktop 通过子进程启动 Server，一切都在同一台机器上。这在开发和个人使用时很方便，但到了团队协作和生产环境，你需要：

让 Server 跑在远程服务器上，多人共享
用 Docker 容器化，一键部署
加上认证授权，防止未授权访问
完善日志监控，出了问题能快速定位

这一章，我们从"能跑"走向"能用在生产"。

10.1 从 STDIO 到 HTTP+SSE：远程部署

为什么需要远程部署

STDIO 模式的限制：

STDIO 模式（我们之前一直用的）：

  Claude Desktop ──(子进程)──▶ MCP Server
        │                         │
        └── 同一台机器 ────────────┘

  ✅ 优点：零配置、启动快、适合开发
  ❌ 限制：
    • Server 只能运行在本地
    • 每个 Client 启动一个独立进程
    • 无法多人共享同一个 Server
    • 无法部署到云端

HTTP+SSE 模式（生产推荐）：

  Claude Desktop ──(HTTP)──▶ 远程 MCP Server
  Cursor IDE     ──(HTTP)──▶   (同一个实例)
  其他 Client    ──(HTTP)──▶

  ✅ 优点：
    • Server 可以部署在任意位置
    • 多个 Client 共享同一个 Server
    • 支持负载均衡和水平扩展
    • 可以加认证、限流、监控

SSE Transport 原理

HTTP+SSE（Server-Sent Events）是 MCP 远程部署的标准传输层：

SSE 通信流程：

  Client                          Server
    │                               │
    │   POST /sse  (建立连接)        │
    │ ─────────────────────────────▶ │
    │                               │
    │   SSE 事件流（持久连接）        │
    │ ◀═════════════════════════════ │
    │   event: endpoint             │
    │   data: /messages?sid=xxx     │
    │                               │
    │   POST /messages?sid=xxx      │
    │ ─────────────────────────────▶ │
    │   (JSON-RPC request)          │
    │                               │
    │   SSE 事件                     │
    │ ◀═════════════════════════════ │
    │   event: message              │
    │   data: {JSON-RPC response}   │
    │                               │

  两条通道：
    POST /messages  → Client 发请求（Request）
    SSE 事件流       → Server 返回响应（Response + Notification）

简单理解：Client 通过 HTTP POST 发送请求，Server 通过 SSE 事件流推送结果。SSE 是一种轻量级的服务器推送技术，基于普通 HTTP，天然支持穿越防火墙和代理。

用 FastMCP 启动 SSE Server

FastMCP 内置了 SSE Transport 支持，切换只需一行代码：

python

# server_sse.py —— 以 SSE 模式运行的 MCP Server
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("我的远程 Server")

@mcp.tool()
def greet(name: str) -> str:
    """向用户问好"""
    return f"你好，{name}！这是来自远程 Server 的问候。"

# ═══════════════════════════════════════════
# 关键：以 SSE 模式运行
# ═══════════════════════════════════════════

if __name__ == "__main__":
    mcp.run(transport="sse")

启动方式：

bash

# 方式 1：直接运行
uv run server_sse.py
# 输出：Uvicorn running on http://0.0.0.0:8000

# 方式 2：指定端口和 host
HOST=0.0.0.0 PORT=9000 uv run server_sse.py

# 方式 3：用 uvicorn 手动控制
uv run uvicorn server_sse:mcp.sse_app --host 0.0.0.0 --port 8000 --workers 4

STDIO vs SSE 启动对比：

  STDIO（本地模式）：
    mcp.run()                    ← 默认值
    mcp.run(transport="stdio")   ← 显式指定

  SSE（远程模式）：
    mcp.run(transport="sse")     ← 一行切换
    → 背后启动 Uvicorn HTTP 服务器
    → 监听 /sse 和 /messages 端点

客户端连接 SSE Server

对于 Claude Desktop，修改配置指向远程地址：

json

{
  "mcpServers": {
    "remote-server": {
      "url": "http://你的服务器IP:8000/sse"
    }
  }
}

对于代码中的 Client 连接：

python

# client_sse.py —— 连接远程 SSE Server
from mcp import ClientSession
from mcp.client.sse import sse_client

async def main():
    # 连接远程 Server
    async with sse_client("http://你的服务器IP:8000/sse") as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            # 列出可用工具
            tools = await session.list_tools()
            print(f"可用工具：{[t.name for t in tools.tools]}")

            # 调用远程工具
            result = await session.call_tool("greet", {"name": "远程用户"})
            print(result.content[0].text)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Streamable HTTP（MCP 2025-03 新增）

MCP 协议在 2025 年 3 月引入了新的传输层——Streamable HTTP，作为 SSE 的升级版：

SSE（旧版） vs Streamable HTTP（新版）：

  SSE Transport：
    • 两个端点：POST /messages + GET /sse
    • 需要维持持久 SSE 连接
    • 有状态（Session ID 绑定连接）
    • ⚠️ 在某些基础设施上有兼容性问题

  Streamable HTTP Transport：
    • 单个端点：POST /mcp
    • 无需持久连接（按需升级为 SSE）
    • 可以无状态运行
    • ✅ 更适合现代云基础设施

  迁移建议：
    → 新项目优先使用 Streamable HTTP
    → 旧项目可以先用 SSE，后续迁移
    → 两种方式在 MCP SDK 中都有支持

python

# FastMCP 启动 Streamable HTTP
if __name__ == "__main__":
    mcp.run(transport="streamable-http")
    # → 监听 POST /mcp 端点

现状：截至本文写作时（2025 年），SSE 和 Streamable HTTP 并存。SDK 两种都支持，Claude Desktop 也都能连接。建议新项目直接用 Streamable HTTP。

10.2 Docker 容器化部署

生产环境不能直接 uv run 裸跑。Docker 容器化是标准做法——环境一致、部署方便、隔离安全。

编写 Dockerfile

以第 9 章的 GitHub 分析 Server 为例：

dockerfile

# Dockerfile —— MCP Server 容器化
FROM python:3.12-slim

# 设置工作目录
WORKDIR /app

# 安装 uv（更快的包管理器）
RUN pip install uv

# 复制依赖声明文件
COPY pyproject.toml .

# 安装依赖（利用 Docker 缓存层）
RUN uv sync --no-dev

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
  CMD curl -f http://localhost:8000/sse || exit 1

# 启动命令（SSE 模式）
CMD ["uv", "run", "server.py"]

Dockerfile 分层策略：

  Layer 1: python:3.12-slim     ← 基础镜像（变化极少）
  Layer 2: pip install uv       ← 工具安装（变化极少）
  Layer 3: COPY pyproject.toml  ← 依赖声明（偶尔变化）
  Layer 4: uv sync              ← 安装依赖（偶尔变化）
  Layer 5: COPY . .             ← 应用代码（频繁变化）

  → 代码改了只重建最后一层，构建速度极快

docker-compose 编排

yaml

# docker-compose.yml
version: "3.9"

services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - GITHUB_TOKEN=${GITHUB_TOKEN}
      - HOST=0.0.0.0
      - PORT=8000
      - LOG_LEVEL=info
    restart: unless-stopped
    # 资源限制
    deploy:
      resources:
        limits:
          memory: 512M
          cpus: "1.0"
    # 日志配置
    logging:
      driver: json-file
      options:
        max-size: "10m"
        max-file: "3"

构建与运行

bash

# 构建镜像
docker build -t my-mcp-server .

# 运行容器（传入环境变量）
docker run -d \
  --name mcp-server \
  -p 8000:8000 \
  -e GITHUB_TOKEN="ghp_你的token" \
  --restart unless-stopped \
  my-mcp-server

# 或者用 docker-compose（推荐）
echo "GITHUB_TOKEN=ghp_你的token" > .env
docker compose up -d

# 查看日志
docker compose logs -f mcp-server

# 停止
docker compose down

环境变量管理

生产环境的秘钥管理：

  ❌ 硬编码在代码里：
    github = GitHubClient(token="ghp_abc123")
    → 泄露到 Git 仓库，极其危险

  ❌ 写在 Dockerfile 里：
    ENV GITHUB_TOKEN="ghp_abc123"
    → 镜像里包含明文秘钥

  ✅ 环境变量（基础方案）：
    docker run -e GITHUB_TOKEN="ghp_xxx" ...
    → 运行时注入，不进入镜像

  ✅ .env 文件（开发方案）：
    echo "GITHUB_TOKEN=ghp_xxx" > .env
    docker compose up
    → 记得 .gitignore 加上 .env

  ✅ Secret Manager（生产方案）：
    AWS Secrets Manager / GCP Secret Manager / Vault
    → 集中管理、自动轮转、审计日志

经验法则：开发用 .env 文件，CI/CD 用环境变量，生产用 Secret Manager。三层递进，逐步加固。

10.3 认证、授权与安全最佳实践

MCP Server 一旦暴露在网络上，安全就是头等大事。你不会希望任何人都能调用你的工具来操作数据库或访问内部 API。

认证方案概览

MCP Server 的认证层级：

  Level 0：无认证（仅限本地开发）
    → STDIO 模式默认无认证
    → 因为只有本地进程能访问

  Level 1：API Key 认证（适合内部工具）
    → 在请求头中携带密钥
    → 简单、够用、好实现

  Level 2：OAuth 2.0（适合公开服务）
    → MCP 规范推荐的标准方案
    → 支持令牌刷新、权限范围（Scope）
    → 适合多用户、多权限场景

  Level 3：mTLS（适合高安全场景）
    → 双向证书认证
    → 金融、医疗等合规要求场景

API Key 认证实现

最简单实用的方案——在 HTTP 层加一个 API Key 校验中间件：

python

# server_auth.py —— 带 API Key 认证的 MCP Server
import os
from mcp.server.fastmcp import FastMCP
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

# 有效的 API Key 列表（生产环境从数据库或 Secret Manager 读取）
VALID_API_KEYS = set(
    os.getenv("MCP_API_KEYS", "").split(",")
)

class ApiKeyMiddleware(BaseHTTPMiddleware):
    """API Key 认证中间件"""
    async def dispatch(self, request: Request, call_next):
        # 跳过健康检查端点
        if request.url.path == "/health":
            return await call_next(request)

        # 从请求头提取 API Key
        api_key = request.headers.get("Authorization", "").replace("Bearer ", "")

        if not api_key or api_key not in VALID_API_KEYS:
            return JSONResponse(
                status_code=401,
                content={"error": "无效的 API Key"}
            )

        return await call_next(request)

# 创建 MCP Server
mcp = FastMCP("安全 Server")

# 注入认证中间件
mcp.settings.host = "0.0.0.0"
mcp.settings.port = 8000

@mcp.tool()
def query_database(sql: str) -> str:
    """执行 SQL 查询（需要认证才能调用）"""
    # ... 实际实现
    return f"查询结果：{sql}"

if __name__ == "__main__":
    app = mcp.sse_app()
    app.add_middleware(ApiKeyMiddleware)

    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

客户端连接时携带 API Key：

python

# 带认证的客户端连接
from mcp.client.sse import sse_client

async with sse_client(
    "http://server:8000/sse",
    headers={"Authorization": "Bearer your-api-key-here"}
) as (read, write):
    # ... 正常使用
    pass

输入校验与安全边界

认证解决了"谁能访问"的问题，但还有"传了什么数据"的问题。永远不要信任用户输入：

python

# ❌ 危险：直接拼接 SQL
@mcp.tool()
def query(sql: str) -> str:
    cursor.execute(sql)  # SQL 注入！
    return str(cursor.fetchall())

# ✅ 安全：参数化查询 + 白名单
@mcp.tool()
def query_table(
    table: str,
    limit: int = 10
) -> str:
    """查询数据表。

    Args:
        table: 表名（仅支持白名单中的表）
        limit: 返回行数，1-100
    """
    # 白名单校验
    ALLOWED_TABLES = {"users", "orders", "products"}
    if table not in ALLOWED_TABLES:
        return f"❌ 不允许查询表 '{table}'。可用表：{ALLOWED_TABLES}"

    # 范围校验
    limit = max(1, min(limit, 100))

    # 参数化查询（防 SQL 注入）
    cursor.execute(f"SELECT * FROM {table} LIMIT ?", (limit,))
    return str(cursor.fetchall())

MCP Server 安全边界原则：

  1. 最小权限
     → Server 只暴露必要的功能
     → 数据库连接用只读账号
     → 文件操作限制在指定目录

  2. 输入校验
     → 所有参数做类型和范围校验
     → 使用白名单而非黑名单
     → 文件路径做防穿越检查（禁止 ../）

  3. 输出过滤
     → 不返回内部错误堆栈给客户端
     → 脱敏处理（隐藏密码、Token 等）
     → 限制返回数据量（防止内存溢出）

  4. 速率限制
     → 限制每个 Key 的调用频率
     → 防止滥用和 DDoS
     → 可用 Redis + 滑动窗口实现

安全检查清单

上线前过一遍这个清单：

MCP Server 安全上线清单：

  [ ] 认证
      [ ] 所有远程端点都需要认证
      [ ] API Key / Token 不硬编码在代码中
      [ ] 认证失败返回 401，不泄露细节

  [ ] 输入
      [ ] 所有参数有类型校验
      [ ] 字符串输入有长度限制
      [ ] 文件路径有穿越保护
      [ ] SQL 使用参数化查询
      [ ] 命令执行使用白名单

  [ ] 输出
      [ ] 错误信息不包含内部细节
      [ ] 敏感数据做脱敏处理
      [ ] 返回数据有大小限制

  [ ] 网络
      [ ] 使用 HTTPS（TLS）加密传输
      [ ] 配置 CORS 限制源域名
      [ ] 启用速率限制

  [ ] 运维
      [ ] 容器以非 root 用户运行
      [ ] 秘钥存储在 Secret Manager
      [ ] 定期轮换 API Key
      [ ] 开启审计日志

10.4 日志、监控与故障排查

Server 上了生产，"能跑"只是及格线。出了问题能快速定位——这才是运维能力的分水岭。

结构化日志

Python 的 logging 模块是基础，但生产环境需要结构化日志——方便机器解析和日志平台检索：

python

# logging_config.py —— 结构化日志配置
import logging
import json
from datetime import datetime

class JsonFormatter(logging.Formatter):
    """JSON 格式的日志输出"""
    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # 附加额外字段
        if hasattr(record, "tool_name"):
            log_data["tool_name"] = record.tool_name
        if hasattr(record, "duration_ms"):
            log_data["duration_ms"] = record.duration_ms
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_data, ensure_ascii=False)

def setup_logging(level: str = "INFO"):
    """初始化日志配置"""
    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())

    root = logging.getLogger()
    root.setLevel(getattr(logging, level.upper()))
    root.addHandler(handler)

    return logging.getLogger("mcp-server")

日志输出示例（JSON 格式）：

  {"timestamp":"2025-03-15T08:30:12Z","level":"INFO",
   "logger":"mcp-server","message":"Tool 调用开始",
   "tool_name":"get_repo_info","duration_ms":null}

  {"timestamp":"2025-03-15T08:30:12Z","level":"INFO",
   "logger":"mcp-server","message":"Tool 调用完成",
   "tool_name":"get_repo_info","duration_ms":234}

  {"timestamp":"2025-03-15T08:30:15Z","level":"ERROR",
   "logger":"mcp-server","message":"GitHub API 调用失败",
   "tool_name":"search_repos",
   "exception":"httpx.HTTPStatusError: 403 Forbidden"}

  → JSON 格式可以直接导入 ELK / Datadog / CloudWatch
  → 结构化字段支持精确搜索：tool_name:"get_repo_info"

使用 MCP Context 日志

FastMCP 提供了 Context 对象，可以向客户端发送日志和进度通知：

python

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("带日志的 Server")

@mcp.tool()
async def analyze_repo(owner: str, repo: str, ctx: Context) -> str:
    """分析仓库（带进度日志）"""

    # 日志级别：debug / info / warning / error
    ctx.info(f"开始分析 {owner}/{repo}")

    # 发送进度通知（Client 可以展示进度条）
    await ctx.report_progress(1, 5, "获取仓库信息...")
    repo_info = await get_repo(owner, repo)

    await ctx.report_progress(2, 5, "读取 README...")
    readme = await get_readme(owner, repo)

    await ctx.report_progress(3, 5, "获取文件结构...")
    files = await get_tree(owner, repo)

    await ctx.report_progress(4, 5, "生成分析报告...")
    report = generate_report(repo_info, readme, files)

    await ctx.report_progress(5, 5, "完成！")
    ctx.info(f"分析完成，报告长度：{len(report)} 字符")

    return report

Context 日志的两个去向：

  ctx.info("消息")
    → 发送到 MCP Client（Claude Desktop 的日志面板）
    → 同时写入 Server 端的标准日志

  await ctx.report_progress(current, total, "描述")
    → 发送进度通知给 Client
    → Client 可以展示进度条或百分比
    → 适合耗时较长的操作

性能监控（Prometheus 指标）

对于需要精细监控的生产环境，可以接入 Prometheus：

python

# metrics.py —— Prometheus 指标收集
import time
import functools
from prometheus_client import Counter, Histogram, generate_latest

# 定义指标
TOOL_CALLS = Counter(
    "mcp_tool_calls_total",
    "MCP Tool 调用总次数",
    ["tool_name", "status"]
)

TOOL_DURATION = Histogram(
    "mcp_tool_duration_seconds",
    "MCP Tool 调用耗时",
    ["tool_name"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

def track_tool(func):
    """工具调用监控装饰器"""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        tool_name = func.__name__
        start = time.time()
        try:
            result = await func(*args, **kwargs)
            TOOL_CALLS.labels(tool_name=tool_name, status="success").inc()
            return result
        except Exception as e:
            TOOL_CALLS.labels(tool_name=tool_name, status="error").inc()
            raise
        finally:
            duration = time.time() - start
            TOOL_DURATION.labels(tool_name=tool_name).observe(duration)
    return wrapper

# 使用方式：
@mcp.tool()
@track_tool
async def get_repo_info(owner: str, repo: str) -> str:
    """查询仓库信息（带监控）"""
    # ... 实际实现

Prometheus 监控面板可以呈现：

  mcp_tool_calls_total
    → 每个 Tool 的调用次数和成功率
    → 发现哪些 Tool 最热门、哪些容易出错

  mcp_tool_duration_seconds
    → 每个 Tool 的响应时间分布
    → P50 / P95 / P99 延迟
    → 发现性能瓶颈

  配合 Grafana 可视化：
    → 实时仪表盘
    → 异常告警（错误率 > 5% 时通知）

常见故障排查

故障 1：Client 连接超时

  症状：Claude Desktop 显示 Server 无法连接
  排查：
    $ curl -v http://server:8000/sse
    → 能拿到 SSE 事件流？
    → 检查防火墙 / 安全组是否放行端口
    → 检查 Server 是否监听 0.0.0.0（而非 127.0.0.1）

  常见原因：
    ❌ mcp.run(transport="sse")  → 默认监听 localhost
    ✅ HOST=0.0.0.0 mcp.run(transport="sse")

─────────────────────────────────────────

故障 2：Tool 调用返回空结果

  症状：AI 调用工具后得到空响应
  排查：
    → 用 MCP Inspector 单独测试该 Tool
    → 检查 async 函数是否漏了 await
    → 检查异常是否被静默吞掉（空 except:pass）

  常见原因：
    ❌ async def my_tool():
         result = some_async_call()  # 漏了 await！
    ✅ async def my_tool():
         result = await some_async_call()

─────────────────────────────────────────

故障 3：Docker 容器启动后立即退出

  症状：docker logs 显示 ModuleNotFoundError
  排查：
    $ docker logs mcp-server
    → 看报错信息

  常见原因：
    ❌ 只 COPY 了代码，没安装依赖
    ✅ 确保 Dockerfile 中先 COPY pyproject.toml + uv sync

─────────────────────────────────────────

故障 4：GitHub API 返回 403

  症状：Tool 返回"❌ 查询失败：403 Forbidden"
  排查：
    → 检查 GITHUB_TOKEN 是否设置
    → Token 是否过期或权限不足
    → 是否触发了 GitHub 速率限制

  快速检查：
    $ curl -H "Authorization: token $GITHUB_TOKEN" \
       https://api.github.com/rate_limit

本章小结

知识点	要点
STDIO vs SSE	STDIO 适合本地、SSE 适合远程部署
SSE 启动	`mcp.run(transport="sse")` 一行切换
Streamable HTTP	新一代传输层，单端点 POST /mcp
Docker 部署	分层 Dockerfile + docker-compose 编排
秘钥管理	开发用 .env、CI 用环境变量、生产用 Secret Manager
API Key 认证	Starlette 中间件 + Bearer Token
安全边界	最小权限、输入白名单、输出脱敏、速率限制
结构化日志	JSON 格式日志 + Context 日志 API
监控	Prometheus 指标 + Grafana 仪表盘
故障排查	先 Inspector 隔离测试，再查网络、依赖、权限

下一章预告：附录 —— MCP API 速查表（Python & TypeScript）、常见报错解决方案、推荐学习资源与社区。

MCP 协议开发实战 ​

10. 生产部署与安全 ​

10.1 从 STDIO 到 HTTP+SSE：远程部署 ​

为什么需要远程部署 ​

SSE Transport 原理 ​

用 FastMCP 启动 SSE Server ​

客户端连接 SSE Server ​

Streamable HTTP（MCP 2025-03 新增） ​

10.2 Docker 容器化部署 ​

编写 Dockerfile ​

docker-compose 编排 ​

构建与运行 ​

环境变量管理 ​

10.3 认证、授权与安全最佳实践 ​

认证方案概览 ​

API Key 认证实现 ​

输入校验与安全边界 ​

安全检查清单 ​

10.4 日志、监控与故障排查 ​

结构化日志 ​

使用 MCP Context 日志 ​

性能监控（Prometheus 指标） ​

常见故障排查 ​

本章小结 ​

MCP 协议开发实战

10. 生产部署与安全

10.1 从 STDIO 到 HTTP+SSE：远程部署

为什么需要远程部署

SSE Transport 原理

用 FastMCP 启动 SSE Server

客户端连接 SSE Server

Streamable HTTP（MCP 2025-03 新增）

10.2 Docker 容器化部署

编写 Dockerfile

docker-compose 编排

构建与运行

环境变量管理

10.3 认证、授权与安全最佳实践

认证方案概览

API Key 认证实现

输入校验与安全边界

安全检查清单

10.4 日志、监控与故障排查

结构化日志

使用 MCP Context 日志

性能监控（Prometheus 指标）

常见故障排查

本章小结