Fix stateless HTTP task accumulation causing memory leak (#756)

In stateless mode, each request spawned a `run_stateless_server` task
into the manager's global `_task_group`. After `handle_request()`
completed and `terminate()` was called, the task continued running
inside `app.run()`, blocked on `async for message in
session.incoming_messages`. These zombie tasks accumulated indefinitely,
leaking memory.

Replace the global task group spawn with a request-scoped task group
so that server tasks are automatically cancelled when their request
completes.

Add a regression test that simulates the real blocking behavior of
`app.run()` using `anyio.sleep_forever()` and verifies no tasks
linger in the global task group after requests finish.

Closes #756

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: wiggzz

src/mcp/server/streamable_http_manager.py

@@ -151,7 +151,12 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await self._handle_stateful_request(scope, receive, send)
 
     async def _handle_stateless_request(self, scope: Scope, receive: Receive, send: Send) -> None:
-        """Process request in stateless mode - creating a new transport for each request."""
+        """Process request in stateless mode - creating a new transport for each request.
+
+        Uses a request-scoped task group so the server task is automatically
+        cancelled when the request completes, preventing task accumulation in
+        the manager's global task group.
+        """
         logger.debug("Stateless mode: Creating new transport for this request")
         # No session ID needed in stateless mode
         http_transport = StreamableHTTPServerTransport(
@@ -176,16 +181,23 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                 except Exception:  # pragma: no cover
                     logger.exception("Stateless session crashed")
 
-        # Assert task group is not None for type checking
-        assert self._task_group is not None
-        # Start the server task
-        await self._task_group.start(run_stateless_server)
+        # Use a request-scoped task group instead of the global one.
+        # This ensures the server task is cancelled when the request
+        # finishes, preventing zombie tasks from accumulating.
+        # See: https://github.com/modelcontextprotocol/python-sdk/issues/756
+        async with anyio.create_task_group() as request_tg:
 
-        # Handle the HTTP request and return the response
-        await http_transport.handle_request(scope, receive, send)
+            async def run_request_handler(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED):
+                task_status.started()
+                # Handle the HTTP request and return the response
+                await http_transport.handle_request(scope, receive, send)
+                # Terminate the transport after the request is handled
+                await http_transport.terminate()
+                # Cancel the request-scoped task group to stop the server task
+                request_tg.cancel_scope.cancel()
 
-        # Terminate the transport after the request is handled
-        await http_transport.terminate()
+            await request_tg.start(run_stateless_server)
+            await request_tg.start(run_request_handler)
 
     async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Process request in stateful mode - maintaining session state between requests."""

tests/server/test_streamable_http_manager.py

@@ -268,6 +268,71 @@ async def mock_receive():
             assert len(transport._request_streams) == 0, "Transport should have no active request streams"
 
 
+@pytest.mark.anyio
+async def test_stateless_requests_task_leak():
+    """Test that stateless request tasks don't accumulate in the global task group.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/756
+
+    In the buggy implementation, each stateless request spawns a
+    ``run_stateless_server`` task into the manager's global ``_task_group``.
+    After ``handle_request()`` completes and ``terminate()`` is called, the
+    task continues to run inside ``self.app.run()`` (blocked on
+    ``async for message in session.incoming_messages``). These zombie tasks
+    accumulate forever, leaking memory.
+
+    The fix uses a request-scoped task group so that all child tasks are
+    cancelled when the request finishes.
+    """
+    app = Server("test-stateless-task-leak")
+    manager = StreamableHTTPSessionManager(app=app, stateless=True)
+
+    async with manager.run():
+        # Replace app.run with a coroutine that blocks until cancelled,
+        # simulating the real behavior of app.run which blocks on
+        # ``async for message in session.incoming_messages``.
+        async def blocking_run(*args: Any, **kwargs: Any) -> None:
+            try:
+                await anyio.sleep_forever()
+            except anyio.get_cancelled_exc_class():
+                pass
+
+        app.run = blocking_run  # type: ignore[assignment]
+
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"accept", b"application/json, text/event-stream"),
+            ],
+        }
+
+        async def mock_receive() -> Message:
+            return {"type": "http.request", "body": b"", "more_body": False}
+
+        async def mock_send(message: Message) -> None:
+            pass
+
+        assert manager._task_group is not None
+
+        num_requests = 5
+        for _ in range(num_requests):
+            await manager.handle_request(scope, mock_receive, mock_send)
+
+        # Allow a checkpoint so cancelled tasks can be reaped
+        await anyio.sleep(0.05)
+
+        # After all requests complete, there should be no lingering tasks.
+        # Before the fix, each request left behind a zombie task, so the
+        # count would equal ``num_requests``.
+        assert len(manager._task_group._tasks) == 0, (
+            f"Expected 0 lingering tasks in the global task group but found "
+            f"{len(manager._task_group._tasks)}. Stateless request tasks are leaking."
+        )
+
+
 @pytest.mark.anyio
 async def test_unknown_session_id_returns_404():
     """Test that requests with unknown session IDs return HTTP 404 per MCP spec."""