NoRaincheck

Python MCP - Mounting to an Existing FastAPI ASGI Server

Python MCP - Mounting to an Existing FastAPI ASGI Server

April 2025

You can mount the SSE server to an existing FastAPI ASGI server dynamically. The way I’ve made it work is as follows:

 1sse = SseServerTransport(...)
 2
 3...
 4app.router.routes.append(Mount("/messages", app=sse.handle_post_message))
 5
 6@app.get("/sse")
 7async def handle_mcp_connection(request: Request):
 8    async with sse.connect(sse(request.scope, request.receive, request.send) as (
 9        read_stream,
10        write_stream,
11    ):
12        await mcp._mcp_server.run(
13            read_stream,
14            write_stream,
15            mcp._mcp_server.create_initialization_options(),
16        )

This seemed to work pretty well, along with having Redis as an intermediary broker for dealing with state.

<< Previous Post

|

Next Post >>

🎲 Random post

|

All posts

#Python #Async #FastAPI #MCP