Skip to content

Commit 680dd7f

Browse files
committed
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 #3307
1 parent 6705402 commit 680dd7f

2 files changed

Lines changed: 56 additions & 2 deletions

File tree

src/mcp/client/streamable_http.py

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -525,9 +525,11 @@ async def _handle_reconnection(
525525
await event_source.response.aclose()
526526
return
527527

528-
# Stream ended again without response - reconnect again (reset attempt counter)
528+
# Stream ended again without response - a clean EOF consumes
529+
# the same per-request budget as a failed reconnect, so a
530+
# no-timeout request (a listen stream) cannot reconnect forever.
529531
logger.info("SSE stream disconnected, reconnecting...")
530-
await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0)
532+
await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, attempt + 1)
531533
except Exception as e: # pragma: no cover
532534
logger.debug(f"Reconnection failed: {e}")
533535
# Try to reconnect again if we still have an event ID

tests/client/test_streamable_http.py

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -748,3 +748,55 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
748748
_abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
749749
)
750750
send.close()
751+
752+
753+
class _PrimingThenEofSSEStream(httpx2.AsyncByteStream):
754+
"""Opens fine, emits one id-bearing event with empty data, then clean EOF.
755+
756+
This is the shape a no-timeout request (a listen stream) sees from a server
757+
that accepts the reconnect, primes the resumption position, and drops the
758+
connection without producing the JSON-RPC response.
759+
"""
760+
761+
def __init__(self, event_id: str) -> None:
762+
self._event = f"id: {event_id}\ndata: \n\n".encode()
763+
764+
async def __aiter__(self) -> AsyncIterator[bytes]:
765+
yield self._event
766+
767+
768+
@pytest.mark.anyio
769+
async def test_clean_eof_reconnects_count_toward_the_request_budget() -> None:
770+
"""A per-request reconnect that opens fine, emits an id-bearing priming event, then hits
771+
clean EOF must consume the reconnection budget instead of resetting it, so the waiter is
772+
resolved with CONNECTION_CLOSED after MAX_RECONNECTION_ATTEMPTS reconnects."""
773+
transport = StreamableHTTPTransport("http://test/mcp")
774+
send, receive = create_context_streams[SessionMessage | Exception](1)
775+
seen_last_event_ids: list[str | None] = []
776+
# evt-0 is the starting position; each reconnect primes the next one then EOFs.
777+
streams: list[httpx2.AsyncByteStream] = [
778+
_PrimingThenEofSSEStream(f"evt-{index}") for index in range(MAX_RECONNECTION_ATTEMPTS + 2)
779+
]
780+
781+
def handler(request: httpx2.Request) -> httpx2.Response:
782+
seen_last_event_ids.append(request.headers.get("last-event-id"))
783+
return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=streams.pop(0))
784+
785+
async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
786+
with anyio.fail_after(5):
787+
await transport._handle_reconnection( # pyright: ignore[reportPrivateUsage]
788+
_abandoned_request_context(http, send), "evt-0", 0
789+
)
790+
reply = await receive.receive()
791+
assert isinstance(reply, SessionMessage)
792+
assert isinstance(reply.message, JSONRPCError)
793+
assert reply.message.id == "listen-1"
794+
assert reply.message.error.code == CONNECTION_CLOSED
795+
# Only the budgeted reconnects happened: starting from evt-0, the first
796+
# reconnect is primed to evt-1 and the second to evt-2, then the budget is
797+
# exhausted without a third request.
798+
assert seen_last_event_ids == ["evt-0", "evt-0"]
799+
# and every priming event id observed - positions advanced only as far as
800+
# the server primed them before the budget ran out.
801+
send.close()
802+
receive.close()

0 commit comments

Comments
 (0)