Try Live
Add Docs
Rankings
Pricing
Enterprise
Docs
Install
Theme
Install
Docs
Pricing
Enterprise
More...
More...
Try Live
Rankings
Create API Key
Add Docs
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
Context
Skills
Chat
Benchmark
89.7
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# 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.