diff --git a/docs/deployment/asgi.mdx b/docs/deployment/asgi.mdx
new file mode 100644
index 000000000..f68f19435
--- /dev/null
+++ b/docs/deployment/asgi.mdx
@@ -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.
+
+
+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.
+
+
+### 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.
+
+
+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.
+
+## 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.
+
+
+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.
+
+
+
+## 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.
\ No newline at end of file
diff --git a/docs/deployment/running-server.mdx b/docs/deployment/running-server.mdx
index 3c1e976fa..ec480aa1f 100644
--- a/docs/deployment/running-server.mdx
+++ b/docs/deployment/running-server.mdx
@@ -189,4 +189,25 @@ if __name__ == "__main__":
```
-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.
\ No newline at end of file
+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()
+```
\ No newline at end of file
diff --git a/docs/docs.json b/docs/docs.json
index c35620263..5982c7da6 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -58,6 +58,7 @@
"group": "Deployment",
"pages": [
"deployment/running-server",
+ "deployment/asgi",
"deployment/authentication"
]
},
diff --git a/pyproject.toml b/pyproject.toml
index dea5d446f..ce86e2df7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -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"]
diff --git a/src/fastmcp/low_level/README.md b/src/fastmcp/low_level/README.md
new file mode 100644
index 000000000..929ebd521
--- /dev/null
+++ b/src/fastmcp/low_level/README.md
@@ -0,0 +1 @@
+Patched low-level objects. When possible, we prefer the official SDK, but we patch bugs here if necessary.
\ No newline at end of file
diff --git a/src/fastmcp/low_level/__init__.py b/src/fastmcp/low_level/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/fastmcp/low_level/sse_server_transport.py b/src/fastmcp/low_level/sse_server_transport.py
new file mode 100644
index 000000000..21df959e7
--- /dev/null
+++ b/src/fastmcp/low_level/sse_server_transport.py
@@ -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)
diff --git a/src/fastmcp/server/http.py b/src/fastmcp/server/http.py
index 385ea5280..a6996bdeb 100644
--- a/src/fastmcp/server/http.py
+++ b/src/fastmcp/server/http.py
@@ -13,7 +13,6 @@
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.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 @@ async def sse_endpoint(request: Request) -> Response:
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 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
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,
+ )
diff --git a/tests/client/test_sse.py b/tests/client/test_sse.py
index 58ba87c79..9c6b87e20 100644
--- a/tests/client/test_sse.py
+++ b/tests/client/test_sse.py
@@ -5,6 +5,8 @@
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
diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py
index 18455c3ff..117f783f0 100644
--- a/tests/client/test_streamable_http.py
+++ b/tests/client/test_streamable_http.py
@@ -5,6 +5,8 @@
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