diff --git a/fastapi_mcp/transport/http.py b/fastapi_mcp/transport/http.py index 47af6f06..a730ee48 100644 --- a/fastapi_mcp/transport/http.py +++ b/fastapi_mcp/transport/http.py @@ -29,6 +29,7 @@ def __init__( self._manager_task: asyncio.Task | None = None self._manager_started = False self._startup_lock = asyncio.Lock() + self._ready_event: asyncio.Event | None = None async def _ensure_session_manager_started(self) -> None: """ @@ -57,10 +58,16 @@ async def _ensure_session_manager_started(self) -> None: security_settings=self.security_settings, ) - # Start the session manager in a background task + # Start the session manager in a background task and wait until its + # task group is initialized. A fixed sleep is unreliable and can + # allow requests before the message router is ready (#259). + ready_event = asyncio.Event() + self._ready_event = ready_event + async def run_session_manager(): try: async with self._session_manager.run(): + ready_event.set() logger.info("StreamableHTTP session manager is running") # Keep running until cancelled await asyncio.Event().wait() @@ -68,15 +75,17 @@ async def run_session_manager(): logger.info("StreamableHTTP session manager is shutting down") raise except Exception: + if not ready_event.is_set(): + ready_event.set() logger.exception("Error in StreamableHTTP session manager") raise self._manager_task = asyncio.create_task(run_session_manager()) + await ready_event.wait() + if self._manager_task.done(): + self._manager_task.result() self._manager_started = True - # Give the session manager a moment to initialize - await asyncio.sleep(0.1) - async def handle_fastapi_request(self, request: Request) -> Response: """ Handle a FastAPI request by delegating to the session manager. @@ -134,3 +143,4 @@ async def shutdown(self) -> None: except asyncio.CancelledError: pass self._manager_started = False + self._ready_event = None diff --git a/tests/test_http_transport.py b/tests/test_http_transport.py new file mode 100644 index 00000000..867e6ce0 --- /dev/null +++ b/tests/test_http_transport.py @@ -0,0 +1,19 @@ +import pytest +from mcp.server.lowlevel.server import Server + +from fastapi_mcp.transport.http import FastApiHttpSessionManager + + +@pytest.mark.anyio +async def test_session_manager_startup_waits_for_task_group() -> None: + """Regression for #259: startup must initialize the session manager task group.""" + mcp_server = Server("test-server") + manager = FastApiHttpSessionManager(mcp_server=mcp_server) + + await manager._ensure_session_manager_started() + + assert manager._manager_started is True + assert manager._session_manager is not None + assert manager._session_manager._task_group is not None + + await manager.shutdown()