client

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

    Pass a URL string (Streamable HTTP), a `StdioServerParameters` (launch the command as a
    subprocess and talk over its stdin/stdout), any `Transport`, or - in tests - a `Server` or
    `MCPServer` instance to connect to it in-process.

    Example:
        ```python
        import asyncio

        from mcp import Client

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

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

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

    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 `StdioServerParameters`, the command is launched with `stdio_client`.
    If the server is a `Transport` instance, it will be used directly.
    If the server is a `Server` or `MCPServer` instance, it will be connected in-process.
    """

    _: 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."""

    sampling_capabilities: types.SamplingCapability | None = None
    """Sampling sub-capabilities (e.g. tools) declared alongside `sampling_callback`; no effect without it."""

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

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

    log_level: LoggingLevel | None = None
    """The log level to opt in to on 2026-07-28+ connections (deprecated logging feature, SEP-2577).

    Modern (2026-07-28+) servers send `notifications/message` only for requests that opt in by
    carrying `io.modelcontextprotocol/logLevel` in `_meta`, and only at or above that level. Setting
    this stamps that opt-in on every request; `None` (the default) means no opt-in, so no log
    messages arrive - a `logging_callback` alone is not an opt-in. No effect on handshake-era
    connections, where the deprecated `logging/setLevel` request governs delivery instead. A
    per-request `_meta` entry with the same key overrides this default."""

    # 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 | None = field(default_factory=CacheConfig)
    """Client-side response caching for the SEP-2549 cacheable methods (2026-07-28).

    The default `CacheConfig()` honors server `ttlMs`/`cacheScope` hints with a
    per-client in-memory store; pass a customized `CacheConfig`, or `None` 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)
    _registered_extensions: tuple[ClientExtension, ...] = 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)
        self._registered_extensions = tuple(self.extensions or ())

        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))
        elif isinstance(srv, StdioServerParameters):
            self._connect = _connect_transport(stdio_client(srv))
        else:
            self._connect = _connect_transport(srv)

        if self.cache is not None:
            config = self.cache
            # 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,
            sampling_capabilities=self.sampling_capabilities,
            list_roots_callback=self.list_roots_callback,
            logging_callback=self.logging_callback,
            log_level=self.log_level,
            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_capabilities are populated (server_info
            # stays optional: 2026-era servers may not identify themselves). 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

    def extension(self, extension_type: type[ClientExtensionBinding[_BoundT]]) -> _BoundT:
        """Return the typed API of a registered extension for this connection.

        Raises:
            RuntimeError: If the client is not connected.
            ValueError: If no extension of this type was registered.
        """
        session = self.session
        for extension in self._registered_extensions:
            if isinstance(extension, extension_type):
                return cast(ClientExtensionBinding[_BoundT], extension).bind(session)
        raise ValueError(f"client extension {extension_type.__name__!r} is not registered")

    # TODO(maxisbey): the by-construction shape is for __aenter__ to return a connected-view
    # type whose protocol_version/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.
    # (server_info stays Optional even connected: the 2026-era stamp is optional.)
    @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 | None:
        """Server name/version, or `None` when the server did not identify itself.

        Legacy connections always carry it (`InitializeResult.serverInfo` is
        required); on 2026-era connections the `_meta` `serverInfo` stamp is
        optional, so an anonymous server reads as `None`.
        """
        return 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)

    def listen(
        self,
        *,
        tools_list_changed: bool = False,
        prompts_list_changed: bool = False,
        resources_list_changed: bool = False,
        resource_subscriptions: Sequence[str] = (),
    ) -> AbstractAsyncContextManager[Subscription]:
        """Open a `subscriptions/listen` stream of typed change events (2026-07-28 only).

        Keyword args mirror the wire `SubscriptionFilter`; entering waits for the ack (honored subset: `sub.honored`):

            async with client.listen(tools_list_changed=True) as sub:
                async for event in sub:
                    tools = await client.list_tools()  # refetch on change

        A graceful close ends the loop; an abrupt drop raises `SubscriptionLost`. No replay: re-listen and refetch.

        Raises:
            ListenNotSupportedError: The negotiated protocol version predates 2026-07-28.
            MCPError: The server rejected the request or the connection failed first.
            SubscriptionLost: The stream ended before it was acknowledged.
            TimeoutError: The read timeout elapsed before the acknowledgment.
        """
        return _listen(
            self.session,
            tools_list_changed=tools_list_changed,
            prompts_list_changed=prompts_list_changed,
            resources_list_changed=resources_list_changed,
            resource_subscriptions=resource_subscriptions,
            on_event=self._evict_for_listen_event if self._response_cache is not None else None,
        )

    async def _evict_for_listen_event(self, event: ServerEvent) -> None:
        """Finish response-cache eviction before a listen consumer can refetch.

        Without it the iterator wakes first and refetches a still-warm entry, with no
        corrective wake (events are deduplicated level triggers). The tee path repeats
        the eviction; deliberate: idempotent, and it covers non-iterating consumers.
        """
        cache = self._response_cache
        assert cache is not None  # installed as the event barrier only when a cache exists
        try:
            await cache.evict_for_notification(event_to_notification(event, {}))
        except Exception:  # boundary: eviction reaches user store code; a cache fault must not block delivery
            logger.exception("Response cache eviction failed; the event is still delivered")

    @deprecated(
        "resources/subscribe is removed as of 2026-07-28; use Client.listen() instead.",
        category=MCPDeprecationWarning,
    )
    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Subscribe to resource updates (2025-era servers only)."""
        return await self.session.subscribe_resource(uri, meta=meta)  # pyright: ignore[reportDeprecated]

    @deprecated(
        "resources/unsubscribe is removed as of 2026-07-28; use Client.listen() instead.",
        category=MCPDeprecationWarning,
    )
    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
        """Unsubscribe from resource updates (2025-era servers only)."""
        return await self.session.unsubscribe_resource(uri, meta=meta)  # pyright: ignore[reportDeprecated]

    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)`.

        On a 2026-07-28 connection, a call the server rejects with `HEADER_MISMATCH`
        (this client has not listed the tool, or its input schema changed since) is
        resent once after refetching the tool listing. A second rejection is raised,
        and so is the first when the listing cannot be refetched.

        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, and
                for the whole re-list after a `HEADER_MISMATCH`. Defaults to this
                client's `read_timeout_seconds`.
            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 send(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,
            )

        async def retry(r: InputResponses | None, s: str | None) -> CallToolResult | InputRequiredResult | Result:
            try:
                return await send(r, s)
            except MCPError as mismatch:
                if mismatch.code != HEADER_MISMATCH or self.protocol_version not in MODERN_PROTOCOL_VERSIONS:
                    raise
                # The spec's recovery: the tool's listed schema is missing or stale, so re-list and resend once.
                timeout = read_timeout_seconds if read_timeout_seconds is not None else self.read_timeout_seconds
                try:
                    with anyio.fail_after(timeout):
                        await self._relist_tool(name)
                except (MCPError, TimeoutError, ValidationError) as relist_error:
                    raise mismatch from relist_error
                return await send(r, s)

        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
            ),
        )

    async def _relist_tool(self, name: str) -> None:
        """Refetch the tool listing from the server, page by page, until a page lists `name`."""
        cursor: str | None = None
        for _ in range(_RELIST_PAGE_CAP):
            page = await self.list_tools(cursor=cursor, cache_mode="refresh")
            cursor = page.next_cursor
            if cursor is None or any(tool.name == name for tool in page.tools):
                return

    @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."""
        await self.session.send_roots_list_changed()  # pyright: ignore[reportDeprecated]