在过去三个月里，我把团队内部的 Agent 工具链全部迁移到了 MCP（Model Context Protocol）协议上，期间踩过无数关于鉴权失效、并发卡死、Token 计费异常的坑。这篇文章把我目前在生产环境运行的 FastMCP Server 完整拆解出来，包含从架构设计、异步并发、OAuth2 鉴权到成本压测的全部细节。文末我会附上我在 4 个主流模型上的实测延迟与价格对比，告诉你为什么最终我把 LLM 后端切到了 HolySheep AI。
一、为什么选 FastMCP 而不是裸写 JSON-RPC
MCP 协议本质上是 JSON-RPC 2.0 的超集，但官方 SDK（modelcontextprotocol/python-sdk）对工具发现、资源订阅、提示模板做了完整封装。FastMCP 是社区在官方 SDK 上的装饰器风格封装，写起来像写 Flask：
- 用
@mcp.tool()装饰函数即可注册工具，无需手写 schema。 - 内置
stdio、sse、streamable-http三种传输层适配。 - 与 Pydantic v2 原生集成，输入输出校验零成本。
我在 2025 年 Q4 做过一次对比：裸写 JSON-RPC 需要 380 行代码的工具注册逻辑，用 FastMCP 只需要 60 行，且 bug 率下降约 70%。
二、生产级架构：反向代理 + Token 桶限流
MCP Server 通常会被多个 Agent 客户端同时调用，如果直接暴露在公网，攻击面非常大。我推荐的生产架构是：
- 边缘层：Nginx 做 TLS 终止与 IP 限速（limit_req zone=mcp:10m rate=30r/s）。
- 鉴权层：JWT 中间件校验 Bearer Token，从 HolySheep 控制台签发的 Key 走 HS256。
- 业务层：FastMCP + uvicorn，4 worker 进程。
- 异步队列：Redis Stream 缓冲工具调用结果，配合 aioredis 做背压。
实测下来，单机 4C8G 能稳定支撑 120 QPS，P99 延迟稳定在 180ms 以内（不含 LLM 调用）。
三、代码实现：从零搭建一个带鉴权的 FastMCP Server
先准备依赖：
pip install fastmcp uvicorn[standard] pyjwt[crypto] redis httpx pydantic>=2.6
下面是我正在使用的 server.py 完整实现，连接到 HolySheep 的 OpenAI 兼容接口：
import os
import time
import jwt
import httpx
import asyncio
from typing import Annotated
from pydantic import Field
from fastmcp import FastMCP, Context
from fastmcp.server.dependencies import get_context
============ 配置 ============
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
JWT_SECRET = os.getenv("MCP_JWT_SECRET", "your-32bytes-secret-key-xxxxxx")
JWT_ALGO = "HS256"
RATE_LIMIT_PER_MIN = 60 # 单租户每分钟最大请求数
mcp = FastMCP(
name="production-mcp-server",
instructions="生产级 MCP Server，提供受鉴权保护的 LLM 工具集",
stateless_http=True,
)
============ 鉴权依赖 ============
def verify_bearer(authorization: str | None) -> str:
"""从 Authorization 头解析并校验 JWT，返回租户 ID"""
if not authorization or not authorization.startswith("Bearer "):
raise PermissionError("missing bearer token")
token = authorization.split(" ", 1)[1]
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGO])
return payload["tenant_id"]
except jwt.PyJWTError as e:
raise PermissionError(f"invalid token: {e}") from e
============ 限流（基于内存令牌桶，生产建议换 Redis） ============
class TokenBucket:
def __init__(self, rate: int, capacity: int):
self.rate, self.capacity = rate, capacity
self.tokens, self.last = capacity, time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self) -> bool:
async with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate / 60)
self.last = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
_buckets: dict[str, TokenBucket] = {}
def get_bucket(tenant: str) -> TokenBucket:
if tenant not in _buckets:
_buckets[tenant] = TokenBucket(RATE_LIMIT_PER_MIN, RATE_LIMIT_PER_MIN)
return _buckets[tenant]
============ LLM 调用封装 ============
async def call_llm(prompt: str, model: str = "gpt-4.1", max_tokens: int = 1024) -> dict:
"""统一调用 HolySheep 兼容 OpenAI 协议的 Chat Completions"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json",
}
body = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": 0.2,
}
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{HOLYSHEEP_BASE_URL}/chat/completions",
json=body, headers=headers)
r.raise_for_status()
return r.json()
============ 注册工具 ============
@mcp.tool(description="带鉴权的文本摘要工具，自动注入调用方 tenant 信息")
async def summarize(
ctx: Context,
text: Annotated[str, Field(min_length=10, max_length=20000, description="待摘要原文")],
style: Annotated[str, Field(pattern="^(brief|detailed)$")] = "brief",
) -> dict:
# 从 HTTP 头中取 Bearer（FastMCP 会把 request 暴露给 ctx）
req = ctx.request_context.request
tenant = verify_bearer(req.headers.get("authorization"))
bucket = get_bucket(tenant)
if not await bucket.acquire():
return {"error": "rate_limited", "tenant": tenant, "retry_after": 5}
prompt = f"请用{'详细' if style=='detailed' else '简洁'}风格摘要:\n{text}"
started = time.monotonic()
resp = await call_llm(prompt, model="gpt-4.1", max_tokens=512)
elapsed_ms = int((time.monotonic() - started) * 1000)
return {
"tenant": tenant,
"summary": resp["choices"][0]["message"]["content"],
"usage": resp.get("usage", {}),
"elapsed_ms": elapsed_ms,
"model": "gpt-4.1",
}
@mcp.tool(description="查询账户余额，便于 Agent 自我决策是否继续调用")
async def quota(ctx: Context) -> dict:
req = ctx.request_context.request
tenant = verify_bearer(req.headers.get("authorization"))
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(f"{HOLYSHEEP_BASE_URL}/dashboard/billing/credit_grants",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"})
return {"tenant": tenant, **r.json()}
============ 启动入口 ============
if __