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

Author: Jianke LIN

src/mcp/client/streamable_http.py

@@ -435,18 +435,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

@@ -17,11 +17,21 @@
 from mcp.client import ClientSession
 from mcp.client.streamable_http import (
     MCP_PROTOCOL_VERSION,
+    RequestContext,
     StreamableHTTPTransport,
     _encode_header_value,
     streamable_http_client,
 )
-from mcp.types import JSONRPCMessage, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse
+from mcp.shared._context_streams import create_context_streams
+from mcp.shared.message import SessionMessage
+from mcp.types import (
+    CONNECTION_CLOSED,
+    JSONRPCError,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+    JSONRPCResponse,
+)
 
 
 @pytest.mark.parametrize(
@@ -98,6 +108,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_pinned_transport_ignores_returned_session_id_and_never_opens_get_or_delete() -> None:
     """A server-issued ``Mcp-Session-Id`` never reaches a pinned client's wire: only POSTs are sent.