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/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_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: 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()