From b3a8a7e4a47bbae53bc65a90404c9f5edec9958b Mon Sep 17 00:00:00 2001 From: David Hayes Date: Sat, 21 Feb 2026 13:12:51 +0000 Subject: [PATCH 1/2] fix: stop infinite GET stream reconnection on repeated empty connections Only reset the attempt counter when events were actually received during the connection. Connections that close immediately without delivering events now count toward MAX_RECONNECTION_ATTEMPTS. Github-Issue:#1401 --- src/mcp/client/streamable_http.py | 12 +++++++++--- tests/client/test_streamable_http.py | 23 +++++++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 5c3501d8c4..7fa60a0728 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: httpx.AsyncClient, read_stream_writer: event_source.response.raise_for_status() logger.debug("GET SSE connection established") + received_events = False async for sse in event_source.aiter_sse(): + received_events = True # Track last event ID for reconnection if sse.id: last_event_id = sse.id @@ -224,14 +226,18 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer: await self._handle_sse_event(sse, read_stream_writer) - # Stream ended normally (server closed) - reset attempt counter - attempt = 0 + # Only reset attempts if we actually received events; + # empty connections count toward MAX_RECONNECTION_ATTEMPTS + if received_events: + attempt = 0 + 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 da17a71c1e..5be0307aef 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -748,3 +748,26 @@ 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() + + +@pytest.mark.anyio +async def test_get_stream_gives_up_after_repeated_empty_connections(monkeypatch: pytest.MonkeyPatch) -> None: + """A GET stream that keeps opening but closing with no events counts each empty connection toward + the reconnection budget, so the loop terminates instead of reconnecting forever.""" + monkeypatch.setattr("mcp.client.streamable_http.DEFAULT_RECONNECTION_DELAY_MS", 0) + get_requests = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal get_requests + get_requests += 1 + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=b"") + + transport = StreamableHTTPTransport("http://test/mcp") + transport.session_id = "sess-1" + send, receive = create_context_streams[SessionMessage | Exception](1) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http: + with anyio.fail_after(5): + await transport.handle_get_stream(http, send) + assert get_requests == MAX_RECONNECTION_ATTEMPTS + send.close() + receive.close() From 85b6fed3210ceb2f7f0874aaf39ee882ece46540 Mon Sep 17 00:00:00 2001 From: David Hayes Date: Sat, 21 Feb 2026 13:12:09 +0000 Subject: [PATCH 2/2] Log warning for unhandled exceptions in default message handler Transport errors that are not tied to a specific pending request (e.g., GET stream failures) were silently swallowed by the default message handler. Add a warning log so these exceptions are at least visible in logs as an observability safety net. Github-Issue: #1401 --- src/mcp/client/session.py | 2 ++ tests/client/test_session.py | 11 ++++++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index 097ade1c91..fe9fa8fcda 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -166,6 +166,8 @@ async def __call__( async def _default_message_handler( message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception, ) -> None: + if isinstance(message, Exception): + logger.warning("Unhandled exception in message handler: %s", message) await anyio.lowlevel.checkpoint() diff --git a/tests/client/test_session.py b/tests/client/test_session.py index 507c8f69e3..3384e4925b 100644 --- a/tests/client/test_session.py +++ b/tests/client/test_session.py @@ -38,7 +38,7 @@ from mcp import MCPError from mcp.client import ClientRequestContext from mcp.client.client import Client -from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession +from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession, _default_message_handler from mcp.client.subscriptions import ToolsListChanged, listen from mcp.server import Server, ServerRequestContext from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair @@ -1030,6 +1030,15 @@ async def handler(msg: object) -> None: assert "message_handler raised on transport exception" in caplog.text +@pytest.mark.anyio +async def test_default_message_handler_logs_unhandled_transport_exceptions(caplog: pytest.LogCaptureFixture): + """The default handler has no per-request waiter to resolve, so a transport-level `Exception` + item (e.g. a GET stream failure) would vanish silently; it logs a warning as a safety net.""" + with caplog.at_level("WARNING", logger="client"): + await _default_message_handler(RuntimeError("boom")) + assert "Unhandled exception in message handler" in caplog.text + + @pytest.mark.anyio async def test_message_handler_awaiting_session_traffic_on_transport_exception_completes(): """A `message_handler` that awaits session traffic on a transport `Exception` item completes: