expose FastAPI as MCP tools with FastAPI-MCP

Author: ks6088ts

Makefile

@@ -124,3 +124,7 @@ dev: ## run FastAPI server in development mode
 .PHONY: azure-functions
 azure-functions: ## run Azure Functions server
 	uv run func start
+
+.PHONY: mcp-insppector
+mcp-inspector: ## run MCP Inspector server
+	npx @modelcontextprotocol/inspector

docs/index.md

@@ -8,3 +8,10 @@
 
 - [Using FastAPI Framework with Azure Functions](https://learn.microsoft.com/en-us/samples/azure-samples/fastapi-on-azure-functions/fastapi-on-azure-functions/)
 - [ks6088ts-labs/azure-functions-python](https://github.com/ks6088ts-labs/azure-functions-python)
+
+## MCP
+
+- [FastAPI-MCP](https://github.com/tadata-org/fastapi_mcp)
+- [FastAPI で作成した API エンドポイントをモデル コンテキスト プロトコル (MCP) ツールとして公開してみる](https://dev.classmethod.jp/articles/fastapi-api-mcp/)
+- [MCP Inspector > docs](https://modelcontextprotocol.io/docs/tools/inspector)
+- [MCP Inspector > codes](https://github.com/modelcontextprotocol/inspector)

function_app.py

@@ -1,6 +1,6 @@
 import azure.functions as func
 
-from main import app as fastapi_app
+from template_fastapi.app import app as fastapi_app
 
 app = func.AsgiFunctionApp(
     app=fastapi_app,

main.py

@@ -1,13 +1,7 @@
-from fastapi import FastAPI
+from fastapi_mcp import FastApiMCP
 
-app = FastAPI()
+from template_fastapi.app import app as fastapi_app
 
+mcp = FastApiMCP(fastapi_app)
 
-@app.get("/")
-def read_root():
-    return {"Hello": "World"}
-
-
-@app.get("/items/{item_id}")
-def read_item(item_id: int, q: str | None = None):
-    return {"item_id": item_id, "q": q}
+mcp.mount()

pyproject.toml

@@ -6,6 +6,7 @@ readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
     "azure-functions>=1.23.0",
+    "fastapi-mcp>=0.3.4",
     "fastapi[standard]>=0.115.12",
 ]
 

template_fastapi/app.py

@@ -0,0 +1,124 @@
+"""
+Simple example of using FastAPI-MCP to add an MCP server to a FastAPI app.
+ref. https://github.com/tadata-org/fastapi_mcp/blob/v0.3.4/examples/shared/apps/items.py
+"""
+
+from fastapi import FastAPI, HTTPException, Query
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+    description: str | None = None
+    price: float
+    tags: list[str] = []
+
+
+items_db: dict[int, Item] = {}
+
+
+@app.get("/items/", response_model=list[Item], tags=["items"], operation_id="list_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"], operation_id="get_item")
+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"], operation_id="create_item")
+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"], operation_id="update_item")
+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"], operation_id="delete_item")
+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"], operation_id="search_items")
+async def search_items(
+    q: str | None = Query(None, description="Search query string"),
+    min_price: float | None = Query(None, description="Minimum price"),
+    max_price: float | None = 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())
+
+    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())
+        ]
+
+    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]
+
+    if tags:
+        results = [item for item in results if all(tag in item.tags for tag in tags)]
+
+    return results
+
+
+sample_items = [
+    Item(id=1, name="Hammer", description="A tool for hammering nails", price=9.99, tags=["tool", "hardware"]),
+    Item(id=2, name="Screwdriver", description="A tool for driving screws", price=7.99, tags=["tool", "hardware"]),
+    Item(id=3, name="Wrench", description="A tool for tightening bolts", price=12.99, tags=["tool", "hardware"]),
+    Item(id=4, name="Saw", description="A tool for cutting wood", price=19.99, tags=["tool", "hardware", "cutting"]),
+    Item(id=5, name="Drill", description="A tool for drilling holes", price=49.99, tags=["tool", "hardware", "power"]),
+]
+for item in sample_items:
+    items_db[item.id] = item