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()