概述

在现代 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}")

    yield

    logger.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 Client

async 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 工具,构建更加复杂的智能应用系统。

Logo

更多推荐