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

Author: Epochex

src/mcp/client/streamable_http.py

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

@@ -0,0 +1,32 @@
+import httpx
+import pytest
+
+from mcp.client.streamable_http import RequestContext, StreamableHTTPTransport
+from mcp.shared._context_streams import create_context_streams
+from mcp.shared.message import SessionMessage
+from mcp.types import CONNECTION_CLOSED, JSONRPCError, JSONRPCRequest
+
+pytestmark = 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