Handle message callbacks in ClientSession

This change adds a message_handler callback to ClientSession to allow for direct handling of incoming messages instead of requiring an async iterator. The change simplifies the client code by removing the need for a separate receive loop task.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: dsp-ant

src/mcp/client/__main__.py

@@ -7,9 +7,11 @@
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
+import mcp.types as types
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.stdio import StdioServerParameters, stdio_client
+from mcp.shared.session import RequestResponder
 from mcp.types import JSONRPCMessage
 
 if not sys.warnoptions:
@@ -21,26 +23,25 @@
 logger = logging.getLogger("client")
 
 
-async def receive_loop(session: ClientSession):
-    logger.info("Starting receive loop")
-    async for message in session.incoming_messages:
-        if isinstance(message, Exception):
-            logger.error("Error: %s", message)
-            continue
+async def message_handler(
+    message: RequestResponder[types.ServerRequest, types.ClientResult]
+    | types.ServerNotification
+    | Exception,
+) -> None:
+    if isinstance(message, Exception):
+        logger.error("Error: %s", message)
+        return
 
-        logger.info("Received message from server: %s", message)
+    logger.info("Received message from server: %s", message)
 
 
 async def run_session(
     read_stream: MemoryObjectReceiveStream[JSONRPCMessage | Exception],
     write_stream: MemoryObjectSendStream[JSONRPCMessage],
 ):
-    async with (
-        ClientSession(read_stream, write_stream) as session,
-        anyio.create_task_group() as tg,
-    ):
-        tg.start_soon(receive_loop, session)
-
+    async with ClientSession(
+        read_stream, write_stream, message_handler=message_handler
+    ) as session:
         logger.info("Initializing session")
         await session.initialize()
         logger.info("Initialized")

src/mcp/client/session.py

@@ -1,6 +1,7 @@
 from datetime import timedelta
 from typing import Any, Protocol
 
+import anyio.lowlevel
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from pydantic import AnyUrl, TypeAdapter
 
@@ -31,6 +32,23 @@ async def __call__(
     ) -> None: ...
 
 
+class MessageHandlerFnT(Protocol):
+    async def __call__(
+        self,
+        message: RequestResponder[types.ServerRequest, types.ClientResult]
+        | types.ServerNotification
+        | Exception,
+    ) -> None: ...
+
+
+async def _default_message_handler(
+    message: RequestResponder[types.ServerRequest, types.ClientResult]
+    | types.ServerNotification
+    | Exception,
+) -> None:
+    await anyio.lowlevel.checkpoint()
+
+
 async def _default_sampling_callback(
     context: RequestContext["ClientSession", Any],
     params: types.CreateMessageRequestParams,
@@ -78,6 +96,7 @@ def __init__(
         sampling_callback: SamplingFnT | None = None,
         list_roots_callback: ListRootsFnT | None = None,
         logging_callback: LoggingFnT | None = None,
+        message_handler: MessageHandlerFnT | None = None,
     ) -> None:
         super().__init__(
             read_stream,
@@ -89,6 +108,7 @@ def __init__(
         self._sampling_callback = sampling_callback or _default_sampling_callback
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
         self._logging_callback = logging_callback or _default_logging_callback
+        self._message_handler = message_handler or _default_message_handler
 
     async def initialize(self) -> types.InitializeResult:
         sampling = types.SamplingCapability()
@@ -337,10 +357,20 @@ async def _received_request(
                         types.ClientResult(root=types.EmptyResult())
                     )
 
+    async def _handle_incoming(
+        self,
+        req: RequestResponder[types.ServerRequest, types.ClientResult]
+        | types.ServerNotification
+        | Exception,
+    ) -> None:
+        """Handle incoming messages by forwarding to the message handler."""
+        await self._message_handler(req)
+
     async def _received_notification(
         self, notification: types.ServerNotification
     ) -> None:
         """Handle notifications from the server."""
+        # Process specific notification types
         match notification.root:
             case types.LoggingMessageNotification(params=params):
                 await self._logging_callback(params)

src/mcp/server/session.py

@@ -92,7 +92,7 @@ def __init__(
         self._initialization_state = InitializationState.NotInitialized
         self._init_options = init_options
         self._incoming_message_stream_writer, self._incoming_message_stream_reader = (
-            anyio.create_memory_object_stream[ServerRequestResponder]()
+            anyio.create_memory_object_stream[ServerRequestResponder](0)
         )
         self._exit_stack.push_async_callback(
             lambda: self._incoming_message_stream_reader.aclose()
@@ -308,7 +308,7 @@ async def send_prompt_list_changed(self) -> None:
         )
 
     async def _handle_incoming(self, req: ServerRequestResponder) -> None:
-        return await self._incoming_message_stream_writer.send(req)
+        await self._incoming_message_stream_writer.send(req)
 
     @property
     def incoming_messages(

src/mcp/shared/memory.py

@@ -9,7 +9,13 @@
 import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 
-from mcp.client.session import ClientSession, ListRootsFnT, LoggingFnT, SamplingFnT
+from mcp.client.session import (
+    ClientSession,
+    ListRootsFnT,
+    LoggingFnT,
+    MessageHandlerFnT,
+    SamplingFnT,
+)
 from mcp.server import Server
 from mcp.types import JSONRPCMessage
 
@@ -57,6 +63,7 @@ async def create_connected_server_and_client_session(
     sampling_callback: SamplingFnT | None = None,
     list_roots_callback: ListRootsFnT | None = None,
     logging_callback: LoggingFnT | None = None,
+    message_handler: MessageHandlerFnT | None = None,
     raise_exceptions: bool = False,
 ) -> AsyncGenerator[ClientSession, None]:
     """Creates a ClientSession that is connected to a running MCP server."""
@@ -86,6 +93,7 @@ async def create_connected_server_and_client_session(
                     sampling_callback=sampling_callback,
                     list_roots_callback=list_roots_callback,
                     logging_callback=logging_callback,
+                    message_handler=message_handler,
                 ) as client_session:
                     await client_session.initialize()
                     yield client_session

src/mcp/shared/session.py

@@ -378,4 +378,4 @@ async def _handle_incoming(
         | Exception,
     ) -> None:
         """A generic handler for incoming messages. Overwritten by subclasses."""
-        await anyio.lowlevel.checkpoint()
+        pass

test.py

@@ -0,0 +1,53 @@
+import logging
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.sse import sse_client
+from mcp.client.stdio import stdio_client
+
+# disable httpx logs
+logging.basicConfig(level=logging.DEBUG)
+logging.getLogger("httpcore.http11").setLevel(logging.CRITICAL)
+logging.getLogger("httpx").setLevel(logging.CRITICAL)
+logging.getLogger("urllib3.connectionpool").setLevel(logging.CRITICAL)
+
+code = """
+import numpy
+a = numpy.array([1, 2, 3])
+print(a)
+a
+"""
+
+
+async def call_tools(session: ClientSession):
+    # await session.initialize()
+    await session.set_logging_level("debug")
+    tools = await session.list_tools()
+    print(f"{tools=}")
+    result = await session.call_tool("run_python_code", {"python_code": code})
+    print(f"{result=}")
+
+
+async def sse():
+    async with sse_client("http://localhost:3001/sse") as (read, write):
+        async with ClientSession(read, write) as session:
+            await call_tools(session)
+
+
+async def stdio():
+    server_params = StdioServerParameters(
+        command="npx",
+        args=[
+            "--registry=https://registry.npmjs.org",
+            "@pydantic/mcp-run-python",
+            "stdio",
+        ],
+    )
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write) as session:
+            await call_tools(session)
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    asyncio.run(sse())

tests/client/test_logging_callback.py

@@ -1,11 +1,12 @@
 from typing import List, Literal
 
-import anyio
 import pytest
 
+import mcp.types as types
 from mcp.shared.memory import (
     create_connected_server_and_client_session as create_session,
 )
+from mcp.shared.session import RequestResponder
 from mcp.types import (
     LoggingMessageNotificationParams,
     TextContent,
@@ -46,40 +47,37 @@ async def test_tool_with_log(
         )
         return True
 
-    async with anyio.create_task_group() as tg:
-        async with create_session(
-            server._mcp_server, logging_callback=logging_collector
-        ) as client_session:
+    # Create a message handler to catch exceptions
+    async def message_handler(
+        message: RequestResponder[types.ServerRequest, types.ClientResult]
+        | types.ServerNotification
+        | Exception,
+    ) -> None:
+        if isinstance(message, Exception):
+            raise message
 
-            async def listen_session():
-                try:
-                    async for message in client_session.incoming_messages:
-                        if isinstance(message, Exception):
-                            raise message
-                except anyio.EndOfStream:
-                    pass
+    async with create_session(
+        server._mcp_server,
+        logging_callback=logging_collector,
+        message_handler=message_handler,
+    ) as client_session:
+        # First verify our test tool works
+        result = await client_session.call_tool("test_tool", {})
+        assert result.isError is False
+        assert isinstance(result.content[0], TextContent)
+        assert result.content[0].text == "true"
 
-            tg.start_soon(listen_session)
-
-            # First verify our test tool works
-            result = await client_session.call_tool("test_tool", {})
-            assert result.isError is False
-            assert isinstance(result.content[0], TextContent)
-            assert result.content[0].text == "true"
-
-            # Now send a log message via our tool
-            log_result = await client_session.call_tool(
-                "test_tool_with_log",
-                {
-                    "message": "Test log message",
-                    "level": "info",
-                    "logger": "test_logger",
-                },
-            )
-            assert log_result.isError is False
-            assert len(logging_collector.log_messages) == 1
-            assert logging_collector.log_messages[
-                0
-            ] == LoggingMessageNotificationParams(
-                level="info", logger="test_logger", data="Test log message"
-            )
+        # Now send a log message via our tool
+        log_result = await client_session.call_tool(
+            "test_tool_with_log",
+            {
+                "message": "Test log message",
+                "level": "info",
+                "logger": "test_logger",
+            },
+        )
+        assert log_result.isError is False
+        assert len(logging_collector.log_messages) == 1
+        assert logging_collector.log_messages[0] == LoggingMessageNotificationParams(
+            level="info", logger="test_logger", data="Test log message"
+        )

tests/client/test_session.py

@@ -1,7 +1,9 @@
 import anyio
 import pytest
 
+import mcp.types as types
 from mcp.client.session import ClientSession
+from mcp.shared.session import RequestResponder
 from mcp.types import (
     LATEST_PROTOCOL_VERSION,
     ClientNotification,
@@ -75,21 +77,28 @@ async def mock_server():
                 )
             )
 
-    async def listen_session():
-        async for message in session.incoming_messages:
-            if isinstance(message, Exception):
-                raise message
+    # Create a message handler to catch exceptions
+    async def message_handler(
+        message: RequestResponder[types.ServerRequest, types.ClientResult]
+        | types.ServerNotification
+        | Exception,
+    ) -> None:
+        if isinstance(message, Exception):
+            raise message
 
     async with (
-        ClientSession(server_to_client_receive, client_to_server_send) as session,
+        ClientSession(
+            server_to_client_receive,
+            client_to_server_send,
+            message_handler=message_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)
-        tg.start_soon(listen_session)
         result = await session.initialize()
 
     # Assert the result