fix: propagate error when SSE stream ends without a response

When an SSE stream ends prematurely (e.g. due to a read timeout), the
client would hang forever waiting for a response that will never arrive.

Now _handle_sse_response checks the return value of _handle_reconnection
and, if reconnection did not deliver a response, sends a JSONRPCError
with INTERNAL_ERROR to the read stream. This unblocks the waiting
request and surfaces the failure as an MCPError to the caller.

Github-Issue: #1401

Author: heyhayes

src/mcp/client/streamable_http.py

@@ -364,12 +364,18 @@ async def _handle_sse_response(
                     await response.aclose()
                     return  # Normal completion, no reconnect needed
         except Exception:
-            logger.debug("SSE stream ended", exc_info=True)  # pragma: no cover
+            logger.debug("SSE stream error", exc_info=True)
 
-        # Stream ended without response - reconnect if we received an event with ID
-        if last_event_id is not None:  # pragma: no branch
+        # Stream ended without a complete response — attempt reconnection if possible
+        if last_event_id is not None:
             logger.info("SSE stream disconnected, reconnecting...")
-            await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
+            if await self._handle_reconnection(ctx, last_event_id, retry_interval_ms):
+                return  # Reconnection delivered the response
+
+        # No response delivered — unblock the waiting request with an error
+        error_data = ErrorData(code=INTERNAL_ERROR, message="SSE stream ended without a response")
+        error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
+        await ctx.read_stream_writer.send(error_msg)
 
     async def _handle_reconnection(
         self,

tests/shared/test_streamable_http.py

@@ -2256,6 +2256,25 @@ async def test_streamable_http_client_preserves_custom_with_mcp_headers(
                 assert headers_data["content-type"] == "application/json"
 
 
+@pytest.mark.anyio
+async def test_sse_read_timeout_propagates_error(basic_server: None, basic_server_url: str):
+    """SSE read timeout should propagate MCPError instead of hanging."""
+    # Create client with very short SSE read timeout
+    short_timeout = httpx.Timeout(30.0, read=0.5)
+    async with httpx.AsyncClient(timeout=short_timeout, follow_redirects=True) as http_client:
+        async with streamable_http_client(f"{basic_server_url}/mcp", http_client=http_client) as (
+            read_stream,
+            write_stream,
+        ):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize()
+
+                # Read a "slow" resource that takes 2s — longer than our 0.5s read timeout
+                with pytest.raises(MCPError):
+                    with anyio.fail_after(10):
+                        await session.read_resource("slow://test")
+
+
 @pytest.mark.anyio
 async def test_handle_reconnection_returns_false_on_max_attempts():
     """_handle_reconnection returns False when max attempts exceeded."""