Accept a pre-built dispatcher via a constructor keyword

Replaces the from_dispatcher classmethod: read_stream/write_stream become
optional and dispatcher is a keyword-only alternative, with mutual
exclusion validated at construction. Drops the __new__-based alternate
constructor and its shared state-init helper.

Author: maxisbey

docs/migration.md

@@ -1168,7 +1168,7 @@ In practice, replace direct `ServerSession` use with `Server.run(read_stream, wr
 
 ### `ClientSession` now runs on `JSONRPCDispatcher`; `BaseSession` removed
 
-`ClientSession` keeps its public surface — the `(read_stream, write_stream, ...)` constructor, every typed method, manual `initialize()`, and the async context-manager lifecycle — but the v1 receive loop (`BaseSession`) underneath it is gone. A new `ClientSession.from_dispatcher(dispatcher, ...)` constructor accepts a pre-built dispatcher (for example a `DirectDispatcher` for in-process embedding).
+`ClientSession` keeps its public surface — the `(read_stream, write_stream, ...)` constructor, every typed method, manual `initialize()`, and the async context-manager lifecycle — but the v1 receive loop (`BaseSession`) underneath it is gone. A new keyword-only `dispatcher=` constructor argument accepts a pre-built dispatcher instead of the stream pair (for example a `DirectDispatcher` for in-process embedding).
 
 Behavior changes:
 

src/mcp/client/session.py

@@ -117,19 +117,25 @@ async def _default_logging_callback(
 
 
 class ClientSession:
-    """Client half of an MCP connection, running on `JSONRPCDispatcher`.
-
-    Construct it over a transport's stream pair, enter it as an async context
-    manager, then call `initialize()`. The receive loop, request correlation,
-    and per-request concurrency live in the dispatcher; this class owns the
-    MCP type layer: typed requests, the initialize handshake, and routing
-    server-initiated traffic to the constructor callbacks.
+    """Client half of an MCP connection, running on a `Dispatcher`.
+
+    Construct it over a transport's stream pair (or pass a pre-built
+    `dispatcher=` instead, e.g. a `DirectDispatcher` for in-process
+    embedding), enter it as an async context manager, then call
+    `initialize()`. The receive loop, request correlation, and per-request
+    concurrency live in the dispatcher; this class owns the MCP type layer:
+    typed requests, the initialize handshake, and routing server-initiated
+    traffic to the constructor callbacks.
+
+    Transport-level `Exception` items reach `message_handler` only when the
+    session builds its own dispatcher from streams, where it wires the
+    dispatcher's `on_stream_exception` itself.
     """
 
     def __init__(
         self,
-        read_stream: ReadStream[SessionMessage | Exception],
-        write_stream: WriteStream[SessionMessage],
+        read_stream: ReadStream[SessionMessage | Exception] | None = None,
+        write_stream: WriteStream[SessionMessage] | None = None,
         read_timeout_seconds: float | None = None,
         sampling_callback: SamplingFnT | None = None,
         elicitation_callback: ElicitationFnT | None = None,
@@ -139,69 +145,7 @@ def __init__(
         client_info: types.Implementation | None = None,
         *,
         sampling_capabilities: types.SamplingCapability | None = None,
-    ) -> None:
-        self._init_state(
-            read_timeout_seconds=read_timeout_seconds,
-            sampling_callback=sampling_callback,
-            elicitation_callback=elicitation_callback,
-            list_roots_callback=list_roots_callback,
-            logging_callback=logging_callback,
-            message_handler=message_handler,
-            client_info=client_info,
-            sampling_capabilities=sampling_capabilities,
-        )
-        # Built here (inert until run() starts in __aenter__) so notifications
-        # can be sent before entering the context manager, as before.
-        self._dispatcher: Dispatcher[Any] = JSONRPCDispatcher(
-            read_stream, write_stream, on_stream_exception=self._on_stream_exception
-        )
-
-    @classmethod
-    def from_dispatcher(
-        cls,
-        dispatcher: Dispatcher[Any],
-        *,
-        read_timeout_seconds: float | None = None,
-        sampling_callback: SamplingFnT | None = None,
-        elicitation_callback: ElicitationFnT | None = None,
-        list_roots_callback: ListRootsFnT | None = None,
-        logging_callback: LoggingFnT | None = None,
-        message_handler: MessageHandlerFnT | None = None,
-        client_info: types.Implementation | None = None,
-        sampling_capabilities: types.SamplingCapability | None = None,
-    ) -> Self:
-        """Build a session over a pre-built dispatcher instead of a stream pair.
-
-        For embedding a server in-process (`DirectDispatcher`) or transports
-        that construct their own dispatcher. Transport-level `Exception` items
-        reach `message_handler` only on the stream constructor, where the
-        session wires the dispatcher's `on_stream_exception` itself.
-        """
-        self = cls.__new__(cls)
-        self._init_state(
-            read_timeout_seconds=read_timeout_seconds,
-            sampling_callback=sampling_callback,
-            elicitation_callback=elicitation_callback,
-            list_roots_callback=list_roots_callback,
-            logging_callback=logging_callback,
-            message_handler=message_handler,
-            client_info=client_info,
-            sampling_capabilities=sampling_capabilities,
-        )
-        self._dispatcher = dispatcher
-        return self
-
-    def _init_state(
-        self,
-        *,
-        read_timeout_seconds: float | None,
-        sampling_callback: SamplingFnT | None,
-        elicitation_callback: ElicitationFnT | None,
-        list_roots_callback: ListRootsFnT | None,
-        logging_callback: LoggingFnT | None,
-        message_handler: MessageHandlerFnT | None,
-        client_info: types.Implementation | None,
-        sampling_capabilities: types.SamplingCapability | None,
+        dispatcher: Dispatcher[Any] | None = None,
     ) -> None:
         self._session_read_timeout_seconds = read_timeout_seconds
         self._client_info = client_info or DEFAULT_CLIENT_INFO
@@ -214,6 +158,18 @@ def _init_state(
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         self._initialize_result: types.InitializeResult | None = None
         self._task_group: anyio.abc.TaskGroup | None = None
+        if dispatcher is not None:
+            if read_stream is not None or write_stream is not None:
+                raise ValueError("pass read_stream/write_stream or dispatcher, not both")
+            self._dispatcher: Dispatcher[Any] = dispatcher
+        else:
+            if read_stream is None or write_stream is None:
+                raise ValueError("read_stream and write_stream are required when no dispatcher is given")
+            # Built here (inert until run() starts in __aenter__) so notifications
+            # can be sent before entering the context manager, as before.
+            self._dispatcher = JSONRPCDispatcher(
+                read_stream, write_stream, on_stream_exception=self._on_stream_exception
+            )
 
     async def __aenter__(self) -> Self:
         self._task_group = anyio.create_task_group()

tests/client/test_session.py

@@ -900,8 +900,8 @@ async def call() -> None:
 
 
 @pytest.mark.anyio
-async def test_from_dispatcher_runs_over_direct_dispatch():
-    """A session built with from_dispatcher works without a stream pair (in-process embedding)."""
+async def test_dispatcher_keyword_runs_over_direct_dispatch():
+    """A session built with dispatcher= works without a stream pair (in-process embedding)."""
     from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
     from mcp.shared.dispatcher import DispatchContext
     from mcp.shared.transport_context import TransportContext
@@ -921,7 +921,7 @@ async def server_on_notify(
     ) -> None:
         notified.append(method)
 
-    session = ClientSession.from_dispatcher(client_side)
+    session = ClientSession(dispatcher=client_side)
     results: list[types.EmptyResult] = []
     async with anyio.create_task_group() as tg:
         await tg.start(server_side.run, server_on_request, server_on_notify)
@@ -938,6 +938,27 @@ async def server_on_notify(
     assert notified == ["notifications/roots/list_changed"]
 
 
+def test_constructor_rejects_streams_and_dispatcher_together():
+    from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
+
+    client_side, _server_side = create_direct_dispatcher_pair()
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    with pytest.raises(ValueError, match="not both"):
+        ClientSession(s2c_recv, dispatcher=client_side)
+    s2c_send.close()
+    s2c_recv.close()
+
+
+def test_constructor_requires_both_streams_without_dispatcher():
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    with pytest.raises(ValueError, match="read_stream and write_stream are required"):
+        ClientSession(s2c_recv)
+    with pytest.raises(ValueError, match="read_stream and write_stream are required"):
+        ClientSession()
+    s2c_send.close()
+    s2c_recv.close()
+
+
 @pytest.mark.anyio
 async def test_send_request_with_server_metadata_routes_related_request_id():
     """ServerMessageMetadata.related_request_id is threaded onto the outgoing message."""