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

Author: radhakrishnan-op

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

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."""