diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..a2ed5b036a 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -214,7 +214,9 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer event_source.response.raise_for_status() logger.debug("GET SSE connection established") + saw_event = False async for sse in event_source: + saw_event = True # Track last event ID for reconnection if sse.id: last_event_id = sse.id @@ -224,14 +226,15 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer await self._handle_sse_event(sse, read_stream_writer) - # Stream ended normally (server closed) - reset attempt counter - attempt = 0 + # A clean stream close is retryable only when the server sent an event. + # An empty stream is a terminated endpoint, so count it against the retry budget. + attempt = 0 if saw_event else attempt + 1 except Exception: logger.debug("GET stream error", exc_info=True) attempt += 1 - if attempt >= MAX_RECONNECTION_ATTEMPTS: # pragma: no cover + if attempt >= MAX_RECONNECTION_ATTEMPTS: logger.debug(f"GET stream max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded") return diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d21f520daf..5763975c04 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -736,6 +736,32 @@ async def test_exhausted_reconnection_attempts_resolve_the_request_with_an_error receive.close() +@pytest.mark.anyio +async def test_empty_get_stream_exhausts_reconnection_attempts(monkeypatch: pytest.MonkeyPatch) -> None: + """An empty, cleanly closed GET stream must not reset the reconnection budget.""" + requests: list[httpx2.Request] = [] + + def handler(request: httpx2.Request) -> httpx2.Response: + requests.append(request) + return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=b"") + + async def no_sleep(_delay: float) -> None: + pass + + monkeypatch.setattr("mcp.client.streamable_http.anyio.sleep", no_sleep) + transport = StreamableHTTPTransport("http://test/mcp") + transport.session_id = "session-1" + send, receive = create_context_streams[SessionMessage | Exception](0) + + async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: + with anyio.fail_after(5): + await transport.handle_get_stream(http, send) + + assert len(requests) == 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."""