在 FastAPI 项目中集成 FastMCP：完整指南与生命周期管理

概述

在现代 AI 应用开发中，将模型上下文协议（MCP）服务与 Web 框架集成已成为构建智能应用的关键需求。本文详细介绍如何在 FastAPI 项目中集成 FastMCP 服务，重点讲解 FastAPI 的生命周期管理机制，并提供完整的配置示例和连接测试工具。通过本文指导，能够构建一个既具备 FastAPI 高性能特性，又拥有 MCP 强大工具扩展能力的完整应用。

项目结构

在深入代码实现之前，首先了解整个项目的组织结构。清晰的项目结构有助于维护和扩展：

project/
├── mcp_server.py      # MCP 服务定义
├── main.py           # FastAPI 主应用
├── lifespan.py       # 生命周期管理
└── test_mcp.py       # 连接测试工具

这种模块化设计使得各个组件职责分明，便于单独测试和维护。

1. MCP 服务定义

首先需要定义 MCP 服务。这是整个集成的基础，负责提供 AI 工具能力：

# mcp_server.py
from fastmcp import FastMCP# 创建 FastMCP 实例，这是所有工具的核心容器
mcp = FastMCP("任务规划工具")@mcp.tool(name="echo", description="返回输入文本")
def echo(text: str) -> str:return f"Echo: {text}"

这段代码创建了一个简单的 MCP 工具。虽然示例中只包含一个基础的 echo 工具，但在实际应用中，可以在这里添加各种复杂的 AI 工具，如数据分析、自然语言处理等功能。

2. 生命周期管理

接下来是集成的核心部分——生命周期管理。这部分确保应用在启动和关闭时能够正确处理资源：

# lifespan.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.logger import logger@asynccontextmanager
async def app_lifespan(app: FastAPI):"""应用生命周期管理"""logger.info("应用启动中...")# 初始化操作try:logger.info("服务初始化完成")except Exception as e:logger.error(f"启动失败: {e}")yieldlogger.info("应用关闭中...")def get_lifespan_with_mcp(mcp_app):"""合并应用生命周期"""@asynccontextmanager    async def combined_lifespan(app: FastAPI):async with app_lifespan(app): async with mcp_app.lifespan(app): yield return combined_lifespan

生命周期管理的关键在于将 FastAPI 的原生生命周期与 MCP 服务的生命周期优雅地合并，确保两者协同工作。

3. FastAPI 主应用

现在将所有组件整合到主应用中。这是整个项目的入口点：

# main.py
from fastapi import FastAPI
from lifespan import get_lifespan_with_mcp
from mcp_server import mcp# 创建 MCP 应用（不在创建时指定路径）
mcp_app = mcp.http_app(transport="streamable-http")# 创建 FastAPI 应用
app = FastAPI(title="任务规划系统",lifespan=get_lifespan_with_mcp(mcp_app)
)@app.get("/")
def hello():return {"message": "系统运行中"}# 挂载 MCP 应用（在此指定路径）
app.mount("/mcp", mcp_app)

通过 mount 方法，将 MCP 服务挂载到 FastAPI 应用的指定路径下，这样既保持了 API 的清晰结构，又实现了功能的完整集成。

4. 连接测试工具

为确保集成成功，提供一个测试工具来验证 MCP 服务的可用性：

# test_mcp.py
import asyncio
from fastmcp import Clientasync def test_mcp_connection():"""测试 MCP 服务连接"""async with Client("http://localhost:8000/mcp") as client:# 测试工具列表tools = await client.list_tools()print("可用工具:")for tool in tools:print(f"- {tool.name}: {tool.description}")# 测试具体工具if tools:result = await client.call_tool("echo", {"text": "测试消息"})print(f"工具测试结果: {result}")if __name__ == "__main__":asyncio.run(test_mcp_connection())

这个测试工具不仅验证连接是否成功，还展示如何在实际应用中使用 MCP 客户端调用工具。

FastAPI 生命周期管理详解

生命周期的重要性

在深入理解代码实现后，探讨生命周期管理的理论基础。生命周期管理允许在应用启动和关闭时执行关键操作，这对于生产级应用至关重要：

• 启动阶段：数据库连接、外部服务初始化、缓存预热
• 运行阶段：保持连接池、监控资源使用情况
• 关闭阶段：优雅释放资源、关闭连接、保存状态

执行流程

理解执行流程有助于调试和优化应用：

应用启动↓
执行 lifespan 启动代码↓
FastAPI 应用就绪↓
处理请求...↓
收到关闭信号↓
执行 lifespan 清理代码↓
应用完全关闭

这个流程确保资源在整个应用生命周期内得到妥善管理。

关键注意事项

1. 对象一致性

在实际部署中，确保对象引用的一致性是最常见的陷阱之一：

# 正确做法
mcp_app = mcp.http_app(transport="streamable-http")
app = FastAPI(lifespan=get_lifespan_with_mcp(mcp_app))
app.mount("/mcp", mcp_app)

确保传递给生命周期函数和挂载的是同一个 MCP 应用实例，避免创建多个实例导致的状态不一致问题。

2. 路径配置

路径配置是另一个需要注意的关键点：

# ❌ 错误：路径重复
mcp_app = mcp.http_app(path="/mcp")  # 创建时指定
app.mount("/mcp", mcp_app)          # 挂载时又指定# ✅ 正确：只在挂载时指定
mcp_app = mcp.http_app()            # 创建时不指定
app.mount("/mcp", mcp_app)          # 挂载时指定

这种配置方式避免路径冲突，确保路由的正确性。

部署与测试

启动服务

完成代码编写后，启动服务进行测试：

uvicorn main:app --reload --port 8000

使用 --reload 参数可以在开发阶段实现热重载，提高开发效率。

运行测试

在另一个终端窗口中运行测试脚本：

python test_mcp.py

预期输出：

可用工具:
- echo: 返回输入文本
工具测试结果: Echo: 测试消息

成功的测试输出确认 MCP 服务已正确集成并可以正常处理请求。

故障排除

常见问题

在实际部署过程中，可能会遇到以下常见问题：

1. 连接失败：检查服务是否启动，端口是否正确，防火墙设置是否允许连接
2. 路径404：确认 MCP 挂载路径配置正确，检查路由冲突
3. 工具不可用：检查 MCP 工具是否正确注册，依赖项是否完整安装
4. 生命周期异常：验证生命周期函数是否正确合并，异常处理是否完善

总结

通过合理的生命周期管理，在 FastAPI 中成功集成了 FastMCP 服务。回顾整个集成过程，有几个关键要点值得强调：

• 使用 @asynccontextmanager 装饰器优雅地管理应用生命周期
• 确保 MCP 应用对象引用的一致性，避免多实例问题
• 避免 URL 路径的重复配置，保持路由清晰
• 通过测试工具验证集成结果，确保功能正常

这种架构设计既保持 FastAPI 的高性能和易用性，又获得 MCP 的强大工具扩展能力，为构建现代化的 AI 应用提供坚实基础。随着业务需求的增长，可以在此基础上轻松扩展更多的 MCP 工具，构建更加复杂的智能应用系统。