一行代碼 FastAPI 秒變 MCP 服務器！企業級 AI 應用首選方案

fastapi_mcp 實戰

使用 FastAPI-MCP 將 MCP 服務器添加到 FastAPI 應用的簡單示例。

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from typing import List, Optional

from fastapi_mcp import add_mcp_server


# Create a simple FastAPI app
app = FastAPI(
    title="Example API",
    description="A simple example API with integrated MCP server",
    version="0.1.0",
)


# Define some models
class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tags: List[str] = []


# In-memory database
items_db: dict[int, Item] = {}


# Define some endpoints
@app.get("/items/", response_model=List[Item], tags=["items"])
async def list_items(skip: int = 0, limit: int = 10):
    """
    List all items in the database.

    Returns a list of items, with pagination support.
    """
    return list(items_db.values())[skip : skip + limit]


@app.get("/items/{item_id}", response_model=Item, tags=["items"])
async def read_item(item_id: int):
    """
    Get a specific item by its ID.

    Raises a 404 error if the item does not exist.
    """
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]


@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):
    """
    Create a new item in the database.

    Returns the created item with its assigned ID.
    """
    items_db[item.id] = item
    return item


@app.put("/items/{item_id}", response_model=Item, tags=["items"])
async def update_item(item_id: int, item: Item):
    """
    Update an existing item.

    Raises a 404 error if the item does not exist.
    """
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    item.id = item_id
    items_db[item_id] = item
    return item


@app.delete("/items/{item_id}", tags=["items"])
async def delete_item(item_id: int):
    """
    Delete an item from the database.

    Raises a 404 error if the item does not exist.
    """
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")

    del items_db[item_id]
    return {"message": "Item deleted successfully"}


@app.get("/items/search/", response_model=List[Item], tags=["search"])
async def search_items(
    q: Optional[str] = Query(None, description="Search query string"),
    min_price: Optional[float] = Query(None, description="Minimum price"),
    max_price: Optional[float] = Query(None, description="Maximum price"),
    tags: List[str] = Query([], description="Filter by tags"),
):
    """
    Search for items with various filters.

    Returns a list of items that match the search criteria.
    """
    results = list(items_db.values())

    # Filter by search query
    if q:
        q = q.lower()
        results = [
            item for item in results if q in item.name.lower() or (item.description and q in item.description.lower())
        ]

    # Filter by price range
    if min_price is not None:
        results = [item for item in results if item.price >= min_price]
    if max_price is not None:
        results = [item for item in results if item.price <= max_price]

    # Filter by tags
    if tags:
        results = [item for item in results if all(tag in item.tags for tag in tags)]

    return results


# Add sample data
sample_items = [
    Item(id=1, , price=9.99, tags=["tool", "hardware"]),
    Item(id=2, , price=7.99, tags=["tool", "hardware"]),
    Item(id=3, , price=12.99, tags=["tool", "hardware"]),
    Item(id=4, , price=19.99, tags=["tool", "hardware", "cutting"]),
    Item(id=5, , price=49.99, tags=["tool", "hardware", "power"]),
]

for item in sample_items:
    items_db[item.id] = item


# Add MCP server to the FastAPI app
mcp_server = add_mcp_server(
    app,
    mount_path="/mcp",        # 掛着mcp服務器的地址
    ,      # mcp服務的名稱
    description="MCP server for the Item API",
    base_url="http://localhost:8000",
    describe_all_responses=False,         # 默認False, Only describe the success response in tool descriptions
    describe_full_response_schema=False,  # Only show LLM-friendly example response in tool descriptions, not the full json schema
)


# Optionally, you can add custom MCP tools not based on FastAPI endpoints
@mcp_server.tool()
async def get_item_count() -> int:
    """Get the total number of items in the database."""
    return len(items_db)


# Run the server if this file is executed directly
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)

特點

直接集成 - 將 MCP 服務器直接掛載到你的 FastAPI 應用程序
無需配置 - 只需將其指向您的 FastAPI 應用程序即可工作
自動發現所有 FastAPI 端點並轉換爲 MCP 工具
保留請求模型和響應模型的架構
保留所有端點的文檔，就像在 Swagger 中一樣
擴展 - 在自動生成的 MCP 工具旁邊添加自定義 MCP 工具

使用 SSE 連接到 MCP 服務器

具有 MCP 集成的 FastAPI 應用程序運行後，您可以使用任何支持 SSE 的 MCP 客戶端（例如 Cursor）連接到它：

運行您的應用程序。
在 Cursor -> Settings -> MCP 中，使用 MCP 服務器端點的 URL（例如）作爲 sse。http://localhost:8000/mcp
Cursor 將自動發現所有可用的工具和資源

使用 mcp-proxy stdio 連接到 MCP 服務器

如果您的 MCP 客戶端不支持 SSE，例如 Claude Desktop：

運行您的應用程序。
安裝 mcp-proxy，例如：uv tool install mcp-proxy
添加 Claude Desktop MCP 配置文件（）：claude_desktop_config.json

在 Windows 上：

{
  "mcpServers": {
    "my-api-mcp-proxy": {
        "command": "mcp-proxy",
        "args": ["http://127.0.0.1:8000/mcp"]
    }
  }
}

https://tadata.com/

https://github.com/tadata-org/fastapi_mcp/tree/main

https://github.com/sparfenyuk/mcp-proxy

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/5kWzAk4h-xHOYB8RzADPlw

fastapi_mcp 實戰

特點

使用 SSE 連接到 MCP 服務器

使用 mcp-proxy stdio 連接到 MCP 服務器

猜你喜歡