From 4c04b9238d6cdedd20e64502f5ab381fdd39074c Mon Sep 17 00:00:00 2001 From: mehmet turac Date: Sun, 30 Aug 2026 16:53:18 +0300 Subject: [PATCH] fix(client): count clean-EOF reconnects toward the request retry budget _handle_reconnection recurses with attempt=0 on the clean-EOF path (stream closed after emitting only a priming event, no JSON-RPC response). The exception path correctly passes attempt+1, but the normal EOF path resets the counter. A no-timeout caller such as subscriptions/listen can therefore reconnect indefinitely instead of resolving with CONNECTION_CLOSED after MAX_RECONNECTION_ATTEMPTS. Pass attempt+1 on both paths so clean EOFs and transport exceptions consume the same budget. Fixes #3307 --- src/mcp/client/streamable_http.py | 4 +-- tests/client/test_streamable_http.py | 50 ++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..4e6056d992 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -525,9 +525,9 @@ async def _handle_reconnection( await event_source.response.aclose() return - # Stream ended again without response - reconnect again (reset attempt counter) + # Stream ended without delivering a JSON-RPC response — count toward the budget. 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..5a6c35ce4c 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -748,3 +748,53 @@ 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() + + +class _PrimingOnlySSEStream(httpx2.AsyncByteStream): + """Emits one id-bearing priming event then EOF — a resumable stream that + never delivers a JSON-RPC response.""" + + _counter = 0 + + def __init__(self) -> None: + _PrimingOnlySSEStream._counter += 1 + self._id = f"evt-{_PrimingOnlySSEStream._counter}" + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield f"id: {self._id}\ndata: \n\n".encode() + + async def aclose(self) -> None: + pass + + +@pytest.mark.anyio +async def test_clean_eof_without_response_counts_toward_reconnection_budget() -> None: + """A resumable stream that reaches EOF without delivering a JSON-RPC response + must consume the reconnection budget — MAX_RECONNECTION_ATTEMPTS total HTTP + requests, not an unbounded sequence of resets.""" + _PrimingOnlySSEStream._counter = 0 + request_count = 0 + + def handler(request: httpx2.Request) -> httpx2.Response: + nonlocal request_count + request_count += 1 + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=_PrimingOnlySSEStream(), + ) + + transport = StreamableHTTPTransport("http://test/mcp") + send, receive = create_context_streams[SessionMessage | Exception](1) + async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: + with anyio.fail_after(5): + await transport._handle_reconnection( # pyright: ignore[reportPrivateUsage] + _abandoned_request_context(http, send), "evt-0", 0 + ) + reply = await receive.receive() + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.error.code == CONNECTION_CLOSED + assert request_count == MAX_RECONNECTION_ATTEMPTS + send.close() + receive.close()