Class ClientSession

src/mcp/client/session.py:183–867

Client half of an MCP connection, running on a `Dispatcher`. Construct it over a transport's stream pair (or pass a pre-built `dispatcher=`), enter as an async context manager, then call `initialize()`. The dispatcher owns the receive loop and request correlation; this class owns th

class ClientSession:
    """Client half of an MCP connection, running on a `Dispatcher`.

    Construct it over a transport's stream pair (or pass a pre-built
    `dispatcher=`), enter as an async context manager, then call
    `initialize()`. The dispatcher owns the receive loop and request
    correlation; this class owns the typed MCP layer and the constructor
    callbacks. Transport `Exception` items reach `message_handler` only when
    the session builds its own dispatcher from a stream pair.
    """

    def __init__(
        self,
        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,
        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,
        dispatcher: Dispatcher[Any] | None = None,
    ) -> None:
        self._session_read_timeout_seconds = read_timeout_seconds
        self._client_info = client_info or DEFAULT_CLIENT_INFO
        self._sampling_callback = sampling_callback or _default_sampling_callback
        self._sampling_capabilities = sampling_capabilities
        self._elicitation_callback = elicitation_callback or _default_elicitation_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
        self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
        self._initialize_result: types.InitializeResult | None = None
        self._discover_result: types.DiscoverResult | None = None
        self._negotiated_version: str | None = None
        self._stamp: Callable[[dict[str, Any], CallOptions], None] = _preconnect_stamp
        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
            if isinstance(dispatcher, JSONRPCDispatcher) and dispatcher.on_stream_exception is None:
                # Route transport-level Exception items into message_handler — only
                # stream-backed dispatchers carry these; DirectDispatcher has none.
                # Don't clobber a caller-supplied hook.
                # TODO(L78): this leaves a bound-method ref on the dispatcher after the
                # session exits (memory pin) and a second wrap of the same dispatcher would
                # skip install. The Transport-as-Dispatcher rework (L77) removes this seam.
                dispatcher.on_stream_exception = self._on_stream_exception
        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 eagerly so notifications can be sent before entering the context manager.
            self._dispatcher = JSONRPCDispatcher(
                read_stream, write_stream, on_stream_exception=self._on_stream_exception