diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 053f1f10a9..32fc1affeb 100644 --- a/src/mcp/client/streamable_http.py +++ b/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, diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 9087394cd6..69f3796eff 100644 --- a/tests/client/test_streamable_http.py +++ b/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)]