From 4a94c918fd321526e06fbcbcdf5c606ff742e018 Mon Sep 17 00:00:00 2001 From: Zeus Date: Mon, 31 Aug 2026 17:29:59 +0000 Subject: [PATCH] fix(client): re-raise transport exceptions in default message handler When a transport-level error (e.g. httpx.ReadTimeout from streamablehttp_client when sse_read_timeout fires) is yielded onto the read stream, the default ClientSession message handler silently checkpointed and continued. That left send_request awaiters to hang indefinitely because no response ever arrived on the per-request response stream. The user-observable contract of message_handler is that it can receive an Exception item, but the default implementation quietly swallowed it. Re-raise so the dispatcher's observer (and any user-supplied handler that mirrors this default) actually surfaces the failure. Refs modelcontextprotocol/python-sdk#1401 --- src/mcp/client/session.py | 10 ++++++++ tests/client/test_session.py | 44 ++++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py index f18cc0ef10..af7a04c792 100644 --- a/src/mcp/client/session.py +++ b/src/mcp/client/session.py @@ -250,6 +250,16 @@ async def __call__(self, message: IncomingMessage) -> None: ... # pragma: no br async def _default_message_handler(message: IncomingMessage) -> None: + """Default handler for incoming server notifications and transport-level exceptions. + + Server notifications are silently acknowledged; transport-level Exception items + (e.g. httpx.ReadTimeout from streamablehttp_client when sse_read_timeout + fires) are re-raised so they propagate out of the receive loop instead of leaving + send_request awaiters to hang indefinitely. See + https://github.com/modelcontextprotocol/python-sdk/issues/1401. + """ + if isinstance(message, Exception): + raise message await anyio.lowlevel.checkpoint() diff --git a/tests/client/test_session.py b/tests/client/test_session.py index 6663fb47a2..241639d4c8 100644 --- a/tests/client/test_session.py +++ b/tests/client/test_session.py @@ -2061,3 +2061,47 @@ def test_intercept_consumes_acks_for_live_routes_and_leaves_malformed_ones(): # Events deliver but are never consumed - they still tee to message_handler. assert intercept("notifications/tools/list_changed", meta) is False assert list(route._pending) == [ToolsListChanged()] # pyright: ignore[reportPrivateUsage] + + +@pytest.mark.anyio +async def test_default_message_handler_raises_on_transport_exception(caplog): + """The default ``message_handler`` re-raises ``Exception`` items it receives from the + transport stream instead of silently swallowing them. See + https://github.com/modelcontextprotocol/python-sdk/issues/1401 for the silent-hang + case this addresses: a transport error (e.g. ``httpx.ReadTimeout``) used to be + checkpoint-and-forget, leaving ``send_request`` awaiters to hang indefinitely.""" + import logging + + from mcp.client.session import _default_message_handler + + boom = RuntimeError("transport went away") + + with caplog.at_level(logging.ERROR, logger="client"): + with pytest.raises(RuntimeError, match="transport went away"): + await _default_message_handler(boom) + + # And a non-exception payload still passes through cleanly. + await _default_message_handler(types.ToolListChangedNotification()) + + +@pytest.mark.anyio +async def test_transport_exception_in_stream_logs_at_error_level(caplog): + """End-to-end: injecting a transport ``Exception`` into the read stream of a + live ``ClientSession`` is now logged at ERROR level (no longer silently + swallowed). See #1401.""" + import logging + + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1) + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage](1) + + try: + with caplog.at_level(logging.ERROR, logger="client"): + async with ClientSession(s2c_recv, c2s_send): + await s2c_send.send(RuntimeError("sse_read_timeout fired")) + # Give the dispatcher a tick to route the exception through message_handler. + await anyio.sleep(0.05) + await anyio.sleep(0) + assert any("sse_read_timeout fired" in rec.message for rec in caplog.records) + finally: + s2c_send.close() + c2s_recv.close()