MCP协议实战：用FastAPI运行Model Context Protocol

如果你需要一种轻量的方式向 LLM 提供商发送丰富的上下文（文档、embeddings、用户元数据），而又不想重复造轮子，那么 Model Context Protocol（MCP）正是你要的。你只需启动一个小型 HTTP 服务器，推送上下文片段，获取一个签名 token，然后将其交给 OpenAI 或 Anthropic。本文接下来将展示从 Docker 到 FastAPI 的 MCP 示例，并附上我在生产环境中踩过的一些坑。

什么是 Model Context Protocol（MCP），为什么它很重要？

MCP 是一个开源规范，定义了一种 JSON-over-HTTP 合约，用于存储、检索和对 prompt 上下文进行版本控制。你不再需要把 32 KB 的 prompt 塞进每一次 LLM 请求，而是将重型内容（大文档、向量索引、用户特定设置）存储在独立服务中，通过一个短 token 来引用它们。

成本控制——你只需为 token 向 LLM 提供商付费，而无需为原始文本的兆字节付费。

一致性——同一上下文可以在多次调用中复用，确保答案的确定性。

安全性——服务可以强制执行访问策略、删除敏感字段、轮换密钥，而无需触碰 LLM 代码。

简而言之，MCP 让你把上下文当作一等公民来对待，而不是靠 hack 式的字符串拼接。

如何用 Docker 在本地搭建 MCP 服务器？

在本地运行 MCP 只需一条命令。官方镜像自带一个用于开发的轻量 SQLite 存储，以及一个可配置的生产 PostgreSQL 后端。

# 1. Pull the image
docker pull ghcr.io/mcp-dev/mcp-server:latest

# 2. Run with an in‑memory store (good for quick tests)
docker run -d \
  -p 8080:8080 \
  -e MCP_STORE=sqlite \
  -e MCP_AUTH_TOKEN=dev-secret \
  ghcr.io/mcp-dev/mcp-server:latest

如果想搭建更真实的环境，可以挂载一个卷并指向 PostgreSQL：

docker run -d \
  -p 8080:8080 \
  -e MCP_STORE=postgres \
  -e POSTGRES_DSN="postgresql://mcp:pwd@db:5432/mcp" \
  -e MCP_AUTH_TOKEN=prod-secret \
  --restart unless-stopped \
  ghcr.io/mcp-dev/mcp-server:latest

容器日志会在 http://localhost:8080/health 显示健康检查。如果看到 {"status":"ok"}，说明服务已就绪，可以开始喂上下文了。

我踩过的坑：默认的 SQLite 文件位于容器内部。当容器重启时，所有数据都会丢失。在超越玩具级演示之前，要么绑定挂载一个主机目录（-v $(pwd)/data:/data），要么切换到 Postgres。

如何在 Python 中使用 MCP 向 OpenAI/Anthropic LLM 传递上下文？

Python 客户端很小（约 30 KB），且同时支持同步和异步代码。以下是一个最小示例：上传一个文档，获取 token，然后调用 OpenAI ChatCompletion 端点。

import httpx
import json
from openai import OpenAI

MCP_URL = "http://localhost:8080"
MCP_TOKEN = "dev-secret"          # same as -e MCP_AUTH_TOKEN
OPENAI_API_KEY = "sk-..."

client = httpx.Client(headers={"Authorization": f"Bearer {MCP_TOKEN}"})

def upload_context(name: str, content: str) -> str:
    resp = client.post(
        f"{MCP_URL}/v1/context",
        json={"name": name, "content": content},
    )
    resp.raise_for_status()
    return resp.json()["context_id"]

def get_context_token(context_id: str) -> str:
    resp = client.post(
        f"{MCP_URL}/v1/token",
        json={"context_id": context_id, "expires_in": 300},
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

# 1️⃣ Upload a long FAQ
ctx_id = upload_context(
    name="support_faq",
    content=open("support_faq.txt").read()
)

# 2️⃣ Get a short-lived token
token = get_context_token(ctx_id)

# 3️⃣ Call OpenAI, passing the token in `metadata`
client_oai = OpenAI(api_key=OPENAI_API_KEY)
completion = client_oai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You have access to supplemental context via the `context_token` field."},
        {"role": "user", "content": "Explain our refund policy."}
    ],
    metadata={"context_token": token}
)

print(completion.choices[0].message.content)

metadata.context_token 字段是 LLM 提供商采用的一种约定，用于按需获取额外数据。Anthropic 也遵循相同的模式，只需换掉客户端库即可。

权衡：你增加了一个额外的网络跳点。在延迟敏感路径（亚 100 毫秒）下，你可能希望将最新的片段保留在进程内，而不是走 MCP。

如何将 MCP 调用集成到 FastAPI 端点中？

将 MCP 嵌入 FastAPI 很直接，因为两者都使用标准 ASGI 模式。以下是一个生产可用的端点，它：

接收用户查询。 根据用户的组织查找或创建上下文记录。 使用生成的 token 调用 OpenAI。 返回 LLM 答案。

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import httpx
from openai import AsyncOpenAI

app = FastAPI()
mcp_client = httpx.AsyncClient(base_url="http://mcp:8080",
                               headers={"Authorization": "Bearer prod-secret"})
openai_client = AsyncOpenAI(api_key="sk-...")

class QueryPayload(BaseModel):
    user_id: str
    organization_id: str
    question: str

async def get_or_create_context(org_id: str) -> str:
    # Try to fetch an existing context_id from DB (pseudo‑code)
    ctx_id = await fetch_context_id_from_db(org_id)
    if ctx_id:
        return ctx_id

    # Fallback: load org‑specific docs from S3, upload to MCP
    docs = await load_org_docs_from_s3(org_id)
    resp = await mcp_client.post(
        "/v1/context",
        json={"name": f"org-{org_id}", "content": docs}
    )
    resp.raise_for_status()
    ctx_id = resp.json()["context_id"]
    await store_context_id_in_db(org_id, ctx_id)
    return ctx_id

@app.post("/answer")
async def answer(payload: QueryPayload):
    ctx_id = await get_or_create_context(payload.organization_id)
    token_resp = await mcp_client.post(
        "/v1/token",
        json={"context_id": ctx_id, "expires_in": 120}
    )
    token_resp.raise_for_status()
    token = token_resp.json()["access_token"]

    try:
        completion = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Use supplemental context via `context_token`."},
                {"role": "user", "content": payload.question}
            ],
            metadata={"context_token": token}
        )
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc))

    return {"answer": completion.choices[0].message.content}

注意职责分离：

• MCP 客户端作为单例存在，复用 HTTP 连接。
• 上下文创建是惰性的——只有给定组织的首次请求才会访问 S3。
• 错误处理向上冒泡为 502，以便调用方知道是 LLM 端出了问题。

如果你已经熟悉异步 SQLAlchemy，可以将 fetch_context_id_from_db 和 store_context_id_in_db 调用接入你用于其他业务数据的同一会话。我在混用同步和异步 DB 调用时遇到了 MissingGreenlet 错误；解决办法是保持全程异步，或将同步路径放到线程池中运行。

生产环境中 MCP 的最佳实践、安全注意事项和故障排除步骤

认证与授权

• 为每个服务使用短期 bearer token（每 24 小时轮换一次）。将密钥存储在 secret manager 中，而不是 Dockerfile 中。
• 将 token 限定于单个组织或用例。MCP 规范支持 audience claims，在网关层强制执行。
• 上传前删除 PII。MCP 不会为你清洗数据。

版本控制

• 对上下文 ID 进行版本控制。文档变更时，创建新 ID 并保留旧 ID 一段宽限期。这可以避免正在飞行中的 LLM 调用被破坏。

监控

• 启用内置 /metrics 端点（Prometheus 格式），抓取 mcp_contexts_total、mcp_token_requests 和 mcp_request_latency_seconds。
• 将 MCP 日志与你的 FastAPI 请求 ID 关联起来。我为每个入站调用添加了 X-Request-ID header 并向下游传播；没有它，我就无法判断是哪个 token 请求导致了超时。

成本

• 每次 token 获取都是一个廉价的 HTTP 调用，但每次上下文上传都会消耗存储。通过 /v1/context/expire 端点为未使用的上下文设置 TTL（如 30 天）。
• 如果你在 LLM 端遇到速率限制，记住 MCP 减少了 prompt 大小，这可以降低基于 token 的定价。

不适用场景

• 实时音频转录，上下文是几秒钟的音频——额外服务的开销抵消了好处。
• 不经常运行的一次性脚本；简单的文件读取更便宜、更简单。

用 MCP 构建的上下文感知 AI 助手长什么样？

以下是我为一家 SaaS 客服团队部署的助手的精简版本。该助手：

从私有 Git 仓库拉取最新的知识库 markdown。 以每日版本化 ID 存储在 MCP 中。 通过 FastAPI 端点提供服务答案，供内部聊天机器人 UI 调用。

# cron_job.py – runs nightly
import httpx, os, subprocess, datetime

MCP_URL = "http://mcp:8080"
TOKEN = os.getenv("MCP_TOKEN")
client = httpx.Client(base_url=MCP_URL,
                     headers={"Authorization": f"Bearer {TOKEN}"})

def refresh_kb():
    # Pull latest docs
    subprocess.run(["git", "pull"], cwd="/opt/kb", check=True)
    with open("/opt/kb/combined.md") as f:
        content = f.read()

    version = datetime.date.today().isoformat()
    resp = client.post(
        "/v1/context",
        json={"name": f"support_kb_{version}", "content": content}
    )
    resp.raise_for_status()
    ctx_id = resp.json()["context_id"]
    # Store the latest ID somewhere reachable by FastAPI
    with open("/tmp/latest_kb_id.txt", "w") as f:
        f.write(ctx_id)

if __name__ == "__main__":
    refresh_kb()

FastAPI 在每次请求时使用最新的 ID：

@app.get("/support")
async def support(question: str):
    with open("/tmp/latest_kb_id.txt") as f:
        ctx_id = f.read().strip()
    token = await get_context_token(ctx_id)   # same helper as earlier
    resp = await openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user","content":question}],
        metadata={"context_token": token}
    )
    return {"answer": resp.choices[0].message.content}

效果如何？助手可以用完整的知识库来回答"如何重置密码？"，而无需向 OpenAI 发送兆字节级的 markdown。即便是流量高峰时，延迟也保持在 300 毫秒以内。

常见问题

MCP 对上下文数据有什么格式要求？ 纯 JSON payload，包含 name（字符串）和 content（字符串）。你也可以发送 metadata 用于自定义索引；规范支持 type 字段用于二进制 blob。

除了 OpenAI 和 Anthropic，我可以将 MCP 与其他 LLM 提供商一起使用吗？ 可以。只要提供商在请求 metadata 中接受 token，并且知道如何通过 MCP 端点解析它，就可以工作。有些供应商要求将 token 放在自定义 header 中——只需调整客户端代码即可。

当我运行多个 FastAPI worker 时，MCP 是线程安全的吗？ 服务器本身除了后端存储外是无状态的，因此并发请求是安全的。客户端必须复用 httpx.AsyncClient 或 requests.Session，以避免 socket 耗尽。

如何自动清理旧上下文？ 使用截止日期调用 /v1/context/expire 端点，或者配置一个后台任务删除超过保留策略期限的行。规范还支持上传时设置 TTL（expires_in），触发自动清理。

结语

MCP 让你将大型 prompt 上下文外部化，从而降低 LLM token 成本并提高一致性。

一条 Docker 命令即可运行开发服务器；任何实际工作负载都切换到 Postgres。

Python 客户端处理上传、token 检索，同时支持 OpenAI 和 Anthropic。

FastAPI 集成只需复用异步 HTTP 客户端并接入一个小型 helper 来获取 token。

用短期 bearer token 保护服务，对上下文进行版本控制，并监控延迟。

适用于任何上下文大小超过 LLM prompt 限制、或需要可复用、可审计数据的场景。

有了这些 MCP 示例，你明天就可以开始原型设计，下一个 sprint 就能交付一个稳定的生产服务。祝你编程愉快。