FastAPI-MCP
by tadata-org
- fastapi_mcp
"""
Server module for FastAPI-MCP.
This module provides functionality for creating and mounting MCP servers to FastAPI applications.
"""
from typing import Dict, Optional, Any
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.requests import Request
from .http_tools import create_mcp_tools_from_openapi
def create_mcp_server(
app: FastAPI,
name: Optional[str] = None,
description: Optional[str] = None,
capabilities: Optional[Dict[str, Any]] = None,
) -> FastMCP:
"""
Create an MCP server from a FastAPI app.
Args:
app: The FastAPI application
name: Name for the MCP server (defaults to app.title)
description: Description for the MCP server (defaults to app.description)
capabilities: Optional capabilities for the MCP server
Returns:
The created FastMCP instance
"""
# Use app details if not provided
server_name = name or app.title or "FastAPI MCP"
server_description = description or app.description
# Create the MCP server
mcp_server = FastMCP(server_name, server_description)
# Configure server capabilities if provided
if capabilities:
for capability, value in capabilities.items():
# TODO: Maybe revise this hacky approach
mcp_server.settings.capabilities[capability] = value # type: ignore
return mcp_server
def mount_mcp_server(
app: FastAPI,
mcp_server: FastMCP,
mount_path: str = "/mcp",
serve_tools: bool = True,
base_url: Optional[str] = None,
describe_all_responses: bool = False,
describe_full_response_schema: bool = False,
) -> None:
"""
Mount an MCP server to a FastAPI app.
Args:
app: The FastAPI application
mcp_server: The MCP server to mount
mount_path: Path where the MCP server will be mounted
serve_tools: Whether to serve tools from the FastAPI app
base_url: Base URL for API requests
describe_all_responses: Whether to include all possible response schemas in tool descriptions. Recommended to keep False, as the LLM will probably derive if there is an error.
describe_full_response_schema: Whether to include full json schema for responses in tool descriptions. Recommended to keep False, as examples are more LLM friendly, and save tokens.
"""
# Normalize mount path
if not mount_path.startswith("/"):
mount_path = f"/{mount_path}"
if mount_path.endswith("/"):
mount_path = mount_path[:-1]
# Create SSE transport for MCP messages
sse_transport = SseServerTransport(f"{mount_path}/messages/")
# Define MCP connection handler
async def handle_mcp_connection(request: Request):
async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams:
await mcp_server._mcp_server.run(
streams[0],
streams[1],
mcp_server._mcp_server.create_initialization_options(),
)
# Mount the MCP connection handler
app.get(mount_path)(handle_mcp_connection)
app.mount(f"{mount_path}/messages/", app=sse_transport.handle_post_message)
# Serve tools from the FastAPI app if requested
if serve_tools:
create_mcp_tools_from_openapi(
app,
mcp_server,
base_url,
describe_all_responses=describe_all_responses,
describe_full_response_schema=describe_full_response_schema,
)
def add_mcp_server(
app: FastAPI,
mount_path: str = "/mcp",
name: Optional[str] = None,
description: Optional[str] = None,
capabilities: Optional[Dict[str, Any]] = None,
serve_tools: bool = True,
base_url: Optional[str] = None,
describe_all_responses: bool = False,
describe_full_response_schema: bool = False,
) -> FastMCP:
"""
Add an MCP server to a FastAPI app.
Args:
app: The FastAPI application
mount_path: Path where the MCP server will be mounted
name: Name for the MCP server (defaults to app.title)
description: Description for the MCP server (defaults to app.description)
capabilities: Optional capabilities for the MCP server
serve_tools: Whether to serve tools from the FastAPI app
base_url: Base URL for API requests (defaults to http://localhost:$PORT)
describe_all_responses: Whether to include all possible response schemas in tool descriptions. Recommended to keep False, as the LLM will probably derive if there is an error.
describe_full_response_schema: Whether to include full json schema for responses in tool descriptions. Recommended to keep False, as examples are more LLM friendly, and save tokens.
Returns:
The FastMCP instance that was created and mounted
"""
# Create MCP server
mcp_server = create_mcp_server(app, name, description, capabilities)
# Mount MCP server to FastAPI app
mount_mcp_server(
app,
mcp_server,
mount_path,
serve_tools,
base_url,
describe_all_responses=describe_all_responses,
describe_full_response_schema=describe_full_response_schema,
)
return mcp_server