From b68785a1cf3496fef6af39f574f43b5b7653b67f Mon Sep 17 00:00:00 2001 From: Radhakrishnan P Date: Mon, 17 Aug 2026 09:58:00 +0530 Subject: [PATCH] fix(streamable-http): count bare-priming-then-EOF reconnects against the request budget A reconnect that reaches EOF without delivering any real data (only a bare id-bearing priming event) was resetting the attempt counter to 0 instead of incrementing it. This let a server that repeatedly opened the resumable stream, emitted only a priming event, and closed again reconnect forever rather than giving up after MAX_RECONNECTION_ATTEMPTS and resolving the waiter with CONNECTION_CLOSED. Track whether any event with non-empty data was received during the reconnect. A reconnect that made real progress (delivered a notification) still earns a fresh budget for the next reconnect; a reconnect that saw only bare priming events counts against the budget the same way a transport exception does. Adds a regression test that drives _handle_reconnection with a mock transport returning priming-then-EOF on every reconnect and asserts the waiter resolves with CONNECTION_CLOSED after exactly MAX_RECONNECTION_ATTEMPTS attempts. Fixes #3307 --- src/mcp/client/streamable_http.py | 13 ++++++-- tests/client/test_streamable_http.py | 50 ++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..2074554bd5 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -508,6 +508,7 @@ async def _handle_reconnection( # Track for potential further reconnection reconnect_last_event_id: str = last_event_id reconnect_retry_ms = retry_interval_ms + made_progress = False async for sse in event_source: if sse.id: # pragma: no branch @@ -525,9 +526,17 @@ async def _handle_reconnection( await event_source.response.aclose() return - # Stream ended again without response - reconnect again (reset attempt counter) + # A real event (notification) earns a fresh budget for the next + # reconnect. A bare priming event (empty data) does not. + if sse.data: + made_progress = True + + # Stream ended again without response. Reset the budget only when this + # reconnect actually delivered a real event; bare-priming-then-EOF counts + # against the budget the same way a transport exception does. logger.info("SSE stream disconnected, reconnecting...") - await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0) + next_attempt = 0 if made_progress else attempt + 1 + await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, next_attempt) 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..4cf02bc86f 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -736,6 +736,56 @@ async def test_exhausted_reconnection_attempts_resolve_the_request_with_an_error receive.close() +class _PrimingOnlySSEStream(httpx2.AsyncByteStream): + """Emits a single bare priming event (id only, empty data) then closes.""" + + def __init__(self, event_id: str) -> None: + self._bytes = f"id: {event_id}\n\n".encode() + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield self._bytes + + async def aclose(self) -> None: + pass + + +@pytest.mark.anyio +async def test_empty_resumable_sse_reconnects_count_toward_the_request_budget() -> None: + """A reconnect that delivers only a bare priming event and reaches EOF must consume + the reconnect budget, the same as the exception path. + + Before the fix, the clean-EOF branch always reset attempt to 0, so a server that + kept sending only priming events could reconnect forever. The fix increments the + counter when no real data arrived during the reconnect, giving up after exactly + MAX_RECONNECTION_ATTEMPTS attempts and resolving the waiter with CONNECTION_CLOSED.""" + call_count = 0 + + def handler(request: httpx2.Request) -> httpx2.Response: + nonlocal call_count + call_count += 1 + return httpx2.Response( + 200, + headers={"content-type": "text/event-stream"}, + stream=_PrimingOnlySSEStream(f"evt-{call_count}"), + ) + + 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.id == "listen-1" + assert reply.message.error.code == CONNECTION_CLOSED + assert call_count == MAX_RECONNECTION_ATTEMPTS + send.close() + receive.close() + + @pytest.mark.anyio async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contained() -> None: """Teardown race: a stream dying after the reader closed resolves best-effort and must not crash."""