Add working example to easily integrate a FastMCP endpoint to a FastAPI app (which is not trivial)

README.md

@@ -676,7 +676,7 @@ The Context object provides the following capabilities:
 - `ctx.session` - Access to the underlying session for advanced communication (see [Session Properties and Methods](#session-properties-and-methods))
 - `ctx.request_context` - Access to request-specific data and lifespan resources (see [Request Context Properties](#request-context-properties))
 - `await ctx.debug(message)` - Send debug log message
-- `await ctx.info(message)` - Send info log message  
+- `await ctx.info(message)` - Send info log message
 - `await ctx.warning(message)` - Send warning log message
 - `await ctx.error(message)` - Send error log message
 - `await ctx.log(level, message, logger_name=None)` - Send log with custom level
@@ -1106,13 +1106,13 @@ The session object accessible via `ctx.session` provides advanced control over c
 async def notify_data_update(resource_uri: str, ctx: Context) -> str:
     """Update data and notify clients of the change."""
     # Perform data update logic here
-    
+
     # Notify clients that this specific resource changed
     await ctx.session.send_resource_updated(AnyUrl(resource_uri))
-    
+
     # If this affects the overall resource list, notify about that too
     await ctx.session.send_resource_list_changed()
-    
+
     return f"Updated {resource_uri} and notified clients"
 ```
 
@@ -1141,11 +1141,11 @@ def query_with_config(query: str, ctx: Context) -> str:
     """Execute a query using shared database and configuration."""
     # Access typed lifespan context
     app_ctx: AppContext = ctx.request_context.lifespan_context
-    
+
     # Use shared resources
     connection = app_ctx.db
     settings = app_ctx.config
-    
+
     # Execute query with configuration
     result = connection.execute(query, timeout=settings.query_timeout)
     return str(result)
@@ -1548,6 +1548,10 @@ app = Starlette(
 _Full example: [examples/snippets/servers/streamable_http_path_config.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_http_path_config.py)_
 <!-- /snippet-source -->
 
+#### Mounting to a FastAPI server
+
+To mount a MCP Streamable HTTP endpoint on a FastAPI application see [`examples/servers/fastapi/`](examples/servers/fastapi/).
+
 #### SSE servers
 
 > **Note**: SSE transport is being superseded by [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http).

examples/servers/fastapi/README.md

@@ -0,0 +1,13 @@
+# FastAPI app with MCP endpoint
+
+A FastAPI application with a Streamable HTTP MCP endpoint mounted on `/mcp`.
+
+The key difference when mounting on FastAPI vs Starlette is that you must manually call `mcp.session_manager.run()` in your FastAPI lifespan, as FastAPI doesn't automatically trigger the lifespan of mounted sub-applications.
+
+## Usage
+
+Start the server:
+
+```bash
+uv run uvicorn main:app --reload
+```

examples/servers/fastapi/main.py

@@ -0,0 +1,81 @@
+"""Example showing how to mount FastMCP on a FastAPI application."""
+
+import contextlib
+from collections.abc import AsyncIterator
+
+from fastapi import FastAPI
+from mcp.server.fastmcp import FastMCP
+
+
+@contextlib.asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncIterator[None]:
+    """FastAPI lifespan that initializes the MCP session manager.
+
+    This is necessary because FastAPI doesn't automatically trigger
+    the lifespan of mounted sub-applications. We need to manually
+    manage the session_manager's lifecycle.
+    """
+    async with mcp.session_manager.run():
+        yield
+
+
+# Create FastAPI app with lifespan
+app = FastAPI(
+    title="My API with MCP",
+    description="Example FastAPI application with mounted MCP endpoint",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+# Create FastMCP instance
+mcp = FastMCP(
+    "MCP Tools",
+    debug=True,
+    streamable_http_path="/",
+    json_response=True,
+    stateless_http=False,  # Required when deploying in production with multiple workers
+)
+
+
+@mcp.tool()
+async def process_data(data: str) -> str:
+    """Process some data and return the result
+
+    Args:
+        data: The data to process
+
+    Returns:
+        The processed data
+    """
+    return f"Processed: {data}"
+
+
+# Get the MCP ASGI Starlette app
+mcp_app = mcp.streamable_http_app()
+
+# Mount the MCP app on FastAPI at /mcp
+app.mount("/mcp", mcp_app)
+
+
+# Add regular FastAPI endpoints
+@app.get("/")
+async def root() -> dict[str, str]:
+    """Root endpoint with API information"""
+    return {
+        "mcp_endpoint": "/mcp",
+        "docs": "/docs",
+        "openapi": "/openapi.json",
+    }
+
+
+@app.post("/hello")
+async def hello(name: str = "World") -> dict[str, str]:
+    """Example FastAPI endpoint
+
+    Args:
+        name: Name to greet
+
+    Returns:
+        A greeting message
+    """
+    return {"message": f"Hello, {name}!"}

examples/servers/fastapi/pyproject.toml

@@ -0,0 +1,23 @@
+[project]
+name = "fastapi-mcp"
+version = "0.1.0"
+description = "A simple FastAPI application with a MCP endpoint."
+readme = "README.md"
+requires-python = ">=3.10"
+keywords = ["mcp", "llm", "fastapi"]
+license = { text = "MIT" }
+dependencies = [
+    "mcp",
+    "fastapi>=0.123.0",
+]
+
+[tool.ruff.lint]
+select = ["E", "F", "I"]
+ignore = []
+
+[tool.ruff]
+line-length = 120
+target-version = "py310"
+
+[dependency-groups]
+dev = ["pyright>=1.1.378", "pytest>=8.3.3", "ruff>=0.6.9"]

pyproject.toml

@@ -94,6 +94,7 @@ packages = ["src/mcp"]
 [tool.pyright]
 typeCheckingMode = "strict"
 include = ["src/mcp", "tests", "examples/servers", "examples/snippets"]
+exclude = ["examples/servers/fastapi"]
 venvPath = "."
 venv = ".venv"
 # The FastAPI style of using decorators in tests gives a `reportUnusedFunction` error.