Index

MCP Client module.

InputRequiredRoundsExceededError

Bases: RuntimeError

The server kept returning InputRequiredResult past the configured max_rounds.

Source code in src/mcp/client/_input_required.py

class InputRequiredRoundsExceededError(RuntimeError):
    """The server kept returning `InputRequiredResult` past the configured `max_rounds`."""

    def __init__(self, max_rounds: int) -> None:
        super().__init__(
            f"Server returned InputRequiredResult for more than {max_rounds} rounds; "
            "raise input_required_max_rounds on the Client, or use "
            "client.session.<method>(..., allow_input_required=True) to drive the loop manually."
        )
        self.max_rounds = max_rounds

Transport

Bases: AbstractAsyncContextManager[TransportStreams], Protocol

Protocol for MCP transports.

A transport is an async context manager that yields read and write streams for bidirectional communication with an MCP server.

Source code in src/mcp/client/_transport.py

class Transport(AbstractAsyncContextManager[TransportStreams], Protocol):
    """Protocol for MCP transports.

    A transport is an async context manager that yields read and write streams
    for bidirectional communication with an MCP server.
    """

CacheConfig dataclass

Configuration for a Client's response cache.

Raises:

Type	Description
ValueError	On a custom store without partition, an empty target_id, or a negative default_ttl_ms.

Source code in src/mcp/client/caching.py

@dataclass(frozen=True, slots=True)
class CacheConfig:
    """Configuration for a `Client`'s response cache.

    Raises:
        ValueError: On a custom `store` without `partition`, an empty `target_id`, or a negative `default_ttl_ms`.
    """

    store: ResponseCacheStore | None = None
    """Backing store; `None` means a per-client `InMemoryResponseCacheStore`.
    A custom store requires an explicit `partition`."""

    partition: str = ""
    """Authorization-context identifier isolating `"private"`-scoped entries
    within a shared store. Derive it from a verified credential - never from
    request-supplied data or the server URL. Fixed for the `Client`'s
    lifetime: construct a new `Client` when the principal changes."""

    target_id: str | None = None
    """Server-identity override for custom transports and proxies where the
    SDK cannot derive one from a URL; must be non-empty when provided."""

    default_ttl_ms: int = 0
    """TTL in milliseconds for results carrying no `ttlMs` hint; the default `0` leaves them uncached."""

    clock: Callable[[], float] = time.time
    """Wall-clock source returning epoch seconds; injectable for expiry tests."""

    share_public: bool = False
    """Serve server-marked `"public"` entries across every partition in the store.

    WARNING: this trusts the server's `"public"` classification for every
    principal sharing the store - a mislabeled response leaks across tenants.
    Constructor-level only: the per-call `cache_mode` can never widen sharing."""

    def __post_init__(self) -> None:
        if self.store is not None and not self.partition:
            raise ValueError("a custom store requires an explicit partition")
        if self.target_id == "":
            raise ValueError("target_id must be a non-empty string or omitted")
        if self.default_ttl_ms < 0:
            raise ValueError(f"default_ttl_ms must be >= 0, got {self.default_ttl_ms}")

store class-attribute instance-attribute

store: ResponseCacheStore | None = None

Backing store; None means a per-client InMemoryResponseCacheStore. A custom store requires an explicit partition.

partition class-attribute instance-attribute

partition: str = ''

Authorization-context identifier isolating "private"-scoped entries within a shared store. Derive it from a verified credential - never from request-supplied data or the server URL. Fixed for the Client's lifetime: construct a new Client when the principal changes.

target_id class-attribute instance-attribute

target_id: str | None = None

Server-identity override for custom transports and proxies where the SDK cannot derive one from a URL; must be non-empty when provided.

default_ttl_ms class-attribute instance-attribute

default_ttl_ms: int = 0

TTL in milliseconds for results carrying no ttlMs hint; the default 0 leaves them uncached.

clock class-attribute instance-attribute

clock: Callable[[], float] = time

Wall-clock source returning epoch seconds; injectable for expiry tests.

share_public class-attribute instance-attribute

share_public: bool = False

Serve server-marked "public" entries across every partition in the store.

WARNING: this trusts the server's "public" classification for every principal sharing the store - a mislabeled response leaks across tenants. Constructor-level only: the per-call cache_mode can never widen sharing.

CacheEntry dataclass

One cached response with its freshness and sharing metadata.

Source code in src/mcp/client/caching.py

@dataclass(frozen=True, slots=True)
class CacheEntry:
    """One cached response with its freshness and sharing metadata."""

    value: Any
    """The cached result; the SDK deep-copies on write and on serve, so a store may hold it as-is."""

    scope: Literal["public", "private"]
    """Server-asserted `cacheScope`: only `"public"` entries may be shared across authorization contexts."""

    expires_at: float | None
    """Epoch seconds after which the entry is stale; `None` is never fresh."""

value instance-attribute

value: Any

The cached result; the SDK deep-copies on write and on serve, so a store may hold it as-is.

scope instance-attribute

scope: Literal['public', 'private']

Server-asserted cacheScope: only "public" entries may be shared across authorization contexts.

expires_at instance-attribute

expires_at: float | None

Epoch seconds after which the entry is stale; None is never fresh.

CacheKey dataclass

Identity of one cached response; compare as the field tuple, never a flattened string (collision hazard).

Source code in src/mcp/client/caching.py

@dataclass(frozen=True, slots=True)
class CacheKey:
    """Identity of one cached response; compare as the field tuple, never a flattened string (collision hazard)."""

    method: str

    params_key: str = ""
    """Result-affecting params discriminator: the uri for `resources/read`, `""` for the list methods."""

    partition: str = ""
    """Coordinator-computed arm identifier; opaque to stores."""

params_key class-attribute instance-attribute

params_key: str = ''

Result-affecting params discriminator: the uri for resources/read, "" for the list methods.

partition class-attribute instance-attribute

partition: str = ''

Coordinator-computed arm identifier; opaque to stores.

CacheMode module-attribute

CacheMode = Literal['use', 'refresh', 'bypass']

Per-call cache behavior: "use" serves and stores, "refresh" stores without serving, "bypass" skips the cache entirely.

InMemoryResponseCacheStore

Default in-process ResponseCacheStore.

Method bodies are synchronous, so concurrent tasks never observe a torn write. max_entries caps the whole store, evicting least-recently-used at the cap (0 disables it); get and set both refresh recency, so a hot entry survives churn from other keys.

Raises:

Type	Description
ValueError	If max_entries is negative.

Source code in src/mcp/client/caching.py

class InMemoryResponseCacheStore:
    """Default in-process `ResponseCacheStore`.

    Method bodies are synchronous, so concurrent tasks never observe a torn
    write. `max_entries` caps the whole store, evicting least-recently-used
    at the cap (`0` disables it); `get` and `set` both refresh recency, so a
    hot entry survives churn from other keys.

    Raises:
        ValueError: If `max_entries` is negative.
    """

    def __init__(self, *, max_entries: int = 1024) -> None:
        if max_entries < 0:
            raise ValueError(f"max_entries must be >= 0, got {max_entries}")
        self._max_entries = max_entries
        self._entries: dict[CacheKey, CacheEntry] = {}

    async def get(self, key: CacheKey) -> CacheEntry | None:
        entry = self._entries.get(key)
        if entry is not None:
            # Pop-and-reinsert moves the key to the back: the dict's insertion order is the LRU ledger.
            self._entries[key] = self._entries.pop(key)
        return entry

    async def set(self, key: CacheKey, entry: CacheEntry) -> None:
        self._entries.pop(key, None)
        self._entries[key] = entry
        if self._max_entries and len(self._entries) > self._max_entries:
            del self._entries[next(iter(self._entries))]

    async def delete(self, key: CacheKey) -> None:
        self._entries.pop(key, None)

    async def clear(self) -> None:
        self._entries.clear()

ResponseCacheStore

Bases: Protocol

Storage contract for the client response cache.

Each Client calls its store from a single event loop; per-operation atomicity is the implementation's responsibility. Operations may raise - the SDK degrades to a miss rather than failing the call. A serializing store must round-trip value back to the result model object (a wrong-shape entry is a miss, never an error). A lookup may issue two sequential get calls (private arm, then public).

Source code in src/mcp/client/caching.py

class ResponseCacheStore(Protocol):
    """Storage contract for the client response cache.

    Each `Client` calls its store from a single event loop; per-operation
    atomicity is the implementation's responsibility. Operations may raise -
    the SDK degrades to a miss rather than failing the call. A serializing
    store must round-trip `value` back to the result model object (a
    wrong-shape entry is a miss, never an error). A lookup may issue two
    sequential `get` calls (private arm, then public).
    """

    async def get(self, key: CacheKey) -> CacheEntry | None: ...

    async def set(self, key: CacheKey, entry: CacheEntry) -> None: ...

    async def delete(self, key: CacheKey) -> None: ...

    async def clear(self) -> None: ...

Client dataclass

A high-level MCP client for connecting to MCP servers.

Supports in-memory transport for testing (pass a Server or MCPServer instance), Streamable HTTP transport (pass a URL string), or a custom Transport instance.

Example

from mcp.client import Client
from mcp.server.mcpserver import MCPServer

server = MCPServer("test")

@server.tool()
def add(a: int, b: int) -> int:
    return a + b

async def main():
    async with Client(server) as client:
        result = await client.call_tool("add", {"a": 1, "b": 2})

asyncio.run(main())

Source code in src/mcp/client/client.py

@dataclass
class Client:
    """A high-level MCP client for connecting to MCP servers.

    Supports in-memory transport for testing (pass a Server or MCPServer instance),
    Streamable HTTP transport (pass a URL string), or a custom Transport instance.

    Example:
        ```python
        from mcp.client import Client
        from mcp.server.mcpserver import MCPServer

        server = MCPServer("test")

        @server.tool()
        def add(a: int, b: int) -> int:
            return a + b

        async def main():
            async with Client(server) as client:
                result = await client.call_tool("add", {"a": 1, "b": 2})

        asyncio.run(main())
        ```
    """

    server: Server[Any] | MCPServer | Transport | str
    """The MCP server to connect to.

    If the server is a `Server` or `MCPServer` instance, it will be connected in-process.
    If the server is a URL string, it will be used as the URL for a `streamable_http_client` transport.
    If the server is a `Transport` instance, it will be used directly.
    """

    _: KW_ONLY

    # TODO(Marcelo): When do `raise_exceptions=True` actually raises?
    raise_exceptions: bool = False
    """Whether to raise exceptions from the server."""

    read_timeout_seconds: float | None = None
    """Timeout for read operations."""

    sampling_callback: SamplingFnT | None = None
    """Callback for handling sampling requests."""

    list_roots_callback: ListRootsFnT | None = None
    """Callback for handling list roots requests."""

    logging_callback: LoggingFnT | None = None
    """Callback for handling logging notifications."""

    # TODO(Marcelo): Why do we have both "callback" and "handler"?
    message_handler: MessageHandlerFnT | None = None
    """Callback for handling raw messages."""

    client_info: Implementation | None = None
    """Client implementation info to send to server."""

    mode: ConnectMode = "auto"
    """How to negotiate the protocol version.

    'auto' (the default) probes `server/discover` and falls back to the initialize handshake on legacy servers;
    for an in-process `Server`/`MCPServer` it dispatches directly without JSON-RPC framing. 'legacy' forces the
    initialize handshake (byte-identical pre-2026 behavior). A modern protocol-version string (e.g. '2026-07-28')
    adopts that version directly without a probe — supply `prior_discover` to reuse a known DiscoverResult, or
    omit it to synthesize a minimal one."""

    prior_discover: types.DiscoverResult | None = None
    """A previously-obtained DiscoverResult to install via .adopt() when mode is a version pin.
    Ignored when mode='legacy'."""

    elicitation_callback: ElicitationFnT | None = None
    """Callback for handling elicitation requests."""

    input_required_max_rounds: int = DEFAULT_INPUT_REQUIRED_MAX_ROUNDS
    """Cap on `InputRequiredResult` retry rounds before `call_tool` / `get_prompt` /
    `read_resource` give up. Use `client.session.<method>(..., allow_input_required=True)`
    to drive the loop manually instead."""

    extensions: Sequence[ClientExtension] | None = None
    """Opt-in client extensions (SEP-2133).

    Each instance contributes its capability ad, its result claims (resolved
    transparently by `call_tool`), and its notification bindings. For an
    ad-only entry use `mcp.client.advertise(identifier, settings)`."""

    cache: CacheConfig | Literal[False] | None = None
    """Client-side response caching for the SEP-2549 cacheable methods (2026-07-28).

    `None` (the default) honors server `ttlMs`/`cacheScope` hints with a per-client
    in-memory store; pass a `CacheConfig` to customize, or `False` to disable. The
    cacheable verbs take a per-call `cache_mode` (see `CacheMode`); calls carrying
    `meta` always reach the server. A `CacheConfig` with a custom `store` requires
    `target_id` when the server is not a URL (no identity can be derived)."""

    _entered: bool = field(init=False, default=False)
    _session: ClientSession | None = field(init=False, default=None)
    _exit_stack: AsyncExitStack | None = field(init=False, default=None)
    _connect: _Connector = field(init=False, repr=False, compare=False)
    _response_cache: ClientResponseCache | None = field(init=False, default=None, repr=False, compare=False)
    _folded_extensions: _FoldedExtensions = field(init=False, repr=False, compare=False)

    def __post_init__(self) -> None:
        if self.mode not in ("legacy", "auto") and self.mode not in MODERN_PROTOCOL_VERSIONS:
            hint = (
                f" ({self.mode!r} is a handshake-era version; use mode='legacy')"
                if self.mode in HANDSHAKE_PROTOCOL_VERSIONS
                else ""
            )
            raise ValueError(
                f"mode must be 'legacy', 'auto', or one of {list(MODERN_PROTOCOL_VERSIONS)}; got {self.mode!r}{hint}"
            )

        self._folded_extensions = _fold_extensions(self.extensions)

        srv = self.server
        if isinstance(srv, MCPServer):
            srv = srv._lowlevel_server  # pyright: ignore[reportPrivateUsage]
        if isinstance(srv, Server):
            self._connect = _connect_inproc(srv)
        elif isinstance(srv, str):
            self._connect = _connect_transport(streamable_http_client(srv))
        else:
            self._connect = _connect_transport(srv)

        if self.cache is not False:
            config = self.cache if self.cache is not None else CacheConfig()
            # Only the hash below leaves this scope - the raw identity may carry credentials; never log or store it.
            target_id = config.target_id
            if target_id is None and isinstance(self.server, str):
                target_id = _strip_userinfo(self.server)
            if target_id is None:
                if config.store is not None:
                    raise ValueError(
                        "a custom cache store requires CacheConfig.target_id when the server is not a URL: "
                        "in-process servers and Transport instances get a random per-client identity, so "
                        "their entries in a shared store could never be served to another client"
                    )
                target_id = uuid.uuid4().hex
            self._response_cache = ClientResponseCache(
                store=config.store if config.store is not None else InMemoryResponseCacheStore(),
                partition=config.partition,
                arm_id=hashlib.sha256(target_id.encode()).hexdigest(),
                default_ttl_ms=config.default_ttl_ms,
                clock=config.clock,
                share_public=config.share_public,
                # Lazy: the negotiated version is unknown until __aenter__'s handshake.
                negotiated_version=lambda: self._session.protocol_version if self._session is not None else None,
            )

    async def _build_session(self, exit_stack: AsyncExitStack) -> ClientSession:
        """Enter the resolved connector and return an un-entered ClientSession."""
        dispatcher = await self._connect(exit_stack, self.mode, self.raise_exceptions)
        message_handler = self.message_handler
        if self._response_cache is not None:
            message_handler = _evicting_message_handler(self._response_cache, self.message_handler)
        return ClientSession(
            dispatcher=dispatcher,
            read_timeout_seconds=self.read_timeout_seconds,
            sampling_callback=self.sampling_callback,
            list_roots_callback=self.list_roots_callback,
            logging_callback=self.logging_callback,
            message_handler=message_handler,
            client_info=self.client_info,
            elicitation_callback=self.elicitation_callback,
            extensions=self._folded_extensions.ad,
            result_claims=self._folded_extensions.claims,
            notification_bindings=self._folded_extensions.bindings,
        )

    async def __aenter__(self) -> Client:
        """Enter the async context manager."""
        if self._entered:
            raise RuntimeError("Client is already entered; cannot reenter")
        self._entered = True

        async with AsyncExitStack() as exit_stack:
            session = await self._build_session(exit_stack)
            session = await exit_stack.enter_async_context(session)

            if self.mode == "legacy":
                await session.initialize()
            elif self.mode == "auto":
                await negotiate_auto(session)
            else:
                session.adopt(self.prior_discover or _synthesize_discover(self.mode))

            # Only publish the session after the handshake succeeds, so `_session is not None`
            # implies the protocol_version/server_info/server_capabilities are populated. If the
            # handshake raised above, the local exit_stack unwinds the transport for us.
            self._session = session
            self._exit_stack = exit_stack.pop_all()
            return self

    async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
        """Exit the async context manager."""
        if self._exit_stack:  # pragma: no branch
            await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
        self._session = None

    @property
    def session(self) -> ClientSession:
        """Get the underlying ClientSession.

        This provides access to the full ClientSession API for advanced use cases.

        Raises:
            RuntimeError: If accessed before entering the context manager.
        """
        if self._session is None:
            raise RuntimeError("Client must be used within an async context manager")
        return self._session

    # TODO(maxisbey): the by-construction shape is for __aenter__ to return a connected-view
    # type whose protocol_version/server_info/server_capabilities are non-Optional fields,
    # eliminating these guards (and the one in .session). Same family as resolving the
    # transport/connector at __post_init__ so the Optional internal fields disappear.
    @property
    def protocol_version(self) -> str:
        """Negotiated protocol version (set by initialize/discover/adopt during ``__aenter__``)."""
        return _connected(self.session.protocol_version)

    @property
    def server_info(self) -> Implementation:
        """Server name/version (set by initialize/discover/adopt during ``__aenter__``)."""
        return _connected(self.session.server_info)

    @property
    def server_capabilities(self) -> ServerCapabilities:
        """Server capabilities (set by initialize/discover/adopt during ``__aenter__``)."""
        return _connected(self.session.server_capabilities)

    @property
    def instructions(self) -> str | None:
        """Server-provided instructions text, if any."""
        return self.session.instructions

    @deprecated(
        "ping is removed as of 2026-07-28; the method only works under mode='legacy'.",
        category=MCPDeprecationWarning,
    )
    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Send a ping request to the server."""
        return await self.session.send_ping(meta=meta)

    @deprecated(
        "Client-to-server progress is deprecated as of 2026-07-28; progress is server-to-client only.",
        category=MCPDeprecationWarning,
    )
    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
    ) -> None:
        """Send a progress notification to the server."""
        await self.session.send_progress_notification(  # pyright: ignore[reportDeprecated]
            progress_token=progress_token,
            progress=progress,
            total=total,
            message=message,
        )

    @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def set_logging_level(self, level: LoggingLevel, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Set the logging level on the server."""
        return await self.session.set_logging_level(level=level, meta=meta)  # pyright: ignore[reportDeprecated]

    async def _cached_fetch(
        self,
        method: str,
        *,
        cursor: str | None,
        meta: RequestParamsMeta | None,
        cache_mode: CacheMode,
        send: Callable[[], Awaitable[_CacheableT]],
        absorb: Callable[[_CacheableT], _CacheableT] | None = None,
    ) -> _CacheableT:
        """Serve one of the four list verbs through the response cache.

        `absorb` (tools/list only) re-applies session-side derived state to a served cache hit.
        """
        cache = self._response_cache
        if cache is None or cache_mode == "bypass":
            return await send()
        # A closed (or never-entered) client must raise, never serve cached entries.
        _ = self.session
        if meta is not None and cache_mode == "use":
            # meta (a progress token, tracing fields) expects a wire request; fetch and replace the entry.
            cache_mode = "refresh"
        if cursor is not None:
            # Continuation pages skip the cache, but an expired cursor means the listing changed (spec SHOULD evict).
            try:
                return await send()
            except MCPError as e:
                if e.code == INVALID_PARAMS:
                    await cache.evict_method(method)
                raise
        if cache_mode == "use" and (hit := await cache.read(method, "")) is not None:
            # The hit is a private deep copy, so absorption may mutate it freely.
            served = cast(_CacheableT, hit)
            return served if absorb is None else absorb(served)
        gen = cache.capture(method, "")
        result = await send()
        await cache.write(method, "", result, gen, cache_mode)
        return result

    async def list_resources(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
        cache_mode: CacheMode = "use",
    ) -> ListResourcesResult:
        """List available resources from the server."""
        return await self._cached_fetch(
            "resources/list",
            cursor=cursor,
            meta=meta,
            cache_mode=cache_mode,
            send=lambda: self.session.list_resources(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
        )

    async def list_resource_templates(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
        cache_mode: CacheMode = "use",
    ) -> ListResourceTemplatesResult:
        """List available resource templates from the server."""
        return await self._cached_fetch(
            "resources/templates/list",
            cursor=cursor,
            meta=meta,
            cache_mode=cache_mode,
            send=lambda: self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
        )

    async def read_resource(
        self,
        uri: str,
        *,
        input_responses: InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        cache_mode: CacheMode = "use",
    ) -> ReadResourceResult:
        """Read a resource from the server.

        If the server returns an `InputRequiredResult`, the embedded input
        requests are dispatched to this client's sampling / elicitation / roots
        callbacks and the read is retried automatically (up to
        `input_required_max_rounds`).

        Args:
            uri: The URI of the resource to read.
            input_responses: Responses to seed the first call with (e.g. when
                resuming from a persisted `InputRequiredResult`).
            request_state: Opaque state to seed the first call with.
            meta: Additional metadata for the request.
            cache_mode: Cache behavior for this call (see `CacheMode`); seeded
                calls (`input_responses` or `request_state` set) ignore it.

        Returns:
            The resource content.

        Raises:
            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
            MCPError: A callback returned `ErrorData` for an embedded input request.
            pydantic.ValidationError: The server returned a result that does not
                conform to the negotiated protocol version.
        """

        async def retry(r: InputResponses | None, s: str | None) -> ReadResourceResult | InputRequiredResult:
            return await self.session.read_resource(
                uri, input_responses=r, request_state=s, meta=meta, allow_input_required=True
            )

        # Seeded calls resume a specific exchange and must never be cached (spec MUST).
        seeded = input_responses is not None or request_state is not None
        cache = None if seeded else self._response_cache
        if cache is None or cache_mode == "bypass":
            return await self._drive_input_required(await retry(input_responses, request_state), retry)
        # A closed (or never-entered) client must raise, never serve cached entries.
        _ = self.session
        if meta is not None and cache_mode == "use":
            # Calls carrying meta always reach the server (mirrors `_cached_fetch`).
            cache_mode = "refresh"
        if cache_mode == "use" and (hit := await cache.read("resources/read", uri)) is not None:
            # Only terminal first-round results are stored, so a hit legitimately skips the driver.
            return cast(ReadResourceResult, hit)
        gen = cache.capture("resources/read", uri)
        first = await retry(None, None)
        if not isinstance(first, InputRequiredResult):
            await cache.write("resources/read", uri, first, gen, cache_mode)
        elif cache_mode == "refresh":
            # The refresh superseded whatever was cached, but an input_required resolution
            # cannot be stored: purge the warm entry so it cannot be served again.
            await cache.evict_key("resources/read", uri)
        # Driver rounds carry inputResponses, so a terminal result reached through them is never cached (spec MUST).
        return await self._drive_input_required(first, retry)

    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Subscribe to resource updates."""
        return await self.session.subscribe_resource(uri, meta=meta)

    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Unsubscribe from resource updates."""
        return await self.session.unsubscribe_resource(uri, meta=meta)

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> CallToolResult:
        """Call a tool on the server.

        If the server returns an `InputRequiredResult`, the embedded input
        requests are dispatched to this client's sampling / elicitation / roots
        callbacks and the call is retried automatically (up to
        `input_required_max_rounds`). To drive the loop yourself — e.g. to
        persist `request_state` across process restarts — use
        `client.session.call_tool(..., allow_input_required=True)`. Persisted
        state is still subject to the server's TTL, request binding, and key
        lifetime; a server on the default process-local key rejects it after a restart.

        Result shapes claimed by this client's `extensions` are finished by the
        owning claim's resolver, whose `CallToolResult` is returned; resolver
        exceptions propagate as-is. To receive the claimed shape yourself, use
        `client.session.call_tool(..., allow_claimed=True)`.

        Args:
            name: The name of the tool to call.
            arguments: Arguments to pass to the tool.
            read_timeout_seconds: Timeout for each underlying `tools/call` round.
            progress_callback: Callback for progress updates.
            input_responses: Responses to seed the first call with (e.g. when
                resuming from a persisted `InputRequiredResult`).
            request_state: Opaque state to seed the first call with.
            meta: Additional metadata for the request.

        Returns:
            The tool result.

        Raises:
            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
            MCPError: A callback returned `ErrorData` for an embedded input request.
            pydantic.ValidationError: The server returned a result that does not
                conform to the negotiated protocol version.
        """

        async def retry(r: InputResponses | None, s: str | None) -> CallToolResult | InputRequiredResult | Result:
            return await self.session.call_tool(
                name,
                arguments,
                read_timeout_seconds=read_timeout_seconds,
                progress_callback=progress_callback,
                input_responses=r,
                request_state=s,
                meta=meta,
                allow_input_required=True,
                # Input rounds resolve before a claimed result, so a claim may end any round.
                allow_claimed=True,
            )

        result = await self._drive_input_required(await retry(input_responses, request_state), retry)
        if isinstance(result, CallToolResult):
            return result
        # Only claimed shapes reach this point, so the lookup is total.
        claim = self._folded_extensions.by_model[type(result)]
        final = await claim.resolve(
            result,
            ClaimContext(session=self.session, tool_name=name, read_timeout_seconds=read_timeout_seconds),
        )
        if not final.is_error:
            # Match the direct path: revalidate the output schema, but never for isError results.
            await self.session.validate_tool_result(name, final)
        return final

    async def list_prompts(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
        cache_mode: CacheMode = "use",
    ) -> ListPromptsResult:
        """List available prompts from the server."""
        return await self._cached_fetch(
            "prompts/list",
            cursor=cursor,
            meta=meta,
            cache_mode=cache_mode,
            send=lambda: self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
        )

    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        input_responses: InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> GetPromptResult:
        """Get a prompt from the server.

        If the server returns an `InputRequiredResult`, the embedded input
        requests are dispatched to this client's sampling / elicitation / roots
        callbacks and the get is retried automatically (up to
        `input_required_max_rounds`).

        Args:
            name: The name of the prompt.
            arguments: Arguments to pass to the prompt.
            input_responses: Responses to seed the first call with (e.g. when
                resuming from a persisted `InputRequiredResult`).
            request_state: Opaque state to seed the first call with.
            meta: Additional metadata for the request.

        Returns:
            The prompt content.

        Raises:
            InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
            MCPError: A callback returned `ErrorData` for an embedded input request.
            pydantic.ValidationError: The server returned a result that does not
                conform to the negotiated protocol version.
        """

        async def retry(r: InputResponses | None, s: str | None) -> GetPromptResult | InputRequiredResult:
            return await self.session.get_prompt(
                name, arguments, input_responses=r, request_state=s, meta=meta, allow_input_required=True
            )

        return await self._drive_input_required(await retry(input_responses, request_state), retry)

    async def _drive_input_required(
        self,
        first: _ResultT | InputRequiredResult,
        retry: Callable[[InputResponses | None, str | None], Awaitable[_ResultT | InputRequiredResult]],
    ) -> _ResultT:
        """Hand an `InputRequiredResult` to the SEP-2322 driver, or pass a terminal result through.

        `dispatch` routes each embedded request through the same callback table
        that serves legacy server→client RPCs, so the two paths stay
        behaviourally identical by construction.
        """
        if not isinstance(first, InputRequiredResult):
            return first
        session = self.session

        async def dispatch(key: str, req: InputRequest) -> InputResponse | ErrorData:
            ctx = ClientRequestContext(session=session, request_id=key, meta=req.params.meta if req.params else None)
            return await session.dispatch_input_request(ctx, req)

        return await run_input_required_driver(
            first, dispatch=dispatch, retry=retry, max_rounds=self.input_required_max_rounds
        )

    async def complete(
        self,
        ref: ResourceTemplateReference | PromptReference,
        argument: dict[str, str],
        context_arguments: dict[str, str] | None = None,
    ) -> CompleteResult:
        """Get completions for a prompt or resource template argument.

        Args:
            ref: Reference to the prompt or resource template
            argument: The argument to complete
            context_arguments: Additional context arguments

        Returns:
            Completion suggestions.
        """
        return await self.session.complete(ref=ref, argument=argument, context_arguments=context_arguments)

    async def list_tools(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
        cache_mode: CacheMode = "use",
    ) -> ListToolsResult:
        """List available tools from the server."""
        return await self._cached_fetch(
            "tools/list",
            cursor=cursor,
            meta=meta,
            cache_mode=cache_mode,
            send=lambda: self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
            # A cache hit skips session.list_tools, so the session re-absorbs the served
            # listing to rebuild its derived per-tool state. Hits are cursorless, but a
            # cached page 1 can carry next_cursor - never prune on a partial listing.
            absorb=lambda hit: self.session._absorb_tool_listing(  # pyright: ignore[reportPrivateUsage]
                hit, complete=hit.next_cursor is None
            ),
        )

    @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def send_roots_list_changed(self) -> None:
        """Send a notification that the roots list has changed."""
        # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
        await self.session.send_roots_list_changed()  # pyright: ignore[reportDeprecated]

server instance-attribute

server: Server[Any] | MCPServer | Transport | str

The MCP server to connect to.

If the server is a Server or MCPServer instance, it will be connected in-process. If the server is a URL string, it will be used as the URL for a streamable_http_client transport. If the server is a Transport instance, it will be used directly.

raise_exceptions class-attribute instance-attribute

raise_exceptions: bool = False

Whether to raise exceptions from the server.

read_timeout_seconds class-attribute instance-attribute

read_timeout_seconds: float | None = None

Timeout for read operations.

sampling_callback class-attribute instance-attribute

sampling_callback: SamplingFnT | None = None

Callback for handling sampling requests.

list_roots_callback class-attribute instance-attribute

list_roots_callback: ListRootsFnT | None = None

Callback for handling list roots requests.

logging_callback class-attribute instance-attribute

logging_callback: LoggingFnT | None = None

Callback for handling logging notifications.

message_handler class-attribute instance-attribute

message_handler: MessageHandlerFnT | None = None

Callback for handling raw messages.

client_info class-attribute instance-attribute

client_info: Implementation | None = None

Client implementation info to send to server.

mode class-attribute instance-attribute

mode: ConnectMode = 'auto'

How to negotiate the protocol version.

'auto' (the default) probes server/discover and falls back to the initialize handshake on legacy servers; for an in-process Server/MCPServer it dispatches directly without JSON-RPC framing. 'legacy' forces the initialize handshake (byte-identical pre-2026 behavior). A modern protocol-version string (e.g. '2026-07-28') adopts that version directly without a probe — supply prior_discover to reuse a known DiscoverResult, or omit it to synthesize a minimal one.

prior_discover class-attribute instance-attribute

prior_discover: DiscoverResult | None = None

A previously-obtained DiscoverResult to install via .adopt() when mode is a version pin. Ignored when mode='legacy'.

elicitation_callback class-attribute instance-attribute

elicitation_callback: ElicitationFnT | None = None

Callback for handling elicitation requests.

input_required_max_rounds class-attribute instance-attribute

input_required_max_rounds: int = (
    DEFAULT_INPUT_REQUIRED_MAX_ROUNDS
)

Cap on InputRequiredResult retry rounds before call_tool / get_prompt / read_resource give up. Use client.session.<method>(..., allow_input_required=True) to drive the loop manually instead.

extensions class-attribute instance-attribute

extensions: Sequence[ClientExtension] | None = None

Opt-in client extensions (SEP-2133).

Each instance contributes its capability ad, its result claims (resolved transparently by call_tool), and its notification bindings. For an ad-only entry use mcp.client.advertise(identifier, settings).

cache class-attribute instance-attribute

cache: CacheConfig | Literal[False] | None = None

Client-side response caching for the SEP-2549 cacheable methods (2026-07-28).

None (the default) honors server ttlMs/cacheScope hints with a per-client in-memory store; pass a CacheConfig to customize, or False to disable. The cacheable verbs take a per-call cache_mode (see CacheMode); calls carrying meta always reach the server. A CacheConfig with a custom store requires target_id when the server is not a URL (no identity can be derived).

__aenter__ async

__aenter__() -> Client

Enter the async context manager.

Source code in src/mcp/client/client.py

async def __aenter__(self) -> Client:
    """Enter the async context manager."""
    if self._entered:
        raise RuntimeError("Client is already entered; cannot reenter")
    self._entered = True

    async with AsyncExitStack() as exit_stack:
        session = await self._build_session(exit_stack)
        session = await exit_stack.enter_async_context(session)

        if self.mode == "legacy":
            await session.initialize()
        elif self.mode == "auto":
            await negotiate_auto(session)
        else:
            session.adopt(self.prior_discover or _synthesize_discover(self.mode))

        # Only publish the session after the handshake succeeds, so `_session is not None`
        # implies the protocol_version/server_info/server_capabilities are populated. If the
        # handshake raised above, the local exit_stack unwinds the transport for us.
        self._session = session
        self._exit_stack = exit_stack.pop_all()
        return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None

Exit the async context manager.

Source code in src/mcp/client/client.py

async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
    """Exit the async context manager."""
    if self._exit_stack:  # pragma: no branch
        await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
    self._session = None

session property

session: ClientSession

Get the underlying ClientSession.

This provides access to the full ClientSession API for advanced use cases.

Raises:

Type	Description
RuntimeError	If accessed before entering the context manager.

protocol_version property

protocol_version: str

Negotiated protocol version (set by initialize/discover/adopt during __aenter__).

server_info property

server_info: Implementation

Server name/version (set by initialize/discover/adopt during __aenter__).

server_capabilities property

server_capabilities: ServerCapabilities

Server capabilities (set by initialize/discover/adopt during __aenter__).

instructions property

instructions: str | None

Server-provided instructions text, if any.

send_ping async

send_ping(
    *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a ping request to the server.

Source code in src/mcp/client/client.py

@deprecated(
    "ping is removed as of 2026-07-28; the method only works under mode='legacy'.",
    category=MCPDeprecationWarning,
)
async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Send a ping request to the server."""
    return await self.session.send_ping(meta=meta)

send_progress_notification async

send_progress_notification(
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None

Send a progress notification to the server.

Source code in src/mcp/client/client.py

@deprecated(
    "Client-to-server progress is deprecated as of 2026-07-28; progress is server-to-client only.",
    category=MCPDeprecationWarning,
)
async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None:
    """Send a progress notification to the server."""
    await self.session.send_progress_notification(  # pyright: ignore[reportDeprecated]
        progress_token=progress_token,
        progress=progress,
        total=total,
        message=message,
    )

set_logging_level async

set_logging_level(
    level: LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None
) -> EmptyResult

Set the logging level on the server.

Source code in src/mcp/client/client.py

@deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def set_logging_level(self, level: LoggingLevel, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Set the logging level on the server."""
    return await self.session.set_logging_level(level=level, meta=meta)  # pyright: ignore[reportDeprecated]

list_resources async

list_resources(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use"
) -> ListResourcesResult

List available resources from the server.

Source code in src/mcp/client/client.py

async def list_resources(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use",
) -> ListResourcesResult:
    """List available resources from the server."""
    return await self._cached_fetch(
        "resources/list",
        cursor=cursor,
        meta=meta,
        cache_mode=cache_mode,
        send=lambda: self.session.list_resources(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
    )

list_resource_templates async

list_resource_templates(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use"
) -> ListResourceTemplatesResult

List available resource templates from the server.

Source code in src/mcp/client/client.py

async def list_resource_templates(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use",
) -> ListResourceTemplatesResult:
    """List available resource templates from the server."""
    return await self._cached_fetch(
        "resources/templates/list",
        cursor=cursor,
        meta=meta,
        cache_mode=cache_mode,
        send=lambda: self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
    )

read_resource async

read_resource(
    uri: str,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use"
) -> ReadResourceResult

Read a resource from the server.

If the server returns an InputRequiredResult, the embedded input requests are dispatched to this client's sampling / elicitation / roots callbacks and the read is retried automatically (up to input_required_max_rounds).

Parameters:

Name	Type	Description	Default
uri	str	The URI of the resource to read.	required
input_responses	InputResponses | None	Responses to seed the first call with (e.g. when resuming from a persisted InputRequiredResult).	None
request_state	str | None	Opaque state to seed the first call with.	None
meta	RequestParamsMeta | None	Additional metadata for the request.	None
cache_mode	CacheMode	Cache behavior for this call (see CacheMode); seeded calls (input_responses or request_state set) ignore it.	'use'

Returns:

Type	Description
ReadResourceResult	The resource content.

Raises:

Type	Description
InputRequiredRoundsExceededError	input_required_max_rounds exhausted.
MCPError	A callback returned ErrorData for an embedded input request.
ValidationError	The server returned a result that does not conform to the negotiated protocol version.

Source code in src/mcp/client/client.py

async def read_resource(
    self,
    uri: str,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use",
) -> ReadResourceResult:
    """Read a resource from the server.

    If the server returns an `InputRequiredResult`, the embedded input
    requests are dispatched to this client's sampling / elicitation / roots
    callbacks and the read is retried automatically (up to
    `input_required_max_rounds`).

    Args:
        uri: The URI of the resource to read.
        input_responses: Responses to seed the first call with (e.g. when
            resuming from a persisted `InputRequiredResult`).
        request_state: Opaque state to seed the first call with.
        meta: Additional metadata for the request.
        cache_mode: Cache behavior for this call (see `CacheMode`); seeded
            calls (`input_responses` or `request_state` set) ignore it.

    Returns:
        The resource content.

    Raises:
        InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
        MCPError: A callback returned `ErrorData` for an embedded input request.
        pydantic.ValidationError: The server returned a result that does not
            conform to the negotiated protocol version.
    """

    async def retry(r: InputResponses | None, s: str | None) -> ReadResourceResult | InputRequiredResult:
        return await self.session.read_resource(
            uri, input_responses=r, request_state=s, meta=meta, allow_input_required=True
        )

    # Seeded calls resume a specific exchange and must never be cached (spec MUST).
    seeded = input_responses is not None or request_state is not None
    cache = None if seeded else self._response_cache
    if cache is None or cache_mode == "bypass":
        return await self._drive_input_required(await retry(input_responses, request_state), retry)
    # A closed (or never-entered) client must raise, never serve cached entries.
    _ = self.session
    if meta is not None and cache_mode == "use":
        # Calls carrying meta always reach the server (mirrors `_cached_fetch`).
        cache_mode = "refresh"
    if cache_mode == "use" and (hit := await cache.read("resources/read", uri)) is not None:
        # Only terminal first-round results are stored, so a hit legitimately skips the driver.
        return cast(ReadResourceResult, hit)
    gen = cache.capture("resources/read", uri)
    first = await retry(None, None)
    if not isinstance(first, InputRequiredResult):
        await cache.write("resources/read", uri, first, gen, cache_mode)
    elif cache_mode == "refresh":
        # The refresh superseded whatever was cached, but an input_required resolution
        # cannot be stored: purge the warm entry so it cannot be served again.
        await cache.evict_key("resources/read", uri)
    # Driver rounds carry inputResponses, so a terminal result reached through them is never cached (spec MUST).
    return await self._drive_input_required(first, retry)

subscribe_resource async

subscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Subscribe to resource updates.

Source code in src/mcp/client/client.py

async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Subscribe to resource updates."""
    return await self.session.subscribe_resource(uri, meta=meta)

unsubscribe_resource async

unsubscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Unsubscribe from resource updates.

Source code in src/mcp/client/client.py

async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
    """Unsubscribe from resource updates."""
    return await self.session.unsubscribe_resource(uri, meta=meta)

call_tool async

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

Call a tool on the server.

If the server returns an InputRequiredResult, the embedded input requests are dispatched to this client's sampling / elicitation / roots callbacks and the call is retried automatically (up to input_required_max_rounds). To drive the loop yourself — e.g. to persist request_state across process restarts — use client.session.call_tool(..., allow_input_required=True). Persisted state is still subject to the server's TTL, request binding, and key lifetime; a server on the default process-local key rejects it after a restart.

Result shapes claimed by this client's extensions are finished by the owning claim's resolver, whose CallToolResult is returned; resolver exceptions propagate as-is. To receive the claimed shape yourself, use client.session.call_tool(..., allow_claimed=True).

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to call.	required
arguments	dict[str, Any] | None	Arguments to pass to the tool.	None
read_timeout_seconds	float | None	Timeout for each underlying tools/call round.	None
progress_callback	ProgressFnT | None	Callback for progress updates.	None
input_responses	InputResponses | None	Responses to seed the first call with (e.g. when resuming from a persisted InputRequiredResult).	None
request_state	str | None	Opaque state to seed the first call with.	None
meta	RequestParamsMeta | None	Additional metadata for the request.	None

Returns:

Type	Description
CallToolResult	The tool result.

Raises:

Type	Description
InputRequiredRoundsExceededError	input_required_max_rounds exhausted.
MCPError	A callback returned ErrorData for an embedded input request.
ValidationError	The server returned a result that does not conform to the negotiated protocol version.

Source code in src/mcp/client/client.py

async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
) -> CallToolResult:
    """Call a tool on the server.

    If the server returns an `InputRequiredResult`, the embedded input
    requests are dispatched to this client's sampling / elicitation / roots
    callbacks and the call is retried automatically (up to
    `input_required_max_rounds`). To drive the loop yourself — e.g. to
    persist `request_state` across process restarts — use
    `client.session.call_tool(..., allow_input_required=True)`. Persisted
    state is still subject to the server's TTL, request binding, and key
    lifetime; a server on the default process-local key rejects it after a restart.

    Result shapes claimed by this client's `extensions` are finished by the
    owning claim's resolver, whose `CallToolResult` is returned; resolver
    exceptions propagate as-is. To receive the claimed shape yourself, use
    `client.session.call_tool(..., allow_claimed=True)`.

    Args:
        name: The name of the tool to call.
        arguments: Arguments to pass to the tool.
        read_timeout_seconds: Timeout for each underlying `tools/call` round.
        progress_callback: Callback for progress updates.
        input_responses: Responses to seed the first call with (e.g. when
            resuming from a persisted `InputRequiredResult`).
        request_state: Opaque state to seed the first call with.
        meta: Additional metadata for the request.

    Returns:
        The tool result.

    Raises:
        InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
        MCPError: A callback returned `ErrorData` for an embedded input request.
        pydantic.ValidationError: The server returned a result that does not
            conform to the negotiated protocol version.
    """

    async def retry(r: InputResponses | None, s: str | None) -> CallToolResult | InputRequiredResult | Result:
        return await self.session.call_tool(
            name,
            arguments,
            read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
            input_responses=r,
            request_state=s,
            meta=meta,
            allow_input_required=True,
            # Input rounds resolve before a claimed result, so a claim may end any round.
            allow_claimed=True,
        )

    result = await self._drive_input_required(await retry(input_responses, request_state), retry)
    if isinstance(result, CallToolResult):
        return result
    # Only claimed shapes reach this point, so the lookup is total.
    claim = self._folded_extensions.by_model[type(result)]
    final = await claim.resolve(
        result,
        ClaimContext(session=self.session, tool_name=name, read_timeout_seconds=read_timeout_seconds),
    )
    if not final.is_error:
        # Match the direct path: revalidate the output schema, but never for isError results.
        await self.session.validate_tool_result(name, final)
    return final

list_prompts async

list_prompts(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use"
) -> ListPromptsResult

List available prompts from the server.

Source code in src/mcp/client/client.py

async def list_prompts(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use",
) -> ListPromptsResult:
    """List available prompts from the server."""
    return await self._cached_fetch(
        "prompts/list",
        cursor=cursor,
        meta=meta,
        cache_mode=cache_mode,
        send=lambda: self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
    )

get_prompt async

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None
) -> GetPromptResult

Get a prompt from the server.

If the server returns an InputRequiredResult, the embedded input requests are dispatched to this client's sampling / elicitation / roots callbacks and the get is retried automatically (up to input_required_max_rounds).

Parameters:

Name	Type	Description	Default
name	str	The name of the prompt.	required
arguments	dict[str, str] | None	Arguments to pass to the prompt.	None
input_responses	InputResponses | None	Responses to seed the first call with (e.g. when resuming from a persisted InputRequiredResult).	None
request_state	str | None	Opaque state to seed the first call with.	None
meta	RequestParamsMeta | None	Additional metadata for the request.	None

Returns:

Type	Description
GetPromptResult	The prompt content.

Raises:

Type	Description
InputRequiredRoundsExceededError	input_required_max_rounds exhausted.
MCPError	A callback returned ErrorData for an embedded input request.
ValidationError	The server returned a result that does not conform to the negotiated protocol version.

Source code in src/mcp/client/client.py

async def get_prompt(
    self,
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
) -> GetPromptResult:
    """Get a prompt from the server.

    If the server returns an `InputRequiredResult`, the embedded input
    requests are dispatched to this client's sampling / elicitation / roots
    callbacks and the get is retried automatically (up to
    `input_required_max_rounds`).

    Args:
        name: The name of the prompt.
        arguments: Arguments to pass to the prompt.
        input_responses: Responses to seed the first call with (e.g. when
            resuming from a persisted `InputRequiredResult`).
        request_state: Opaque state to seed the first call with.
        meta: Additional metadata for the request.

    Returns:
        The prompt content.

    Raises:
        InputRequiredRoundsExceededError: `input_required_max_rounds` exhausted.
        MCPError: A callback returned `ErrorData` for an embedded input request.
        pydantic.ValidationError: The server returned a result that does not
            conform to the negotiated protocol version.
    """

    async def retry(r: InputResponses | None, s: str | None) -> GetPromptResult | InputRequiredResult:
        return await self.session.get_prompt(
            name, arguments, input_responses=r, request_state=s, meta=meta, allow_input_required=True
        )

    return await self._drive_input_required(await retry(input_responses, request_state), retry)

complete async

complete(
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult

Get completions for a prompt or resource template argument.

Parameters:

Name	Type	Description	Default
ref	ResourceTemplateReference | PromptReference	Reference to the prompt or resource template	required
argument	dict[str, str]	The argument to complete	required
context_arguments	dict[str, str] | None	Additional context arguments	None

Returns:

Type	Description
CompleteResult	Completion suggestions.

Source code in src/mcp/client/client.py

async def complete(
    self,
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult:
    """Get completions for a prompt or resource template argument.

    Args:
        ref: Reference to the prompt or resource template
        argument: The argument to complete
        context_arguments: Additional context arguments

    Returns:
        Completion suggestions.
    """
    return await self.session.complete(ref=ref, argument=argument, context_arguments=context_arguments)

list_tools async

list_tools(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use"
) -> ListToolsResult

List available tools from the server.

Source code in src/mcp/client/client.py

async def list_tools(
    self,
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None,
    cache_mode: CacheMode = "use",
) -> ListToolsResult:
    """List available tools from the server."""
    return await self._cached_fetch(
        "tools/list",
        cursor=cursor,
        meta=meta,
        cache_mode=cache_mode,
        send=lambda: self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta)),
        # A cache hit skips session.list_tools, so the session re-absorbs the served
        # listing to rebuild its derived per-tool state. Hits are cursorless, but a
        # cached page 1 can carry next_cursor - never prune on a partial listing.
        absorb=lambda hit: self.session._absorb_tool_listing(  # pyright: ignore[reportPrivateUsage]
            hit, complete=hit.next_cursor is None
        ),
    )

send_roots_list_changed async

send_roots_list_changed() -> None

Send a notification that the roots list has changed.

Source code in src/mcp/client/client.py

@deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def send_roots_list_changed(self) -> None:
    """Send a notification that the roots list has changed."""
    # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
    await self.session.send_roots_list_changed()  # pyright: ignore[reportDeprecated]

ClientRequestContext dataclass

Context for a server-initiated request, passed to the sampling/elicitation/list-roots callbacks.

Source code in src/mcp/client/session.py

@dataclass(kw_only=True)
class ClientRequestContext:
    """Context for a server-initiated request, passed to the sampling/elicitation/list-roots callbacks."""

    session: ClientSession
    request_id: RequestId
    meta: RequestParamsMeta | None = None

ClaimContext dataclass

Host-injected context for one ResultClaim.resolve call.

Source code in src/mcp/client/extension.py

@dataclass(frozen=True, kw_only=True)
class ClaimContext:
    """Host-injected context for one `ResultClaim.resolve` call."""

    session: ClientSession
    tool_name: str
    read_timeout_seconds: float | None

ClientExtension

Base class for an opt-in client extension; override only what you need.

The surface is declarative, fixed at construction, and never receives the client.

Source code in src/mcp/client/extension.py

class ClientExtension:
    """Base class for an opt-in client extension; override only what you need.

    The surface is declarative, fixed at construction, and never receives the client.
    """

    #: Reverse-DNS extension identifier, advertised under `ClientCapabilities.extensions`.
    identifier: str

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        # Per-instance identifiers (assigned in __init__) are validated at consumption instead.
        if (identifier := cls.__dict__.get("identifier")) is not None:
            validate_extension_identifier(identifier, owner=cls.__name__)

    def settings(self) -> dict[str, Any]:
        """Per-extension settings advertised at `ClientCapabilities.extensions[identifier]`.

        Read once at `Client` construction. A claim-bearing extension is
        advertised only at protocol versions where at least one of its claims
        is active.
        """
        return {}

    def claims(self) -> Sequence[ResultClaim[Any]]:
        """Extra result shapes this extension claims, with their resolvers."""
        return ()

    def notifications(self) -> Sequence[NotificationBinding[Any]]:
        """Server notifications this extension observes."""
        return ()

settings

settings() -> dict[str, Any]

Per-extension settings advertised at ClientCapabilities.extensions[identifier].

Read once at Client construction. A claim-bearing extension is advertised only at protocol versions where at least one of its claims is active.

Source code in src/mcp/client/extension.py

def settings(self) -> dict[str, Any]:
    """Per-extension settings advertised at `ClientCapabilities.extensions[identifier]`.

    Read once at `Client` construction. A claim-bearing extension is
    advertised only at protocol versions where at least one of its claims
    is active.
    """
    return {}

claims

claims() -> Sequence[ResultClaim[Any]]

Extra result shapes this extension claims, with their resolvers.

Source code in src/mcp/client/extension.py

def claims(self) -> Sequence[ResultClaim[Any]]:
    """Extra result shapes this extension claims, with their resolvers."""
    return ()

notifications

notifications() -> Sequence[NotificationBinding[Any]]

Server notifications this extension observes.

Source code in src/mcp/client/extension.py

def notifications(self) -> Sequence[NotificationBinding[Any]]:
    """Server notifications this extension observes."""
    return ()

NotificationBinding dataclass

Bases: Generic[NotifyParamsT]

Deliver server notifications for method (the bare wire name) to handler.

Observation-only: validated params arrive one at a time per binding, in dispatch order, through a bounded queue that drops the oldest with a warning on overflow. Stream transports dispatch each notification independently, so near-simultaneous notifications may be dispatched out of wire order. Methods the negotiated version's core tables handle are never delivered to bindings.

Source code in src/mcp/client/extension.py

@dataclass(frozen=True, kw_only=True)
class NotificationBinding(Generic[NotifyParamsT]):
    """Deliver server notifications for `method` (the bare wire name) to `handler`.

    Observation-only: validated params arrive one at a time per binding, in
    dispatch order, through a bounded queue that drops the oldest with a warning
    on overflow. Stream transports dispatch each notification independently, so
    near-simultaneous notifications may be dispatched out of wire order. Methods
    the negotiated version's core tables handle are never delivered to bindings.
    """

    method: str
    params_type: type[NotifyParamsT]
    handler: Callable[[NotifyParamsT], Awaitable[None]]

ResultClaim dataclass

Bases: Generic[ClaimedT]

One extra result shape on one spec verb, keyed by the wire resultType.

Active only while the declaring extension is constructed into the client and the negotiated protocol version admits it. resolve finishes a claimed result, may send follow-ups through ctx.session, and must return the verb's ordinary result. All field constraints are enforced at construction.

Source code in src/mcp/client/extension.py

@dataclass(frozen=True, kw_only=True)
class ResultClaim(Generic[ClaimedT]):
    """One extra result shape on one spec verb, keyed by the wire `resultType`.

    Active only while the declaring extension is constructed into the client and
    the negotiated protocol version admits it. `resolve` finishes a claimed
    result, may send follow-ups through `ctx.session`, and must return the
    verb's ordinary result. All field constraints are enforced at construction.
    """

    result_type: str
    model: type[ClaimedT]
    resolve: Callable[[ClaimedT, ClaimContext], Awaitable[CallToolResult]]
    method: Literal["tools/call"] = "tools/call"
    protocol_versions: frozenset[str] | None = None

    def __post_init__(self) -> None:
        if self.method not in _CLAIM_METHODS:
            raise ValueError(f"claims attach to {sorted(_CLAIM_METHODS)} only; got method {self.method!r}")
        if self.result_type in CORE_RESULT_TYPES:
            raise ValueError(f"resultType {self.result_type!r} is core protocol vocabulary")
        if Result not in self.model.__mro__:  # runtime guard; the ClaimedT bound only constrains checked callers
            raise ValueError(f"{self.model.__name__} must subclass mcp_types.Result")
        if issubclass(self.model, CallToolResult | InputRequiredResult):
            raise ValueError("claim models must not subclass core result types")
        for name, model_field in self.model.model_fields.items():
            for clash in sorted(_wire_keys(name, model_field) & _RESERVED_WIRE_ALIASES):
                raise ValueError(
                    f"{self.model.__name__}.{name} aliases {clash!r}, a typed field of the core "
                    "result surface; a colliding value would fail core validation before the "
                    "claim adapter runs"
                )
        field = self.model.model_fields.get("result_type")
        if field is None or get_args(field.annotation) != (self.result_type,):
            raise ValueError(f"{self.model.__name__}.result_type must be Literal[{self.result_type!r}]")
        if self.protocol_versions is not None and not self.protocol_versions:
            raise ValueError("empty protocol_versions could never activate; use None for all")
        if self.protocol_versions is not None and not self.protocol_versions.issubset(MODERN_PROTOCOL_VERSIONS):
            unrecognized = sorted(self.protocol_versions.difference(MODERN_PROTOCOL_VERSIONS))
            raise ValueError(
                f"protocol_versions {unrecognized} are not modern protocol revisions; claimed shapes "
                "cannot be delivered on a legacy wire (None means every modern version)"
            )

UnexpectedClaimedResult

Bases: RuntimeError

A claimed (extension) result arrived on a call_tool that did not opt in.

The parsed value is carried as result; the server may already hold state it references. Opt in via Client(extensions=[...]) or allow_claimed=True.

Source code in src/mcp/client/extension.py

class UnexpectedClaimedResult(RuntimeError):
    """A claimed (extension) result arrived on a `call_tool` that did not opt in.

    The parsed value is carried as `result`; the server may already hold state it
    references. Opt in via `Client(extensions=[...])` or `allow_claimed=True`.
    """

    def __init__(self, result: Result) -> None:
        super().__init__(
            f"Server returned a claimed result ({type(result).__name__}); pass the owning extension to "
            "Client(extensions=[...]) for transparent resolution, or call with allow_claimed=True "
            "and handle the shape. The carried result may reference server-side state needing cleanup."
        )
        self.result = result

advertise

advertise(
    identifier: str, settings: dict[str, Any] | None = None
) -> ClientExtension

Advertise an extension identifier (with optional settings) and nothing else.

Advertising an extension you do not implement asserts wire support you do not have; for behavioral extensions construct the real extension instead.

Source code in src/mcp/client/extension.py

def advertise(identifier: str, settings: dict[str, Any] | None = None) -> ClientExtension:
    """Advertise an extension identifier (with optional settings) and nothing else.

    Advertising an extension you do not implement asserts wire support you do
    not have; for behavioral extensions construct the real extension instead.
    """
    validate_extension_identifier(identifier, owner="advertise")
    return _AdvertiseOnly(identifier, {} if settings is None else settings)

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.

Extension result_claims fold into tools/call parsing at adopt(); notification_bindings observe vendor notifications via bounded FIFOs.

Source code in src/mcp/client/session.py

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.

    Extension `result_claims` fold into tools/call parsing at `adopt()`;
    `notification_bindings` observe vendor notifications via bounded FIFOs.
    """

    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,
        extensions: dict[str, dict[str, Any]] | None = None,
        result_claims: Mapping[str, Sequence[ResultClaim[Any]]] | None = None,
        notification_bindings: Sequence[NotificationBinding[Any]] | 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._extensions = dict(extensions) if extensions is not None else None
        self._result_claims = _index_claims(result_claims, extensions)
        self._notification_bindings = _index_bindings(notification_bindings)
        self._active_claims: dict[str, ResultClaim[Any]] = {}
        self._call_tool_adapter = _CallToolResultAdapter
        self._binding_queues: dict[
            str, tuple[MemoryObjectSendStream[BaseModel], MemoryObjectReceiveStream[BaseModel]]
        ] = {}
        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._x_mcp_header_maps: dict[str, dict[tuple[str, ...], str]] = {}
        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
            )

    async def __aenter__(self) -> Self:
        self._task_group = anyio.create_task_group()
        await self._task_group.__aenter__()
        try:
            # Queues must exist before the dispatcher starts: _on_notify enqueues into this dict.
            for binding in self._notification_bindings.values():
                send, receive = anyio.create_memory_object_stream[BaseModel](_NOTIFICATION_QUEUE_SIZE)
                self._binding_queues[binding.method] = (send, receive)
            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
            for binding in self._notification_bindings.values():
                _, receive = self._binding_queues[binding.method]
                self._task_group.start_soon(self._deliver_bound_notifications, binding, receive)
        except BaseException:
            # Unwind the entered task group before propagating: a cancellation
            # landing here (e.g. `move_on_after` around connect) would abandon
            # it and anyio would later raise "exited non-innermost cancel scope".
            task_group = self._task_group
            self._task_group = None
            task_group.cancel_scope.cancel()
            # Shield the group's own scope (a new one would break LIFO exit)
            # so a pending outer cancellation cannot re-fire inside __aexit__.
            task_group.cancel_scope.shield = True
            try:
                await task_group.__aexit__(None, None, None)
            finally:
                self._close_binding_queues()
            raise
        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> bool | None:
        # Exit must not block: cancel the dispatcher, binding consumers, and in-flight callbacks.
        assert self._task_group is not None
        self._task_group.cancel_scope.cancel()
        try:
            result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
        finally:
            self._close_binding_queues()
        await resync_tracer()
        return result

    def _close_binding_queues(self) -> None:
        # Unclosed memory object streams warn at garbage collection; close is idempotent.
        for send, receive in self._binding_queues.values():
            send.close()
            receive.close()
        self._binding_queues.clear()

    async def _deliver_bound_notifications(
        self, binding: NotificationBinding[Any], receive: MemoryObjectReceiveStream[BaseModel]
    ) -> None:
        """Consume one binding's FIFO, decoupled from the dispatcher so handlers can do session I/O."""
        while True:
            params = await receive.receive()
            try:
                await binding.handler(params)
            except Exception:
                # A raising handler costs only that delivery, as in _on_notify.
                logger.exception("notification binding handler for %r raised", binding.method)

    async def send_request(
        self,
        request: types.ClientRequest | types.Request[Any, Any],
        result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT],
        request_read_timeout_seconds: float | None = None,
        metadata: ClientMessageMetadata | None = None,
        progress_callback: ProgressFnT | None = None,
    ) -> ReceiveResultT:
        """Send a request and wait for its typed result.

        Args:
            metadata: Streamable HTTP resumption hints.

        Raises:
            MCPError: Error response, read timeout, or connection closed.
            RuntimeError: Called before entering the context manager.
            ValueError: The request declares `name_param` but its params carry no string name.
            pydantic.ValidationError: The server returned a result that does not
                conform to the negotiated protocol version.
        """
        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
        method: str = data["method"]
        opts: CallOptions = {}
        self._stamp(data, opts)
        # The stamp runs first, so its NAME_BEARING_METHODS rows win; a missing name fails loud.
        headers = opts.setdefault("headers", {})
        if (key := type(request).name_param) is not None and MCP_NAME_HEADER not in headers:
            params_data: dict[str, Any] = data.get("params") or {}
            name = params_data.get(key)
            if not isinstance(name, str):
                raise ValueError(f"{method} requires params[{key!r}] for Mcp-Name")
            headers[MCP_NAME_HEADER] = encode_header_value(name)
        timeout = (
            request_read_timeout_seconds
            if request_read_timeout_seconds is not None
            else self._session_read_timeout_seconds
        )
        if timeout is not None:
            opts["timeout"] = timeout
        if progress_callback is not None:
            opts["on_progress"] = progress_callback
        if metadata is not None:
            if metadata.resumption_token is not None:
                opts["resumption_token"] = metadata.resumption_token
            if metadata.on_resumption_token_update is not None:
                opts["on_resumption_token"] = metadata.on_resumption_token_update
        raw = await self._dispatcher.send_raw_request(method, data.get("params"), opts)
        _clamp_inbound_ttl(raw)
        # Literal fallback covers pre-handshake and stateless; matches runner.py.
        version = self._negotiated_version or "2025-11-25"
        try:
            _methods.validate_server_result(method, version, raw)
        except KeyError:
            pass
        if isinstance(result_type, TypeAdapter):
            return result_type.validate_python(raw, by_name=False)
        return result_type.model_validate(raw, by_name=False)

    async def send_notification(self, notification: types.ClientNotification) -> None:
        """Send a one-way notification. Usable before entering the context manager.

        Fire-and-forget: after the connection has closed, the notification is
        dropped with a debug log instead of raising.
        """
        data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
        opts: CallOptions = {}
        self._stamp(data, opts)
        await self._dispatcher.notify(data["method"], data.get("params"), opts)

    def _build_capabilities(self, version: str) -> types.ClientCapabilities:
        """Build the capability ad for a wire speaking `version`.

        Claim-bearing identifiers whose claims are all inactive at `version` drop, so
        the client never advertises result shapes it would reject; claim-less
        identifiers always advertise.
        """
        extensions = self._extensions
        if extensions is not None and self._result_claims:
            extensions = {
                identifier: settings
                for identifier, settings in extensions.items()
                if identifier not in self._result_claims
                or any(_claim_active(claim, version) for claim in self._result_claims[identifier])
            } or None
        sampling = (
            (self._sampling_capabilities or types.SamplingCapability())
            if self._sampling_callback is not _default_sampling_callback
            else None
        )
        elicitation = (
            types.ElicitationCapability(form=types.FormElicitationCapability(), url=types.UrlElicitationCapability())
            if self._elicitation_callback is not _default_elicitation_callback
            else None
        )
        roots = (
            # TODO: Should this be based on whether we
            # _will_ send notifications, or only whether
            # they're supported?
            types.RootsCapability(list_changed=True)
            if self._list_roots_callback is not _default_list_roots_callback
            else None
        )
        return types.ClientCapabilities(
            sampling=sampling, elicitation=elicitation, experimental=None, extensions=extensions, roots=roots
        )

    async def initialize(self) -> types.InitializeResult:
        if self._initialize_result is not None:
            return self._initialize_result
        result = await self.send_request(
            types.InitializeRequest(
                params=types.InitializeRequestParams(
                    protocol_version=LATEST_HANDSHAKE_VERSION,
                    # The handshake negotiates only legacy versions, where no claim is active.
                    capabilities=self._build_capabilities(LATEST_HANDSHAKE_VERSION),
                    client_info=self._client_info,
                ),
            ),
            types.InitializeResult,
        )

        if result.protocol_version not in HANDSHAKE_PROTOCOL_VERSIONS:
            raise RuntimeError(f"Unsupported protocol version from the server: {result.protocol_version}")

        self.adopt(result)

        await self.send_notification(types.InitializedNotification())

        return result

    def adopt(self, result: types.InitializeResult | types.DiscoverResult) -> None:
        """Install negotiated state from a result the caller already holds (no wire traffic).

        Clears the opposite slot, so at most one of `initialize_result` /
        `discover_result` is ever non-None.

        Raises:
            RuntimeError: `result` is a `DiscoverResult` whose `supported_versions`
                shares nothing with this client's `MODERN_PROTOCOL_VERSIONS`.
        """
        if isinstance(result, types.DiscoverResult):
            # ordered oldest→newest via MODERN_PROTOCOL_VERSIONS
            mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in result.supported_versions]
            if not mutual:
                raise RuntimeError(
                    f"No mutually supported modern protocol version "
                    f"(server: {result.supported_versions}, client: {list(MODERN_PROTOCOL_VERSIONS)})"
                )
            version = mutual[-1]
            client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
            capabilities = self._build_capabilities(version).model_dump(by_alias=True, mode="json", exclude_none=True)
            self._stamp = _make_modern_stamp(version, client_info, capabilities, self._resolve_param_headers)
            self._discover_result = result
            self._initialize_result = None
        else:
            version = result.protocol_version
            self._stamp = _make_handshake_stamp(version)
            self._initialize_result = result
            self._discover_result = None
        self._negotiated_version = version
        # Both arms reach here, so re-adoption resets cleanly; legacy versions activate no claims.
        # Core-vocabulary tags are unconstructible (ResultClaim.__post_init__), so no exclusion needed.
        self._active_claims = _active_claims_at(self._result_claims, version)
        self._call_tool_adapter = _build_call_tool_adapter(self._active_claims)
        for method in self._notification_bindings:
            # Bindings are consulted only for methods core does not know, so this one can never fire.
            if (method, version) in _methods.SERVER_NOTIFICATIONS:
                logger.warning(
                    "notification binding for %r will never fire at %s: the core protocol defines this method",
                    method,
                    version,
                )

    async def send_discover(self, version: str) -> dict[str, Any]:
        """Send a single ``server/discover`` at ``version`` and return the raw result dict.

        No retry, no ``adopt()``. The ``_meta`` envelope and the
        ``Mcp-Protocol-Version`` header are stamped at ``version`` so the
        server-side era router sees a coherent probe. Used by ``discover()`` and
        the connect-time auto-negotiation policy.

        Raises:
            MCPError: The server returned a JSON-RPC error, or the transport
                bounced the request at its own layer (a bare HTTP 4xx is
                synthesized into a JSON-RPC error by the transport).
        """
        client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
        capabilities = self._build_capabilities(version).model_dump(by_alias=True, mode="json", exclude_none=True)
        request = types.DiscoverRequest(
            params=types.RequestParams(
                _meta={
                    PROTOCOL_VERSION_META_KEY: version,
                    CLIENT_INFO_META_KEY: client_info,
                    CLIENT_CAPABILITIES_META_KEY: capabilities,
                }
            )
        )
        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
        opts: CallOptions = {
            "timeout": DISCOVER_TIMEOUT_SECONDS,
            "cancel_on_abandon": False,
            "headers": {MCP_PROTOCOL_VERSION_HEADER: version, MCP_METHOD_HEADER: data["method"]},
        }
        raw = await self._dispatcher.send_raw_request(data["method"], data.get("params"), opts)
        # Un-floored, a negative ttl fails the mode='auto' probe's validation and silently downgrades the handshake.
        _clamp_inbound_ttl(raw)
        return raw

    async def discover(self) -> types.DiscoverResult:
        """Probe `server/discover` and adopt the result.

        Sends a single `server/discover` proposing the newest modern protocol
        version. On `UNSUPPORTED_PROTOCOL_VERSION` (-32022) the server's
        `supported` list is intersected with `MODERN_PROTOCOL_VERSIONS` and the
        probe is retried once at the highest mutual version. Any other error —
        including `METHOD_NOT_FOUND` (-32601) and `REQUEST_TIMEOUT` (-32001) —
        propagates; the legacy `initialize()` fallback is the caller's policy.

        Raises:
            MCPError: The server rejected `server/discover`, the probe timed
                out, or the -32022 retry found no mutual version / failed again.
            RuntimeError: `adopt()` found no mutual version in the returned
                `supported_versions`.
        """
        if self._discover_result is not None:
            return self._discover_result

        try:
            raw = await self.send_discover(LATEST_MODERN_VERSION)
        except MCPError as e:
            if e.code != UNSUPPORTED_PROTOCOL_VERSION:
                raise
            try:
                data = types.UnsupportedProtocolVersionErrorData.model_validate(e.error.data)
            except ValidationError:
                raise e from None
            # ordered oldest→newest via MODERN_PROTOCOL_VERSIONS
            mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in data.supported]
            if not mutual:
                raise
            raw = await self.send_discover(mutual[-1])

        result = types.DiscoverResult.model_validate(raw)
        self.adopt(result)
        return result

    @property
    def initialize_result(self) -> types.InitializeResult | None:
        """The server's InitializeResult. None unless `initialize()` ran (or was adopted)."""
        return self._initialize_result

    @property
    def discover_result(self) -> types.DiscoverResult | None:
        """The server's DiscoverResult. None unless `discover()` ran (or was adopted).

        Retained intact (supported_versions, ttl_ms, cache_scope) so callers
        can round-trip it as ``prior_discover=``.
        """
        return self._discover_result

    @property
    def protocol_version(self) -> str | None:
        """Negotiated protocol version. None until `initialize()`, `discover()`, or `adopt()`."""
        return self._negotiated_version

    @property
    def server_info(self) -> types.Implementation | None:
        """Server name/version. None until `initialize()`, `discover()`, or `adopt()`."""
        if self._discover_result is not None:
            return self._discover_result.server_info
        if self._initialize_result is not None:
            return self._initialize_result.server_info
        return None

    @property
    def server_capabilities(self) -> types.ServerCapabilities | None:
        """Server capabilities. None until `initialize()`, `discover()`, or `adopt()`."""
        if self._discover_result is not None:
            return self._discover_result.capabilities
        if self._initialize_result is not None:
            return self._initialize_result.capabilities
        return None

    @property
    def instructions(self) -> str | None:
        """Server-provided instructions text, if any."""
        if self._discover_result is not None:
            return self._discover_result.instructions
        if self._initialize_result is not None:
            return self._initialize_result.instructions
        return None

    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a ping request."""
        return await self.send_request(types.PingRequest(params=types.RequestParams(_meta=meta)), types.EmptyResult)

    @deprecated(
        "Client-to-server progress is deprecated as of 2026-07-28; progress is server-to-client only.",
        category=MCPDeprecationWarning,
    )
    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> None:
        """Send a progress notification."""
        await self.send_notification(
            types.ProgressNotification(
                params=types.ProgressNotificationParams(
                    progress_token=progress_token,
                    progress=progress,
                    total=total,
                    message=message,
                    _meta=meta,
                ),
            )
        )

    @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def set_logging_level(
        self,
        level: types.LoggingLevel,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> types.EmptyResult:
        """Send a logging/setLevel request."""
        return await self.send_request(
            types.SetLevelRequest(params=types.SetLevelRequestParams(level=level, _meta=meta)),
            types.EmptyResult,
        )

    async def list_resources(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListResourcesResult:
        """Send a resources/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(types.ListResourcesRequest(params=params), types.ListResourcesResult)

    async def list_resource_templates(
        self, *, params: types.PaginatedRequestParams | None = None
    ) -> types.ListResourceTemplatesResult:
        """Send a resources/templates/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(
            types.ListResourceTemplatesRequest(params=params),
            types.ListResourceTemplatesResult,
        )

    @overload
    async def read_resource(
        self,
        uri: str,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: Literal[False] = False,
    ) -> types.ReadResourceResult: ...

    @overload
    async def read_resource(
        self,
        uri: str,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool,
    ) -> types.ReadResourceResult | types.InputRequiredResult: ...

    async def read_resource(
        self,
        uri: str,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool = False,
    ) -> types.ReadResourceResult | types.InputRequiredResult:
        """Send a resources/read request.

        Args:
            input_responses: Responses to a prior `InputRequiredResult.input_requests`.
            request_state: Opaque state echoed from a prior `InputRequiredResult`.
            allow_input_required: When `False` (default), an `InputRequiredResult`
                from the server raises `RuntimeError`; when `True`, it is returned
                so the caller can resolve the requests and retry.

        Raises:
            RuntimeError: If the server returns an `InputRequiredResult` and
                `allow_input_required` is `False`.
        """
        result = await self.send_request(
            types.ReadResourceRequest(
                params=types.ReadResourceRequestParams(
                    uri=uri,
                    input_responses=input_responses,
                    request_state=request_state,
                    _meta=meta,
                ),
            ),
            _ReadResourceResultAdapter,
        )
        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
            raise _input_required_unexpected("read_resource")
        return result

    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a resources/subscribe request."""
        return await self.send_request(
            types.SubscribeRequest(params=types.SubscribeRequestParams(uri=uri, _meta=meta)),
            types.EmptyResult,
        )

    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
        """Send a resources/unsubscribe request."""
        return await self.send_request(
            types.UnsubscribeRequest(params=types.UnsubscribeRequestParams(uri=uri, _meta=meta)),
            types.EmptyResult,
        )

    @overload
    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: Literal[False] = False,
        allow_claimed: Literal[False] = False,
    ) -> types.CallToolResult: ...

    @overload
    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool,
        allow_claimed: Literal[False] = False,
    ) -> types.CallToolResult | types.InputRequiredResult: ...

    @overload
    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: Literal[False] = False,
        allow_claimed: bool,
    ) -> types.CallToolResult | types.Result: ...

    @overload
    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool,
        allow_claimed: bool,
    ) -> types.CallToolResult | types.InputRequiredResult | types.Result: ...

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool = False,
        allow_claimed: bool = False,
    ) -> types.CallToolResult | types.InputRequiredResult | types.Result:
        """Send a tools/call request with optional progress callback support.

        On a modern (2026-07-28) connection, arguments annotated with `x-mcp-header`
        in the tool's input schema are mirrored into `Mcp-Param-*` request headers.
        The annotations are read from the tool's last `list_tools` entry, so list
        the tool before calling it to enable header emission.

        Args:
            input_responses: Responses to a prior `InputRequiredResult.input_requests`.
            request_state: Opaque state echoed from a prior `InputRequiredResult`.
            allow_input_required: When ``False`` (default), an `InputRequiredResult`
                from the server raises `RuntimeError`; when ``True``, it is returned
                so the caller can resolve the requests and retry.
            allow_claimed: When `False` (default), a claimed extension result raises
                `UnexpectedClaimedResult`; when `True`, the parsed claim model is returned.

        Raises:
            RuntimeError: If the server returns an `InputRequiredResult` and
                ``allow_input_required`` is ``False``.
            UnexpectedClaimedResult: Claimed result with `allow_claimed` False; carries the parsed value.
        """
        result = await self.send_request(
            types.CallToolRequest(
                params=types.CallToolRequestParams(
                    name=name,
                    arguments=arguments,
                    input_responses=input_responses,
                    request_state=request_state,
                    _meta=meta,
                ),
            ),
            self._call_tool_adapter,
            request_read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
        )

        if isinstance(result, types.CallToolResult) and not result.is_error:
            await self.validate_tool_result(name, result)

        # The input_required arm stays first; a claimed shape is terminal for the multi-round-trip driver.
        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
            raise _input_required_unexpected("call_tool")
        if not isinstance(result, types.CallToolResult | types.InputRequiredResult) and not allow_claimed:
            raise UnexpectedClaimedResult(result)
        return result

    def _resolve_param_headers(self, name: str, arguments: Mapping[str, Any]) -> dict[str, str]:
        """`Mcp-Param-*` headers for a `tools/call`, or empty when the tool was never listed."""
        header_map = self._x_mcp_header_maps.get(name)
        if header_map is None:
            return {}
        return mcp_param_headers(header_map, arguments)

    async def validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
        """Revalidate a `CallToolResult` against the tool's declared output schema.

        Raises:
            RuntimeError: Structured content is missing or does not conform to the schema.
        """
        if name not in self._tool_output_schemas:
            # refresh output schema cache
            await self.list_tools()

        output_schema = None
        if name in self._tool_output_schemas:
            output_schema = self._tool_output_schemas.get(name)
        else:
            logger.warning(f"Tool {name} not listed by server, cannot validate any structured content")

        if output_schema is not None:
            from jsonschema import SchemaError, ValidationError, validate

            if result.structured_content is None:
                raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
            try:
                validate(result.structured_content, output_schema)
            except ValidationError as e:
                raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
            except SchemaError as e:  # pragma: no cover
                raise RuntimeError(f"Invalid schema for tool {name}: {e}")  # pragma: no cover

    async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
        """Send a prompts/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)

    @overload
    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: Literal[False] = False,
    ) -> types.GetPromptResult: ...

    @overload
    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool,
    ) -> types.GetPromptResult | types.InputRequiredResult: ...

    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        input_responses: types.InputResponses | None = None,
        request_state: str | None = None,
        meta: RequestParamsMeta | None = None,
        allow_input_required: bool = False,
    ) -> types.GetPromptResult | types.InputRequiredResult:
        """Send a prompts/get request.

        Args:
            input_responses: Responses to a prior `InputRequiredResult.input_requests`.
            request_state: Opaque state echoed from a prior `InputRequiredResult`.
            allow_input_required: When `False` (default), an `InputRequiredResult`
                from the server raises `RuntimeError`; when `True`, it is returned
                so the caller can resolve the requests and retry.

        Raises:
            RuntimeError: If the server returns an `InputRequiredResult` and
                `allow_input_required` is `False`.
        """
        result = await self.send_request(
            types.GetPromptRequest(
                params=types.GetPromptRequestParams(
                    name=name,
                    arguments=arguments,
                    input_responses=input_responses,
                    request_state=request_state,
                    _meta=meta,
                ),
            ),
            _GetPromptResultAdapter,
        )
        if isinstance(result, types.InputRequiredResult) and not allow_input_required:
            raise _input_required_unexpected("get_prompt")
        return result

    async def complete(
        self,
        ref: types.ResourceTemplateReference | types.PromptReference,
        argument: dict[str, str],
        context_arguments: dict[str, str] | None = None,
    ) -> types.CompleteResult:
        """Send a completion/complete request."""
        context = None
        if context_arguments is not None:
            context = types.CompletionContext(arguments=context_arguments)

        return await self.send_request(
            types.CompleteRequest(
                params=types.CompleteRequestParams(
                    ref=ref,
                    argument=types.CompletionArgument(**argument),
                    context=context,
                ),
            ),
            types.CompleteResult,
        )

    async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult:
        """Send a tools/list request.

        Args:
            params: Full pagination parameters including cursor and any future fields
        """
        result = await self.send_request(
            types.ListToolsRequest(params=params),
            types.ListToolsResult,
        )
        complete = (params is None or params.cursor is None) and result.next_cursor is None
        return self._absorb_tool_listing(result, complete=complete)

    def _absorb_tool_listing(self, result: types.ListToolsResult, *, complete: bool) -> types.ListToolsResult:
        """Filter the listing per the 2026 x-mcp-header MUST and rebuild derived per-tool state, in place.

        Idempotent: cached values are already post-filter, so the response cache can re-absorb a served listing.
        `complete` (an uncursored single-page listing) prunes per-tool state down to the listing's tools.
        """
        if self._negotiated_version in MODERN_PROTOCOL_VERSIONS:
            # 2026-07-28: clients MUST drop tools whose x-mcp-header annotations are invalid.
            kept: list[types.Tool] = []
            for tool in result.tools:
                if (reason := find_invalid_x_mcp_header(tool.input_schema)) is not None:
                    logger.warning("dropping tool %r: invalid x-mcp-header (%s)", tool.name, reason)
                    # Evict any map cached from a prior valid listing so a stale entry can't
                    # mirror headers for a tool this listing dropped.
                    self._x_mcp_header_maps.pop(tool.name, None)
                    continue
                # Cache the arg→header map so a later tools/call mirrors it into Mcp-Param-* headers.
                self._x_mcp_header_maps[tool.name] = x_mcp_header_map(tool.input_schema)
                kept.append(tool)
            result.tools = kept

        # Cache tool output schemas for future validation; cursor pages only ever add.
        for tool in result.tools:
            self._tool_output_schemas[tool.name] = tool.output_schema

        if complete:
            # The listing is the full tool universe, so state for unlisted tools is stale
            # (the server dropped them, or a shared-cache writer's filter did).
            names = {tool.name for tool in result.tools}
            self._x_mcp_header_maps = {k: v for k, v in self._x_mcp_header_maps.items() if k in names}
            self._tool_output_schemas = {k: v for k, v in self._tool_output_schemas.items() if k in names}

        return result

    @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def send_roots_list_changed(self) -> None:
        """Send a roots/list_changed notification."""
        await self.send_notification(types.RootsListChangedNotification())

    async def _on_request(
        self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
    ) -> dict[str, Any]:
        """Answer a server-initiated request via the registered callbacks."""
        # Literal, not LATEST_PROTOCOL_VERSION: the fallback covers the initialize
        # handshake (which only exists at <=2025) and stateless until the header
        # is plumbed; its meaning is fixed regardless of LATEST bumps.
        version = self._negotiated_version or "2025-11-25"
        try:
            request = cast(types.ServerRequest, _methods.parse_server_request(method, version, params))
        except KeyError:
            raise MCPError(code=METHOD_NOT_FOUND, message="Method not found", data=method) from None

        response: types.ClientResult | types.ErrorData
        if isinstance(request, types.PingRequest):
            # Answered without a context: ping has no callback that would need one.
            response = types.EmptyResult()
        else:
            assert dctx.request_id is not None  # the callback-driving dispatchers always assign ids
            ctx = ClientRequestContext(
                session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None
            )
            response = await self.dispatch_input_request(ctx, request)
        client_response = ClientResponse.validate_python(response)
        if isinstance(client_response, types.ErrorData):
            raise MCPError.from_error_data(client_response)
        dumped = client_response.model_dump(by_alias=True, mode="json", exclude_none=True)
        try:
            _methods.validate_client_result(method, version, dumped)
        except ValidationError:
            logger.exception("client callback for %r returned an invalid result", method)
            raise MCPError(code=INTERNAL_ERROR, message="Client callback returned an invalid result") from None
        return dumped

    async def dispatch_input_request(
        self, ctx: ClientRequestContext, request: types.InputRequest
    ) -> types.InputResponse | types.ErrorData:
        """Route an input request through the client's callback table.

        Shared by the legacy server→client RPC path (`_on_request`) and the
        2026-07-28 multi-round-trip driver, which dispatches the embedded
        `InputRequiredResult.input_requests` through the same callbacks.

        Returns the callback's `InputResponse`, or `ErrorData` when the callback declines.
        """
        match request:
            case types.CreateMessageRequest(params=p):
                return await self._sampling_callback(ctx, p)
            case types.ElicitRequest(params=p):
                return await self._elicitation_callback(ctx, p)
            case types.ListRootsRequest():  # pragma: no branch
                return await self._list_roots_callback(ctx)

    async def _on_notify(
        self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
    ) -> None:
        """Route a server notification: validate, run the typed callback, tee to message_handler."""
        # Same fallback as `_on_request`: covers pre-handshake and stateless.
        version = self._negotiated_version or "2025-11-25"
        try:
            notification = cast(types.ServerNotification, _methods.parse_server_notification(method, version, params))
        except KeyError:
            # Only methods unknown to the negotiated version's core tables reach the bindings.
            binding = self._notification_bindings.get(method)
            if binding is None:
                logger.debug("dropped %r: not defined at %s", method, version)
                return
            try:
                bound_params = binding.params_type.model_validate(params or {})
            except ValidationError:
                logger.warning("Failed to validate notification: %s", method, exc_info=True)
                return
            send, receive = self._binding_queues[method]
            try:
                # Must not await: DirectDispatcher calls _on_notify inline; blocking deadlocks in-process servers.
                send.send_nowait(bound_params)
            except anyio.WouldBlock:
                # Evict the oldest event; no checkpoint since the failed send,
                # so the buffer is still full and the retry cannot block.
                receive.receive_nowait()
                logger.warning("notification queue for %r is full; dropped the oldest event", method)
                send.send_nowait(bound_params)
            return
        except ValidationError:
            logger.warning("Failed to validate notification: %s", method, exc_info=True)
            return
        if isinstance(notification, types.CancelledNotification):
            # The dispatcher already applied the cancellation; not surfaced to message_handler.
            return
        try:
            if isinstance(notification, types.LoggingMessageNotification):
                await self._logging_callback(notification.params)
            await self._message_handler(notification)
        except Exception:
            # Contain here, not in the dispatcher: DirectDispatcher awaits this
            # handler inline in the peer's notify() call, so a raising callback
            # would otherwise fail the peer's send. A raising logging_callback
            # skips the message_handler tee for that notification (v1 parity).
            logger.exception("notification callback for %r raised", method)

    async def _on_stream_exception(self, exc: Exception) -> None:
        """Deliver a transport-level fault to message_handler via a spawned task.

        Running the handler inline would park the dispatcher's read loop and
        deadlock handlers that await session I/O.
        """
        assert self._task_group is not None
        self._task_group.start_soon(self._deliver_stream_exception, exc)

    async def _deliver_stream_exception(self, exc: Exception) -> None:
        try:
            await self._message_handler(exc)
        except Exception:
            logger.exception("message_handler raised on transport exception")

send_request async

send_request(
    request: ClientRequest | Request[Any, Any],
    result_type: (
        type[ReceiveResultT] | TypeAdapter[ReceiveResultT]
    ),
    request_read_timeout_seconds: float | None = None,
    metadata: ClientMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ReceiveResultT

Send a request and wait for its typed result.

Parameters:

Name	Type	Description	Default
metadata	ClientMessageMetadata | None	Streamable HTTP resumption hints.	None

Raises:

Type	Description
MCPError	Error response, read timeout, or connection closed.
RuntimeError	Called before entering the context manager.
ValueError	The request declares name_param but its params carry no string name.
ValidationError	The server returned a result that does not conform to the negotiated protocol version.

Source code in src/mcp/client/session.py

async def send_request(
    self,
    request: types.ClientRequest | types.Request[Any, Any],
    result_type: type[ReceiveResultT] | TypeAdapter[ReceiveResultT],
    request_read_timeout_seconds: float | None = None,
    metadata: ClientMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ReceiveResultT:
    """Send a request and wait for its typed result.

    Args:
        metadata: Streamable HTTP resumption hints.

    Raises:
        MCPError: Error response, read timeout, or connection closed.
        RuntimeError: Called before entering the context manager.
        ValueError: The request declares `name_param` but its params carry no string name.
        pydantic.ValidationError: The server returned a result that does not
            conform to the negotiated protocol version.
    """
    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
    method: str = data["method"]
    opts: CallOptions = {}
    self._stamp(data, opts)
    # The stamp runs first, so its NAME_BEARING_METHODS rows win; a missing name fails loud.
    headers = opts.setdefault("headers", {})
    if (key := type(request).name_param) is not None and MCP_NAME_HEADER not in headers:
        params_data: dict[str, Any] = data.get("params") or {}
        name = params_data.get(key)
        if not isinstance(name, str):
            raise ValueError(f"{method} requires params[{key!r}] for Mcp-Name")
        headers[MCP_NAME_HEADER] = encode_header_value(name)
    timeout = (
        request_read_timeout_seconds
        if request_read_timeout_seconds is not None
        else self._session_read_timeout_seconds
    )
    if timeout is not None:
        opts["timeout"] = timeout
    if progress_callback is not None:
        opts["on_progress"] = progress_callback
    if metadata is not None:
        if metadata.resumption_token is not None:
            opts["resumption_token"] = metadata.resumption_token
        if metadata.on_resumption_token_update is not None:
            opts["on_resumption_token"] = metadata.on_resumption_token_update
    raw = await self._dispatcher.send_raw_request(method, data.get("params"), opts)
    _clamp_inbound_ttl(raw)
    # Literal fallback covers pre-handshake and stateless; matches runner.py.
    version = self._negotiated_version or "2025-11-25"
    try:
        _methods.validate_server_result(method, version, raw)
    except KeyError:
        pass
    if isinstance(result_type, TypeAdapter):
        return result_type.validate_python(raw, by_name=False)
    return result_type.model_validate(raw, by_name=False)

send_notification async

send_notification(notification: ClientNotification) -> None

Send a one-way notification. Usable before entering the context manager.

Fire-and-forget: after the connection has closed, the notification is dropped with a debug log instead of raising.

Source code in src/mcp/client/session.py

async def send_notification(self, notification: types.ClientNotification) -> None:
    """Send a one-way notification. Usable before entering the context manager.

    Fire-and-forget: after the connection has closed, the notification is
    dropped with a debug log instead of raising.
    """
    data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
    opts: CallOptions = {}
    self._stamp(data, opts)
    await self._dispatcher.notify(data["method"], data.get("params"), opts)

adopt

adopt(result: InitializeResult | DiscoverResult) -> None

Install negotiated state from a result the caller already holds (no wire traffic).

Clears the opposite slot, so at most one of initialize_result / discover_result is ever non-None.

Raises:

Type	Description
RuntimeError	result is a DiscoverResult whose supported_versions shares nothing with this client's MODERN_PROTOCOL_VERSIONS.

Source code in src/mcp/client/session.py

def adopt(self, result: types.InitializeResult | types.DiscoverResult) -> None:
    """Install negotiated state from a result the caller already holds (no wire traffic).

    Clears the opposite slot, so at most one of `initialize_result` /
    `discover_result` is ever non-None.

    Raises:
        RuntimeError: `result` is a `DiscoverResult` whose `supported_versions`
            shares nothing with this client's `MODERN_PROTOCOL_VERSIONS`.
    """
    if isinstance(result, types.DiscoverResult):
        # ordered oldest→newest via MODERN_PROTOCOL_VERSIONS
        mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in result.supported_versions]
        if not mutual:
            raise RuntimeError(
                f"No mutually supported modern protocol version "
                f"(server: {result.supported_versions}, client: {list(MODERN_PROTOCOL_VERSIONS)})"
            )
        version = mutual[-1]
        client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
        capabilities = self._build_capabilities(version).model_dump(by_alias=True, mode="json", exclude_none=True)
        self._stamp = _make_modern_stamp(version, client_info, capabilities, self._resolve_param_headers)
        self._discover_result = result
        self._initialize_result = None
    else:
        version = result.protocol_version
        self._stamp = _make_handshake_stamp(version)
        self._initialize_result = result
        self._discover_result = None
    self._negotiated_version = version
    # Both arms reach here, so re-adoption resets cleanly; legacy versions activate no claims.
    # Core-vocabulary tags are unconstructible (ResultClaim.__post_init__), so no exclusion needed.
    self._active_claims = _active_claims_at(self._result_claims, version)
    self._call_tool_adapter = _build_call_tool_adapter(self._active_claims)
    for method in self._notification_bindings:
        # Bindings are consulted only for methods core does not know, so this one can never fire.
        if (method, version) in _methods.SERVER_NOTIFICATIONS:
            logger.warning(
                "notification binding for %r will never fire at %s: the core protocol defines this method",
                method,
                version,
            )

send_discover async

send_discover(version: str) -> dict[str, Any]

Send a single server/discover at version and return the raw result dict.

No retry, no adopt(). The _meta envelope and the Mcp-Protocol-Version header are stamped at version so the server-side era router sees a coherent probe. Used by discover() and the connect-time auto-negotiation policy.

Raises:

Type	Description
MCPError	The server returned a JSON-RPC error, or the transport bounced the request at its own layer (a bare HTTP 4xx is synthesized into a JSON-RPC error by the transport).

Source code in src/mcp/client/session.py

async def send_discover(self, version: str) -> dict[str, Any]:
    """Send a single ``server/discover`` at ``version`` and return the raw result dict.

    No retry, no ``adopt()``. The ``_meta`` envelope and the
    ``Mcp-Protocol-Version`` header are stamped at ``version`` so the
    server-side era router sees a coherent probe. Used by ``discover()`` and
    the connect-time auto-negotiation policy.

    Raises:
        MCPError: The server returned a JSON-RPC error, or the transport
            bounced the request at its own layer (a bare HTTP 4xx is
            synthesized into a JSON-RPC error by the transport).
    """
    client_info = self._client_info.model_dump(by_alias=True, mode="json", exclude_none=True)
    capabilities = self._build_capabilities(version).model_dump(by_alias=True, mode="json", exclude_none=True)
    request = types.DiscoverRequest(
        params=types.RequestParams(
            _meta={
                PROTOCOL_VERSION_META_KEY: version,
                CLIENT_INFO_META_KEY: client_info,
                CLIENT_CAPABILITIES_META_KEY: capabilities,
            }
        )
    )
    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
    opts: CallOptions = {
        "timeout": DISCOVER_TIMEOUT_SECONDS,
        "cancel_on_abandon": False,
        "headers": {MCP_PROTOCOL_VERSION_HEADER: version, MCP_METHOD_HEADER: data["method"]},
    }
    raw = await self._dispatcher.send_raw_request(data["method"], data.get("params"), opts)
    # Un-floored, a negative ttl fails the mode='auto' probe's validation and silently downgrades the handshake.
    _clamp_inbound_ttl(raw)
    return raw

discover async

discover() -> DiscoverResult

Probe server/discover and adopt the result.

Sends a single server/discover proposing the newest modern protocol version. On UNSUPPORTED_PROTOCOL_VERSION (-32022) the server's supported list is intersected with MODERN_PROTOCOL_VERSIONS and the probe is retried once at the highest mutual version. Any other error — including METHOD_NOT_FOUND (-32601) and REQUEST_TIMEOUT (-32001) — propagates; the legacy initialize() fallback is the caller's policy.

Raises:

Type	Description
MCPError	The server rejected server/discover, the probe timed out, or the -32022 retry found no mutual version / failed again.
RuntimeError	adopt() found no mutual version in the returned supported_versions.

Source code in src/mcp/client/session.py

async def discover(self) -> types.DiscoverResult:
    """Probe `server/discover` and adopt the result.

    Sends a single `server/discover` proposing the newest modern protocol
    version. On `UNSUPPORTED_PROTOCOL_VERSION` (-32022) the server's
    `supported` list is intersected with `MODERN_PROTOCOL_VERSIONS` and the
    probe is retried once at the highest mutual version. Any other error —
    including `METHOD_NOT_FOUND` (-32601) and `REQUEST_TIMEOUT` (-32001) —
    propagates; the legacy `initialize()` fallback is the caller's policy.

    Raises:
        MCPError: The server rejected `server/discover`, the probe timed
            out, or the -32022 retry found no mutual version / failed again.
        RuntimeError: `adopt()` found no mutual version in the returned
            `supported_versions`.
    """
    if self._discover_result is not None:
        return self._discover_result

    try:
        raw = await self.send_discover(LATEST_MODERN_VERSION)
    except MCPError as e:
        if e.code != UNSUPPORTED_PROTOCOL_VERSION:
            raise
        try:
            data = types.UnsupportedProtocolVersionErrorData.model_validate(e.error.data)
        except ValidationError:
            raise e from None
        # ordered oldest→newest via MODERN_PROTOCOL_VERSIONS
        mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in data.supported]
        if not mutual:
            raise
        raw = await self.send_discover(mutual[-1])

    result = types.DiscoverResult.model_validate(raw)
    self.adopt(result)
    return result

initialize_result property

initialize_result: InitializeResult | None

The server's InitializeResult. None unless initialize() ran (or was adopted).

discover_result property

discover_result: DiscoverResult | None

The server's DiscoverResult. None unless discover() ran (or was adopted).

Retained intact (supported_versions, ttl_ms, cache_scope) so callers can round-trip it as prior_discover=.

protocol_version property

protocol_version: str | None

Negotiated protocol version. None until initialize(), discover(), or adopt().

server_info property

server_info: Implementation | None

Server name/version. None until initialize(), discover(), or adopt().

server_capabilities property

server_capabilities: ServerCapabilities | None

Server capabilities. None until initialize(), discover(), or adopt().

instructions property

instructions: str | None

Server-provided instructions text, if any.

send_ping async

send_ping(
    *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a ping request.

Source code in src/mcp/client/session.py

async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a ping request."""
    return await self.send_request(types.PingRequest(params=types.RequestParams(_meta=meta)), types.EmptyResult)

send_progress_notification async

send_progress_notification(
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    *,
    meta: RequestParamsMeta | None = None
) -> None

Send a progress notification.

Source code in src/mcp/client/session.py

@deprecated(
    "Client-to-server progress is deprecated as of 2026-07-28; progress is server-to-client only.",
    category=MCPDeprecationWarning,
)
async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> None:
    """Send a progress notification."""
    await self.send_notification(
        types.ProgressNotification(
            params=types.ProgressNotificationParams(
                progress_token=progress_token,
                progress=progress,
                total=total,
                message=message,
                _meta=meta,
            ),
        )
    )

set_logging_level async

set_logging_level(
    level: LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a logging/setLevel request.

Source code in src/mcp/client/session.py

@deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def set_logging_level(
    self,
    level: types.LoggingLevel,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.EmptyResult:
    """Send a logging/setLevel request."""
    return await self.send_request(
        types.SetLevelRequest(params=types.SetLevelRequestParams(level=level, _meta=meta)),
        types.EmptyResult,
    )

list_resources async

list_resources(
    *, params: PaginatedRequestParams | None = None
) -> ListResourcesResult

Send a resources/list request.

Parameters:

Name	Type	Description	Default
params	PaginatedRequestParams | None	Full pagination parameters including cursor and any future fields	None

Source code in src/mcp/client/session.py

async def list_resources(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListResourcesResult:
    """Send a resources/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(types.ListResourcesRequest(params=params), types.ListResourcesResult)

list_resource_templates async

list_resource_templates(
    *, params: PaginatedRequestParams | None = None
) -> ListResourceTemplatesResult

Send a resources/templates/list request.

Parameters:

Name	Type	Description	Default
params	PaginatedRequestParams | None	Full pagination parameters including cursor and any future fields	None

Source code in src/mcp/client/session.py

async def list_resource_templates(
    self, *, params: types.PaginatedRequestParams | None = None
) -> types.ListResourceTemplatesResult:
    """Send a resources/templates/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(
        types.ListResourceTemplatesRequest(params=params),
        types.ListResourceTemplatesResult,
    )

read_resource async

read_resource(
    uri: str,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: Literal[False] = False
) -> ReadResourceResult

read_resource(
    uri: str,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool
) -> ReadResourceResult | InputRequiredResult

read_resource(
    uri: str,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False
) -> ReadResourceResult | InputRequiredResult

Send a resources/read request.

Parameters:

Name	Type	Description	Default
input_responses	InputResponses | None	Responses to a prior InputRequiredResult.input_requests.	None
request_state	str | None	Opaque state echoed from a prior InputRequiredResult.	None
allow_input_required	bool	When False (default), an InputRequiredResult from the server raises RuntimeError; when True, it is returned so the caller can resolve the requests and retry.	False

Raises:

Type	Description
RuntimeError	If the server returns an InputRequiredResult and allow_input_required is False.

Source code in src/mcp/client/session.py

async def read_resource(
    self,
    uri: str,
    *,
    input_responses: types.InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False,
) -> types.ReadResourceResult | types.InputRequiredResult:
    """Send a resources/read request.

    Args:
        input_responses: Responses to a prior `InputRequiredResult.input_requests`.
        request_state: Opaque state echoed from a prior `InputRequiredResult`.
        allow_input_required: When `False` (default), an `InputRequiredResult`
            from the server raises `RuntimeError`; when `True`, it is returned
            so the caller can resolve the requests and retry.

    Raises:
        RuntimeError: If the server returns an `InputRequiredResult` and
            `allow_input_required` is `False`.
    """
    result = await self.send_request(
        types.ReadResourceRequest(
            params=types.ReadResourceRequestParams(
                uri=uri,
                input_responses=input_responses,
                request_state=request_state,
                _meta=meta,
            ),
        ),
        _ReadResourceResultAdapter,
    )
    if isinstance(result, types.InputRequiredResult) and not allow_input_required:
        raise _input_required_unexpected("read_resource")
    return result

subscribe_resource async

subscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a resources/subscribe request.

Source code in src/mcp/client/session.py

async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a resources/subscribe request."""
    return await self.send_request(
        types.SubscribeRequest(params=types.SubscribeRequestParams(uri=uri, _meta=meta)),
        types.EmptyResult,
    )

unsubscribe_resource async

unsubscribe_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> EmptyResult

Send a resources/unsubscribe request.

Source code in src/mcp/client/session.py

async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
    """Send a resources/unsubscribe request."""
    return await self.send_request(
        types.UnsubscribeRequest(params=types.UnsubscribeRequestParams(uri=uri, _meta=meta)),
        types.EmptyResult,
    )

call_tool async

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: Literal[False] = False,
    allow_claimed: Literal[False] = False
) -> CallToolResult

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool,
    allow_claimed: Literal[False] = False
) -> CallToolResult | InputRequiredResult

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: Literal[False] = False,
    allow_claimed: bool
) -> CallToolResult | Result

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool,
    allow_claimed: bool
) -> CallToolResult | InputRequiredResult | Result

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False,
    allow_claimed: bool = False
) -> CallToolResult | InputRequiredResult | Result

Send a tools/call request with optional progress callback support.

On a modern (2026-07-28) connection, arguments annotated with x-mcp-header in the tool's input schema are mirrored into Mcp-Param-* request headers. The annotations are read from the tool's last list_tools entry, so list the tool before calling it to enable header emission.

Parameters:

Name	Type	Description	Default
input_responses	InputResponses | None	Responses to a prior InputRequiredResult.input_requests.	None
request_state	str | None	Opaque state echoed from a prior InputRequiredResult.	None
allow_input_required	bool	When False (default), an InputRequiredResult from the server raises RuntimeError; when True, it is returned so the caller can resolve the requests and retry.	False
allow_claimed	bool	When False (default), a claimed extension result raises UnexpectedClaimedResult; when True, the parsed claim model is returned.	False

Raises:

Type	Description
RuntimeError	If the server returns an InputRequiredResult and allow_input_required is False.
UnexpectedClaimedResult	Claimed result with allow_claimed False; carries the parsed value.

Source code in src/mcp/client/session.py

async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    input_responses: types.InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False,
    allow_claimed: bool = False,
) -> types.CallToolResult | types.InputRequiredResult | types.Result:
    """Send a tools/call request with optional progress callback support.

    On a modern (2026-07-28) connection, arguments annotated with `x-mcp-header`
    in the tool's input schema are mirrored into `Mcp-Param-*` request headers.
    The annotations are read from the tool's last `list_tools` entry, so list
    the tool before calling it to enable header emission.

    Args:
        input_responses: Responses to a prior `InputRequiredResult.input_requests`.
        request_state: Opaque state echoed from a prior `InputRequiredResult`.
        allow_input_required: When ``False`` (default), an `InputRequiredResult`
            from the server raises `RuntimeError`; when ``True``, it is returned
            so the caller can resolve the requests and retry.
        allow_claimed: When `False` (default), a claimed extension result raises
            `UnexpectedClaimedResult`; when `True`, the parsed claim model is returned.

    Raises:
        RuntimeError: If the server returns an `InputRequiredResult` and
            ``allow_input_required`` is ``False``.
        UnexpectedClaimedResult: Claimed result with `allow_claimed` False; carries the parsed value.
    """
    result = await self.send_request(
        types.CallToolRequest(
            params=types.CallToolRequestParams(
                name=name,
                arguments=arguments,
                input_responses=input_responses,
                request_state=request_state,
                _meta=meta,
            ),
        ),
        self._call_tool_adapter,
        request_read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
    )

    if isinstance(result, types.CallToolResult) and not result.is_error:
        await self.validate_tool_result(name, result)

    # The input_required arm stays first; a claimed shape is terminal for the multi-round-trip driver.
    if isinstance(result, types.InputRequiredResult) and not allow_input_required:
        raise _input_required_unexpected("call_tool")
    if not isinstance(result, types.CallToolResult | types.InputRequiredResult) and not allow_claimed:
        raise UnexpectedClaimedResult(result)
    return result

validate_tool_result async

validate_tool_result(
    name: str, result: CallToolResult
) -> None

Revalidate a CallToolResult against the tool's declared output schema.

Raises:

Type	Description
RuntimeError	Structured content is missing or does not conform to the schema.

Source code in src/mcp/client/session.py

async def validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
    """Revalidate a `CallToolResult` against the tool's declared output schema.

    Raises:
        RuntimeError: Structured content is missing or does not conform to the schema.
    """
    if name not in self._tool_output_schemas:
        # refresh output schema cache
        await self.list_tools()

    output_schema = None
    if name in self._tool_output_schemas:
        output_schema = self._tool_output_schemas.get(name)
    else:
        logger.warning(f"Tool {name} not listed by server, cannot validate any structured content")

    if output_schema is not None:
        from jsonschema import SchemaError, ValidationError, validate

        if result.structured_content is None:
            raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
        try:
            validate(result.structured_content, output_schema)
        except ValidationError as e:
            raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
        except SchemaError as e:  # pragma: no cover
            raise RuntimeError(f"Invalid schema for tool {name}: {e}")  # pragma: no cover

list_prompts async

list_prompts(
    *, params: PaginatedRequestParams | None = None
) -> ListPromptsResult

Send a prompts/list request.

Parameters:

Name	Type	Description	Default
params	PaginatedRequestParams | None	Full pagination parameters including cursor and any future fields	None

Source code in src/mcp/client/session.py

async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
    """Send a prompts/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    return await self.send_request(types.ListPromptsRequest(params=params), types.ListPromptsResult)

get_prompt async

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: Literal[False] = False
) -> GetPromptResult

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool
) -> GetPromptResult | InputRequiredResult

get_prompt(
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False
) -> GetPromptResult | InputRequiredResult

Send a prompts/get request.

Parameters:

Name	Type	Description	Default
input_responses	InputResponses | None	Responses to a prior InputRequiredResult.input_requests.	None
request_state	str | None	Opaque state echoed from a prior InputRequiredResult.	None
allow_input_required	bool	When False (default), an InputRequiredResult from the server raises RuntimeError; when True, it is returned so the caller can resolve the requests and retry.	False

Raises:

Type	Description
RuntimeError	If the server returns an InputRequiredResult and allow_input_required is False.

Source code in src/mcp/client/session.py

async def get_prompt(
    self,
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    input_responses: types.InputResponses | None = None,
    request_state: str | None = None,
    meta: RequestParamsMeta | None = None,
    allow_input_required: bool = False,
) -> types.GetPromptResult | types.InputRequiredResult:
    """Send a prompts/get request.

    Args:
        input_responses: Responses to a prior `InputRequiredResult.input_requests`.
        request_state: Opaque state echoed from a prior `InputRequiredResult`.
        allow_input_required: When `False` (default), an `InputRequiredResult`
            from the server raises `RuntimeError`; when `True`, it is returned
            so the caller can resolve the requests and retry.

    Raises:
        RuntimeError: If the server returns an `InputRequiredResult` and
            `allow_input_required` is `False`.
    """
    result = await self.send_request(
        types.GetPromptRequest(
            params=types.GetPromptRequestParams(
                name=name,
                arguments=arguments,
                input_responses=input_responses,
                request_state=request_state,
                _meta=meta,
            ),
        ),
        _GetPromptResultAdapter,
    )
    if isinstance(result, types.InputRequiredResult) and not allow_input_required:
        raise _input_required_unexpected("get_prompt")
    return result

complete async

complete(
    ref: ResourceTemplateReference | PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> CompleteResult

Send a completion/complete request.

Source code in src/mcp/client/session.py

async def complete(
    self,
    ref: types.ResourceTemplateReference | types.PromptReference,
    argument: dict[str, str],
    context_arguments: dict[str, str] | None = None,
) -> types.CompleteResult:
    """Send a completion/complete request."""
    context = None
    if context_arguments is not None:
        context = types.CompletionContext(arguments=context_arguments)

    return await self.send_request(
        types.CompleteRequest(
            params=types.CompleteRequestParams(
                ref=ref,
                argument=types.CompletionArgument(**argument),
                context=context,
            ),
        ),
        types.CompleteResult,
    )

list_tools async

list_tools(
    *, params: PaginatedRequestParams | None = None
) -> ListToolsResult

Send a tools/list request.

Parameters:

Name	Type	Description	Default
params	PaginatedRequestParams | None	Full pagination parameters including cursor and any future fields	None

Source code in src/mcp/client/session.py

async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult:
    """Send a tools/list request.

    Args:
        params: Full pagination parameters including cursor and any future fields
    """
    result = await self.send_request(
        types.ListToolsRequest(params=params),
        types.ListToolsResult,
    )
    complete = (params is None or params.cursor is None) and result.next_cursor is None
    return self._absorb_tool_listing(result, complete=complete)

send_roots_list_changed async

send_roots_list_changed() -> None

Send a roots/list_changed notification.

Source code in src/mcp/client/session.py

@deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def send_roots_list_changed(self) -> None:
    """Send a roots/list_changed notification."""
    await self.send_notification(types.RootsListChangedNotification())

dispatch_input_request async

dispatch_input_request(
    ctx: ClientRequestContext, request: InputRequest
) -> InputResponse | ErrorData

Route an input request through the client's callback table.

Shared by the legacy server→client RPC path (_on_request) and the 2026-07-28 multi-round-trip driver, which dispatches the embedded InputRequiredResult.input_requests through the same callbacks.

Returns the callback's InputResponse, or ErrorData when the callback declines.

Source code in src/mcp/client/session.py

async def dispatch_input_request(
    self, ctx: ClientRequestContext, request: types.InputRequest
) -> types.InputResponse | types.ErrorData:
    """Route an input request through the client's callback table.

    Shared by the legacy server→client RPC path (`_on_request`) and the
    2026-07-28 multi-round-trip driver, which dispatches the embedded
    `InputRequiredResult.input_requests` through the same callbacks.

    Returns the callback's `InputResponse`, or `ErrorData` when the callback declines.
    """
    match request:
        case types.CreateMessageRequest(params=p):
            return await self._sampling_callback(ctx, p)
        case types.ElicitRequest(params=p):
            return await self._elicitation_callback(ctx, p)
        case types.ListRootsRequest():  # pragma: no branch
            return await self._list_roots_callback(ctx)