alihan/fastapi_mcp
1
0
Fork 0
mirror of https://github.com/tadata-org/fastapi_mcp.git synced 2025-04-13 23:32:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

change to class based

Browse Source
This commit is contained in:
Shahar Abramov
2025-04-08 18:13:39 +03:00
parent 222e104d13
commit 97ffd8e7c4
4 changed files with 139 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
examples/full_schema_description_example.py
View File

@@ -4,11 +4,11 @@ Simple example of using FastAPI-MCP to add an MCP server to a FastAPI app.
from examples.apps import items from examples.apps import items
from fastapi_mcp import add_mcp_server from fastapi_mcp import FastApiMCP
# Add MCP server to the FastAPI app # Add MCP server to the FastAPI app
mcp = add_mcp_server( mcp = FastApiMCP(
items.app, items.app,
mount_path="/mcp", mount_path="/mcp",
name="Item API MCP", name="Item API MCP",
@@ -18,6 +18,9 @@ mcp = add_mcp_server(
describe_all_responses=True, describe_all_responses=True,
) )
mcp.mount()
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn

6
examples/simple_example.py
View File

@@ -4,11 +4,11 @@ Simple example of using FastAPI-MCP to add an MCP server to a FastAPI app.
from examples.apps import items from examples.apps import items
from fastapi_mcp import add_mcp_server from fastapi_mcp import FastApiMCP
# Add MCP server to the FastAPI app # Add MCP server to the FastAPI app
mcp = add_mcp_server( mcp = FastApiMCP(
items.app, items.app,
mount_path="/mcp", mount_path="/mcp",
name="Item API MCP", name="Item API MCP",
@@ -16,6 +16,8 @@ mcp = add_mcp_server(
base_url="http://localhost:8000", base_url="http://localhost:8000",
) )
mcp.mount()
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn

6
fastapi_mcp/__init__.py
View File

@@ -12,11 +12,9 @@ except Exception:
# Fallback for local development # Fallback for local development
__version__ = "0.0.0.dev0" __version__ = "0.0.0.dev0"
from .server import add_mcp_server, create_mcp_server, mount_mcp_server from .server import FastApiMCP
__all__ = [ __all__ = [
"add_mcp_server", "FastApiMCP",
"create_mcp_server",
"mount_mcp_server",
] ]

283
fastapi_mcp/server.py
View File

@@ -1,5 +1,5 @@
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Dict, Optional, Any, Tuple, List, Union, AsyncIterator from typing import Dict, Optional, Any, List, Union, AsyncIterator
from fastapi import FastAPI, Request from fastapi import FastAPI, Request
from fastapi.openapi.utils import get_openapi from fastapi.openapi.utils import get_openapi
@@ -11,179 +11,152 @@ from fastapi_mcp.openapi.convert import convert_openapi_to_mcp_tools
from fastapi_mcp.execute import execute_api_tool from fastapi_mcp.execute import execute_api_tool
def create_mcp_server( class FastApiMCP:
app: FastAPI, def __init__(
name: Optional[str] = None, self,
description: Optional[str] = None, fastapi: FastAPI,
base_url: Optional[str] = None, mount_path: str = "/mcp",
describe_all_responses: bool = False, name: Optional[str] = None,
describe_full_response_schema: bool = False, description: Optional[str] = None,
) -> Tuple[Server, Dict[str, Dict[str, Any]]]: base_url: Optional[str] = None,
""" describe_all_responses: bool = False,
Create an MCP server from a FastAPI app. describe_full_response_schema: bool = False,
):
self.operation_map: Dict[str, Dict[str, Any]]
self.tools: List[types.Tool]
Args: self.fastapi = fastapi
app: The FastAPI application self.name = name
name: Name for the MCP server (defaults to app.title) self.description = description
description: Description for the MCP server (defaults to app.description)
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
describe_full_response_schema: Whether to include full json schema for responses in tool descriptions
Returns: self._mount_path = mount_path
A tuple containing: self._base_url = base_url
- The created MCP Server instance (NOT mounted to the app) self._describe_all_responses = describe_all_responses
- A mapping of operation IDs to operation details for HTTP execution self._describe_full_response_schema = describe_full_response_schema
"""
# Get OpenAPI schema from FastAPI app
openapi_schema = get_openapi(
title=app.title,
version=app.version,
openapi_version=app.openapi_version,
description=app.description,
routes=app.routes,
)
# Get server name and description from app if not provided self.mcp_server = self.create_server()
server_name = name or app.title or "FastAPI MCP"
server_description = description or app.description
# Convert OpenAPI schema to MCP tools def create_server(self) -> Server:
tools, operation_map = convert_openapi_to_mcp_tools( """
openapi_schema, Create an MCP server from the FastAPI app.
describe_all_responses=describe_all_responses,
describe_full_response_schema=describe_full_response_schema,
)
# Determine base URL if not provided Args:
if not base_url: app: The FastAPI application
# Try to determine the base URL from FastAPI config name: Name for the MCP server (defaults to app.title)
if hasattr(app, "root_path") and app.root_path: description: Description for the MCP server (defaults to app.description)
base_url = app.root_path base_url: Base URL for API requests (defaults to http://localhost:$PORT)
else: describe_all_responses: Whether to include all possible response schemas in tool descriptions
# Default to localhost with FastAPI default port describe_full_response_schema: Whether to include full json schema for responses in tool descriptions
port = 8000
for route in app.routes:
if hasattr(route, "app") and hasattr(route.app, "port"):
port = route.app.port
break
base_url = f"http://localhost:{port}"
# Normalize base URL Returns:
if base_url.endswith("/"): A tuple containing:
base_url = base_url[:-1] - The created MCP Server instance (NOT mounted to the app)
- A mapping of operation IDs to operation details for HTTP execution
"""
# Get OpenAPI schema from FastAPI app
openapi_schema = get_openapi(
title=self.fastapi.title,
version=self.fastapi.version,
openapi_version=self.fastapi.openapi_version,
description=self.fastapi.description,
routes=self.fastapi.routes,
)
# Create the MCP server # Get server name and description from app if not provided
mcp_server: Server = Server(server_name, server_description) server_name = self.name or self.fastapi.title or "FastAPI MCP"
server_description = self.description or self.fastapi.description
# Create a lifespan context manager to store the base_url and operation_map # Convert OpenAPI schema to MCP tools
@asynccontextmanager self.tools, self.operation_map = convert_openapi_to_mcp_tools(
async def server_lifespan(server) -> AsyncIterator[Dict[str, Any]]: openapi_schema,
# Store context data that will be available to all server handlers describe_all_responses=self._describe_all_responses,
context = {"base_url": base_url, "operation_map": operation_map} describe_full_response_schema=self._describe_full_response_schema,
yield context )
# Use our custom lifespan # Determine base URL if not provided
mcp_server.lifespan = server_lifespan if not self._base_url:
# Try to determine the base URL from FastAPI config
if hasattr(self.fastapi, "root_path") and self.fastapi.root_path:
self._base_url = self.fastapi.root_path
else:
# Default to localhost with FastAPI default port
port = 8000
for route in self.fastapi.routes:
if hasattr(route, "app") and hasattr(route.app, "port"):
port = route.app.port
break
self._base_url = f"http://localhost:{port}"
# Register handlers for tools # Normalize base URL
@mcp_server.list_tools() if self._base_url.endswith("/"):
async def handle_list_tools() -> List[types.Tool]: self._base_url = self._base_url[:-1]
"""Handler for the tools/list request"""
return tools
# Register the tool call handler # Create the MCP server
@mcp_server.call_tool() mcp_server: Server = Server(server_name, server_description)
async def handle_call_tool(
name: str, arguments: Dict[str, Any]
) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
"""Handler for the tools/call request"""
# Get context from server lifespan
ctx = mcp_server.request_context
base_url = ctx.lifespan_context["base_url"]
operation_map = ctx.lifespan_context["operation_map"]
# Execute the tool # Create a lifespan context manager to store the base_url and operation_map
return await execute_api_tool(base_url, name, arguments, operation_map) @asynccontextmanager
async def server_lifespan(server) -> AsyncIterator[Dict[str, Any]]:
# Store context data that will be available to all server handlers
context = {"base_url": self._base_url, "operation_map": self.operation_map}
yield context
return mcp_server, operation_map # Use our custom lifespan
mcp_server.lifespan = server_lifespan
# Register handlers for tools
@mcp_server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
"""Handler for the tools/list request"""
return self.tools
def mount_mcp_server( # Register the tool call handler
app: FastAPI, @mcp_server.call_tool()
mcp_server: Server, async def handle_call_tool(
operation_map: Dict[str, Dict[str, Any]], name: str, arguments: Dict[str, Any]
mount_path: str = "/mcp", ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
base_url: Optional[str] = None, """Handler for the tools/call request"""
) -> None: # Get context from server lifespan
""" ctx = mcp_server.request_context
Mount an MCP server to a FastAPI app. base_url = ctx.lifespan_context["base_url"]
operation_map = ctx.lifespan_context["operation_map"]
Args: # Execute the tool
app: The FastAPI application return await execute_api_tool(base_url, name, arguments, operation_map)
mcp_server: The MCP server to mount
operation_map: A mapping of operation IDs to operation details
mount_path: Path where the MCP server will be mounted
base_url: Base URL for API requests
"""
# 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 return mcp_server
sse_transport = SseServerTransport(f"{mount_path}/messages/")
# Define MCP connection handler def mount(self) -> None:
async def handle_mcp_connection(request: Request): """
async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams: Mount the MCP server to the FastAPI app.
await mcp_server.run(
streams[0],
streams[1],
mcp_server.create_initialization_options(notification_options=None, experimental_capabilities={}),
)
# Mount the MCP connection handler Args:
app.get(mount_path)(handle_mcp_connection) app: The FastAPI application
app.mount(f"{mount_path}/messages/", app=sse_transport.handle_post_message) mcp_server: The MCP server to mount
operation_map: A mapping of operation IDs to operation details
mount_path: Path where the MCP server will be mounted
base_url: Base URL for API requests
"""
# Normalize mount path
if not self._mount_path.startswith("/"):
self._mount_path = f"/{self._mount_path}"
if self._mount_path.endswith("/"):
self._mount_path = self._mount_path[:-1]
# Create SSE transport for MCP messages
sse_transport = SseServerTransport(f"{self._mount_path}/messages/")
def add_mcp_server( # Define MCP connection handler
app: FastAPI, async def handle_mcp_connection(request: Request):
mount_path: str = "/mcp", async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams:
name: Optional[str] = None, await self.mcp_server.run(
description: Optional[str] = None, streams[0],
base_url: Optional[str] = None, streams[1],
describe_all_responses: bool = False, self.mcp_server.create_initialization_options(
describe_full_response_schema: bool = False, notification_options=None, experimental_capabilities={}
) -> Server: ),
""" )
Add an MCP server to a FastAPI app.
Args: # Mount the MCP connection handler
app: The FastAPI application self.fastapi.get(self._mount_path)(handle_mcp_connection)
mount_path: Path where the MCP server will be mounted self.fastapi.mount(f"{self._mount_path}/messages/", app=sse_transport.handle_post_message)
name: Name for the MCP server (defaults to app.title)
description: Description for the MCP server (defaults to app.description)
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
describe_full_response_schema: Whether to include full json schema for responses in tool descriptions
Returns:
The MCP server instance that was created and mounted
"""
# Create MCP server
mcp_server, operation_map = create_mcp_server(
app,
name,
description,
base_url,
describe_all_responses=describe_all_responses,
describe_full_response_schema=describe_full_response_schema,
)
# Mount MCP server
mount_mcp_server(app, mcp_server, operation_map, mount_path, base_url)
return mcp_server