Documentation Index
Fetch the complete documentation index at: https://docs.mcp-use.com/llms.txt
Use this file to discover all available pages before exploring further.
FastAPI Mounting Guide
You can mount an MCPServer (or FastMCP) instance within an existing FastAPI application. This allows you to serve your MCP tools alongside your existing API endpoints.
Basic Integration
When mounting MCPServer, we need to ensure its internal task group (used for managing SSE streams) is properly initialized. We do this using FastAPI’s lifespan events.
from contextlib import asynccontextmanager
from fastapi import FastAPI
from mcp_use.server import MCPServer
import uvicorn
import anyio
# 1. Create your MCP Server
mcp = MCPServer("my-agent-server")
@mcp.tool()
def add(a: int, b: int) -> int:
return a + b
# 2. Define lifespan to manage MCP server startup/shutdown
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize the SSE task group
async with anyio.create_task_group() as tg:
# Inject the task group into the session manager
mcp.session_manager._task_group = tg
yield
# 3. Create your existing FastAPI app with lifespan
app = FastAPI(title="My Existing App", lifespan=lifespan)
@app.get("/")
def read_root():
return {"message": "Hello from the main app"}
# 4. Mount the MCP Server
mount_path = "/agent"
app.mount(mount_path, mcp.app)
if __name__ == "__main__":
print(f"Serving MCP at http://localhost:8000{mount_path}/mcp")
uvicorn.run(app, port=8000)
Important Considerations
- Lifespan Management: The critical part is wrapping the application lifespan with an
anyio.create_task_group() and assigning it to mcp.session_manager._task_group. This works around FastMCP’s assumption that it controls the main loop.
- Transport: Mounting implies using the Streamable HTTP transport pattern.
- Paths: The
MCPServer has internal routes like /mcp (for SSE/POST) and /docs. When mounted at /agent, these become /agent/mcp and /agent/docs.
Example with Shared State
If you want your MCP tools to access the main application’s state (like a database connection), you can pass that context into the tool or use a closure.
fake_db = {"items": []}
@mcp.tool()
def add_item(name: str):
fake_db["items"].append(name)
return f"Added {name}"