fix: propagate SSE stream errors to waiting requests

Fixes #1401. Also fixes #1789 (closed as duplicate).

When an SSE read timeout occurs during a StreamableHTTP POST request, the
pending send_request call hangs indefinitely. The transport catches the
exception but never sends an error back through the read stream, leaving
the caller blocked on response_stream_reader.receive() with nothing to
receive.

This fix propagates SSE stream failures as JSONRPCError to the waiting
request, keeping failures isolated to the affected request rather than
tearing down the entire session.

Changes:
- _handle_sse_response now sends JSONRPCError when SSE stream ends without
  a complete response
- _handle_reconnection returns bool to indicate success/failure
- handle_get_stream tracks received_events to properly count reconnection
  attempts
- _default_message_handler logs warnings for unhandled exceptions

Author: internet-dot

src/mcp/client/session.py

@@ -57,6 +57,8 @@ async def __call__(
 async def _default_message_handler(
     message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
 ) -> None:
+    if isinstance(message, Exception):
+        logger.warning("Unhandled exception in message handler: %s", message)
     await anyio.lowlevel.checkpoint()
 
 

src/mcp/client/streamable_http.py

@@ -47,7 +47,7 @@
 
 # Reconnection defaults
 DEFAULT_RECONNECTION_DELAY_MS = 1000  # 1 second fallback when server doesn't provide retry
-MAX_RECONNECTION_ATTEMPTS = 2  # Max retry attempts before giving up
+MAX_RECONNECTION_ATTEMPTS = 5  # Max retry attempts before giving up
 
 
 class StreamableHTTPError(Exception):
@@ -197,6 +197,7 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
+                    received_events = False
                     async for sse in event_source.aiter_sse():
                         # Track last event ID for reconnection
                         if sse.id:
@@ -205,10 +206,12 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                         if sse.retry is not None:
                             retry_interval_ms = sse.retry
 
+                        received_events = True
                         await self._handle_sse_event(sse, read_stream_writer)
 
-                    # Stream ended normally (server closed) - reset attempt counter
-                    attempt = 0
+                    # Only reset attempts if we actually received events;
+                    # empty connections count toward MAX_RECONNECTION_ATTEMPTS
+                    attempt = 0 if received_events else attempt + 1
 
             except Exception:  # pragma: lax no cover
                 logger.debug("GET stream error", exc_info=True)
@@ -364,25 +367,36 @@ 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,
         ctx: RequestContext,
         last_event_id: str,
         retry_interval_ms: int | None = None,
         attempt: int = 0,
-    ) -> None:
-        """Reconnect with Last-Event-ID to resume stream after server disconnect."""
+    ) -> bool:
+        """Reconnect with Last-Event-ID to resume stream after server disconnect.
+
+        Returns:
+            True if the response was successfully delivered, False if max
+            reconnection attempts were exceeded without delivering a response.
+        """
         # Bail if max retries exceeded
-        if attempt >= MAX_RECONNECTION_ATTEMPTS:  # pragma: no cover
+        if attempt >= MAX_RECONNECTION_ATTEMPTS:
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
-            return
+            return False
 
         # Always wait - use server value or default
         delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
@@ -419,15 +433,15 @@ async def _handle_reconnection(
                     )
                     if is_complete:
                         await event_source.response.aclose()
-                        return
+                        return True
 
-                # Stream ended again without response - reconnect again (reset attempt counter)
+                # Stream ended again without response - reconnect again
                 logger.info("SSE stream disconnected, reconnecting...")
-                await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
+                return await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, attempt + 1)
         except Exception as e:  # pragma: no cover
             logger.debug(f"Reconnection failed: {e}")
             # Try to reconnect again if we still have an event ID
-            await self._handle_reconnection(ctx, last_event_id, retry_interval_ms, attempt + 1)
+            return await self._handle_reconnection(ctx, last_event_id, retry_interval_ms, attempt + 1)
 
     async def post_writer(
         self,