From dfbe6ba37477f97e02fc829e5a16bf3912d1626e Mon Sep 17 00:00:00 2001 From: "OpenSourceMaintenance-Luccc-grok-4.5" Date: Sat, 18 Jul 2026 06:48:00 +0000 Subject: [PATCH 1/4] fix(server): terminate streamable HTTP sessions on manager shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: #2150 — active SSE sessions were cancelled without terminate(), and terminate() did not close _sse_stream_writers, so ASGI/EventSourceResponse could hang on shutdown. Local Atlas work only; no public PR until operator approval. --- src/mcp/server/streamable_http.py | 8 ++++ src/mcp/server/streamable_http_manager.py | 12 +++++ .../test_2150_shutdown_terminates_sessions.md | 42 ++++++++++++++++ tests/server/test_streamable_http_manager.py | 48 +++++++++++++++++++ 4 files changed, 110 insertions(+) create mode 100644 tests/issues/test_2150_shutdown_terminates_sessions.md diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index d316345c7..ce805b46b 100644 --- a/src/mcp/server/streamable_http.py +++ b/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()) diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py index 31f587ee6..8a47df911 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/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 diff --git a/tests/issues/test_2150_shutdown_terminates_sessions.md b/tests/issues/test_2150_shutdown_terminates_sessions.md new file mode 100644 index 000000000..6e03ae81b --- /dev/null +++ b/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. diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index 70440d9d0..aea60dc3e 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/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 From 60e6030426d4ccd224dc48558922c426d512dc68 Mon Sep 17 00:00:00 2001 From: "OpenSourceMaintenance-Luccc-grok-4.5" Date: Sat, 18 Jul 2026 09:43:27 +0000 Subject: [PATCH 2/4] style: satisfy markdownlint on #2150 notes Why: CI pre-commit failed on markdownlint auto-fixes for the issue notes file. --- tests/issues/test_2150_shutdown_terminates_sessions.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/issues/test_2150_shutdown_terminates_sessions.md b/tests/issues/test_2150_shutdown_terminates_sessions.md index 6e03ae81b..d60862b06 100644 --- a/tests/issues/test_2150_shutdown_terminates_sessions.md +++ b/tests/issues/test_2150_shutdown_terminates_sessions.md @@ -1,6 +1,6 @@ # Issue #2150 local verification notes -Upstream: https://github.com/modelcontextprotocol/python-sdk/issues/2150 +Upstream: SHA baseline: 3a6f2996cdd8358957479791e8b26198c07d6a75 ## Bug (still present on main at scout time) From 60c67a4123cf2c8757f932c12121c6ee9948490a Mon Sep 17 00:00:00 2001 From: "OpenSourceMaintenance-Luccc-grok-4.5" Date: Sat, 18 Jul 2026 10:05:48 +0000 Subject: [PATCH 3/4] fix(server): register standalone GET and replay SSE writers for terminate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: cubic P2 on #3125 — terminate only closed writers already in _sse_stream_writers. Standalone GET never registered its writer, and replay only registered after event store returned stream_id, so those ASGI responses could still hang across manager shutdown. Register writers when each response starts (provisional key during replay until stream_id is known), pop on exit, and cover with tests. --- src/mcp/server/streamable_http.py | 22 ++++++++++-- tests/server/test_streamable_http_manager.py | 35 ++++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index ce805b46b..628d2c58c 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -700,6 +700,8 @@ async def _handle_get_request(self, request: Request, send: Send) -> None: # Create SSE stream sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0) + # Register immediately so session terminate() can close this ASGI response. + self._sse_stream_writers[GET_STREAM_KEY] = sse_stream_writer async def standalone_sse_writer(): try: @@ -723,11 +725,13 @@ async def standalone_sse_writer(): await sse_stream_writer.send(event_data) except anyio.ClosedResourceError: # Session teardown can close the stream while the writer is between dequeues. + # Also expected when terminate()/close_sse_stream() closes the writer. pass except Exception: logger.exception("Error in standalone SSE writer") # pragma: no cover finally: logger.debug("Closing standalone SSE writer") + self._sse_stream_writers.pop(GET_STREAM_KEY, None) await self._clean_up_memory_streams(GET_STREAM_KEY) # Create and start EventSourceResponse @@ -744,6 +748,7 @@ async def standalone_sse_writer(): logger.exception("Error in standalone SSE response") await self._clean_up_memory_streams(GET_STREAM_KEY) finally: + self._sse_stream_writers.pop(GET_STREAM_KEY, None) await sse_stream_writer.aclose() await sse_stream_reader.aclose() @@ -884,8 +889,13 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send) # Create SSE stream for replay sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0) + # Provisional key so terminate() can close the response during replay itself + # (before stream_id is known from event_store.replay_events_after). + replay_writer_key: RequestId = f"_replay:{last_event_id}:{id(sse_stream_writer)}" + self._sse_stream_writers[replay_writer_key] = sse_stream_writer async def replay_sender(): + registered_stream_id: RequestId | None = None try: async with sse_stream_writer: # Define an async callback for sending events @@ -899,8 +909,10 @@ async def send_event(event_message: EventMessage) -> None: # If stream ID not in mapping, create it if stream_id and stream_id not in self._request_streams: # pragma: no branch try: - # Register SSE writer so close_sse_stream() can close it + # Re-key from provisional → stream_id for close_sse_stream(stream_id) + self._sse_stream_writers.pop(replay_writer_key, None) self._sse_stream_writers[stream_id] = sse_stream_writer + registered_stream_id = stream_id # Prime the resumed connection so the client sees the stream # is re-registered. The replay→live-tail ordering window here @@ -922,13 +934,18 @@ async def send_event(event_message: EventMessage) -> None: await sse_stream_writer.send(event_data) finally: - self._sse_stream_writers.pop(stream_id, None) + if registered_stream_id is not None: + self._sse_stream_writers.pop(registered_stream_id, None) await self._clean_up_memory_streams(stream_id) except anyio.ClosedResourceError: # pragma: lax no cover # Expected when close_sse_stream() is called logger.debug("Replay SSE stream closed by close_sse_stream()") except Exception: # pragma: lax no cover logger.exception("Error in replay sender") + finally: + self._sse_stream_writers.pop(replay_writer_key, None) + if registered_stream_id is not None: + self._sse_stream_writers.pop(registered_stream_id, None) # Create and start EventSourceResponse response = EventSourceResponse( @@ -942,6 +959,7 @@ async def send_event(event_message: EventMessage) -> None: except Exception: # pragma: lax no cover logger.exception("Error in replay response") finally: + self._sse_stream_writers.pop(replay_writer_key, None) await sse_stream_writer.aclose() await sse_stream_reader.aclose() diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index aea60dc3e..4ccc4df17 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/tests/server/test_streamable_http_manager.py @@ -794,3 +794,38 @@ async def counting_terminate() -> None: assert transport.is_terminated assert transport.mcp_session_id not in manager._server_instances assert not manager._server_instances + + +@pytest.mark.anyio +async def test_terminate_closes_standalone_get_sse_writer_when_registered(): + """GET standalone SSE writers must be registered so terminate can close them (#2150/cubic).""" + transport = StreamableHTTPServerTransport(mcp_session_id="get-session-2150") + send_stream, receive_stream = anyio.create_memory_object_stream[object](1) + # Simulate standalone GET registration under GET_STREAM_KEY + from mcp.server.streamable_http import GET_STREAM_KEY + + transport._sse_stream_writers[GET_STREAM_KEY] = send_stream # type: ignore[assignment] + + await transport.terminate() + + assert transport.is_terminated + assert GET_STREAM_KEY 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_terminate_closes_provisional_replay_sse_writer(): + """Replay writers registered under provisional keys must close on terminate.""" + transport = StreamableHTTPServerTransport(mcp_session_id="replay-session-2150") + send_stream, receive_stream = anyio.create_memory_object_stream[object](1) + key = "_replay:evt-1:123" + transport._sse_stream_writers[key] = send_stream # type: ignore[assignment] + + await transport.terminate() + + assert key not in transport._sse_stream_writers + with pytest.raises(anyio.ClosedResourceError): + await send_stream.send(object()) # type: ignore[arg-type] + await receive_stream.aclose() From a6976e1a981e1550ac5b990bf678ce826e86bb8f Mon Sep 17 00:00:00 2001 From: AgentSmithClaw <13392688435a@gmail.com> Date: Sun, 19 Jul 2026 11:11:11 +0800 Subject: [PATCH 4/4] test(server): cover terminate cleanup of orphaned request streams Exercise terminate()'s remaining _request_streams loop so CI fail-under=100% coverage passes for streamable_http (#2150 / PR #3125). Also mark the defensive registered_stream_id guard as no-branch. --- src/mcp/server/streamable_http.py | 4 +++- tests/server/test_streamable_http_manager.py | 5 +++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py index 628d2c58c..eb7d95dc2 100644 --- a/src/mcp/server/streamable_http.py +++ b/src/mcp/server/streamable_http.py @@ -934,7 +934,9 @@ async def send_event(event_message: EventMessage) -> None: await sse_stream_writer.send(event_data) finally: - if registered_stream_id is not None: + # registered_stream_id is set immediately on try entry; keep the + # guard for defensive cleanup if re-key is later reordered. + if registered_stream_id is not None: # pragma: no branch self._sse_stream_writers.pop(registered_stream_id, None) await self._clean_up_memory_streams(stream_id) except anyio.ClosedResourceError: # pragma: lax no cover diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py index 4ccc4df17..5cca45e98 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/tests/server/test_streamable_http_manager.py @@ -758,11 +758,16 @@ async def test_terminate_closes_active_sse_stream_writers(): 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] + # Orphaned request stream with a key NOT also in _sse_stream_writers: close_sse_stream + # only pops matching request streams, so terminate's remaining-keys loop must run. + req_send, req_recv = anyio.create_memory_object_stream[object](1) + transport._request_streams["req-orphan"] = (req_send, req_recv) # type: ignore[assignment] await transport.terminate() assert transport.is_terminated assert "req-1" not in transport._sse_stream_writers + assert not transport._request_streams with pytest.raises(anyio.ClosedResourceError): await send_stream.send(object()) # type: ignore[arg-type] await receive_stream.aclose()