fix(client/streamable_http): bound SSE reconnect loop with exponential backoff + jitter

src/mcp/client/streamable_http.py

@@ -4,6 +4,7 @@
 
 import contextlib
 import logging
+import random
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -52,6 +53,23 @@
 # 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_DELAY_MS = 30_000  # Cap exponential backoff at 30s to avoid pathological waits
+RECONNECTION_BACKOFF_FACTOR = 2  # Exponential factor applied per retry attempt
+RECONNECTION_JITTER_RATIO = 0.25  # +/-25% jitter on backoff delay to avoid thundering-herd
+
+
+def _compute_backoff_delay_ms(attempt: int, base_delay_ms: int) -> int:
+    """Compute a bounded exponential-backoff delay with jitter for a retry attempt.
+
+    The base delay (typically the server-provided retry interval, or
+    ``DEFAULT_RECONNECTION_DELAY_MS``) is multiplied by ``2 ** attempt`` and capped
+    at ``MAX_RECONNECTION_DELAY_MS``. Jitter in the +/- ``RECONNECTION_JITTER_RATIO``
+    band is applied to decorrelate reconnects from concurrent clients and avoid a
+    thundering-herd reconnect storm against a freshly-recovered upstream.
+    """
+    exponential = min(base_delay_ms * (RECONNECTION_BACKOFF_FACTOR**attempt), MAX_RECONNECTION_DELAY_MS)
+    jitter = exponential * RECONNECTION_JITTER_RATIO
+    return max(0, int(exponential + random.uniform(-jitter, jitter)))
 
 
 class StreamableHTTPError(Exception):
@@ -224,8 +242,8 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
 
                         await self._handle_sse_event(sse, read_stream_writer)
 
-                    # Stream ended normally (server closed) - reset attempt counter
-                    attempt = 0
+                    # Stream ended normally (server closed) - increment attempt to respect MAX_RECONNECTION_ATTEMPTS
+                    attempt += 1
 
             except Exception:
                 logger.debug("GET stream error", exc_info=True)
@@ -235,8 +253,10 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
                 logger.debug(f"GET stream max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
                 return
 
-            # Wait before reconnecting
-            delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
+            # Wait before reconnecting, using bounded exponential backoff with jitter
+            # to avoid thundering-herd storms when many clients reconnect at once.
+            base_delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
+            delay_ms = _compute_backoff_delay_ms(attempt, base_delay_ms)
             logger.info(f"GET stream disconnected, reconnecting in {delay_ms}ms...")
             await anyio.sleep(delay_ms / 1000.0)
 
@@ -493,8 +513,12 @@ async def _handle_reconnection(
             )
             return
 
-        # Always wait - use server value or default
-        delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
+        # Always wait - use bounded exponential backoff with jitter on top of the
+        # server-provided retry interval (or fallback default) so a freshly
+        # recovered upstream is not pummelled by synchronized reconnects.
+        base_delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
+        delay_ms = _compute_backoff_delay_ms(attempt, base_delay_ms)
+        logger.debug(f"Reconnection attempt {attempt + 1}/{MAX_RECONNECTION_ATTEMPTS}, waiting {delay_ms}ms...")
         await anyio.sleep(delay_ms / 1000.0)
 
         headers = self._prepare_headers()
@@ -525,9 +549,12 @@ async def _handle_reconnection(
                         await event_source.response.aclose()
                         return
 
-                # Stream ended again without response - reconnect again (reset attempt counter)
+                # Stream ended again without response. Do NOT reset the attempt
+                # counter to 0 here; respect MAX_RECONNECTION_ATTEMPTS to prevent
+                # an infinite reconnect loop when an upstream keeps closing the
+                # stream (see issue #3356).
                 logger.info("SSE stream disconnected, reconnecting...")
-                await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
+                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

tests/client/test_streamable_http.py

@@ -34,6 +34,7 @@
     MAX_RECONNECTION_ATTEMPTS,
     RequestContext,
     StreamableHTTPTransport,
+    _compute_backoff_delay_ms,
     streamable_http_client,
 )
 from mcp.server import Server
@@ -748,3 +749,33 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
                 _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
             )
     send.close()
+
+
+@pytest.mark.parametrize(
+    ("attempt", "base_delay_ms", "expected_floor", "expected_ceiling"),
+    [
+        # Attempt 0: a single base delay ± 25% jitter
+        (0, 1000, 750, 1250),
+        # Attempt 1: 2x base = 2000ms ± 25%
+        (1, 1000, 1500, 2500),
+        # Attempt 3: 8x = 8000ms ± 25%
+        (3, 1000, 6000, 10000),
+        # Large attempt is capped at MAX_RECONNECTION_DELAY_MS (30s) ± 25%
+        (10, 1000, 22500, 37500),
+    ],
+)
+def test_compute_backoff_delay_ms_grows_then_caps(
+    attempt: int, base_delay_ms: int, expected_floor: int, expected_ceiling: int
+) -> None:
+    """Bounded exponential backoff: doubles per attempt, capped, with ±25% jitter."""
+    for _ in range(50):  # sample jitter range
+        delay = _compute_backoff_delay_ms(attempt, base_delay_ms)
+        assert expected_floor <= delay <= expected_ceiling, (
+            f"attempt={attempt} base={base_delay_ms} -> {delay} not in [{expected_floor}, {expected_ceiling}]"
+        )
+
+
+def test_compute_backoff_delay_ms_no_regression_for_zero_attempt() -> None:
+    """Attempt 0 stays close to base (no doubling) so a healthy re-establishment isn't penalised."""
+    samples = [_compute_backoff_delay_ms(0, 100) for _ in range(20)]
+    assert all(75 <= s <= 125 for s in samples), f"jittered 100ms base out of range: {samples}"