一、FastAPI 🤝 FastMCP

将 FastMCP 与 FastAPI 应用集成

提示：2.11 版本新增： FastMCP 正在引入新一代 OpenAPI 解析器。新解析器具有显著提升的性能和兼容性，并且更易于维护。要启用它，请设置环境变量

FASTMCP_EXPERIMENTAL_ENABLE_NEW_OPENAPI_PARSER=true。

新解析器与现有实现在 API 上基本兼容，并将在未来版本中成为默认选项。我们鼓励所有用户在其成为默认选项之前进行测试并报告任何问题。

FastMCP 提供两种强大的方式与 FastAPI 应用集成：

提示：从 OpenAPI 生成 MCP 服务器是开始使用 FastMCP 的好方法，但在实践中，与精心设计和整理的 MCP 服务器相比，LLM 在使用自动转换的 OpenAPI 服务器时性能显得较差。对于具有许多端点和参数的复杂 API 尤其如此。

我们建议将 FastAPI 集成用于引导和原型设计，而不是将 API 镜像给 LLM 客户端。更多详细信息请参阅 文章停止将 REST API 转换为 MCP。

注意：FastMCP 不将 FastAPI 作为依赖项包含；要使用此集成，必须单独安装它。

二、FastAPI 应用示例

在本指南中，我们将使用这个电子商务 API 作为示例（点击 复制 按钮将其复制到其他代码块中使用）：

# 将此 FastAPI 服务器复制到本指南的其他代码块中

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# 数据模型
class Product(BaseModel):
    name: str
    price: float
    category: str
    description: str | None = None

class ProductResponse(BaseModel):
    id: int
    name: str
    price: float
    category: str
    description: str | None = None

# 创建 FastAPI 应用
app = FastAPI(title="电子商务 API", version="1.0.0")

# 内存数据库
products_db = {
    1: ProductResponse(
        id=1, name="笔记本电脑", price=999.99, category="电子产品"
    ),
    2: ProductResponse(
        id=2, name="鼠标", price=29.99, category="电子产品"
    ),
    3: ProductResponse(
        id=3, name="办公椅", price=299.99, category="家具"
    ),
}
next_id = 4

@app.get("/products", response_model=list[ProductResponse])
def list_products(
    category: str | None = None,
    max_price: float | None = None,
) -> list[ProductResponse]:
    """列出所有产品，支持可选过滤。"""
    products = list(products_db.values())
    if category:
        products = [p for p in products if p.category == category]
    if max_price:
        products = [p for p in products if p.price <= max_price]
    return products

@app.get("/products/{product_id}", response_model=ProductResponse)
def get_product(product_id: int):
    """根据 ID 获取特定产品。"""
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="产品未找到")
    return products_db[product_id]

@app.post("/products", response_model=ProductResponse)
def create_product(product: Product):
    """创建新产品。"""
    global next_id
    product_response = ProductResponse(id=next_id, **product.model_dump())
    products_db[next_id] = product_response
    next_id += 1
    return product_response

@app.put("/products/{product_id}", response_model=ProductResponse)
def update_product(product_id: int, product: Product):
    """更新现有产品。"""
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="产品未找到")
    products_db[product_id] = ProductResponse(
        id=product_id,
        **product.model_dump(),
    )
    return products_db[product_id]

@app.delete("/products/{product_id}")
def delete_product(product_id: int):
    """删除产品。"""
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="产品未找到")
    del products_db[product_id]
    return {"message": "产品已删除"}

提示：本指南中所有后续代码示例均假设已定义了上述 FastAPI 应用代码。每个示例都基于此基础应用 app 构建。

三、生成 MCP 服务器

新版本：2.0.0 功能

引导 MCP 服务器的最常见方法之一是从现有 FastAPI 应用生成。FastMCP 将您的 FastAPI 端点公开为 MCP 组件（默认为工具），以便向 LLM 客户端公开您的 API。

3.1 基本转换

一行代码将 FastAPI 应用转换为 MCP 服务器：

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP

# 转换为 MCP 服务器
mcp = FastMCP.from_fastapi(app=app)

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

3.2 添加组件

转换后的 MCP 服务器是一个完整的 FastMCP 实例，这意味着您可以像处理任何其他 FastMCP 实例一样向其添加新工具、资源和其他组件。

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP

# 转换为 MCP 服务器
mcp = FastMCP.from_fastapi(app=app)

# 添加新工具
@mcp.tool
def get_product(product_id: int) -> ProductResponse:
    """根据 ID 获取产品。"""
    return products_db[product_id]

# 运行 MCP 服务器
if __name__ == "__main__":
    mcp.run()

3.3 与 MCP 服务器交互

将 FastAPI 应用转换为 MCP 服务器后，可以使用 FastMCP 客户端与其交互，在部署到基于 LLM 的应用之前测试功能。

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP
from fastmcp.client import Client
import asyncio

# 转换为 MCP 服务器
mcp = FastMCP.from_fastapi(app=app)

async def demo():
    async with Client(mcp) as client:
        # 列出可用工具
        tools = await client.list_tools()
        print(f"可用工具: {[t.name for t in tools]}")
        
        # 创建产品
        result = await client.call_tool(
            "create_product_products_post",
            {
                "name": "无线键盘",
                "price": 79.99,
                "category": "电子产品",
                "description": "蓝牙机械键盘"
            }
        )
        print(f"创建的产品: {result.data}")
        
        # 列出 100 美元以下的电子产品
        result = await client.call_tool(
            "list_products_products_get",
            {"category": "电子产品", "max_price": 100}
        )
        print(f"经济型电子产品: {result.data}")

if __name__ == "__main__":
    asyncio.run(demo())

3.4 自定义路由映射

由于 FastMCP 的 FastAPI 集成基于其 OpenAPI 集成，您可以用完全相同的方式自定义端点如何转换为 MCP 组件。例如，这里我们使用 RouteMap 将所有 GET 请求映射到 MCP 资源，所有 POST/PUT/DELETE 请求映射到 MCP 工具：

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP
from fastmcp.server.openapi import RouteMap, MCPType

# 如果使用实验性解析器，从实验性模块导入：
# from fastmcp.experimental.server.openapi import RouteMap, MCPType

# 自定义映射规则
mcp = FastMCP.from_fastapi(
    app=app,
    route_maps=[
        # 带路径参数的 GET → ResourceTemplates
        RouteMap(
            methods=["GET"], 
            pattern=r".*\{.*\}.*", 
            mcp_type=MCPType.RESOURCE_TEMPLATE
        ),
        # 其他 GET → Resources
        RouteMap(
            methods=["GET"], 
            pattern=r".*", 
            mcp_type=MCPType.RESOURCE
        ),
        # POST/PUT/DELETE → Tools（默认）
    ],
)

# 现在：
# - GET /products → Resource
# - GET /products/{id} → ResourceTemplate
# - POST/PUT/DELETE → Tools

提示：要了解有关自定义转换过程的更多信息，请参阅 OpenAPI 集成指南。

3.5 身份验证和请求头

可以通过 httpx_client_kwargs 参数配置请求头和其他客户端选项。例如，要为 FastAPI 应用添加身份验证，可以将 headers 字典传递给 httpx_client_kwargs 参数：

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP

# 为 FastAPI 应用添加身份验证
from fastapi import Depends, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    if credentials.credentials != "secret-token":
        raise HTTPException(status_code=401, detail="身份验证无效")
    return credentials.credentials

# 添加受保护的端点
@app.get("/admin/stats", dependencies=[Depends(verify_token)])
def get_admin_stats():
    return {
        "产品总数": len(products_db),
        "分类": list(set(p.category for p in products_db.values()))
    }

# 创建带身份验证请求头的 MCP 服务器
mcp = FastMCP.from_fastapi(
    app=app,
    httpx_client_kwargs={
        "headers": {
            "Authorization": "Bearer secret-token",
        }
    }
)

四、挂载 MCP 服务器

新版本：2.3.1 功能

除了生成服务器外，FastMCP 还可以促进将 MCP 服务器添加到现有 FastAPI 应用中。可以通过挂载 MCP ASGI 应用程序来实现。

基本挂载

要挂载 MCP 服务器，可以在 FastMCP 实例上使用 http_app 方法。这将返回一个可以挂载到 FastAPI 应用的 ASGI 应用程序。

from fastmcp import FastMCP
from fastapi import FastAPI

# 创建 MCP 服务器
mcp = FastMCP("分析工具")

@mcp.tool
def analyze_pricing(category: str) -> dict:
    """分析指定分类的定价。"""
    products = [p for p in products_db.values() if p.category == category]
    if not products:
        return {"error": f"{category} 分类中无产品"}
    
    prices = [p.price for p in products]
    return {
        "category": category,
        "avg_price": round(sum(prices) / len(prices), 2),
        "min": min(prices),
        "max": max(prices),
    }

# 从 MCP 服务器创建 ASGI 应用
mcp_app = mcp.http_app(path='/mcp')

# 关键：将 lifespan 传递给 FastAPI
app = FastAPI(title="电子商务 API", lifespan=mcp_app.lifespan)

# 挂载 MCP 服务器
app.mount("/analytics", mcp_app)

# 现在：API 在 /products/*，MCP 在 /analytics/mcp/

五、提供 LLM 友好的 API

一个常见模式是从 FastAPI 应用生成 MCP 服务器，并从同一应用提供两个接口。这提供了与常规 API 并行的 LLM 优化接口：

# 假设上述 FastAPI 应用已定义
from fastmcp import FastMCP
from fastapi import FastAPI

# 1. 从 API 生成 MCP 服务器
mcp = FastMCP.from_fastapi(app=app, name="电子商务 MCP")

# 2. 创建 MCP 的 ASGI 应用
mcp_app = mcp.http_app(path='/mcp')

# 3. 创建结合两组路由的新 FastAPI 应用
combined_app = FastAPI(
    title="带 MCP 的电子商务 API",
    routes=[
        *mcp_app.routes,  # MCP 路由
        *app.routes,      # 原始 API 路由
    ],
    lifespan=mcp_app.lifespan,
)

# 现在拥有：
# - 常规 API: http://localhost:8000/products
# - LLM 友好的 MCP: http://localhost:8000/mcp
# 两者从同一 FastAPI 应用提供服务！

这种方法允许您维护单一代码库，同时为 LLM 客户端提供传统 REST 端点和 MCP 兼容端点。

六、关键注意事项

6.1 操作 ID

FastAPI 操作 ID 成为 MCP 组件名称。始终指定有意义的操作 ID：

# 良好 - 显式 operation_id
@app.get("/users/{user_id}", operation_id="get_user_by_id")
def get_user(user_id: int):
    return {"id": user_id}

# 不太理想 - 自动生成的名称
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id}

6.2 生命周期管理

挂载 MCP 服务器时，始终传递生命周期上下文：

# 正确 - 传递了 lifespan
mcp_app = mcp.http_app(path='/mcp')
app = FastAPI(lifespan=mcp_app.lifespan)
app.mount("/mcp", mcp_app)

# 错误 - 缺少 lifespan
app = FastAPI()
app.mount("/mcp", mcp.http_app())  # 会话管理器不会初始化

如果在路径前缀下挂载经过身份验证的 MCP 服务器，请参阅 挂载经过身份验证的服务器 了解重要的 OAuth 路由注意事项。

6.3 CORS 中间件

如果 FastAPI 应用使用 CORSMiddleware 并且挂载了 OAuth 保护的 FastMCP 服务器，请避免添加应用范围的 CORS 中间件。FastMCP 和 MCP SDK 已经处理了 OAuth 路由的 CORS，叠加 CORS 中间件可能导致冲突（例如 .well-known 路由或 OPTIONS 请求出现 404 错误）。

如果需要在自己的 FastAPI 路由上使用 CORS，请使用子应用模式：将 API 和 FastMCP 作为单独的应用挂载，每个应用都有自己的中间件，而不是向组合应用添加顶层 CORSMiddleware。

6.4 组合生命周期

如果 FastAPI 应用已有生命周期（用于数据库连接、启动任务等），不能简单地用 MCP 生命周期替换它。相反，需要创建一个新的生命周期函数来管理两个上下文。这确保应用初始化逻辑和 MCP 服务器的会话管理器都能正确运行：

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP

# 现有生命周期
@asynccontextmanager
async def app_lifespan(app: FastAPI):
    # 启动
    print("启动应用...")
    # 初始化数据库、缓存等
    yield
    # 关闭
    print("关闭应用...")

# 创建 MCP 服务器
mcp = FastMCP("工具")
mcp_app = mcp.http_app(path='/mcp')

# 组合两个生命周期
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
    # 运行两个生命周期
    async with app_lifespan(app):
        async with mcp_app.lifespan(app):
            yield

# 使用组合生命周期
app = FastAPI(lifespan=combined_lifespan)
app.mount("/mcp", mcp_app)

这种模式确保应用初始化逻辑和 MCP 服务器的会话管理器都得到正确管理。关键是使用嵌套的 async with 语句 - 内部上下文（MCP）将在外部上下文（应用）之后初始化，并在其之前清理。这为所有资源维护了正确的初始化和清理顺序。

6.5 性能提示

  • 测试使用内存传输 - 直接将 MCP 服务器传递给客户端

  • 设计专用 MCP 工具 - 比自动转换复杂 API 更好

  • 保持工具参数简单 - LLM 在使用专注接口时表现更好

有关配置选项的更多详细信息，请参阅 OpenAPI 集成指南。

# fastapi
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

企业级. Web考编论坛网站管理系统源码｜SpringBoot+Vue+MyBatis架构+MySQL数据库【完整版】

avatar 2048 AI社区
cover

Java Web 房屋租赁系统系统源码-SpringBoot2+Vue3+MyBatis-Plus+MySQL8.0【含文档】

avatar 2048 AI社区
cover

SpringBoot+Vue web新能源充电系统平台完整项目源码+SQL脚本+接口文档【Java Web毕设】

avatar 2048 AI社区