name: fastmcp description: FastMCP Python 框架,用于构建具有工具、资源、存储后端(内存/磁盘/Redis/DynamoDB)的 MCP 服务器。适用于 Claude 工具暴露、OAuth 代理、云部署,或遇到存储、生命周期、中间件、循环导入、异步错误时使用。
关键词:FastMCP, MCP 服务器 Python, Model Context Protocol Python, fastmcp 框架, mcp 工具, mcp 资源, mcp 提示, fastmcp 存储, fastmcp 内存存储, fastmcp 磁盘存储, fastmcp redis, fastmcp dynamodb, fastmcp 生命周期, fastmcp 中间件, fastmcp oauth 代理, 服务器组合 mcp, fastmcp 导入, fastmcp 挂载, fastmcp 云, fastmcp 部署, mcp 认证, fastmcp 图标, openapi mcp, claude mcp 服务器, fastmcp 测试, 存储配置错误, 生命周期问题, 中间件顺序, 循环导入, 模块级服务器, 异步等待 mcp license: MIT metadata: version: “2.0.0” package_version: “fastmcp>=2.13.0” python_version: “>=3.10” token_savings: “90-95%” errors_prevented: 25 production_tested: true last_updated: “2025-11-18”
FastMCP - 用 Python 构建 MCP 服务器
FastMCP 是一个 Python 框架,用于构建模型上下文协议 (MCP) 服务器,将工具、资源和提示暴露给像 Claude 这样的大型语言模型。
快速开始
安装
pip install fastmcp
# 或: uv pip install fastmcp
最小化服务器
from fastmcp import FastMCP
# 必须放在模块级别以用于 FastMCP Cloud
mcp = FastMCP("我的服务器")
@mcp.tool()
async def hello(name: str) -> str:
"""向某人问好。"""
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run()
运行:
python server.py # 本地开发
fastmcp dev server.py # 使用 FastMCP CLI
python server.py --transport http --port 8000 # HTTP 模式
复制粘贴模板: 参见 templates/basic-server.py
核心概念
工具
LLM 可以调用的函数:
@mcp.tool()
def calculate(operation: str, a: float, b: float) -> float:
"""执行数学运算。"""
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)
最佳实践:
- 清晰、描述性的函数名称
- 全面的文档字符串(LLM 会读取这些!)
- 强类型提示(Pydantic 自动验证)
- 返回结构化数据(字典/列表)
- 优雅处理错误
资源
暴露静态或动态数据:
@mcp.resource("data://config")
def get_config() -> dict:
"""提供应用程序配置。"""
return {"version": "1.0.0", "features": ["auth", "api"]}
# 带参数的动态资源
@mcp.resource("user://{user_id}/profile")
async def get_user_profile(user_id: str) -> dict:
"""通过 ID 获取用户配置文件。"""
return {"id": user_id, "name": f"User {user_id}"}
URI 方案: data://, file://, resource://, info://, api://, 或自定义
提示
为 LLM 预配置的提示:
@mcp.prompt("analyze")
def analyze_prompt(topic: str) -> str:
"""生成分析提示。"""
return f"""分析 {topic},考虑:
1. 当前状态
2. 挑战
3. 机会
4. 建议"""
上下文功能
进度跟踪:
from fastmcp import Context
@mcp.tool()
async def batch_process(items: list, context: Context) -> dict:
"""处理项目并更新进度。"""
for i, item in enumerate(items):
await context.report_progress(i + 1, len(items), f"处理 {item}")
await process_item(item)
return {"processed": len(items)}
用户输入:
@mcp.tool()
async def confirm_action(action: str, context: Context) -> dict:
"""执行操作前请求用户确认。"""
confirmed = await context.request_elicitation(
prompt=f"确认 {action}? (是/否)",
response_type=str
)
return {"confirmed": confirmed.lower() == "是"}
存储后端
根据部署选择存储:
from key_value.stores import DiskStore, RedisStore
from key_value.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# 内存(默认)- 仅用于开发
mcp = FastMCP("开发服务器")
# 磁盘 - 单实例
mcp = FastMCP(
"生产服务器",
storage=FernetEncryptionWrapper(
key_value=DiskStore(path="/var/lib/mcp/storage"),
fernet=Fernet(os.getenv("STORAGE_ENCRYPTION_KEY"))
)
)
# Redis - 多实例
mcp = FastMCP(
"生产服务器",
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):
"""服务器启动时运行一次(v2.13.0+)。"""
db = await Database.connect()
print("服务器启动")
try:
yield {"db": db}
finally:
await db.disconnect()
print("服务器停止")
mcp = FastMCP("我的服务器", lifespan=app_lifespan)
关键点: v2.13.0+ 生命周期每服务器运行(非每会话)。对于每会话逻辑,使用中间件。
中间件系统
8 种内置中间件类型:
from fastmcp.middleware import (
LoggingMiddleware,
TimingMiddleware,
RateLimitingMiddleware,
ResponseCachingMiddleware
)
# 顺序重要!
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"之前: {tool_name}")
result = await self.next(tool_name, arguments, context) # 必须调用 next()
print(f"之后: {tool_name}")
return result
服务器组合
导入服务器(静态,一次性复制):
main_server.import_server(vendor_server) # 静态捆绑
挂载服务器(动态,运行时委托):
main_server.mount(api_server, prefix="api") # 更改立即生效
云部署
FastMCP Cloud 要求:
- 服务器必须放在模块级别
- 使用磁盘/Redis 存储(非内存)
- 无导入时执行
# ✅ 云就绪模式
mcp = FastMCP("我的服务器") # 模块级别
@mcp.tool()
async def my_tool(): pass
if __name__ == "__main__":
mcp.run()
部署:
fastmcp deploy server.py
五大关键错误
1. 缺少服务器对象
错误: RuntimeError: No server object found at module level
修复:
# ❌ 错误
def create_server():
return FastMCP("服务器")
# ✅ 正确
mcp = FastMCP("服务器") # 放在模块级别
2. 异步/等待混淆
错误: RuntimeError: no running event loop
修复:
# ❌ 错误: 同步函数调用异步
@mcp.tool()
def bad_tool():
result = await async_function() # 错误!
# ✅ 正确: 异步工具
@mcp.tool()
async def good_tool():
result = await async_function()
return result
3. 上下文未注入
错误: TypeError: missing 1 required positional argument: 'context'
修复:
from fastmcp import Context
# ❌ 错误: 无类型提示
@mcp.tool()
async def bad_tool(context): # 缺少类型!
await context.report_progress(...)
# ✅ 正确: 正确类型提示
@mcp.tool()
async def good_tool(context: Context):
await context.report_progress(0, 100, "开始")
4. 存储后端未配置
错误: RuntimeError: OAuth tokens lost on restart
修复: 在生产中使用磁盘或 Redis 存储(见上方存储后端部分)
5. 循环导入错误
错误: ImportError: cannot import name 'X' from partially initialized module
修复:
# ❌ 错误: 工厂函数导致循环依赖
# shared/__init__.py
def get_client():
from .api_client import APIClient # 循环!
return APIClient()
# ✅ 正确: 直接导入
# 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 # 开发模式,热重载
fastmcp run server.py # 生产模式
fastmcp deploy server.py # 部署到 FastMCP Cloud
fastmcp test server.py # 运行测试
最佳实践
- 服务器结构: 保持模块级服务器,在单独文件中组织工具
- 类型提示: 使用 Pydantic 模型进行复杂验证
- 文档: 编写详细文档字符串(LLM 会读取!)
- 错误处理: 捕获并返回结构化错误
- 存储: 在生产中使用加密磁盘/Redis
- 生命周期: 每服务器初始化连接一次
- 中间件: 顺序重要(错误处理 → 计时 → 日志 → 速率限制 → 缓存)
- 测试: 使用
FastMCP.test_tool()单元测试工具
捆绑资源
参考资料 (references/):
cli-commands.md- 完整 CLI 命令参考(dev, run, deploy, test)cloud-deployment.md- FastMCP Cloud 部署指南,含模块级要求common-errors.md- 所有 25 个文档化错误及解决方案和预防context-features.md- 进度跟踪、用户输入和上下文 API 模式error-catalog.md- 全面错误目录,含修复方法integration-patterns.md- 服务器组合(导入/挂载)、OAuth 代理、OpenAPIproduction-patterns.md- 存储后端、生命周期、中间件、架构模式
模板 (templates/):
basic-server.py- 最小化 MCP 服务器,含工具、资源、提示client-example.py- MCP 客户端集成示例api-client-pattern.py- API 集成模式error-handling.py- 错误处理最佳实践openapi-integration.py- OpenAPI 模式集成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", // 用于存储后端
"cryptography": ">=41.0.0", // 用于加密
"redis": ">=5.0.0" // 用于 Redis 存储
}
}
官方文档
- FastMCP GitHub: https://github.com/jlowin/fastmcp
- MCP 协议: https://modelcontextprotocol.io
- FastMCP Cloud: https://fastmcp.com
验证清单
- [ ] 已安装 FastMCP (
fastmcp>=2.13.0) - [ ] 服务器对象放在模块级别
- [ ] 工具文档字符串全面
- [ ] 上下文参数有类型提示
- [ ] 资源 URI 有方案
- [ ] 存储后端已配置(生产)
- [ ] 生命周期模式正确(v2.13.0+)
- [ ] 中间件顺序合理
- [ ] 客户端配置已测试
- [ ] 生产部署成功
令牌节省: 90-95% 对比从头学习 错误预防: 25 个文档化问题 生产测试: ✅ 多次部署