ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 05:54:19 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #390 from jlowin/mount

Browse source 
Ensure FastMCP handles nested SSE and SHTTP apps properly in ASGI frameworks
This commit is contained in:
Jeremiah Lowin 2025-05-09 17:45:35 -04:00 • committed by GitHub
commit 5abdf7e686
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 387 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

171
docs/deployment/asgi.mdx Normal file
View file

@ -0,0 +1,171 @@
---
title: Integrating FastMCP in ASGI Applications
sidebarTitle: ASGI Integration
description: Integrate FastMCP servers into existing Starlette, FastAPI, or other ASGI applications
icon: plug
---
While FastMCP provides standalone server capabilities, you can also integrate your FastMCP server into existing web applications. This approach is useful for:
- Adding MCP functionality to an existing website or API
- Mounting MCP servers under specific URL paths
- Combining multiple services in a single application
- Leveraging existing authentication and middleware
Please note that all FastMCP servers have a `run()` method that can be used to start the server. This guide focuses on integration with broader ASGI frameworks.
## ASGI Server
FastMCP servers can be created as [Starlette](https://www.starlette.io/) ASGI apps for straightforward hosting or integration into existing applications.
The first step is to obtain a Starlette application instance from your FastMCP server using either the `streamable_http_app()` (preferred) or `sse_app()` (legacy) methods:
```python
from fastmcp import FastMCP
mcp = FastMCP("MyServer")
@mcp.tool()
def hello(name: str) -> str:
return f"Hello, {name}!"
# Get a Starlette app instance for the preferred transport
http_app = mcp.streamable_http_app() # For Streamable HTTP transport
sse_app = mcp.sse_app() # For SSE transport
```
Both methods return a Starlette application that can be integrated with other ASGI-compatible web frameworks.
The MCP server's endpoint is mounted at the root path `/mcp` for Streamable HTTP transport, and `/sse` for SSE transport, though you can change these paths by passing a `path` argument to the `streamable_http_app()` or `sse_app()` methods:
```python
http_app = mcp.streamable_http_app(path="/custom-mcp-path")
sse_app = mcp.sse_app(path="/custom-sse-path")
```
### Running the Server
To run the FastMCP server, you can use the `uvicorn` ASGI server:
```python
import uvicorn
# (define the app here)
if __name__ == "__main__":
uvicorn.run(http_app, host="0.0.0.0", port=8000)
```
Or, from the command line:
```bash
uvicorn path.to.your.app:http_app --host 0.0.0.0 --port 8000
```
## Starlette Integration
You can mount your FastMCP server in another Starlette application using the `Mount` class.
```python
from fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Mount
# Create your FastMCP server as well as any tools, resources, etc.
mcp = FastMCP("MyServer")
# Create the ASGI app
mcp_app = mcp.streamable_http_app(path='/mcp')
# Create a Starlette app and mount the MCP server
app = Starlette(
routes=[
Mount("/mcp-server", app=mcp_app),
# Add other routes as needed
],
lifespan=mcp_app.router.lifespan_context,
)
```
The MCP endpoint will be available at `/mcp-server/mcp` of the resulting Starlette app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the resulting Starlette app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.
</Warning>
### Nested Mounts
You can create complex routing structures by nesting mounts:
```python
from fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.routing import Mount
# Create your FastMCP server as well as any tools, resources, etc.
mcp = FastMCP("MyServer")
# Create the ASGI app
mcp_app = mcp.streamable_http_app(path='/mcp')
# Create nested application structure
inner_app = Starlette(routes=[Mount("/inner", app=mcp_app)])
app = Starlette(
routes=[Mount("/outer", app=inner_app)],
lifespan=mcp_app.router.lifespan_context,
)
```
In this setup, the MCP server is accessible at the `/outer/inner/mcp` path of the resulting Starlette app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the *outer* Starlette app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.
</Warning>
## FastAPI Integration
FastAPI is built on Starlette, so you can mount your FastMCP server in a similar way:
```python
from fastmcp import FastMCP
from fastapi import FastAPI
from starlette.routing import Mount
# Create your FastMCP server as well as any tools, resources, etc.
mcp = FastMCP("MyServer")
# Create the ASGI app
mcp_app = mcp.streamable_http_app(path='/mcp')
# Create a FastAPI app and mount the MCP server
app = FastAPI(lifespan=mcp_app.router.lifespan_context)
app.mount("/mcp-server", mcp_app)
```
The MCP endpoint will be available at `/mcp-server/mcp` of the resulting FastAPI app.
<Warning>
For Streamable HTTP transport, you **must** pass the lifespan context from the FastMCP app to the resulting FastAPI app, as nested lifespans are not recognized. Otherwise, the FastMCP server's session manager will not be properly initialized.
</Warning>
## Custom Routes
In addition to adding your FastMCP server to an existing ASGI app, you can also add custom web routes to your FastMCP server, which will be exposed alongside the MCP endpoint. To do so, use the `@custom_route` decorator. Note that this is less flexible than using a full ASGI framework, but can be useful for adding simple endpoints like health checks to your standalone server.
```python
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
mcp = FastMCP("MyServer")
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> JSONResponse:
return JSONResponse({"status": "healthy"})
```
These routes will be included in the FastMCP app when mounted in your web application.

23
docs/deployment/running-server.mdx
View file

@ -189,4 +189,25 @@ if __name__ == "__main__":
```
</CodeGroup>
Your client only needs to know the host, port, and "main" path; the message path will be transmitted to it as part of the connection handshake.
Your client only needs to know the host, port, and "main" path; the message path will be transmitted to it as part of the connection handshake.
## Custom Routes
You can also add custom web routes to your FastMCP server, which will be exposed alongside the MCP endpoint. To do so, use the `@custom_route` decorator. Note that this is less flexible than using a full ASGI framework, but can be useful for adding simple endpoints like health checks to your standalone server.
```python
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
mcp = FastMCP("MyServer")
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> JSONResponse:
return JSONResponse({"status": "healthy"})
if __name__ == "__main__":
mcp.run()
```

1
docs/docs.json
View file

@ -58,6 +58,7 @@
"group": "Deployment",
"pages": [
"deployment/running-server",
"deployment/asgi",
"deployment/authentication"
]
},

2
pyproject.toml
View file

@ -83,7 +83,7 @@ asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
asyncio_default_test_loop_scope = "session"
filterwarnings = []
timeout = 5
timeout = 3
[tool.pyright]
include = ["src", "tests"]

1
src/fastmcp/low_level/README.md Normal file
View file

@ -0,0 +1 @@
Patched low-level objects. When possible, we prefer the official SDK, but we patch bugs here if necessary.

0
src/fastmcp/low_level/__init__.py Normal file
View file

104
src/fastmcp/low_level/sse_server_transport.py Normal file
View file

@ -0,0 +1,104 @@
import logging
from contextlib import asynccontextmanager
from typing import Any
from urllib.parse import quote
from uuid import uuid4
import anyio
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from mcp.server.sse import SseServerTransport as LowLevelSSEServerTransport
from mcp.shared.message import SessionMessage
from sse_starlette import EventSourceResponse
from starlette.types import Receive, Scope, Send
logger = logging.getLogger(__name__)
class SseServerTransport(LowLevelSSEServerTransport):
"""
Patched SSE server transport
"""
@asynccontextmanager
async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
"""
See https://github.com/modelcontextprotocol/python-sdk/pull/659/
"""
if scope["type"] != "http":
logger.error("connect_sse received non-HTTP request")
raise ValueError("connect_sse can only handle HTTP requests")
logger.debug("Setting up SSE connection")
read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
write_stream: MemoryObjectSendStream[SessionMessage]
write_stream_reader: MemoryObjectReceiveStream[SessionMessage]
read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
session_id = uuid4()
self._read_stream_writers[session_id] = read_stream_writer
logger.debug(f"Created new session with ID: {session_id}")
# Determine the full path for the message endpoint to be sent to the client.
# scope['root_path'] is the prefix where the current Starlette app
# instance is mounted.
# e.g., "" if top-level, or "/api_prefix" if mounted under "/api_prefix".
root_path = scope.get("root_path", "")
# self._endpoint is the path *within* this app, e.g., "/messages".
# Concatenating them gives the full absolute path from the server root.
# e.g., "" + "/messages" -> "/messages"
# e.g., "/api_prefix" + "/messages" -> "/api_prefix/messages"
full_message_path_for_client = root_path.rstrip("/") + self._endpoint
# This is the URI (path + query) the client will use to POST messages.
client_post_uri_data = (
f"{quote(full_message_path_for_client)}?session_id={session_id.hex}"
)
sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[
dict[str, Any]
](0)
async def sse_writer():
logger.debug("Starting SSE writer")
async with sse_stream_writer, write_stream_reader:
await sse_stream_writer.send(
{"event": "endpoint", "data": client_post_uri_data}
)
logger.debug(f"Sent endpoint event: {client_post_uri_data}")
async for session_message in write_stream_reader:
logger.debug(f"Sending message via SSE: {session_message}")
await sse_stream_writer.send(
{
"event": "message",
"data": session_message.message.model_dump_json(
by_alias=True, exclude_none=True
),
}
)
async with anyio.create_task_group() as tg:
async def response_wrapper(scope: Scope, receive: Receive, send: Send):
"""
The EventSourceResponse returning signals a client close / disconnect.
In this case we close our side of the streams to signal the client that
the connection has been closed.
"""
await EventSourceResponse(
content=sse_stream_reader, data_sender_callable=sse_writer
)(scope, receive, send)
await read_stream_writer.aclose()
await write_stream_reader.aclose()
logging.debug(f"Client session disconnected {session_id}")
logger.debug("Starting SSE response task")
tg.start_soon(response_wrapper, scope, receive, send)
logger.debug("Yielding read and write streams")
yield (read_stream, write_stream)

34
src/fastmcp/server/http.py
View file

@ -13,7 +13,6 @@ from mcp.server.auth.middleware.bearer_auth import (
from mcp.server.auth.provider import OAuthAuthorizationServerProvider
from mcp.server.auth.routes import create_auth_routes
from mcp.server.auth.settings import AuthSettings
from mcp.server.sse import SseServerTransport
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.middleware import Middleware
@ -23,6 +22,7 @@ from starlette.responses import Response
from starlette.routing import Mount, Route
from starlette.types import Receive, Scope, Send
from fastmcp.low_level.sse_server_transport import SseServerTransport
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
@ -110,7 +110,7 @@ def setup_auth_middleware_and_routes(
def create_base_app(
routes: list[Route | Mount],
middleware: list[Middleware],
debug: bool,
debug: bool = False,
lifespan: Callable | None = None,
) -> Starlette:
"""Create a base Starlette app with common middleware and routes.
@ -127,17 +127,12 @@ def create_base_app(
# Always add RequestContextMiddleware as the outermost middleware
middleware.append(Middleware(RequestContextMiddleware))
# Create the app
app_kwargs = {
"debug": debug,
"routes": routes,
"middleware": middleware,
}
if lifespan:
app_kwargs["lifespan"] = lifespan
return Starlette(**app_kwargs)
return Starlette(
routes=routes,
middleware=middleware,
debug=debug,
lifespan=lifespan,
)
def create_sse_app(
@ -224,7 +219,11 @@ def create_sse_app(
routes.extend(cast(list[Route | Mount], additional_routes))
# Create and return the app
return create_base_app(routes, middleware, debug)
return create_base_app(
routes=routes,
middleware=middleware,
debug=debug,
)
def create_streamable_http_app(
@ -305,4 +304,9 @@ def create_streamable_http_app(
yield
# Create and return the app with lifespan
return create_base_app(routes, middleware, debug, lifespan)
return create_base_app(
routes=routes,
middleware=middleware,
debug=debug,
lifespan=lifespan,
)

29
tests/client/test_sse.py
View file

@ -5,6 +5,8 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp.types import TextResourceContents
from starlette.applications import Starlette
from starlette.routing import Mount
from fastmcp.client import Client
from fastmcp.client.transports import SSETransport
@ -90,3 +92,30 @@ async def test_http_headers(sse_server: str):
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
def run_nested_server(host: str, port: int) -> None:
try:
app = fastmcp_server().sse_app()
mount = Starlette(routes=[Mount("/nest-inner", app=app)])
mount2 = Starlette(routes=[Mount("/nest-outer", app=mount)])
server = uvicorn.Server(
config=uvicorn.Config(app=mount2, host=host, port=port, log_level="error")
)
server.run()
except Exception as e:
print(f"Server error: {e}")
sys.exit(1)
sys.exit(0)
async def test_nested_sse_server_resolves_correctly():
# tests patch for
# https://github.com/modelcontextprotocol/python-sdk/pull/659
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=SSETransport(f"{url}/nest-outer/nest-inner/sse")
) as client:
result = await client.ping()
assert result is True

39
tests/client/test_streamable_http.py
View file

@ -5,6 +5,8 @@ from collections.abc import Generator
import pytest
import uvicorn
from mcp.types import TextResourceContents
from starlette.applications import Starlette
from starlette.routing import Mount
from fastmcp.client import Client
from fastmcp.client.transports import StreamableHttpTransport
@ -100,3 +102,40 @@ async def test_http_headers(streamable_http_server: str):
json_result = json.loads(raw_result[0].text)
assert "x-demo-header" in json_result
assert json_result["x-demo-header"] == "ABC"
def run_nested_server(host: str, port: int) -> None:
try:
mcp_app = fastmcp_server().streamable_http_app()
mount = Starlette(routes=[Mount("/nest-inner", app=mcp_app)])
mount2 = Starlette(
routes=[Mount("/nest-outer", app=mount)],
lifespan=mcp_app.router.lifespan_context,
)
server = uvicorn.Server(
config=uvicorn.Config(
app=mount2,
host=host,
port=port,
log_level="error",
lifespan="on",
)
)
server.run()
except Exception as e:
print(f"Server error: {e}")
sys.exit(1)
sys.exit(0)
async def test_nested_streamable_http_server_resolves_correctly():
# tests patch for
# https://github.com/modelcontextprotocol/python-sdk/pull/659
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/mcp")
) as client:
result = await client.ping()
assert result is True