diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..55e454cdfa 100644 --- a/src/mcp/client/streamable_http.py +++ b/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 diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d21f520daf..baf8c61da3 100644 --- a/tests/client/test_streamable_http.py +++ b/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}"