I'm not sure if you've thought about the use case, but we're hoping to be able to use one FastAPI app with multiple routes that all have separate stateless MCP servers.

On your branch this works:

 """Example Echo MCP Server."""

 from fastapi import APIRouter
 from mcp.server.fastmcp import FastMCP
 from starlette.applications import Starlette

 mcp = FastMCP(name="EchoServer")


 @mcp.tool(description="A simple echo tool")
 def echo(message: str) -> str:
     return f"Echo: {message}"


 # Get the Starlette app and convert to router
 app: Starlette = mcp.streamable_http_app()

with a parent app like

from routers import echo

app = FastAPI(lifespan=echo.app.router.lifespan_context)
app.mount("/echo", echo.app)

Although that involves kind of digging into your implementation to know that the lifespan function sets up the session manager and that it ends up internally in the starlette router. It might be helpful to have this exposed on the FastMCP object to make that more clear/explicit? either way it'll take a bit of documentation

from routers import echo

app = FastAPI(lifespan=lambda app: echo.mcp.session_manager.run())
app.mount("/echo", echo.mcp.streamable_http_app())

@baxen, thank you, yeah, haven't considered that use case, would definitely add documentation!
Do you have any suggestions on how to make it more clear/explicit?

i haven't thought of anything better than having the FastMCP object having the StreamableHTTPSessionManager as a member. The error it already raises when run hasn't been called is a good hint, and as a member on FastMCP it'll be easier to find it and call it's run method (like in that last snippet)

right, not super happy to add session_manager but also don't have a better option. Added to fast mcp, also adding to README this example: #664

thank you @baxen, let me know if this works!

nitpick: it might be better to have another elif here and then throw an exception in the else block. If someone introduces a new transport but misses updating this code they might believe their code is working even though it is really using a different transport.

question: what's the intention behind importing these things inside this method? Is it to not force dev to have these dependencies if they are not using this method?

yes, that's right, if one only uses run(transport="stdio"), they never need those dependencies

got it -- makes sense and is pretty clean since this method is only invoked during server startup as far as i understand. we should just be careful that people dont start doing this in methods that are called after server startup since then import errors will only become visible in the runtime.

nitpick: Does this mean that StreamableHTTPSessionManager is not thread safe (two threads using same manager instance will overwrite eachother's self._task_group)? Might be worth documenting.

+1 to this

same, maybe not worth a fix but def worth a note in the comment header

updated, making only once instance being able to use run + docs

+1 to this