fix(streamable-http): fail request when resumption can't complete

Author: Epochex

src/mcp/client/streamable_http.py

@@ -19,6 +19,7 @@
 from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
+    CONNECTION_CLOSED,
     INTERNAL_ERROR,
     INVALID_REQUEST,
     PARSE_ERROR,
@@ -366,10 +367,16 @@ async def _handle_sse_response(
         except Exception:
             logger.debug("SSE stream ended", exc_info=True)  # pragma: no cover
 
-        # Stream ended without response - reconnect if we received an event with ID
-        if last_event_id is not None:  # pragma: no branch
-            logger.info("SSE stream disconnected, reconnecting...")
-            await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
+        # Stream ended without a terminal response/error. If the server provided an event id,
+        # try resuming; otherwise fail the request instead of hanging forever.
+        if last_event_id is None:
+            error_data = ErrorData(code=CONNECTION_CLOSED, message="SSE stream disconnected before response completed")
+            error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
+            await ctx.read_stream_writer.send(error_msg)
+            return
+
+        logger.info("SSE stream disconnected, reconnecting...")
+        await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
 
     async def _handle_reconnection(
         self,
@@ -380,7 +387,19 @@ async def _handle_reconnection(
     ) -> None:
         """Reconnect with Last-Event-ID to resume stream after server disconnect."""
         # Bail if max retries exceeded
-        if attempt >= MAX_RECONNECTION_ATTEMPTS:  # pragma: no cover
+        if attempt >= MAX_RECONNECTION_ATTEMPTS:
+            original_request_id = None
+            if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
+                original_request_id = ctx.session_message.message.id
+
+            if original_request_id is not None:
+                error_data = ErrorData(
+                    code=CONNECTION_CLOSED,
+                    message="SSE stream disconnected and could not be resumed",
+                    data={"last_event_id": last_event_id},
+                )
+                error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
+                await ctx.read_stream_writer.send(error_msg)
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
             return
 

tests/interaction/transports/test_hosting_resume.py

@@ -296,6 +296,78 @@ async def call() -> None:
     assert received == snapshot(["before close", "after close"])
 
 
+async def test_a_call_whose_stream_closes_and_cannot_be_resumed_fails_instead_of_hanging() -> None:
+    """If a resumable response stream disconnects and the server session is gone, the client fails
+    the request instead of hanging forever.
+
+    The server closes the call's SSE stream after emitting one related notification. The test then
+    deletes the active server-side session to force the client's reconnect GET to return 404.
+    Without a terminal response/error on the read stream, ClientSession.send_request waits forever
+    (read timeout defaults to None). The transport must surface a request-scoped error when it
+    gives up reconnecting.
+    """
+    reconnect_attempted = anyio.Event()
+    allow_exit = anyio.Event()
+    done = anyio.Event()
+    raised: list[BaseException] = []
+    manager_ref = None
+    deleted_session = False
+
+    mcp = MCPServer("resumable")
+
+    @mcp.tool()
+    async def interrupt(ctx: Context) -> str:
+        await ctx.info("before close")
+        await ctx.close_sse_stream()
+        await allow_exit.wait()
+        return "unreachable"
+
+    async def record_request(request: httpx.Request) -> None:
+        nonlocal deleted_session
+        if request.method != "GET":
+            return
+        if request.headers.get("last-event-id") is None:
+            return
+        reconnect_attempted.set()
+        if deleted_session or manager_ref is None:
+            return
+        session_ids = list(manager_ref._server_instances.keys())
+        if session_ids:  # pragma: no branch
+            del manager_ref._server_instances[session_ids[0]]
+            deleted_session = True
+
+    async with mounted_app(mcp, event_store=SequencedEventStore(), retry_interval=0, on_request=record_request) as (
+        http,
+        manager,
+    ):
+        manager_ref = manager
+        with anyio.fail_after(5):  # pragma: no branch
+            async with (
+                streamable_http_client(f"{BASE_URL}/mcp", http_client=http, terminate_on_close=False) as (r, w),
+                ClientSession(r, w) as session,
+                anyio.create_task_group() as tg,
+            ):
+                await session.initialize()
+
+                async def call() -> None:
+                    try:
+                        await session.call_tool("interrupt", {})
+                    except BaseException as exc:
+                        raised.append(exc)
+                    finally:
+                        done.set()
+
+                tg.start_soon(call)
+                await reconnect_attempted.wait()
+                await done.wait()
+                allow_exit.set()
+                tg.cancel_scope.cancel()
+
+    assert len(raised) == 1
+    assert isinstance(raised[0], Exception)
+    assert "disconnected" in str(raised[0]).lower()
+
+
 @requirement("client-transport:http:resume-stream-api")
 async def test_a_captured_resumption_token_replays_missed_messages_on_a_new_connection() -> None:
     """A resumption token captured via on_resumption_token_update on one connection lets a fresh