DI-5377 Preserve stateless task cleanup on MCP SDK 1.30

pyproject.toml

@@ -88,6 +88,7 @@ source = "uv-dynamic-versioning"
 vcs = "git"
 style = "pep440"
 bump = true
+tagged-metadata = true
 
 [project.urls]
 Homepage = "https://modelcontextprotocol.io"

src/mcp/server/streamable_http_manager.py

@@ -121,6 +121,7 @@ def __init__(
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
         self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
+        self._stateless_transports: set[StreamableHTTPServerTransport] = set()
         # Identity of the credential that created each session; requests for a
         # session must present the same credential.
         self._session_owners: dict[str, AuthorizationContext] = {}
@@ -166,11 +167,19 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
                 yield  # Let the application run
             finally:
                 logger.info("StreamableHTTP session manager shutting down")
+                # Close in-flight streams before cancelling the manager. This
+                # lets SSE responses send their final HTTP body chunk.
+                for transport in (*self._server_instances.values(), *self._stateless_transports):
+                    try:
+                        await transport.terminate()
+                    except Exception:  # pragma: no cover
+                        logger.debug("Error terminating transport during shutdown", exc_info=True)
                 # Cancel task group to stop all spawned tasks
                 tg.cancel_scope.cancel()
                 self._task_group = None
                 # Clear any remaining server instances
                 self._server_instances.clear()
+                self._stateless_transports.clear()
                 self._session_owners.clear()
 
     async def handle_request(
@@ -244,15 +253,20 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                 except Exception:  # pragma: no cover
                     logger.exception("Stateless session crashed")
 
-        # The per-request server task only ends once the transport is
-        # terminated, so terminate it even if the request was cancelled.
-        assert self._task_group is not None
+        self._stateless_transports.add(http_transport)
         try:
-            await self._task_group.start(run_stateless_server)
-            await http_transport.handle_request(scope, receive, send)
+            # A request-scoped group ensures its server task cannot outlive a
+            # disconnected client and accumulate in the manager's task group.
+            async with anyio.create_task_group() as request_tg:
+                try:
+                    await request_tg.start(run_stateless_server)
+                    await http_transport.handle_request(scope, receive, send)
+                finally:
+                    with anyio.CancelScope(shield=True):
+                        await http_transport.terminate()
+                    request_tg.cancel_scope.cancel()
         finally:
-            with anyio.CancelScope(shield=True):
-                await http_transport.terminate()
+            self._stateless_transports.discard(http_transport)
 
     async def _handle_stateful_request(
         self,

tests/client/test_http_unicode.py

@@ -35,7 +35,7 @@
 }
 
 
-def run_unicode_server(port: int) -> None:  # pragma: no cover
+def run_unicode_server(server_socket: socket.socket) -> None:  # pragma: no cover
     """Run the Unicode test server in a separate process."""
     # Import inside the function since this runs in a separate process
     from collections.abc import AsyncGenerator
@@ -137,25 +137,34 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     config = uvicorn.Config(
         app=app,
         host="127.0.0.1",
-        port=port,
+        port=server_socket.getsockname()[1],
         log_level="error",
     )
     uvicorn_server = uvicorn.Server(config)
-    uvicorn_server.run()
+    uvicorn_server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def unicode_server_port() -> int:
-    """Find an available port for the Unicode test server."""
+def unicode_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the Unicode test server's port until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
+def unicode_server_port(unicode_server_socket: socket.socket) -> int:
+    return unicode_server_socket.getsockname()[1]
+
+
+@pytest.fixture
+def running_unicode_server(
+    unicode_server_port: int, unicode_server_socket: socket.socket
+) -> Generator[str, None, None]:
     """Start a Unicode test server in a separate process."""
-    proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
+    proc = multiprocessing.Process(
+        target=run_unicode_server, kwargs={"server_socket": unicode_server_socket}, daemon=True
+    )
     proc.start()
 
     # Wait for server to be ready

tests/client/test_notification_response.py

@@ -67,31 +67,38 @@ async def handle_mcp_request(request: Request) -> Response:
     return app
 
 
-def run_non_sdk_server(port: int) -> None:  # pragma: no cover
+def run_non_sdk_server(server_socket: socket.socket) -> None:  # pragma: no cover
     """Run the non-SDK server in a separate process."""
     app = create_non_sdk_server_app()
     config = uvicorn.Config(
         app=app,
         host="127.0.0.1",
-        port=port,
+        port=server_socket.getsockname()[1],
         log_level="error",  # Reduce noise in tests
     )
     server = uvicorn.Server(config=config)
-    server.run()
+    server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def non_sdk_server_port() -> int:
-    """Get an available port for the test server."""
+def non_sdk_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the test server's port until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def non_sdk_server(non_sdk_server_port: int) -> Generator[None, None, None]:
+def non_sdk_server_port(non_sdk_server_socket: socket.socket) -> int:
+    return non_sdk_server_socket.getsockname()[1]
+
+
+@pytest.fixture
+def non_sdk_server(non_sdk_server_port: int, non_sdk_server_socket: socket.socket) -> Generator[None, None, None]:
     """Start a non-SDK server for testing."""
-    proc = multiprocessing.Process(target=run_non_sdk_server, kwargs={"port": non_sdk_server_port}, daemon=True)
+    proc = multiprocessing.Process(
+        target=run_non_sdk_server, kwargs={"server_socket": non_sdk_server_socket}, daemon=True
+    )
     proc.start()
 
     # Wait for server to be ready

tests/server/fastmcp/test_integration.py

@@ -88,11 +88,16 @@ async def handle_generic_notification(
 
 # Common fixtures
 @pytest.fixture
-def server_port() -> int:
-    """Get a free port for testing."""
+def server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve a port for the test server until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -101,7 +106,9 @@ def server_url(server_port: int) -> str:
     return f"http://127.0.0.1:{server_port}"
 
 
-def run_server_with_transport(module_name: str, port: int, transport: str) -> None:  # pragma: no cover
+def run_server_with_transport(
+    module_name: str, server_socket: socket.socket, transport: str
+) -> None:  # pragma: no cover
     """Run server with specified transport."""
     # Get the MCP instance based on module name
     if module_name == "basic_tool":
@@ -135,13 +142,17 @@ def run_server_with_transport(module_name: str, port: int, transport: str) -> No
     else:
         raise ValueError(f"Invalid transport for test server: {transport}")
 
-    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=port, log_level="error"))
-    print(f"Starting {transport} server on port {port}")
-    server.run()
+    server = uvicorn.Server(
+        config=uvicorn.Config(app=app, host="127.0.0.1", port=server_socket.getsockname()[1], log_level="error")
+    )
+    print(f"Starting {transport} server on port {server_socket.getsockname()[1]}")
+    server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def server_transport(request: pytest.FixtureRequest, server_port: int) -> Generator[str, None, None]:
+def server_transport(
+    request: pytest.FixtureRequest, server_port: int, server_socket: socket.socket
+) -> Generator[str, None, None]:
     """Start server in a separate process with specified MCP instance and transport.
 
     Args:
@@ -155,7 +166,7 @@ def server_transport(request: pytest.FixtureRequest, server_port: int) -> Genera
 
     proc = multiprocessing.Process(
         target=run_server_with_transport,
-        args=(module_name, server_port, transport),
+        args=(module_name, server_socket, transport),
         daemon=True,
     )
     proc.start()