From 680dd7f82f0707adf775b0ded77c5a5771144f95 Mon Sep 17 00:00:00 2001 From: yuwk <1729065730@qq.com> Date: Fri, 28 Aug 2026 21:37:38 +0800 Subject: [PATCH] Fix per-request SSE reconnection budget on clean EOF A per-request reconnect that opens successfully, emits an id-bearing priming event, and then reaches EOF without a JSON-RPC response used to recurse with attempt=0, resetting the reconnection budget. A no-timeout request (subscriptions/listen) could therefore reconnect forever instead of resolving its waiter with CONNECTION_CLOSED after MAX_RECONNECTION_ATTEMPTS. The clean EOF path now consumes the same per-request budget as the failed-reconnect path. A regression test drives StreamableHTTPTransport._handle_reconnection() with mock streams that always open fine, emit one id-bearing priming event with empty data, then EOF; the waiter must be resolved with CONNECTION_CLOSED after exactly MAX_RECONNECTION_ATTEMPTS reconnects, and no further request may be sent after the budget is exhausted. Fixes modelcontextprotocol/python-sdk#3307 --- src/mcp/client/streamable_http.py | 6 ++-- tests/client/test_streamable_http.py | 52 ++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..61a3f93a06 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -525,9 +525,11 @@ async def _handle_reconnection( await event_source.response.aclose() return - # Stream ended again without response - reconnect again (reset attempt counter) + # Stream ended again without response - a clean EOF consumes + # the same per-request budget as a failed reconnect, so a + # no-timeout request (a listen stream) cannot reconnect forever. 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..4c3d0ecc38 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -748,3 +748,55 @@ 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 _PrimingThenEofSSEStream(httpx2.AsyncByteStream): + """Opens fine, emits one id-bearing event with empty data, then clean EOF. + + This is the shape a no-timeout request (a listen stream) sees from a server + that accepts the reconnect, primes the resumption position, and drops the + connection without producing the JSON-RPC response. + """ + + def __init__(self, event_id: str) -> None: + self._event = f"id: {event_id}\ndata: \n\n".encode() + + async def __aiter__(self) -> AsyncIterator[bytes]: + yield self._event + + +@pytest.mark.anyio +async def test_clean_eof_reconnects_count_toward_the_request_budget() -> None: + """A per-request reconnect that opens fine, emits an id-bearing priming event, then hits + clean EOF must consume the reconnection budget instead of resetting it, so the waiter is + resolved with CONNECTION_CLOSED after MAX_RECONNECTION_ATTEMPTS reconnects.""" + transport = StreamableHTTPTransport("http://test/mcp") + send, receive = create_context_streams[SessionMessage | Exception](1) + seen_last_event_ids: list[str | None] = [] + # evt-0 is the starting position; each reconnect primes the next one then EOFs. + streams: list[httpx2.AsyncByteStream] = [ + _PrimingThenEofSSEStream(f"evt-{index}") for index in range(MAX_RECONNECTION_ATTEMPTS + 2) + ] + + def handler(request: httpx2.Request) -> httpx2.Response: + seen_last_event_ids.append(request.headers.get("last-event-id")) + return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=streams.pop(0)) + + 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 + # Only the budgeted reconnects happened: starting from evt-0, the first + # reconnect is primed to evt-1 and the second to evt-2, then the budget is + # exhausted without a third request. + assert seen_last_event_ids == ["evt-0", "evt-0"] + # and every priming event id observed - positions advanced only as far as + # the server primed them before the budget ran out. + send.close() + receive.close()