New in version: 2.0.0

FastMCP 可以自动将 FastAPI 应用程序转换为 MCP 服务器。

FastMCP 包含FastAPI作为依赖项;您必须单独安装它才能运行这些示例。

from fastapi import FastAPI
from fastmcp import FastMCP


# A FastAPI app
app = FastAPI()

@app.get("/items")
def list_items():
    return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]

@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"id": item_id, "name": f"Item {item_id}"}

@app.post("/items")
def create_item(name: str):
    return {"id": 3, "name": name}


# Create an MCP server from your FastAPI app
mcp = FastMCP.from_fastapi(app=app)

if __name__ == "__main__":
    mcp.run()  # Start the MCP server

配置选项

超时

你可以为所有API请求设置超时时间:

# Set a 5 second timeout for all requests
mcp = FastMCP.from_fastapi(app=app, timeout=5.0)

此超时设置适用于所有由工具、资源和资源模板发出的请求。

路由映射

默认情况下,FastMCP会根据以下规则将FastAPI路由映射到MCP组件:

FastAPI路由类型FastAPI示例MCP组件备注
GET without path params@app.get("/stats")ResourceSimple resources for fetching data
GET with path params@app.get("/users/{id}")Resource TemplatePath parameters become template parameters
POST, PUT, DELETE, etc.@app.post("/users")ToolOperations that modify data

有关路由映射或自定义映射规则的更多详细信息,请参阅OpenAPI集成文档;FastMCP对FastAPI和OpenAPI集成使用相同的映射规则。

完整示例

以下是一个更详细的数据模型示例:

import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from fastmcp import FastMCP, Client

# Define your Pydantic model
class Item(BaseModel):
    name: str
    price: float

# Create your FastAPI app
app = FastAPI()
items = {}  # In-memory database

@app.get("/items")
def list_items():
    """List all items"""
    return list(items.values())

@app.get("/items/{item_id}")
def get_item(item_id: int):
    """Get item by ID"""
    if item_id not in items:
        raise HTTPException(404, "Item not found")
    return items[item_id]

@app.post("/items")
def create_item(item: Item):
    """Create a new item"""
    item_id = len(items) + 1
    items[item_id] = {"id": item_id, **item.model_dump()}
    return items[item_id]

# Test your MCP server with a client
async def check_mcp(mcp: FastMCP):
    # List the components that were created
    tools = await mcp.get_tools()
    resources = await mcp.get_resources()
    templates = await mcp.get_resource_templates()
    
    print(
        f"{len(tools)} Tool(s): {', '.join([t.name for t in tools.values()])}"
    )
    print(
        f"{len(resources)} Resource(s): {', '.join([r.name for r in resources.values()])}"
    )
    print(
        f"{len(templates)} Resource Template(s): {', '.join([t.name for t in templates.values()])}"
    )
    
    return mcp

if __name__ == "__main__":
    # Create MCP server from FastAPI app
    mcp = FastMCP.from_fastapi(app=app)
    
    asyncio.run(check_mcp(mcp))
    
    # In a real scenario, you would run the server:
    mcp.run()

优势

  • 利用现有FastAPI应用 - 无需重写您的API逻辑
  • 模式复用 - 继承FastAPI的Pydantic模型和验证功能
  • 完整功能支持 - 兼容FastAPI的认证、依赖项等功能
  • ASGI传输 - 直接通信,无需额外的HTTP开销