fix(client): re-raise transport exceptions in default message handler

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()
 
 

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()