test(streamable-http): cover disconnect without resumption anchor

Author: Jianke LIN

src/mcp/client/streamable_http.py

@@ -459,18 +459,15 @@ async def _handle_reconnection(
         """Reconnect with Last-Event-ID to resume stream after server disconnect."""
         # Bail if max retries exceeded
         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)
+            assert isinstance(ctx.session_message.message, JSONRPCRequest)
+            original_request_id = ctx.session_message.message.id
+            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/client/test_streamable_http.py

@@ -18,6 +18,7 @@
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
+    CONNECTION_CLOSED,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     JSONRPCError,
@@ -28,10 +29,11 @@
 from mcp_types.version import LATEST_MODERN_VERSION
 from starlette.types import Receive, Scope, Send
 
-from mcp.client.streamable_http import streamable_http_client
+from mcp.client.streamable_http import RequestContext, StreamableHTTPTransport, streamable_http_client
 from mcp.server import Server
 from mcp.server._streamable_http_modern import handle_modern_request
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, ServerEvent
+from mcp.shared._context_streams import create_context_streams
 from mcp.shared.dispatcher import CallOptions, DispatchContext
 from mcp.shared.inbound import MCP_METHOD_HEADER, MCP_PROTOCOL_VERSION_HEADER, encode_header_value
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
@@ -73,6 +75,30 @@ def test_mcp_name_header_values_are_base64_wrapped_when_unsafe_for_an_http_field
         assert encoded == raw
 
 
+@pytest.mark.anyio
+async def test_sse_response_disconnect_before_any_event_id_fails_request() -> None:
+    transport = StreamableHTTPTransport("http://example.com/mcp")
+    async with httpx.AsyncClient() as client:
+        read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1)
+        request = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "noop", "arguments": {}})
+        ctx = RequestContext(
+            client=client,
+            session_id=None,
+            session_message=SessionMessage(request),
+            metadata=None,
+            read_stream_writer=read_stream_writer,
+        )
+        response = httpx.Response(200, headers={"content-type": "text/event-stream"}, content=b"")
+
+        async with read_stream_writer, read_stream:
+            await transport._handle_sse_response(response, ctx)
+            message = await read_stream.receive()
+
+    assert isinstance(message.message, JSONRPCError)
+    assert message.message.id == 1
+    assert message.message.error.code == CONNECTION_CLOSED
+
+
 @pytest.mark.anyio
 async def test_post_request_merges_per_message_metadata_headers() -> None:
     """`ClientMessageMetadata.headers` on a `SessionMessage` are merged into the outgoing POST headers