fix(client): avoid stack growth during SSE reconnection

Author: namtran1812

src/mcp/client/streamable_http.py

@@ -545,66 +545,73 @@ async def _handle_reconnection(
         assert isinstance(ctx.session_message.message, JSONRPCRequest)
         original_request_id = ctx.session_message.message.id
 
-        if attempt >= MAX_RECONNECTION_ATTEMPTS:
-            # Resolve on give-up: a request with no read timeout (a listen
-            # stream) would otherwise hang its caller forever.
-            logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
-            await self._resolve_abandoned_request(
-                ctx.read_stream_writer, original_request_id, "SSE stream ended and reconnection attempts were exhausted"
-            )
-            return
+        while True:
+            if attempt >= MAX_RECONNECTION_ATTEMPTS:
+                # Resolve on give-up: a request with no read timeout (a listen
+                # stream) would otherwise hang its caller forever.
+                logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
+                await self._resolve_abandoned_request(
+                    ctx.read_stream_writer,
+                    original_request_id,
+                    "SSE stream ended and reconnection attempts were exhausted",
+                )
+                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
-        await anyio.sleep(delay_ms / 1000.0)
+            # Always wait - use server value or default
+            delay_ms = retry_interval_ms if retry_interval_ms is not None else DEFAULT_RECONNECTION_DELAY_MS
+            await anyio.sleep(delay_ms / 1000.0)
 
-        headers = self._prepare_headers()
-        headers[LAST_EVENT_ID] = last_event_id
+            headers = self._prepare_headers()
+            headers[LAST_EVENT_ID] = last_event_id
 
-        is_sse_response = False
-        try:
-            async with sse_within_origin(
-                ctx.client, self.url, headers=headers, max_event_size=self.max_sse_event_size
-            ) as event_source:
-                event_source.response.raise_for_status()
-                content_type = event_source.response.headers.get("content-type", "").partition(";")[0]
-                is_sse_response = content_type.strip().lower() == "text/event-stream"
-                logger.info("Reconnected to SSE stream")
-
-                # Track for potential further reconnection
-                reconnect_last_event_id: str = last_event_id
-                reconnect_retry_ms = retry_interval_ms
-
-                async for sse in event_source:
-                    if sse.id:  # pragma: no branch
-                        reconnect_last_event_id = sse.id
-                    if sse.retry is not None:
-                        reconnect_retry_ms = sse.retry
+            is_sse_response = False
+            try:
+                async with sse_within_origin(
+                    ctx.client, self.url, headers=headers, max_event_size=self.max_sse_event_size
+                ) as event_source:
+                    event_source.response.raise_for_status()
+                    content_type = event_source.response.headers.get("content-type", "").partition(";")[0]
+                    is_sse_response = content_type.strip().lower() == "text/event-stream"
+                    logger.info("Reconnected to SSE stream")
 
-                    is_complete = await self._handle_sse_event(
-                        sse,
-                        ctx.read_stream_writer,
-                        original_request_id,
-                        ctx.metadata.on_resumption_token_update if ctx.metadata else None,
+                    # Track for potential further reconnection
+                    reconnect_last_event_id: str = last_event_id
+                    reconnect_retry_ms = retry_interval_ms
+
+                    async for sse in event_source:
+                        if sse.id:  # pragma: no branch
+                            reconnect_last_event_id = sse.id
+                        if sse.retry is not None:
+                            reconnect_retry_ms = sse.retry
+
+                        is_complete = await self._handle_sse_event(
+                            sse,
+                            ctx.read_stream_writer,
+                            original_request_id,
+                            ctx.metadata.on_resumption_token_update if ctx.metadata else None,
+                        )
+                        if is_complete:
+                            await event_source.response.aclose()
+                            return
+
+                    # Stream ended again without response - reconnect again (reset attempt counter)
+                    logger.info("SSE stream disconnected, reconnecting...")
+            except httpx2.SSEError as exc:
+                if is_sse_response:
+                    await self._resolve_abandoned_request(
+                        ctx.read_stream_writer, original_request_id, f"SSE stream failed: {exc}"
                     )
-                    if is_complete:
-                        await event_source.response.aclose()
-                        return
+                    return
+                attempt += 1
+                continue
+            except Exception as e:  # pragma: no cover
+                logger.debug(f"Reconnection failed: {e}")
+                attempt += 1
+                continue
 
-                # Stream ended again without response - reconnect again (reset attempt counter)
-                logger.info("SSE stream disconnected, reconnecting...")
-                await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
-        except httpx2.SSEError as exc:
-            if is_sse_response:
-                await self._resolve_abandoned_request(
-                    ctx.read_stream_writer, original_request_id, f"SSE stream failed: {exc}"
-                )
-            else:
-                await self._handle_reconnection(ctx, last_event_id, retry_interval_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)
+            last_event_id = reconnect_last_event_id
+            retry_interval_ms = reconnect_retry_ms
+            attempt = 0
 
     async def post_writer(
         self,

tests/client/test_streamable_http.py

@@ -1279,3 +1279,62 @@ async def test_https_endpoint_redirected_to_plain_http_elsewhere_never_suggests_
 The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust,
 often combined with a trailing-slash difference. Try https://backend.lan:8000/mcp/ instead, or fix the proxy settings.\
 """)
+
+
+@pytest.mark.anyio
+async def test_productive_reconnections_deliver_the_response_beyond_python_stack_depth() -> None:
+    """SDK-defined: productive resumptions must not accumulate Python stack frames.
+
+    Raw HTTP fixtures model a peer closing every response after one checkpoint;
+    the final response must arrive without changing the interpreter recursion limit.
+    """
+    connections = 1500
+    seen_cursors: list[str | None] = []
+    tokens: list[str] = []
+
+    async def on_token(token: str) -> None:
+        tokens.append(token)
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        seen_cursors.append(request.headers.get("last-event-id"))
+        index = len(seen_cursors)
+        payload: dict[str, Any] = (
+            {"jsonrpc": "2.0", "id": "long-call", "result": {}}
+            if index == connections
+            else {
+                "jsonrpc": "2.0",
+                "method": "notifications/message",
+                "params": {"level": "info", "data": index},
+            }
+        )
+        return httpx2.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            content=f"id: checkpoint-{index}\nretry: 0\ndata: {json.dumps(payload)}\n\n".encode(),
+        )
+
+    with anyio.fail_after(5):
+        async with (
+            httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(
+                    JSONRPCRequest(jsonrpc="2.0", id="long-call", method="tools/list"),
+                    metadata=ClientMessageMetadata(on_resumption_token_update=on_token),
+                )
+            )
+            for index in range(1, connections):
+                notification = await read.receive()
+                assert isinstance(notification, SessionMessage)
+                assert notification.message == JSONRPCNotification(
+                    jsonrpc="2.0",
+                    method="notifications/message",
+                    params={"level": "info", "data": index},
+                )
+            response = await read.receive()
+            assert isinstance(response, SessionMessage)
+            assert response.message == JSONRPCResponse(jsonrpc="2.0", id="long-call", result={})
+
+    assert seen_cursors == [None, *(f"checkpoint-{index}" for index in range(1, connections))]
+    assert tokens == [f"checkpoint-{index}" for index in range(1, connections + 1)]