fastmcp

FastMCP - 使用 Python 构建 MCP 服务

FastMCP 是一个 Python 框架，用于构建 Model Context Protocol（MCP）服务，向 Claude 等大模型提供工具、资源和提示词。

快速开始

安装

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()

运行：

python server.py              # Local development
fastmcp dev server.py         # With FastMCP CLI
python server.py --transport http --port 8000  # HTTP mode

复制粘贴模板：参见 templates/basic-server.py

核心概念

工具

大模型可以调用的函数：

@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
    """Perform mathematical operations."""
    operations = {
        "add": lambda x, y: x + y,
        "subtract": lambda x, y: x - y,
        "multiply": lambda x, y: x * y,
        "divide": lambda x, y: x / y if y != 0 else None
    }
    return operations.get(operation, lambda x, y: None)(a, b)

最佳实践：

• 使用清晰、描述性强的函数名称
• 编写全面的文档字符串（大模型会读取这些内容！）
• 使用严格的类型提示（Pydantic 会自动验证）
• 返回结构化数据（字典/列表）
• 妥善处理错误

资源

公开静态或动态数据：

@mcp.resource("data://config")
def get_config() -> dict:
    """Provide application configuration."""
    return {"version": "1.0.0", "features": ["auth", "api"]}

# Dynamic resource with parameters
@mcp.resource("user://{user_id}/profile")
async def get_user_profile(user_id: str) -> dict:
    """Get user profile by ID."""
    return {"id": user_id, "name": f"User {user_id}"}

URI 方案：data://、file://、resource://、info://、api:// 或自定义方案

提示词

为大模型预先配置的提示词：

@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
    """Generate analysis prompt."""
    return f"""Analyze {topic} considering:
    1. Current state
    2. Challenges
    3. Opportunities
    4. Recommendations"""

上下文功能

进度跟踪：

from fastmcp import Context

@mcp.tool()
async def batch_process(items: list, context: Context) -> dict:
    """Process items with progress updates."""
    for i, item in enumerate(items):
        await context.report_progress(i + 1, len(items), f"Processing {item}")
        await process_item(item)
    return {"processed": len(items)}

用户输入：

@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
    """Perform action with user confirmation."""
    confirmed = await context.request_elicitation(
        prompt=f"Confirm {action}? (yes/no)",
        response_type=str
    )
    return {"confirmed": confirmed.lower() == "yes"}

存储后端

根据部署方式选择存储：

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

# Memory (default) - Development only
mcp = FastMCP("Dev Server")

# Disk - Single instance
mcp = FastMCP(
    "Production Server",
    storage=FernetEncryptionWrapper(
        key_value=DiskStore(path="/var/lib/mcp/storage"),
        fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
    )
)

# Redis - Multi-instance
mcp = FastMCP(
    "Production Server",
    storage=FernetEncryptionWrapper(
        key_value=RedisStore(
            host=os.getenv("REDIS_HOST"),
            password=os.getenv("REDIS_PASSWORD")
        ),
        fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
    )
)

服务生命周期

在服务启动时初始化资源：

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan(server: FastMCP):
    """Runs ONCE when server starts (v2.13.0+)."""
    db = await Database.connect()
    print("Server starting")
    
    try:
        yield {"db": db}
    finally:
        await db.disconnect()
        print("Server stopping")

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

重要：v2.13.0 及更高版本的生命周期按服务运行（而非按会话运行）。对于按会话运行的逻辑，请使用中间件。

中间件系统

8 种内置中间件类型：

from fastmcp.middleware import (
    LoggingMiddleware,
    TimingMiddleware,
    RateLimitingMiddleware,
    ResponseCachingMiddleware
)

# Order matters!
mcp.add_middleware(LoggingMiddleware())
mcp.add_middleware(TimingMiddleware())
mcp.add_middleware(RateLimitingMiddleware(max_requests=100, window_seconds=60))
mcp.add_middleware(ResponseCachingMiddleware(ttl_seconds=3600))

自定义中间件：

from fastmcp.middleware import BaseMiddleware

class CustomMiddleware(BaseMiddleware):
    async def on_call_tool(self, tool_name, arguments, context):
        print(f"Before: {tool_name}")
        result = await self.next(tool_name, arguments, context)  # MUST call next()
        print(f"After: {tool_name}")
        return result

服务组合

导入服务（静态，一次性复制）：

main_server.import_server(vendor_server)  # Static bundle

挂载服务（动态，运行时委派）：

main_server.mount(api_server, prefix="api")  # Changes appear immediately

云端部署

FastMCP Cloud 要求：

1. 服务 MUST 位于模块级别
2. 使用磁盘/Redis 存储（而非内存）
3. 不要在导入时执行代码

# ✅ Cloud-ready pattern
mcp = FastMCP("My Server")  # Module level

@mcp.tool()
async def my_tool(): pass

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

部署：

fastmcp deploy server.py

5 大关键错误

1. 缺少服务对象

错误： RuntimeError: No server object found at module level

修复：

# ❌ WRONG
def create_server():
    return FastMCP("server")

# ✅ CORRECT
mcp = FastMCP("server")  # At module level

2. Async/Await 混淆

错误： RuntimeError: no running event loop

修复：

# ❌ WRONG: Sync function calling async
@mcp.tool()
def bad_tool():
    result = await async_function()  # Error!

# ✅ CORRECT: Async tool
@mcp.tool()
async def good_tool():
    result = await async_function()
    return result

3. 未注入上下文

错误： TypeError: missing 1 required positional argument: 'context'

修复：

from fastmcp import Context

# ❌ WRONG: No type hint
@mcp.tool()
async def bad_tool(context):  # Missing type!
    await context.report_progress(...)

# ✅ CORRECT: Proper type hint
@mcp.tool()
async def good_tool(context: Context):
    await context.report_progress(0, 100, "Starting")

4. 未配置存储后端

错误： RuntimeError: OAuth tokens lost on restart

修复： 在生产环境中使用磁盘或 Redis 存储（参见上文“存储后端”部分）

5. 循环导入错误

错误： ImportError: cannot import name 'X' from partially initialized module

修复：

# ❌ WRONG: Factory function creating circular dependency
# shared/__init__.py
def get_client():
    from .api_client import APIClient  # Circular!
    return APIClient()

# ✅ CORRECT: Direct imports
# shared/__init__.py
from .api_client import APIClient
from .cache import CacheManager

# shared/monitoring.py
from .api_client import APIClient
client = APIClient()

查看全部 25 个错误：references/error-catalog.md

客户端配置

Claude Desktop

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

Claude Code CLI

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["server.py"]
    }
  }
}

CLI 命令

fastmcp dev server.py              # Development mode with hot reload
fastmcp run server.py              # Production mode
fastmcp deploy server.py           # Deploy to FastMCP Cloud
fastmcp test server.py             # Run tests

最佳实践

1. 服务结构：将服务保留在模块级别，并把工具整理到独立文件中
2. 类型提示：对复杂验证使用 Pydantic 模型
3. 文档：编写详细的文档字符串（大模型会读取这些内容！）
4. 错误处理：捕获错误并返回结构化错误信息
5. 存储：在生产环境中使用加密的磁盘/Redis 存储
6. 生命周期：每个服务只初始化一次连接
7. 中间件：顺序很重要（错误处理 → 计时 → 日志记录 → 速率限制 → 缓存）
8. 测试：使用 FastMCP.test_tool() 对工具进行单元测试

捆绑资源

参考文档（references/）：

• cli-commands.md - 完整的 CLI 命令参考（dev、run、deploy、test）
• cloud-deployment.md - FastMCP Cloud 部署指南及模块级要求
• common-errors.md - 全部 25 个已记录错误及其解决方案和预防措施
• context-features.md - 进度跟踪、用户输入和 Context API 模式
• error-catalog.md - 全面的错误目录及修复方法
• integration-patterns.md - 服务组合（import/mount）、OAuth Proxy、OpenAPI
• production-patterns.md - 存储后端、生命周期、中间件和架构模式

模板（templates/）：

• basic-server.py - 含工具、资源和提示词的最简 MCP 服务
• client-example.py - MCP 客户端集成示例
• api-client-pattern.py - API 集成模式
• error-handling.py - 错误处理最佳实践
• openapi-integration.py - OpenAPI schema 集成
• prompts-examples.py - 提示词模板模式
• resources-examples.py - 资源 URI 模式和示例
• tools-examples.py - 工具定义模式
• self-contained-server.py - 完整的生产就绪型独立服务
• .env.example - 环境变量模板
• requirements.txt - Python 依赖项
• pyproject.toml - Python 项目配置

依赖项

{
  "dependencies": {
    "fastmcp": ">=2.13.0",
    "pydantic": ">=2.0.0"
  },
  "optionalDependencies": {
    "py-key-value-aio": ">=0.1.0",  // For storage backends
    "cryptography": ">=41.0.0",     // For encryption
    "redis": ">=5.0.0"              // For Redis storage
  }
}

官方文档

• FastMCP GitHub：https://github.com/jlowin/fastmcp
• MCP 协议：https://modelcontextprotocol.io
• FastMCP Cloud：https://fastmcp.com

验证清单

• 已安装 FastMCP（fastmcp>=2.13.0）
• 模块级别定义了服务对象
• 工具文档字符串内容全面
• Context 参数包含类型提示
• 资源 URI 包含 scheme
• 已配置存储后端（生产环境）
• 生命周期模式正确（v2.13.0+）
• 中间件顺序合理
• 已测试客户端配置
• 生产环境部署成功

Token 节省：相比从零开始学习，节省 90-95% 避免的错误：25 个已记录问题 生产环境测试：✅ 已多次部署