fix(server): terminate streamable HTTP sessions on manager shutdown

src/mcp/server/streamable_http.py

@@ -774,11 +774,19 @@ async def terminate(self) -> None:
         """Terminate the current session, closing all streams.
 
         Once terminated, all requests with this session ID will receive 404 Not Found.
+
+        Active SSE writers are closed first so EventSourceResponse / ASGI callables can
+        complete instead of hanging until the task group is cancelled (see #2150).
         """
 
         self._terminated = True
         logger.info(f"Terminating session: {self.mcp_session_id}")
 
+        # Close SSE stream writers first so long-lived GET/POST SSE responses finish.
+        # Copy keys: close_sse_stream mutates the dict (includes GET_STREAM_KEY).
+        for request_id in list(self._sse_stream_writers.keys()):
+            self.close_sse_stream(request_id)
+
         # We need a copy of the keys to avoid modification during iteration
         request_stream_keys = list(self._request_streams.keys())
 

src/mcp/server/streamable_http_manager.py

@@ -153,6 +153,18 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
                 yield  # Let the application run
             finally:
                 logger.info("StreamableHTTP session manager shutting down")
+                # Terminate active transports before cancelling the task group so
+                # in-flight SSE responses can complete cleanly (issue #2150).
+                active_transports = list(self._server_instances.values())
+                for transport in active_transports:
+                    if not transport.is_terminated:  # pragma: no branch
+                        try:
+                            await transport.terminate()
+                        except Exception:  # pragma: no cover
+                            logger.exception(
+                                "Error terminating streamable HTTP session %s during shutdown",
+                                transport.mcp_session_id,
+                            )
                 # Cancel task group to stop all spawned tasks
                 tg.cancel_scope.cancel()
                 self._task_group = None

tests/issues/test_2150_shutdown_terminates_sessions.md

@@ -0,0 +1,42 @@
+# Issue #2150 local verification notes
+
+Upstream: https://github.com/modelcontextprotocol/python-sdk/issues/2150
+SHA baseline: 3a6f2996cdd8358957479791e8b26198c07d6a75
+
+## Bug (still present on main at scout time)
+
+1. `StreamableHTTPSessionManager.run()` finally only:
+   - `tg.cancel_scope.cancel()`
+   - `_server_instances.clear()`
+   without calling `transport.terminate()` on active sessions.
+
+2. `StreamableHTTPServerTransport.terminate()` closed request/read/write streams but
+   did **not** close `_sse_stream_writers`, leaving EventSourceResponse hung.
+
+## Fix (local branch `atlas/fix-2150-shutdown-sessions`)
+
+1. Manager shutdown: terminate each non-terminated transport before cancel.
+2. Transport.terminate: close all SSE writers via close_sse_stream / close_standalone_sse_stream first.
+
+## Tests added
+
+- `test_terminate_closes_active_sse_stream_writers`
+- `test_manager_shutdown_terminates_active_sessions`
+
+in `tests/server/test_streamable_http_manager.py`.
+
+## Local env note
+
+This Atlas sandbox lacked `pip`/`uv`; tests were not executed here. Run upstream:
+
+```bash
+uv sync
+uv run pytest tests/server/test_streamable_http_manager.py -k 2150 -q
+# or by test name:
+uv run pytest tests/server/test_streamable_http_manager.py::test_terminate_closes_active_sse_stream_writers -q
+uv run pytest tests/server/test_streamable_http_manager.py::test_manager_shutdown_terminates_active_sessions -q
+```
+
+## Publication
+
+Operator approval: none. Do not open PR until approved.

tests/server/test_streamable_http_manager.py

@@ -746,3 +746,51 @@ async def test_anonymous_session_accepts_anonymous_requests(
     session_id = await _open_session(manager, None)
 
     assert await _request_session(manager, session_id, None) != 404
+
+
+@pytest.mark.anyio
+async def test_terminate_closes_active_sse_stream_writers():
+    """Regression for #2150: terminate must close SSE writers so ASGI can finish.
+
+    Without this, manager shutdown cancels the task group while EventSourceResponse
+    is still open and uvicorn logs "ASGI callable returned without completing response".
+    """
+    transport = StreamableHTTPServerTransport(mcp_session_id="test-session-2150")
+    send_stream, receive_stream = anyio.create_memory_object_stream[object](1)
+    transport._sse_stream_writers["req-1"] = send_stream  # type: ignore[assignment]
+
+    await transport.terminate()
+
+    assert transport.is_terminated
+    assert "req-1" not in transport._sse_stream_writers
+    with pytest.raises(anyio.ClosedResourceError):
+        await send_stream.send(object())  # type: ignore[arg-type]
+    await receive_stream.aclose()
+
+
+@pytest.mark.anyio
+async def test_manager_shutdown_terminates_active_sessions():
+    """Regression for #2150: run() finally should terminate tracked transports."""
+    app = Server("test-shutdown-terminate")
+    manager = StreamableHTTPSessionManager(app=app)
+    transport = StreamableHTTPServerTransport(mcp_session_id="shutdown-session")
+    # Inject a live session as if a client still held an SSE connection.
+    manager._server_instances[transport.mcp_session_id] = transport  # type: ignore[index]
+    original_terminate = transport.terminate
+    terminate_calls = 0
+
+    async def counting_terminate() -> None:
+        nonlocal terminate_calls
+        terminate_calls += 1
+        await original_terminate()
+
+    transport.terminate = counting_terminate  # type: ignore[method-assign]
+
+    async with manager.run():
+        assert transport.mcp_session_id in manager._server_instances
+        # Exit context -> shutdown path should terminate then clear.
+
+    assert terminate_calls == 1
+    assert transport.is_terminated
+    assert transport.mcp_session_id not in manager._server_instances
+    assert not manager._server_instances