fastmcp

FastMCP - 使用 Python 构建 MCP 服务

FastMCP 是一个 Python 框架，用于构建 Model Context Protocol（MCP）服务，向 Claude 等大语言模型公开工具、资源和提示词。本技能提供经过生产环境验证的模式、错误预防方法和部署策略，助你构建稳健的 MCP 服务。

快速入门

安装

pip install fastmcp
# or
uv pip install fastmcp

最简服务

from fastmcp import FastMCP

# MUST be at module level for FastMCP Cloud
mcp = FastMCP("My Server")

@mcp.tool()
async def hello(name: str) -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"

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

运行：

# Local development
python server.py

# With FastMCP CLI
fastmcp dev server.py

# HTTP mode
python server.py --transport http --port 8000

v2.14.x 新增内容（2025 年 12 月）

v2.14.2（2024 年 12 月 31 日）

• 将 MCP SDK 固定为 <2.x，以确保兼容性
• Supabase 提供程序新增 auth_route 参数
• 错误修复：outputSchema $ref 解析、OAuth Proxy 验证、支持 OpenAPI 3.1

v2.14.1：使用工具进行采样（SEP-1577）

• ctx.sample() 现在接受工具，支持 Agent 工作流
• AnthropicSamplingHandler 从实验性功能升级
• ctx.sample_step() 用于执行单次 LLM 调用并返回 SampleStep
• 新增 Python 3.13 支持

v2.14.0：后台任务（SEP-1686）

• 协议原生后台任务，用于长时间运行的操作
• 为异步装饰器添加 task=True；可跟踪进度而不阻塞
• 支持 MCP 2025-11-25 规范
• SEP-1699：SSE 轮询和事件恢复
• SEP-1330：多选枚举引导模式
• SEP-1034：引导模式的默认值

⚠️ 重大变更（v2.14.0）：

• 已移除 BearerAuthProvider 模块（请使用 JWTVerifier 或 OAuthProxy）
• 已移除 Context.get_http_request() 方法
• 已移除顶层导入 fastmcp.Image（请使用 from fastmcp.utilities import Image）
• 已移除 enable_docket、enable_tasks 设置（始终启用）
• 已移除 run_streamable_http_async()、sse_app()、streamable_http_app()、run_sse_async() 方法
• 已从装饰器中移除 dependencies 参数
• 已取消对 output_schema=False 的支持
• 已弃用 FASTMCP_SERVER_ 环境变量前缀

已知兼容性：

• MCP SDK 固定为 <2.x（v2.14.2+）

v3.0.0 新增内容（Beta - 2026 年 1 月）

⚠️ 重大破坏性变更 - FastMCP 3.0 进行了完整的架构重构。

提供程序架构

现在所有组件均通过提供程序获取：

• FileSystemProvider - 从目录中发现带装饰器的函数，并支持热重载
• SkillsProvider - 将 Agent 技能文件公开为 MCP 资源
• OpenAPIProvider - 根据 OpenAPI 规范自动生成
• ProxyProvider - 代理到远程 MCP 服务

from fastmcp import FastMCP
from fastmcp.providers import FileSystemProvider

mcp = FastMCP("server")
mcp.add_provider(FileSystemProvider(path="./tools", reload=True))

转换（组件中间件）

无需更改源代码即可修改组件：

• 按命名空间、重命名、按版本筛选
• ResourcesAsTools - 将资源公开为工具
• PromptsAsTools - 将提示词公开为工具

from fastmcp.transforms import Namespace, VersionFilter

mcp.add_transform(Namespace(prefix="api"))
mcp.add_transform(VersionFilter(min_version="2.0"))

组件版本控制

@mcp.tool(version="2.0")
async def fetch_data(query: str) -> dict:
    # Clients see highest version by default
    # Can request specific version
    return {"data": [...]}

会话范围状态

@mcp.tool()
async def set_preference(key: str, value: str, ctx: Context) -> dict:
    await ctx.set_state(key, value)  # Persists across session
    return {"saved": True}

@mcp.tool()
async def get_preference(key: str, ctx: Context) -> dict:
    value = await ctx.get_state(key, default=None)
    return {"value": value}

其他功能

• --reload 标志，用于开发期间自动重启
• 自动为同步函数分派线程池
• 工具超时
• OpenTelemetry 跟踪
• 组件授权：@tool(auth=require_scopes("admin"))

迁移指南

如尚未准备就绪，请固定使用 v2：

# requirements.txt
fastmcp<3

对于大多数服务，只需更新导入即可：

# v2.x and v3.0 compatible
from fastmcp import FastMCP

mcp = FastMCP("server")
# ... rest of code works the same

参见：官方迁移指南

---

核心概念

工具

大模型可以调用的函数。最佳实践：名称清晰、文档字符串详尽（大模型会读取这些内容！）、类型提示明确（由 Pydantic 验证）、结构化返回值、错误处理。

@mcp.tool()
async def async_tool(url: str) -> dict:  # Use async for I/O
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).json()

资源

向大模型公开数据。URI 方案：data://、file://、resource://、info://、api:// 或自定义方案。

@mcp.resource("user://{user_id}/profile")  # Template with parameters
async def get_user(user_id: str) -> dict:  # CRITICAL: param names must match
    return await fetch_user_from_db(user_id)

提示词

带参数的预配置提示词。

@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
    return f"Analyze {topic} considering: state, challenges, opportunities, recommendations."

上下文功能

注入 Context 参数（须带类型提示！）以使用高级功能：

引导（用户输入）：

from fastmcp import Context

@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
    confirmed = await context.request_elicitation(prompt=f"Confirm {action}?", response_type=str)
    return {"status": "completed" if confirmed.lower() == "yes" else "cancelled"}

进度跟踪：

@mcp.tool()
async def batch_import(file_path: str, context: Context) -> dict:
    data = await read_file(file_path)
    for i, item in enumerate(data):
        await context.report_progress(i + 1, len(data), f"Importing {i + 1}/{len(data)}")
        await import_item(item)
    return {"imported": len(data)}

采样（从工具调用 LLM）：

@mcp.tool()
async def enhance_text(text: str, context: Context) -> str:
    response = await context.request_sampling(
        messages=[{"role": "user", "content": f"Enhance: {text}"}],
        temperature=0.7
    )
    return response["content"]

后台任务（v2.14.0+）

长时间运行的操作，可报告进度而不阻塞客户端。使用 Docket 任务调度器（在 v2.14.0+ 中始终启用）。

基本用法：

@mcp.tool(task=True)  # Enable background task mode
async def analyze_large_dataset(dataset_id: str, context: Context) -> dict:
    """Analyze large dataset with progress tracking."""
    data = await fetch_dataset(dataset_id)

    for i, chunk in enumerate(data.chunks):
        # Report progress to client
        await context.report_progress(
            current=i + 1,
            total=len(data.chunks),
            message=f"Processing chunk {i + 1}/{len(data.chunks)}"
        )
        await process_chunk(chunk)

    return {"status": "complete", "records_processed": len(data)}

任务状态： pending → running → completed / failed / cancelled

适用场景：

• 耗时超过 30 秒的操作（存在 LLM 超时风险）
• 可逐项更新状态的批处理
• 执行过程中可能需要用户输入的操作
• 耗时较长的 API 调用或数据处理

已知限制（v2.14.x）：

• 后台任务轮询期间，ctx.report_progress() 中的 statusMessage 不会转发给客户端（GitHub Issue #2904）
• 进度消息会显示在服务日志中，但不会显示在客户端 UI 中
• 解决方法：目前改用官方 MCP SDK（mcp>=1.10.0），不要使用 FastMCP
• 状态：PR #2906 中的修复尚待处理

**重要提示：**任务通过 Docket 调度器执行。无法通过代理执行任务（会引发错误）。

使用工具进行采样（v2.14.1+）

服务可以将工具传递给 ctx.sample()，用于 Agent 工作流，让大模型在采样期间调用工具。

Agent 式采样：

from fastmcp import Context
from fastmcp.sampling import AnthropicSamplingHandler

# Configure sampling handler
mcp = FastMCP("Agent Server")
mcp.add_sampling_handler(AnthropicSamplingHandler(api_key=os.getenv("ANTHROPIC_API_KEY")))

@mcp.tool()
async def research_topic(topic: str, context: Context) -> dict:
    """Research a topic using agentic sampling with tools."""

    # Define tools available during sampling
    research_tools = [
        {
            "name": "search_web",
            "description": "Search the web for information",
            "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}}
        },
        {
            "name": "fetch_url",
            "description": "Fetch content from a URL",
            "inputSchema": {"type": "object", "properties": {"url": {"type": "string"}}}
        }
    ]

    # Sample with tools - LLM can call these tools during reasoning
    result = await context.sample(
        messages=[{"role": "user", "content": f"Research: {topic}"}],
        tools=research_tools,
        max_tokens=4096
    )

    return {"research": result.content, "tools_used": result.tool_calls}

单步采样：

@mcp.tool()
async def get_single_response(prompt: str, context: Context) -> dict:
    """Get a single LLM response without tool loop."""

    # sample_step() returns SampleStep for inspection
    step = await context.sample_step(
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7
    )

    return {
        "content": step.content,
        "model": step.model,
        "stop_reason": step.stop_reason
    }

采样处理器：

• AnthropicSamplingHandler - 用于 Claude 模型（v2.14.1+）
• OpenAISamplingHandler - 用于 GPT 模型

已知限制： 客户端连接到单个服务时，ctx.sample() 可以正常工作；但客户端配置了多个服务时，会失败并报错“Sampling not supported”。不使用采样的工具可以正常工作。（社区提供的发现）

存储后端

基于 py-key-value-aio 构建，用于存储 OAuth 令牌、缓存响应和持久化状态。

可用后端：

• 内存（默认）：临时存储、速度快，仅用于开发
• 磁盘：持久化存储，使用 FernetEncryptionWrapper 加密，适配平台（Mac/Windows 默认使用）
• Redis：分布式，适用于生产环境和多实例
• 其他：DynamoDB、MongoDB、Elasticsearch、Memcached、RocksDB、Valkey

基本用法：

from key_value.stores import DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

# Disk (persistent, single instance)
mcp = FastMCP("Server", storage=DiskStore(path="/app/data/storage"))

# Redis (distributed, production)
mcp = FastMCP("Server", storage=RedisStore(
    host=os.getenv("REDIS_HOST"), password=os.getenv("REDIS_PASSWORD")
))

# Encrypted storage (recommended)
mcp = FastMCP("Server", storage=FernetEncryptionWrapper(
    key_value=DiskStore(path="/app/data"),
    fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
))

**平台默认设置：**Mac/Windows 使用磁盘，Linux 使用内存。可通过 storage 参数覆盖。

服务生命周期

⚠️ v2.13.0 中的破坏性变更：生命周期行为从每个会话一次改为每个服务实例一次。

资源仅初始化/清理一次，而非每个会话一次——这对数据库连接、API 客户端至关重要。

from contextlib import asynccontextmanager
from dataclasses import dataclass

@dataclass
class AppContext:
    db: Database
    api_client: httpx.AsyncClient

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Runs ONCE per server instance."""
    db = await Database.connect(os.getenv("DATABASE_URL"))
    api_client = httpx.AsyncClient(base_url=os.getenv("API_BASE_URL"), timeout=30.0)

    try:
        yield AppContext(db=db, api_client=api_client)
    finally:
        await db.disconnect()
        await api_client.aclose()

mcp = FastMCP("Server", lifespan=app_lifespan)

# Access in tools
@mcp.tool()
async def query_db(sql: str, context: Context) -> list:
    app_ctx = context.fastmcp_context.lifespan_context
    return await app_ctx.db.query(sql)

ASGI 集成（FastAPI/Starlette）：

mcp = FastMCP("Server", lifespan=mcp_lifespan)
app = FastAPI(lifespan=mcp.lifespan)  # ✅ MUST pass lifespan!

状态管理：

context.fastmcp_context.set_state(key, value)  # Store
context.fastmcp_context.get_state(key, default=None)  # Retrieve

中间件系统

**8 种内置类型：**TimingMiddleware、ResponseCachingMiddleware、LoggingMiddleware、RateLimitingMiddleware、ErrorHandlingMiddleware、ToolInjectionMiddleware、PromptToolMiddleware、ResourceToolMiddleware

执行顺序（顺序很重要！）：

Request Flow:
  → ErrorHandlingMiddleware (catches errors)
    → TimingMiddleware (starts timer)
      → LoggingMiddleware (logs request)
        → RateLimitingMiddleware (checks rate limit)
          → ResponseCachingMiddleware (checks cache)
            → Tool/Resource Handler

基本用法：

from fastmcp.middleware import ErrorHandlingMiddleware, TimingMiddleware, LoggingMiddleware

mcp.add_middleware(ErrorHandlingMiddleware())  # First: catch errors
mcp.add_middleware(TimingMiddleware())         # Second: time requests
mcp.add_middleware(LoggingMiddleware(level="INFO"))
mcp.add_middleware(RateLimitingMiddleware(max_requests=100, window_seconds=60))
mcp.add_middleware(ResponseCachingMiddleware(ttl_seconds=300, storage=RedisStore()))

自定义中间件：

from fastmcp.middleware import BaseMiddleware

class AccessControlMiddleware(BaseMiddleware):
    async def on_call_tool(self, tool_name, arguments, context):
        user = context.fastmcp_context.get_state("user_id")
        if user not in self.allowed_users:
            raise PermissionError(f"User not authorized")
        return await self.next(tool_name, arguments, context)

钩子层级：on_message（全部）→ on_request/on_notification → on_call_tool/on_read_resource/on_get_prompt → on_list_*（列表操作）

服务组合

两种策略：

1. import_server() - 静态快照：导入时复制一次，变更不会传播，速度快（无运行时委托）。适用于：已定稿的组件包。

2. mount() - 动态链接：实时运行时链接，变更立即可见，使用运行时委托（速度较慢）。适用于：模块化运行时组合。

基本用法：

# Import (static)
main_server.import_server(api_server)  # One-time copy

# Mount (dynamic)
main_server.mount(api_server, prefix="api")  # Tools: api.fetch_data
main_server.mount(db_server, prefix="db")    # Resources: resource://db/path

标签筛选：

@api_server.tool(tags=["public"])
def public_api(): pass

main_server.import_server(api_server, include_tags=["public"])  # Only public
main_server.mount(api_server, prefix="api", exclude_tags=["admin"])  # No admin

资源前缀格式：

• 路径（自 v2.4.0 起为默认）：resource://prefix/path
• 协议（旧版）：prefix+resource://path

main_server.mount(subserver, prefix="api", resource_prefix_format="path")

OAuth 与身份验证

4 种身份验证模式：

1. 令牌验证（JWTVerifier）：验证外部令牌
2. 外部身份提供商（RemoteAuthProvider）：支持 DCR 的 OAuth 2.0/OIDC
3. OAuth 代理（OAuthProxy）：连接不支持 DCR 的提供商（GitHub、Google、Azure、AWS、Discord、Facebook）
4. 完整 OAuth（OAuthProvider）：完整的授权服务器

模式 1：令牌验证

from fastmcp.auth import JWTVerifier

auth = JWTVerifier(issuer="https://auth.example.com", audience="my-server",
                   public_key=os.getenv("JWT_PUBLIC_KEY"))
mcp = FastMCP("Server", auth=auth)

模式 3：OAuth 代理（生产环境）

from fastmcp.auth import OAuthProxy
from key_value.stores import RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

auth = OAuthProxy(
    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
    client_storage=FernetEncryptionWrapper(
        key_value=RedisStore(host=os.getenv("REDIS_HOST"), password=os.getenv("REDIS_PASSWORD")),
        fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
    ),
    upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
    upstream_token_endpoint="https://github.com/login/oauth/access_token",
    upstream_client_id=os.getenv("GITHUB_CLIENT_ID"),
    upstream_client_secret=os.getenv("GITHUB_CLIENT_SECRET"),
    enable_consent_screen=True  # CRITICAL: Prevents confused deputy attacks
)
mcp = FastMCP("GitHub Auth", auth=auth)

**OAuth 代理功能：**令牌工厂模式（签发自己的 JWT）、同意屏幕（防止绕过）、支持 PKCE、RFC 7662 令牌内省

**支持的提供商：**GitHub、Google、Azure、AWS Cognito、Discord、Facebook、WorkOS、AuthKit、Descope、Scalekit、OCI（v2.13.1）

Supabase 提供商（v2.14.2+）：

from fastmcp.auth import SupabaseProvider

auth = SupabaseProvider(
    auth_route="/custom-auth",  # Custom auth route (new in v2.14.2)
    # ... other config
)

图标、API 集成、云端部署

**图标：**可添加到服务器、工具、资源和提示词。使用 Icon(url, size)，通过 Icon.from_file() 或 Image.to_data_uri() 使用数据 URI（v2.13.1）。

API 集成（3 种模式）：

1. 手动：使用带有 base_url/headers/timeout 的 httpx.AsyncClient
2. OpenAPI 自动生成：FastMCP.from_openapi(spec, client, route_maps) - GET→资源/模板，POST/PUT/DELETE→工具
3. FastAPI 转换：FastMCP.from_fastapi(app, httpx_client_kwargs)

云端部署的关键要求：

1. ❗ 模块级服务器，名称为 mcp、server 或 app
2. requirements.txt 中仅包含 PyPI 依赖项
3. 公开的 GitHub 仓库（或可访问的仓库）
4. 使用环境变量进行配置

# ✅ CORRECT: Module-level export
mcp = FastMCP("server")  # At module level!

# ❌ WRONG: Function-wrapped
def create_server():
    return FastMCP("server")  # Too late for cloud!

**部署：**https://fastmcp.cloud → 登录 → 创建项目 → 选择仓库 → 部署

客户端配置（Claude Desktop）：

{"mcpServers": {"my-server": {"url": "https://project.fastmcp.app/mcp", "transport": "http"}}}

30 个常见错误（及解决方案）

错误 1：缺少服务器对象

错误：RuntimeError: No server object found at module level **原因：**服务器未在模块级导出（FastMCP Cloud 的要求） **解决方案：**在模块级定义 mcp = FastMCP("server")，不要将其放在函数内部

错误 2：Async/Await 混淆

错误：RuntimeError: no running event loop、TypeError: object coroutine can't be used in 'await' **原因：**错误地混用同步和异步代码 **解决方案：**对带有 await 的工具使用 async def，对非异步代码使用同步的 def

错误 3：未注入上下文

错误：TypeError: missing 1 required positional argument: 'context' **原因：**缺少 Context 类型注解 解决方案：async def tool(context: Context) - 必须添加类型提示！

错误 4：资源 URI 语法

错误：ValueError: Invalid resource URI: missing scheme **原因：**资源 URI 缺少方案前缀 **解决方案：**使用 @mcp.resource("data://config")，而不是 @mcp.resource("config")

错误 5：资源模板参数不匹配

错误：TypeError: get_user() missing 1 required positional argument **原因：**函数参数名称与 URI 模板不匹配 解决方案：@mcp.resource("user://{user_id}/profile") → def get_user(user_id: str) - 名称必须完全匹配

---

错误 6：Pydantic 验证错误

错误：ValidationError: value is not a valid integer **原因：**类型提示与提供的数据不匹配 **解决方案：**使用 Pydantic 模型：class Params(BaseModel): query: str = Field(min_length=1)

错误 7：传输方式/协议不匹配

错误：ConnectionError: Server using different transport **原因：**客户端和服务器使用了不兼容的传输方式 **解决方案：**确保传输方式匹配 - stdio：mcp.run() + {"command": "python", "args": ["server.py"]}，HTTP：mcp.run(transport="http", port=8000) + {"url": "http://localhost:8000/mcp", "transport": "http"}

HTTP 超时问题（已在 v2.14.3 中修复）：

• HTTP 传输的默认超时时间错误地设为 5 秒，而非 MCP 默认的 30 秒（GitHub Issue #2845）
• 在 v2.14.2 及更早版本中，执行时间超过 5 秒的工具会静默失败
• 解决方案：升级到 fastmcp>=2.14.3（超时时间现在遵循 MCP 的 30 秒默认值）

错误 8：导入错误（可编辑包）

错误： ModuleNotFoundError: No module named 'my_package' 原因： 包未正确安装 解决方案： pip install -e .，或使用绝对导入，或执行 export PYTHONPATH="/path/to/project"

错误 9：弃用警告

错误： DeprecationWarning: 'mcp.settings' is deprecated 原因： 使用旧版 FastMCP v1 API 解决方案： 使用 os.getenv("API_KEY") 替代 mcp.settings.get("API_KEY")

错误 10：端口已被占用

错误： OSError: [Errno 48] Address already in use 原因： 端口 8000 已被占用 解决方案： 使用其他端口 --port 8001，或终止进程 lsof -ti:8000 | xargs kill -9

错误 11：架构生成失败

错误： TypeError: Object of type 'ndarray' is not JSON serializable 原因： 类型提示不受支持（NumPy 数组、自定义类） 解决方案： 返回兼容 JSON 的类型：list[float]，或进行转换：{"values": np_array.tolist()}

不支持自定义类（来自社区）： FastMCP 支持所有兼容 Pydantic 的类型，但工具返回值中的自定义类必须转换为字典或 Pydantic 模型：

# ❌ NOT SUPPORTED
class MyCustomClass:
    def __init__(self, value: str):
        self.value = value

@mcp.tool()
async def get_custom() -> MyCustomClass:
    return MyCustomClass("test")  # Serialization error

# ✅ SUPPORTED - Use dict or Pydantic
@mcp.tool()
async def get_custom() -> dict[str, str]:
    obj = MyCustomClass("test")
    return {"value": obj.value}

# OR use Pydantic BaseModel
from pydantic import BaseModel
class MyModel(BaseModel):
    value: str

@mcp.tool()
async def get_model() -> MyModel:
    return MyModel(value="test")  # Works!

OutputSchema $ref 解析（已在 v2.14.2 中修复）：

• outputSchema 中根级别的 $ref 未被解引用（GitHub Issue #2720）
• 导致不符合 MCP 规范，并引发客户端兼容性问题
• 解决方案：升级到 fastmcp>=2.14.2（自动解引用 $ref）

错误 12：JSON 序列化

错误： TypeError: Object of type 'datetime' is not JSON serializable 原因： 返回了无法进行 JSON 序列化的对象 解决方案： 转换：datetime.now().isoformat()，字节：.decode('utf-8')

错误 13：循环导入错误

错误： ImportError: cannot import name 'X' from partially initialized module 原因： 存在循环依赖（云部署中很常见） 解决方案： 在 __init__.py 中使用直接导入：from .api_client import APIClient，或在函数中使用延迟导入

错误 14：Python 版本兼容性

错误： DeprecationWarning: datetime.utcnow() is deprecated 原因： 使用了 Python 3.12+ 中已弃用的方法 解决方案： 使用 datetime.now(timezone.utc) 替代 datetime.utcnow()

错误 15：导入时执行

错误： RuntimeError: Event loop is closed 原因： 在模块导入时创建了异步资源 解决方案： 使用延迟初始化——创建带有异步 connect() 方法的连接类，并在需要时于工具中调用

---

错误 16：未配置存储后端

错误： RuntimeError: OAuth tokens lost on restart、ValueError: Cache not persisting 原因： 在生产环境中使用默认的内存存储，且未启用持久化 解决方案： 使用加密的 DiskStore（单实例）或 RedisStore（多实例），并搭配 FernetEncryptionWrapper

错误 17：未将 Lifespan 传递给 ASGI 应用

错误： RuntimeError: Database connection never initialized、Warning: MCP lifespan hooks not running 原因： 使用 FastMCP 和 FastAPI/Starlette 时未传递 lifespan（v2.13.0 的要求） 解决方案： app = FastAPI(lifespan=mcp.lifespan)——必须传递 lifespan！

错误 18：中间件执行顺序错误

错误： RuntimeError: Rate limit not checked before caching 原因： 中间件顺序不正确（顺序很重要！） 解决方案： ErrorHandling → Timing → Logging → RateLimiting → ResponseCaching（按此顺序）

错误 19：中间件循环依赖

错误： RecursionError: maximum recursion depth exceeded 原因： 中间件未调用 self.next()，或调用方式不正确 解决方案： 始终在中间件钩子中调用 result = await self.next(tool_name, arguments, context)

错误 20：导入与挂载混淆

错误： RuntimeError: Subserver changes not reflected、ValueError: Unexpected tool namespacing 原因： 需要使用 mount() 时却使用了 import_server()（或反之） 解决方案： 静态捆绑使用 import_server()（一次性复制），动态组合使用 mount()（实时链接）

错误 21：资源前缀格式不匹配

错误： ValueError: Resource not found: resource://api/users 原因： 资源前缀格式错误 解决方案： 路径格式（默认，v2.4.0+）：resource://prefix/path；协议格式（旧版）：prefix+resource://path —— 使用 resource_prefix_format="path" 设置

错误 22：OAuth 代理未启用同意屏幕

错误： SecurityWarning: Authorization bypass possible 原因： OAuth 代理未启用同意屏幕（安全漏洞） 解决方案： 始终设置 enable_consent_screen=True，以防止混淆代理攻击（严重）

错误 23：生产环境缺少 JWT 签名密钥

错误： ValueError: JWT signing key required for OAuth Proxy 原因： OAuth 代理缺少 jwt_signing_key 解决方案： 生成密钥：secrets.token_urlsafe(32)，将其存储在 FASTMCP_JWT_SIGNING_KEY 环境变量中，并传递给 OAuthProxy(jwt_signing_key=...)

错误 24：图标数据 URI 格式错误

错误： ValueError: Invalid data URI format 原因： 图标的数据 URI 格式不正确 解决方案： 使用 Icon.from_file("/path/icon.png", size="medium") 或 Image.to_data_uri()（v2.13.1），不要手动设置格式

错误 25：生命周期行为变更（v2.13.0）

错误： Warning: Lifespan runs per-server, not per-session 原因： 在 v2.13.0+ 中仍预期 v2.12 的行为（每个会话运行一次），而实际行为已变为每个服务运行一次 解决方案： v2.13.0+ 中，生命周期在每个服务中只运行一次，而非每个会话运行一次；每个会话的逻辑请使用中间件

错误 26：BearerAuthProvider 已移除（v2.14.0）

错误： ImportError: cannot import name 'BearerAuthProvider' from 'fastmcp.auth' 原因： BearerAuthProvider 模块已在 v2.14.0 中移除 解决方案： 使用 JWTVerifier 验证令牌，或使用 OAuthProxy 实现完整的 OAuth 流程：

# Before (v2.13.x)
from fastmcp.auth import BearerAuthProvider

# After (v2.14.0+)
from fastmcp.auth import JWTVerifier
auth = JWTVerifier(issuer="...", audience="...", public_key="...")

错误 27：Context.get_http_request() 已移除（v2.14.0）

错误： AttributeError: 'Context' object has no attribute 'get_http_request' 原因： Context.get_http_request() 方法已在 v2.14.0 中移除 解决方案： 通过中间件访问请求信息，或使用向中间件公开的 InitializeResult

错误 28：Image 导入路径已变更（v2.14.0）

错误： ImportError: cannot import name 'Image' from 'fastmcp' 原因： fastmcp.Image 顶层导入已在 v2.14.0 中移除 解决方案： 使用新的导入路径：

# Before (v2.13.x)
from fastmcp import Image

# After (v2.14.0+)
from fastmcp.utilities import Image

错误 29：FastAPI 挂载路径重复

错误： 客户端无法连接到 /mcp 端点，收到 404 错误 来源： GitHub Issue #2961 原因： 将 FastMCP 挂载到 /mcp 会因路径前缀重复而创建 /mcp/mcp 端点 解决方案： 挂载到根路径 /，或调整客户端配置

# ❌ WRONG - Creates /mcp/mcp endpoint
from fastapi import FastAPI
from fastmcp import FastMCP

mcp = FastMCP("server")
app = FastAPI(lifespan=mcp.lifespan)
app.mount("/mcp", mcp)  # Endpoint becomes /mcp/mcp

# ✅ CORRECT - Mount at root
app.mount("/", mcp)  # Endpoint is /mcp

# ✅ OR adjust client config
# In claude_desktop_config.json:
{"url": "http://localhost:8000/mcp/mcp", "transport": "http"}

重要： 还必须将 lifespan=mcp.lifespan 传递给 FastAPI（参见错误 #17）。

错误 30：后台任务因“No Active Context”而失败（ASGI 挂载）

错误： RuntimeError: No active context found 来源： GitHub Issue #2877 原因： FastMCP 挂载到 FastAPI/Starlette 且使用后台任务（task=True）时，ContextVar 传播出现问题 解决方案： 升级到 fastmcp>=2.14.3

# In v2.14.2 and earlier - FAILS
from fastapi import FastAPI
from fastmcp import FastMCP, Context

mcp = FastMCP("server")
app = FastAPI(lifespan=mcp.lifespan)

@mcp.tool(task=True)
async def sample(name: str, ctx: Context) -> dict:
    # RuntimeError: No active context found
    await ctx.report_progress(1, 1, "Processing")
    return {"status": "OK"}

app.mount("/", mcp)

# ✅ FIXED in v2.14.3
# pip install fastmcp>=2.14.3

注意： 此问题与错误 #17（未向 ASGI 应用传递生命周期）相关。

---

生产模式、测试与 CLI

4 种生产模式：

1. 工具模块：使用单个 utils.py，包含 Config 类和 format_success/error 辅助函数
2. 连接池：使用单例 httpx.AsyncClient，并提供 get_client() 类方法
3. 带退避的重试：retry_with_backoff(func, max_retries=3, initial_delay=1.0, exponential_base=2.0)
4. 基于时间的缓存：TimeBasedCache(ttl=300)，提供 .get() 和 .set() 方法

测试：

• 单元测试：pytest + create_test_client(test_server) + await client.call_tool()
• 集成测试：Client("server.py") + list_tools() + call_tool() + list_resources()

CLI 命令：

fastmcp dev server.py                # Run with inspector
fastmcp install server.py             # Install to Claude Desktop
FASTMCP_LOG_LEVEL=DEBUG fastmcp dev  # Debug logging

**最佳实践：**采用工厂模式并在模块级导出，使用经过验证的环境配置，编写全面的文档字符串（大模型会读取这些内容！），提供健康检查资源

项目结构：

• 简单项目：server.py、requirements.txt、.env、README.md
• 生产项目：src/（server.py、utils.py、tools/、resources/、prompts/）、tests/、pyproject.toml

---

参考资料与总结

**官方资源：**https://github.com/jlowin/fastmcp、https://fastmcp.cloud、https://modelcontextprotocol.io，Context7：`/jlowin/fastmcp` **相关技能：**openai-api、claude-api、cloudflare-worker-base、typescript-mcp **软件包版本：**fastmcp>=2.14.2（PyPI）、Python>=3.10（v2.14.1+ 支持 3.13）、httpx、pydantic、py-key-value-aio、cryptography 最后更新：2026-01-21

17 项关键要点：

1. 在模块级导出服务器（FastMCP Cloud）
2. 使用持久化存储（Disk/Redis）处理 OAuth/缓存
3. 使用服务器生命周期管理资源
4. 中间件顺序：错误 → 计时 → 日志记录 → 速率限制 → 缓存
5. 组合方式：import_server()（静态）与 mount()（动态）
6. OAuth 安全：同意页面 + 加密存储 + JWT 签名
7. 正确使用 async/await（不要阻塞事件循环）
8. 采用结构化错误处理
9. 避免循环导入
10. 在本地测试（fastmcp dev）
11. 使用环境变量（绝不硬编码机密）
12. 编写全面的文档字符串（大模型会读取！）
13. 采用生产模式（工具、连接池、重试、缓存）
14. 自动生成 OpenAPI
15. 健康检查 + 监控
16. 使用后台任务处理长时间运行的操作（task=True）
17. 使用工具进行采样以支持 Agent 工作流（ctx.sample(tools=[...])）

**生产就绪能力：**加密存储、4 种身份验证模式、8 种中间件类型、模块化组合、OAuth 安全（同意页面、PKCE、RFC 7662）、响应缓存、连接池、计时中间件、后台任务、Agent 采样、FastAPI/Starlette 挂载、v3.0 提供方架构

可避免 30 多个错误。节省 90%–95% 的 Token。