FastAPI MCP (tadata-org/fastapi_mcp)

FastAPI MCP

https://github.com/tadata-org/fastapi_mcp
Admin
Expose your FastAPI endpoints as Model Context Protocol (MCP) tools with built-in authentication and...

Tokens:12,452
Snippets:99
Trust Score:7
Update:6 hours ago
Show doc for...
Context Summary (auto-generated)
Raw
# FastAPI-MCP

FastAPI-MCP is a Python library that automatically generates Model Context Protocol (MCP) servers from FastAPI applications. It enables FastAPI endpoints to be exposed as MCP tools that can be used by AI assistants like Claude, allowing LLMs to interact directly with your APIs. The library is built by Tadata Inc. and designed as a native FastAPI extension rather than just an OpenAPI converter.

The library provides a FastAPI-first approach with native dependency injection for authentication, ASGI transport for efficient communication, and zero-configuration setup. It automatically converts your FastAPI routes into MCP tools while preserving request/response schemas, endpoint documentation, and supporting both HTTP and SSE transports. FastAPI-MCP supports flexible deployment options, allowing you to mount the MCP server to the same FastAPI app or deploy it separately.

## APIs and Key Functions

### Basic MCP Server Setup

Initialize and mount an MCP server to your FastAPI application with minimal configuration.

```python
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI()

# Define your API endpoints
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id, "name": "Example Item"}

# Create MCP server from FastAPI app
mcp = FastApiMCP(app)

# Mount MCP server with HTTP transport (recommended)
mcp.mount_http()

# Run the application
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # MCP server now available at http://localhost:8000/mcp
```

### FastApiMCP Constructor with Schema Configuration

Create an MCP server with custom naming and detailed response schema descriptions for better LLM understanding.

```python
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float

@app.get("/items/", response_model=List[Item])
async def list_items(skip: int = 0, limit: int = 10):
    """List all items with pagination support."""
    return [
        Item(id=1, name="Hammer", description="Tool for hammering", price=9.99),
        Item(id=2, name="Wrench", description="Tool for tightening", price=12.99)
    ]

# Configure MCP server with full schema descriptions
mcp = FastApiMCP(
    app,
    name="Item API MCP",
    description="MCP server for the Item API",
    describe_full_response_schema=True,  # Include complete JSON schema for responses
    describe_all_responses=True,         # Include all response types (2XX, 4XX, 5XX)
)

mcp.mount_http()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

### Filtering Endpoints by Operations and Tags

Selectively expose specific API endpoints as MCP tools using operation IDs or tags.

```python
from fastapi import FastAPI, HTTPException
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str
    price: float

@app.get("/items/", tags=["items"], operation_id="list_items")
async def list_items():
    """List all items."""
    return [{"id": 1, "name": "Hammer", "price": 9.99}]

@app.get("/items/{item_id}", tags=["items"], operation_id="get_item")
async def get_item(item_id: int):
    """Get a specific item by ID."""
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.post("/items/", tags=["items"], operation_id="create_item")
async def create_item(item: Item):
    """Create a new item."""
    return item

@app.delete("/items/{item_id}", tags=["items"], operation_id="delete_item")
async def delete_item(item_id: int):
    """Delete an item."""
    return {"message": "Item deleted"}

@app.get("/items/search/", tags=["search"], operation_id="search_items")
async def search_items(q: str):
    """Search for items."""
    return []

# Include only specific operations
readonly_mcp = FastApiMCP(
    app,
    name="Read-Only Item API",
    include_operations=["get_item", "list_items"]
)
readonly_mcp.mount_http(mount_path="/readonly-mcp")

# Exclude specific operations
no_delete_mcp = FastApiMCP(
    app,
    name="No Delete Item API",
    exclude_operations=["delete_item"]
)
no_delete_mcp.mount_http(mount_path="/no-delete-mcp")

# Include only specific tags
items_only_mcp = FastApiMCP(
    app,
    name="Items Only API",
    include_tags=["items"]
)
items_only_mcp.mount_http(mount_path="/items-only-mcp")

# Combine filters (greedy approach - includes endpoints matching either criteria)
combined_mcp = FastApiMCP(
    app,
    name="Combined Filter API",
    include_operations=["delete_item"],
    include_tags=["search"]
)
combined_mcp.mount_http(mount_path="/combined-mcp")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

### Token-Based Authentication with Header Passthrough

Protect MCP endpoints using FastAPI dependencies and forward authorization headers to tool invocations.

```python
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi_mcp import FastApiMCP, AuthConfig

app = FastAPI()

# Define authentication scheme
token_auth_scheme = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(token_auth_scheme)):
    """Verify the Bearer token."""
    # In production, validate token against your auth system
    if credentials.credentials != "valid-secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return credentials.credentials

@app.get("/public")
async def public_endpoint():
    """Public endpoint - no auth required."""
    return {"message": "This is public"}

@app.get("/private")
async def private_endpoint(token: str = Depends(verify_token)):
    """Private endpoint - requires authentication."""
    return {"message": "This is private", "token": token}

# Create MCP server with authentication
mcp = FastApiMCP(
    app,
    name="Protected MCP",
    auth_config=AuthConfig(
        dependencies=[Depends(token_auth_scheme)]  # Apply auth to all MCP requests
    ),
    headers=["authorization"]  # Forward authorization header to tool calls
)

mcp.mount_http()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Configure MCP client with:
# {
#   "mcpServers": {
#     "protected-api": {
#       "command": "npx",
#       "args": ["mcp-remote", "http://localhost:8000/mcp",
#                "--header", "Authorization:${AUTH_HEADER}"]
#     }
#   },
#   "env": {
#     "AUTH_HEADER": "Bearer valid-secret-token"
#   }
# }
```

### OAuth 2.0 Authentication with Auth0

Integrate OAuth 2.0 authentication using Auth0 with automatic proxy setup and JWT validation.

```python
from fastapi import FastAPI, Depends, HTTPException, Request, status
from pydantic_settings import BaseSettings
from fastapi_mcp import FastApiMCP, AuthConfig
import jwt
import logging

logger = logging.getLogger(__name__)

class Settings(BaseSettings):
    auth0_domain: str              # e.g., "your-tenant.auth0.com"
    auth0_audience: str            # e.g., "https://your-tenant.auth0.com/api/v2/"
    auth0_client_id: str
    auth0_client_secret: str

    @property
    def auth0_jwks_url(self):
        return f"https://{self.auth0_domain}/.well-known/jwks.json"

    @property
    def auth0_oauth_metadata_url(self):
        return f"https://{self.auth0_domain}/.well-known/openid-configuration"

    class Config:
        env_file = ".env"

settings = Settings()

async def fetch_jwks_key(jwks_url: str):
    """Fetch JWKS public key for JWT validation."""
    import httpx
    async with httpx.AsyncClient() as client:
        response = await client.get(jwks_url)
        jwks = response.json()
        return jwt.PyJWKClient(jwks_url).get_signing_key_from_jwk(jwks["keys"][0])

async def lifespan(app: FastAPI):
    app.state.jwks_public_key = await fetch_jwks_key(settings.auth0_jwks_url)
    logger.info(f"Loaded JWKS for {settings.auth0_domain}")
    yield

app = FastAPI(lifespan=lifespan)

async def verify_auth(request: Request) -> dict:
    """Verify JWT token from Auth0."""
    try:
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authorization header"
            )

        token = auth_header.split(" ")[1]

        # Decode and validate JWT
        claims = jwt.decode(
            token,
            app.state.jwks_public_key,
            algorithms=["RS256"],
            audience=settings.auth0_audience,
            issuer=f"https://{settings.auth0_domain}/",
            options={"verify_signature": True}
        )
        return claims
    except Exception as e:
        logger.error(f"Auth error: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized"
        )

async def get_current_user(claims: dict = Depends(verify_auth)) -> str:
    """Extract user ID from validated token."""
    user_id = claims.get("sub")
    if not user_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="No user ID in token"
        )
    return user_id

@app.get("/api/public")
async def public():
    return {"message": "Public endpoint"}

@app.get("/api/protected")
async def protected(user_id: str = Depends(get_current_user)):
    return {"message": f"Hello, {user_id}!", "user_id": user_id}

# Configure MCP with Auth0 authentication and proxy setup
mcp = FastApiMCP(
    app,
    name="MCP With Auth0",
    description="FastAPI-MCP with Auth0 authentication",
    auth_config=AuthConfig(
        issuer=f"https://{settings.auth0_domain}/",
        authorize_url=f"https://{settings.auth0_domain}/authorize",
        oauth_metadata_url=settings.auth0_oauth_metadata_url,
        audience=settings.auth0_audience,
        client_id=settings.auth0_client_id,
        client_secret=settings.auth0_client_secret,
        dependencies=[Depends(verify_auth)],
        setup_proxies=True,                    # Setup MCP-compliant OAuth proxies
        setup_fake_dynamic_registration=True   # Enable client registration endpoint
    ),
)

mcp.mount_http()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # OAuth metadata available at: /.well-known/oauth-authorization-server
```

### Separate Server Deployment

Deploy the MCP server independently from the original FastAPI application for isolated scaling and deployment.

```python
from fastapi import FastAPI, HTTPException
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel
import httpx

# Original API application (could be running on different server)
api_app = FastAPI(title="Items API")

class Item(BaseModel):
    id: int
    name: str
    price: float

items_db = {
    1: Item(id=1, name="Hammer", price=9.99),
    2: Item(id=2, name="Wrench", price=12.99)
}

@api_app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@api_app.post("/items/", response_model=Item)
async def create_item(item: Item):
    items_db[item.id] = item
    return item

# Create MCP server from the API app schema
mcp = FastApiMCP(
    api_app,
    name="Items MCP Server",
    http_client=httpx.AsyncClient(
        base_url="http://api-server:8001"  # Point to actual API server
    )
)

# Create separate FastAPI app for MCP server
mcp_server_app = FastAPI(title="MCP Server")
mcp.mount_http(mcp_server_app, mount_path="/mcp")

# Deploy separately - the original API endpoints are NOT exposed
# Only the MCP interface is available
if __name__ == "__main__":
    import uvicorn

    # Run MCP server on port 8000
    # Original API would run separately on port 8001
    uvicorn.run(mcp_server_app, host="0.0.0.0", port=8000)
    # MCP tools available at: http://localhost:8000/mcp
    # Original API not exposed through this server
```

### Custom HTTP Client Configuration

Configure custom HTTP client settings including timeouts and ASGI transport for direct app communication.

```python
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
import httpx

app = FastAPI()

@app.get("/slow-endpoint")
async def slow_endpoint():
    """Endpoint that might take longer to respond."""
    import asyncio
    await asyncio.sleep(5)
    return {"status": "completed"}

@app.get("/fast-endpoint")
async def fast_endpoint():
    return {"status": "ready"}

# Configure custom HTTP client with extended timeout
custom_client = httpx.AsyncClient(
    transport=httpx.ASGITransport(
        app=app,
        raise_app_exceptions=False  # Handle exceptions gracefully
    ),
    base_url="http://apiserver",
    timeout=30.0,  # 30 second timeout for slow operations
    limits=httpx.Limits(
        max_keepalive_connections=10,
        max_connections=20
    )
)

mcp = FastApiMCP(
    app,
    name="Custom Client MCP",
    http_client=custom_client  # Use custom configured client
)

mcp.mount_http()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

### Multiple MCP Servers with Different Configurations

Mount multiple MCP servers with different filtering and configurations to the same FastAPI app.

```python
from fastapi import FastAPI, APIRouter
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str

@app.get("/items/", tags=["read"], operation_id="list_items")
async def list_items():
    return []

@app.post("/items/", tags=["write"], operation_id="create_item")
async def create_item(item: Item):
    return item

@app.delete("/items/{item_id}", tags=["write"], operation_id="delete_item")
async def delete_item(item_id: int):
    return {"deleted": True}

# Read-only MCP server
readonly_mcp = FastApiMCP(
    app,
    name="Read-Only MCP",
    include_tags=["read"]
)
readonly_mcp.mount_http(mount_path="/mcp-readonly")

# Full access MCP server
full_mcp = FastApiMCP(
    app,
    name="Full Access MCP",
    description="Complete API access including write operations"
)
full_mcp.mount_http(mount_path="/mcp-full")

# Mount to custom router
admin_router = APIRouter(prefix="/admin")
admin_mcp = FastApiMCP(
    app,
    name="Admin MCP",
    include_operations=["delete_item"]
)
admin_mcp.mount_http(router=admin_router, mount_path="/mcp")
app.include_router(admin_router)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # Available endpoints:
    # - /mcp-readonly: Read operations only
    # - /mcp-full: All operations
    # - /admin/mcp: Delete operations only
```

### SSE Transport Configuration

Configure Server-Sent Events (SSE) transport as an alternative to HTTP transport for streaming connections.

```python
from fastapi import FastAPI
from fastapi_mcp import FastApiMCP

app = FastAPI()

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

mcp = FastApiMCP(app, name="SSE MCP Server")

# Mount with SSE transport (legacy support)
# Note: mount_sse() is the recommended method for SSE
mcp.mount_sse(mount_path="/sse")

# Alternative: Use deprecated mount() method with SSE
# mcp.mount(mount_path="/mcp", transport="sse")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # SSE endpoints available at:
    # - GET /sse (connection endpoint)
    # - POST /sse/messages/ (message endpoint)
```

## Summary

FastAPI-MCP serves as a bridge between FastAPI web applications and Model Context Protocol, enabling AI assistants to interact with REST APIs as native tools. The primary use cases include exposing existing FastAPI services to LLM-powered applications, creating AI-accessible microservices, and building authenticated API wrappers for Claude and other MCP-compatible AI systems. The library is particularly valuable for teams that want to make their FastAPI backends LLM-accessible without rewriting endpoints or managing separate tool definitions.

Integration patterns include mounting MCP servers directly to existing FastAPI applications for unified deployment, deploying MCP servers separately for independent scaling and security isolation, using FastAPI dependencies for authentication and authorization, filtering endpoints by operation IDs or tags to control API surface area, and forwarding authentication headers from MCP clients to underlying API calls. The library's FastAPI-native design ensures seamless integration with existing authentication systems, middleware, and deployment infrastructure while maintaining type safety and automatic schema generation from Pydantic models.