fix(session): log exceptions in default message_handler instead of silently swallowing

The default _default_message_handler in ClientSession silently discarded
all messages including exceptions, making transport errors (e.g. SSE read
timeouts) impossible to diagnose. Callers had no indication that an error
occurred, leading to silent hangs that were extremely difficult to debug.

The handler now logs exceptions at WARNING level with full traceback via
exc_info. This provides observability while preserving backward
compatibility — callers who want to re-raise exceptions or implement
custom error handling can still pass their own message_handler.

Github-Issue: #1401
Reported-by: Unshure

Author: Aboudjem

src/mcp/client/session.py

@@ -57,6 +57,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", exc_info=message)
     await anyio.lowlevel.checkpoint()
 
 

tests/client/test_session.py

@@ -705,3 +705,142 @@ async def mock_server():
         await session.initialize()
 
         await session.call_tool(name=mocked_tool.name, arguments={"foo": "bar"}, meta=meta)
+
+
+@pytest.mark.anyio
+async def test_default_message_handler_logs_exceptions(caplog: pytest.LogCaptureFixture):
+    """Test that the default message handler logs exceptions instead of silently swallowing them.
+
+    When an exception (e.g. a transport error) is delivered through the read stream,
+    the default handler should log it at WARNING level so the error is observable.
+    Previously, exceptions were silently discarded, making transport failures
+    impossible to diagnose.
+    """
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](1)
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    async def mock_server():
+        # Receive the initialization request
+        session_message = await client_to_server_receive.receive()
+        jsonrpc_request = session_message.message
+        assert isinstance(jsonrpc_request, JSONRPCRequest)
+
+        result = InitializeResult(
+            protocol_version=LATEST_PROTOCOL_VERSION,
+            capabilities=ServerCapabilities(),
+            server_info=Implementation(name="mock-server", version="0.1.0"),
+        )
+
+        # Send init response
+        await server_to_client_send.send(
+            SessionMessage(
+                JSONRPCResponse(
+                    jsonrpc="2.0",
+                    id=jsonrpc_request.id,
+                    result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
+                )
+            )
+        )
+
+        # Receive initialized notification
+        await client_to_server_receive.receive()
+
+        # Inject an exception into the read stream (simulating a transport error)
+        await server_to_client_send.send(RuntimeError("SSE stream read timeout"))
+
+        # Close the stream so the session can exit cleanly
+        await server_to_client_send.aclose()
+
+    async with (
+        ClientSession(
+            server_to_client_receive,
+            client_to_server_send,
+            # Use the default message_handler (no override)
+        ) as session,
+        anyio.create_task_group() as tg,
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        tg.start_soon(mock_server)
+        await session.initialize()
+
+        # Wait for the receive loop to process the exception
+        await anyio.sleep(0.1)
+
+    # Verify the exception was logged instead of silently swallowed
+    warning_records = [r for r in caplog.records if "Unhandled exception in message handler" in r.message]
+    assert len(warning_records) >= 1
+    # The exception details are attached via exc_info, visible in the formatted output
+    assert warning_records[0].exc_info is not None
+    assert warning_records[0].exc_info[1] is not None
+    assert "SSE stream read timeout" in str(warning_records[0].exc_info[1])
+
+
+@pytest.mark.anyio
+async def test_custom_message_handler_can_suppress_exceptions():
+    """Test that a custom message handler can suppress exceptions if desired."""
+    client_to_server_send, client_to_server_receive = anyio.create_memory_object_stream[SessionMessage](1)
+    server_to_client_send, server_to_client_receive = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    suppressed_exceptions: list[Exception] = []
+
+    async def suppressing_handler(
+        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+    ) -> None:
+        if isinstance(message, Exception):
+            suppressed_exceptions.append(message)
+            # Intentionally NOT re-raising — old silent behavior
+
+    async def mock_server():
+        # Receive the initialization request
+        session_message = await client_to_server_receive.receive()
+        jsonrpc_request = session_message.message
+        assert isinstance(jsonrpc_request, JSONRPCRequest)
+
+        result = InitializeResult(
+            protocol_version=LATEST_PROTOCOL_VERSION,
+            capabilities=ServerCapabilities(),
+            server_info=Implementation(name="mock-server", version="0.1.0"),
+        )
+
+        # Send init response
+        await server_to_client_send.send(
+            SessionMessage(
+                JSONRPCResponse(
+                    jsonrpc="2.0",
+                    id=jsonrpc_request.id,
+                    result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
+                )
+            )
+        )
+
+        # Receive initialized notification
+        await client_to_server_receive.receive()
+
+        # Inject an exception, then close the stream
+        await server_to_client_send.send(RuntimeError("transport error"))
+        await server_to_client_send.aclose()
+
+    async with (
+        ClientSession(
+            server_to_client_receive,
+            client_to_server_send,
+            message_handler=suppressing_handler,
+        ) as session,
+        anyio.create_task_group() as tg,
+        client_to_server_send,
+        client_to_server_receive,
+        server_to_client_send,
+        server_to_client_receive,
+    ):
+        tg.start_soon(mock_server)
+        await session.initialize()
+
+        # Give the receive loop time to process the exception
+        await anyio.sleep(0.1)
+
+    # The custom handler captured the exception instead of crashing
+    assert len(suppressed_exceptions) == 1
+    assert str(suppressed_exceptions[0]) == "transport error"