Is your feature request related to a problem? Please describe.
Found that it has been useful in my fork of this repo to split out the creation of the starlette app into its own function so I can create it and customize it in my code before running it.
Describe the solution you'd like
async def _create_sse_app(self):
"""Run the server using SSE transport."""
from starlette.applications import Starlette
from starlette.routing import Mount, Route
sse = SseServerTransport("/messages/")
async def handle_sse(request):
async with sse.connect_sse(
request.scope, request.receive, request._send
) as streams:
await self._mcp_server.run(
streams[0],
streams[1],
self._mcp_server.create_initialization_options(),
)
starlette_app = Starlette(
debug=self.settings.debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse.handle_post_message),
],
)
return starlette_app
async def run_sse_async(self) -> None:
"""Run the server using SSE transport."""
starlette_app = self._create_sse_app()
# ... rest is sameand then use this like:
mcp_server = FastMCP("example", transport="sse")
#...server set up
app = mcp_server._create_sse_app()
app.add_route(...)
app.add_middleware(...)
# run with this or the uvicorn cli
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)Describe alternatives you've considered
none since it is currently contained in the run_sse_async function
Is your feature request related to a problem? Please describe.
Found that it has been useful in my fork of this repo to split out the creation of the starlette app into its own function so I can create it and customize it in my code before running it.
Describe the solution you'd like
and then use this like:
Describe alternatives you've considered
none since it is currently contained in the run_sse_async function