Index

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

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]

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)

ClientSessionGroup

Client for managing connections to multiple MCP servers.

This class is responsible for encapsulating management of server connections. It aggregates tools, resources, and prompts from all connected servers.

For auxiliary handlers, such as resource subscription, this is delegated to the client and can be accessed via the session.

Example

name_fn = lambda name, server_info: f"{(server_info.name)}_{name}"
async with ClientSessionGroup(component_name_hook=name_fn) as group:
    for server_param in server_params:
        await group.connect_to_server(server_param)
    ...

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

class ClientSessionGroup:
    """Client for managing connections to multiple MCP servers.

    This class is responsible for encapsulating management of server connections.
    It aggregates tools, resources, and prompts from all connected servers.

    For auxiliary handlers, such as resource subscription, this is delegated to
    the client and can be accessed via the session.

    Example:
        ```python
        name_fn = lambda name, server_info: f"{(server_info.name)}_{name}"
        async with ClientSessionGroup(component_name_hook=name_fn) as group:
            for server_param in server_params:
                await group.connect_to_server(server_param)
            ...
        ```
    """

    class _ComponentNames(BaseModel):
        """Used for reverse index to find components."""

        prompts: set[str] = Field(default_factory=set)
        resources: set[str] = Field(default_factory=set)
        tools: set[str] = Field(default_factory=set)

    # Standard MCP components.
    _prompts: dict[str, types.Prompt]
    _resources: dict[str, types.Resource]
    _tools: dict[str, types.Tool]

    # Client-server connection management.
    _sessions: dict[mcp.ClientSession, _ComponentNames]
    _tool_to_session: dict[str, mcp.ClientSession]
    _exit_stack: contextlib.AsyncExitStack
    _session_exit_stacks: dict[mcp.ClientSession, contextlib.AsyncExitStack]

    # Optional fn consuming (component_name, server_info) for custom names.
    # This is to provide a means to mitigate naming conflicts across servers.
    # Example: (tool_name, server_info) => "{result.server_info.name}.{tool_name}"
    _ComponentNameHook: TypeAlias = Callable[[str, types.Implementation], str]
    _component_name_hook: _ComponentNameHook | None

    def __init__(
        self,
        exit_stack: contextlib.AsyncExitStack | None = None,
        component_name_hook: _ComponentNameHook | None = None,
    ) -> None:
        """Initializes the MCP client."""

        self._tools = {}
        self._resources = {}
        self._prompts = {}

        self._sessions = {}
        self._tool_to_session = {}
        if exit_stack is None:
            self._exit_stack = contextlib.AsyncExitStack()
            self._owns_exit_stack = True
        else:
            self._exit_stack = exit_stack
            self._owns_exit_stack = False
        self._session_exit_stacks = {}
        self._component_name_hook = component_name_hook

    async def __aenter__(self) -> Self:  # pragma: no cover
        # Enter the exit stack only if we created it ourselves
        if self._owns_exit_stack:
            await self._exit_stack.__aenter__()
        return self

    async def __aexit__(
        self,
        _exc_type: type[BaseException] | None,
        _exc_val: BaseException | None,
        _exc_tb: TracebackType | None,
    ) -> bool | None:  # pragma: no cover
        """Closes session exit stacks and main exit stack upon completion."""

        # Only close the main exit stack if we created it
        if self._owns_exit_stack:
            await self._exit_stack.aclose()

        # Concurrently close session stacks.
        async with anyio.create_task_group() as tg:
            for exit_stack in self._session_exit_stacks.values():
                tg.start_soon(exit_stack.aclose)

    @property
    def sessions(self) -> list[mcp.ClientSession]:
        """Returns the list of sessions being managed."""
        return list(self._sessions.keys())  # pragma: no cover

    @property
    def prompts(self) -> dict[str, types.Prompt]:
        """Returns the prompts as a dictionary of names to prompts."""
        return self._prompts

    @property
    def resources(self) -> dict[str, types.Resource]:
        """Returns the resources as a dictionary of names to resources."""
        return self._resources

    @property
    def tools(self) -> dict[str, types.Tool]:
        """Returns the tools as a dictionary of names to tools."""
        return self._tools

    @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: types.RequestParamsMeta | None = None,
        allow_input_required: 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: types.RequestParamsMeta | None = None,
        allow_input_required: bool,
    ) -> types.CallToolResult | types.InputRequiredResult: ...

    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: types.RequestParamsMeta | None = None,
        allow_input_required: bool = False,
    ) -> types.CallToolResult | types.InputRequiredResult:
        """Executes a tool given its name and arguments.

        Raises:
            RuntimeError: If the server returns an `InputRequiredResult` and
                ``allow_input_required`` is ``False``.
        """
        session = self._tool_to_session[name]
        session_tool_name = self.tools[name].name
        return await session.call_tool(
            session_tool_name,
            arguments=arguments,
            read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
            input_responses=input_responses,
            request_state=request_state,
            meta=meta,
            allow_input_required=allow_input_required,
        )

    async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
        """Disconnects from a single MCP server."""

        session_known_for_components = session in self._sessions
        session_known_for_stack = session in self._session_exit_stacks

        if not session_known_for_components and not session_known_for_stack:
            raise MCPError(
                code=types.INVALID_PARAMS,
                message="Provided session is not managed or already disconnected.",
            )

        if session_known_for_components:  # pragma: no branch
            component_names = self._sessions.pop(session)  # Pop from _sessions tracking

            # Remove prompts associated with the session.
            for name in component_names.prompts:
                if name in self._prompts:  # pragma: no branch
                    del self._prompts[name]
            # Remove resources associated with the session.
            for name in component_names.resources:
                if name in self._resources:  # pragma: no branch
                    del self._resources[name]
            # Remove tools associated with the session.
            for name in component_names.tools:
                if name in self._tools:  # pragma: no branch
                    del self._tools[name]
                if name in self._tool_to_session:  # pragma: no branch
                    del self._tool_to_session[name]

        # Clean up the session's resources via its dedicated exit stack
        if session_known_for_stack:
            session_stack_to_close = self._session_exit_stacks.pop(session)  # pragma: no cover
            await session_stack_to_close.aclose()  # pragma: no cover

    async def connect_with_session(
        self, server_info: types.Implementation, session: mcp.ClientSession
    ) -> mcp.ClientSession:
        """Connects to a single MCP server."""
        await self._aggregate_components(server_info, session)
        return session

    async def connect_to_server(
        self,
        server_params: ServerParameters,
        session_params: ClientSessionParameters | None = None,
    ) -> mcp.ClientSession:
        """Connects to a single MCP server."""
        server_info, session = await self._establish_session(server_params, session_params or ClientSessionParameters())
        return await self.connect_with_session(server_info, session)

    async def _establish_session(
        self,
        server_params: ServerParameters,
        session_params: ClientSessionParameters,
    ) -> tuple[types.Implementation, mcp.ClientSession]:
        """Establish a client session to an MCP server."""

        session_stack = contextlib.AsyncExitStack()
        try:
            # Create read and write streams that facilitate io with the server.
            if isinstance(server_params, StdioServerParameters):
                client = mcp.stdio_client(server_params)
                read, write = await session_stack.enter_async_context(client)
            elif isinstance(server_params, SseServerParameters):
                client = sse_client(
                    url=server_params.url,
                    headers=server_params.headers,
                    timeout=server_params.timeout,
                    sse_read_timeout=server_params.sse_read_timeout,
                )
                read, write = await session_stack.enter_async_context(client)
            else:
                httpx_client = create_mcp_http_client(
                    headers=server_params.headers,
                    timeout=httpx.Timeout(
                        server_params.timeout,
                        read=server_params.sse_read_timeout,
                    ),
                )
                await session_stack.enter_async_context(httpx_client)

                client = streamable_http_client(
                    url=server_params.url,
                    http_client=httpx_client,
                    terminate_on_close=server_params.terminate_on_close,
                )
                read, write = await session_stack.enter_async_context(client)

            session = await session_stack.enter_async_context(
                mcp.ClientSession(
                    read,
                    write,
                    read_timeout_seconds=session_params.read_timeout_seconds,
                    sampling_callback=session_params.sampling_callback,
                    elicitation_callback=session_params.elicitation_callback,
                    list_roots_callback=session_params.list_roots_callback,
                    logging_callback=session_params.logging_callback,
                    message_handler=session_params.message_handler,
                    client_info=session_params.client_info,
                )
            )

            result = await session.initialize()

            # Session successfully initialized.
            # Store its stack and register the stack with the main group stack.
            self._session_exit_stacks[session] = session_stack
            # session_stack itself becomes a resource managed by the
            # main _exit_stack.
            await self._exit_stack.enter_async_context(session_stack)

            return result.server_info, session
        except Exception:  # pragma: no cover
            # If anything during this setup fails, ensure the session-specific
            # stack is closed.
            await session_stack.aclose()
            raise

    async def _aggregate_components(self, server_info: types.Implementation, session: mcp.ClientSession) -> None:
        """Aggregates prompts, resources, and tools from a given session."""

        # Create a reverse index so we can find all prompts, resources, and
        # tools belonging to this session. Used for removing components from
        # the session group via self.disconnect_from_server.
        component_names = self._ComponentNames()

        # Temporary components dicts. We do not want to modify the aggregate
        # lists in case of an intermediate failure.
        prompts_temp: dict[str, types.Prompt] = {}
        resources_temp: dict[str, types.Resource] = {}
        tools_temp: dict[str, types.Tool] = {}
        tool_to_session_temp: dict[str, mcp.ClientSession] = {}

        # Query the server for its prompts and aggregate to list.
        try:
            prompts = (await session.list_prompts()).prompts
            for prompt in prompts:
                name = self._component_name(prompt.name, server_info)
                prompts_temp[name] = prompt
                component_names.prompts.add(name)
        except MCPError as err:  # pragma: no cover
            logging.warning(f"Could not fetch prompts: {err}")

        # Query the server for its resources and aggregate to list.
        try:
            resources = (await session.list_resources()).resources
            for resource in resources:
                name = self._component_name(resource.name, server_info)
                resources_temp[name] = resource
                component_names.resources.add(name)
        except MCPError as err:  # pragma: no cover
            logging.warning(f"Could not fetch resources: {err}")

        # Query the server for its tools and aggregate to list.
        try:
            tools = (await session.list_tools()).tools
            for tool in tools:
                name = self._component_name(tool.name, server_info)
                tools_temp[name] = tool
                tool_to_session_temp[name] = session
                component_names.tools.add(name)
        except MCPError as err:  # pragma: no cover
            logging.warning(f"Could not fetch tools: {err}")

        # Clean up exit stack for session if we couldn't retrieve anything
        # from the server.
        if not any((prompts_temp, resources_temp, tools_temp)):
            del self._session_exit_stacks[session]  # pragma: no cover

        # Check for duplicates.
        matching_prompts = prompts_temp.keys() & self._prompts.keys()
        if matching_prompts:
            raise MCPError(  # pragma: no cover
                code=types.INVALID_PARAMS,
                message=f"{matching_prompts} already exist in group prompts.",
            )
        matching_resources = resources_temp.keys() & self._resources.keys()
        if matching_resources:
            raise MCPError(  # pragma: no cover
                code=types.INVALID_PARAMS,
                message=f"{matching_resources} already exist in group resources.",
            )
        matching_tools = tools_temp.keys() & self._tools.keys()
        if matching_tools:
            raise MCPError(code=types.INVALID_PARAMS, message=f"{matching_tools} already exist in group tools.")

        # Aggregate components.
        self._sessions[session] = component_names
        self._prompts.update(prompts_temp)
        self._resources.update(resources_temp)
        self._tools.update(tools_temp)
        self._tool_to_session.update(tool_to_session_temp)

    def _component_name(self, name: str, server_info: types.Implementation) -> str:
        if self._component_name_hook:
            return self._component_name_hook(name, server_info)
        return name

__init__

__init__(
    exit_stack: AsyncExitStack | None = None,
    component_name_hook: _ComponentNameHook | None = None,
) -> None

Initializes the MCP client.

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

def __init__(
    self,
    exit_stack: contextlib.AsyncExitStack | None = None,
    component_name_hook: _ComponentNameHook | None = None,
) -> None:
    """Initializes the MCP client."""

    self._tools = {}
    self._resources = {}
    self._prompts = {}

    self._sessions = {}
    self._tool_to_session = {}
    if exit_stack is None:
        self._exit_stack = contextlib.AsyncExitStack()
        self._owns_exit_stack = True
    else:
        self._exit_stack = exit_stack
        self._owns_exit_stack = False
    self._session_exit_stacks = {}
    self._component_name_hook = component_name_hook

__aexit__ async

__aexit__(
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: TracebackType | None,
) -> bool | None

Closes session exit stacks and main exit stack upon completion.

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

async def __aexit__(
    self,
    _exc_type: type[BaseException] | None,
    _exc_val: BaseException | None,
    _exc_tb: TracebackType | None,
) -> bool | None:  # pragma: no cover
    """Closes session exit stacks and main exit stack upon completion."""

    # Only close the main exit stack if we created it
    if self._owns_exit_stack:
        await self._exit_stack.aclose()

    # Concurrently close session stacks.
    async with anyio.create_task_group() as tg:
        for exit_stack in self._session_exit_stacks.values():
            tg.start_soon(exit_stack.aclose)

sessions property

sessions: list[ClientSession]

Returns the list of sessions being managed.

prompts property

prompts: dict[str, Prompt]

Returns the prompts as a dictionary of names to prompts.

resources property

resources: dict[str, Resource]

Returns the resources as a dictionary of names to resources.

tools property

tools: dict[str, Tool]

Returns the tools as a dictionary of names to tools.

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
) -> 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
) -> 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: bool = False
) -> CallToolResult | InputRequiredResult

Executes a tool given its name and arguments.

Raises:

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

Source code in src/mcp/client/session_group.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: types.RequestParamsMeta | None = None,
    allow_input_required: bool = False,
) -> types.CallToolResult | types.InputRequiredResult:
    """Executes a tool given its name and arguments.

    Raises:
        RuntimeError: If the server returns an `InputRequiredResult` and
            ``allow_input_required`` is ``False``.
    """
    session = self._tool_to_session[name]
    session_tool_name = self.tools[name].name
    return await session.call_tool(
        session_tool_name,
        arguments=arguments,
        read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
        input_responses=input_responses,
        request_state=request_state,
        meta=meta,
        allow_input_required=allow_input_required,
    )

disconnect_from_server async

disconnect_from_server(session: ClientSession) -> None

Disconnects from a single MCP server.

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

async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
    """Disconnects from a single MCP server."""

    session_known_for_components = session in self._sessions
    session_known_for_stack = session in self._session_exit_stacks

    if not session_known_for_components and not session_known_for_stack:
        raise MCPError(
            code=types.INVALID_PARAMS,
            message="Provided session is not managed or already disconnected.",
        )

    if session_known_for_components:  # pragma: no branch
        component_names = self._sessions.pop(session)  # Pop from _sessions tracking

        # Remove prompts associated with the session.
        for name in component_names.prompts:
            if name in self._prompts:  # pragma: no branch
                del self._prompts[name]
        # Remove resources associated with the session.
        for name in component_names.resources:
            if name in self._resources:  # pragma: no branch
                del self._resources[name]
        # Remove tools associated with the session.
        for name in component_names.tools:
            if name in self._tools:  # pragma: no branch
                del self._tools[name]
            if name in self._tool_to_session:  # pragma: no branch
                del self._tool_to_session[name]

    # Clean up the session's resources via its dedicated exit stack
    if session_known_for_stack:
        session_stack_to_close = self._session_exit_stacks.pop(session)  # pragma: no cover
        await session_stack_to_close.aclose()  # pragma: no cover

connect_with_session async

connect_with_session(
    server_info: Implementation, session: ClientSession
) -> ClientSession

Connects to a single MCP server.

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

async def connect_with_session(
    self, server_info: types.Implementation, session: mcp.ClientSession
) -> mcp.ClientSession:
    """Connects to a single MCP server."""
    await self._aggregate_components(server_info, session)
    return session

connect_to_server async

connect_to_server(
    server_params: ServerParameters,
    session_params: ClientSessionParameters | None = None,
) -> ClientSession

Connects to a single MCP server.

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

async def connect_to_server(
    self,
    server_params: ServerParameters,
    session_params: ClientSessionParameters | None = None,
) -> mcp.ClientSession:
    """Connects to a single MCP server."""
    server_info, session = await self._establish_session(server_params, session_params or ClientSessionParameters())
    return await self.connect_with_session(server_info, session)

StdioServerParameters

Bases: BaseModel

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

class StdioServerParameters(BaseModel):
    command: str
    """The executable to run to start the server."""

    args: list[str] = Field(default_factory=list)
    """Command line arguments to pass to the executable."""

    env: dict[str, str] | None = None
    """Extra environment variables, merged over get_default_environment()."""

    cwd: str | Path | None = None
    """The working directory to use when spawning the process."""

    encoding: str = "utf-8"
    """Text encoding for messages to and from the server."""

    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
    """Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers."""

command instance-attribute

command: str

The executable to run to start the server.

args class-attribute instance-attribute

args: list[str] = Field(default_factory=list)

Command line arguments to pass to the executable.

env class-attribute instance-attribute

env: dict[str, str] | None = None

Extra environment variables, merged over get_default_environment().

cwd class-attribute instance-attribute

cwd: str | Path | None = None

The working directory to use when spawning the process.

encoding class-attribute instance-attribute

encoding: str = 'utf-8'

Text encoding for messages to and from the server.

encoding_error_handler class-attribute instance-attribute

encoding_error_handler: Literal[
    "strict", "ignore", "replace"
] = "strict"

Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers.

stdio_client async

stdio_client(
    server: StdioServerParameters, errlog: TextIO = stderr
) -> AsyncGenerator[TransportStreams, None]

Spawns an MCP server subprocess and connects to it over stdin/stdout.

Raises:

Type	Description
OSError	If the server process cannot be spawned.
ValueError	If the spawn parameters are invalid (embedded NUL bytes).

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

@asynccontextmanager
async def stdio_client(
    server: StdioServerParameters, errlog: TextIO = sys.stderr
) -> AsyncGenerator[TransportStreams, None]:
    """Spawns an MCP server subprocess and connects to it over stdin/stdout.

    Raises:
        OSError: If the server process cannot be spawned.
        ValueError: If the spawn parameters are invalid (embedded NUL bytes).
    """
    command = _get_executable_command(server.command)

    process = await _create_platform_compatible_process(
        command=command,
        args=server.args,
        env=get_default_environment() | (server.env or {}),
        errlog=errlog,
        cwd=server.cwd,
    )

    # The spawn succeeded; no awaits until the task group is entered, or a
    # cancellation delivered in the gap would leak the live process.
    read_stream_writer, read_stream = anyio.create_memory_object_stream[SessionMessage | Exception](0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)

    shutting_down = False
    writer_done = anyio.Event()

    async def stdout_reader() -> None:
        assert process.stdout, "Opened process is missing stdout"

        stdout = TextReceiveStream(process.stdout, encoding=server.encoding, errors=server.encoding_error_handler)
        try:
            async with read_stream_writer:
                try:
                    # One line at a time; no read-ahead while a delivery is blocked.
                    buffer = ""
                    async for chunk in stdout:
                        lines = (buffer + chunk).split("\n")
                        buffer = lines.pop()
                        for line in lines:
                            try:
                                await read_stream_writer.send(_parse_line(line))
                            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
                                return  # the session is gone; only the drain below remains
                finally:
                    await _drain_stdout(process)
        except anyio.ClosedResourceError:
            pass  # our own shutdown closed the stdout stream under the read
        except (anyio.BrokenResourceError, ConnectionError):
            # Teardown noise during shutdown, a real failure otherwise; either way
            # the session sees clean closure when the read stream closes.
            if not shutting_down:
                logger.exception("Reading from the MCP server's stdout failed mid-session")

    async def stdin_writer() -> None:
        assert process.stdin, "Opened process is missing stdin"

        try:
            async with write_stream_reader:
                async for session_message in write_stream_reader:
                    json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
                    data = (json + "\n").encode(encoding=server.encoding, errors=server.encoding_error_handler)
                    await process.stdin.send(data)
        except (anyio.ClosedResourceError, anyio.BrokenResourceError, OSError):
            # The server may still be alive: close the read stream so the session
            # sees the connection end instead of a request hanging forever.
            await read_stream_writer.aclose()
        finally:
            writer_done.set()

    async def shutdown() -> None:
        """Winds the transport down: stop traffic, flush, stop the server, release the streams."""
        # Unblock the reader into its drain: a server stuck writing stdout cannot
        # read its stdin, so draining is what lets the flush below complete.
        read_stream.close()
        # Bounded window for the writer to flush already-accepted messages.
        write_stream.close()
        with anyio.move_on_after(_WRITER_FLUSH_TIMEOUT) as flush_scope:
            await writer_done.wait()
        if flush_scope.cancelled_caught:
            await anyio.lowlevel.cancel_shielded_checkpoint()  # resync coverage on 3.11 (gh-106749)
        await _stop_server_process(process)
        await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader)
        # One pass so unblocked tasks exit via their except paths before the cancel.
        await anyio.lowlevel.checkpoint()

    async with anyio.create_task_group() as tg:
        tg.start_soon(stdout_reader)
        tg.start_soon(stdin_writer)
        try:
            yield read_stream, write_stream
        finally:
            shutting_down = True
            # Shutdown must finish even under caller cancellation, or the server
            # process would leak; every wait inside is bounded. (Native
            # task.cancel() and the fallback's worker threads can still defeat it.)
            with anyio.CancelScope(shield=True):
                await shutdown()
            # Unstick pipe tasks a kill survivor's open pipe end could still block.
            tg.cancel_scope.cancel()
    # The cancel lands via throw(); one yield resyncs 3.11 coverage (gh-106749).
    await anyio.lowlevel.cancel_shielded_checkpoint()

ServerSession

Per-request proxy for server-to-client requests and notifications.

Built once per inbound request by the kernel's _make_context. Holds two Outbound channels: the request-scoped one (the per-request DispatchContext, which on streamable HTTP routes onto the originating POST's response stream) and the connection's standalone channel (connection.outbound). related_request_id on the public methods is the selector — present means request-scoped, absent means standalone — and never crosses the Outbound Protocol.

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

class ServerSession:
    """Per-request proxy for server-to-client requests and notifications.

    Built once per inbound request by the kernel's `_make_context`. Holds two
    `Outbound` channels: the request-scoped one (the per-request
    `DispatchContext`, which on streamable HTTP routes onto the originating
    POST's response stream) and the connection's standalone channel
    (`connection.outbound`). `related_request_id` on the public methods is the
    selector — present means request-scoped, absent means standalone — and
    never crosses the `Outbound` Protocol.
    """

    def __init__(self, request_outbound: DispatchContext[Any], connection: Connection) -> None:
        self._request_outbound = request_outbound
        self._connection = connection

    @property
    def client_params(self) -> types.InitializeRequestParams | None:
        """The client's `initialize` request params; `None` when no client info was supplied."""
        return self._connection.client_params

    @property
    def protocol_version(self) -> str:
        """The protocol version this connection speaks.

        Populated at `Connection` construction and overwritten once the
        handshake commits on the loop path; never `None`.
        """
        return self._connection.protocol_version

    async def send_request(
        self,
        request: types.ServerRequest,
        result_type: type[ResultT],
        request_read_timeout_seconds: float | None = None,
        metadata: ServerMessageMetadata | None = None,
        progress_callback: ProgressFnT | None = None,
    ) -> ResultT:
        """Send a typed server-to-client request and validate the result.

        Raises:
            MCPError: The peer responded with an error.
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests (raised by the held `Outbound`).
            pydantic.ValidationError: The peer's result does not match `result_type`.
        """
        related = metadata.related_request_id if metadata is not None else None
        channel = self._request_outbound if related is not None else self._connection.outbound
        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
        opts: CallOptions = {}
        if request_read_timeout_seconds is not None:
            opts["timeout"] = request_read_timeout_seconds
        if progress_callback is not None:
            opts["on_progress"] = progress_callback
        result = await channel.send_raw_request(data["method"], data.get("params"), opts or None)
        try:
            _methods.validate_client_result(request.method, self.protocol_version, result)
        except KeyError:
            pass
        return result_type.model_validate(result, by_name=False)

    async def send_notification(
        self,
        notification: types.ServerNotification,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send a typed server-to-client notification."""
        channel = self._request_outbound if related_request_id is not None else self._connection.outbound
        data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
        await channel.notify(data["method"], data.get("params"))

    def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
        """Check if the client supports a specific capability."""
        return self._connection.check_capability(capability)

    @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def send_log_message(
        self,
        level: types.LoggingLevel,
        data: Any,
        logger: str | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send a log message notification."""
        await self.send_notification(
            types.LoggingMessageNotification(
                params=types.LoggingMessageNotificationParams(
                    level=level,
                    data=data,
                    logger=logger,
                ),
            ),
            related_request_id,
        )

    async def send_resource_updated(self, uri: str | AnyUrl) -> None:
        """Send a resource updated notification."""
        await self.send_notification(
            types.ResourceUpdatedNotification(
                params=types.ResourceUpdatedNotificationParams(uri=str(uri)),
            )
        )

    @overload
    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: None = None,
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResult:
        """Overload: Without tools, returns single content."""
        ...

    @overload
    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: list[types.Tool],
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResultWithTools:
        """Overload: With tools, returns array-capable content."""
        ...

    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: list[types.Tool] | None = None,
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResult | types.CreateMessageResultWithTools:
        """Send a sampling/create_message request.

        Args:
            messages: The conversation messages to send.
            max_tokens: Maximum number of tokens to generate.
            system_prompt: Optional system prompt.
            include_context: Optional context inclusion setting.
                Should only be set to "thisServer" or "allServers"
                if the client has sampling.context capability.
            temperature: Optional sampling temperature.
            stop_sequences: Optional stop sequences.
            metadata: Optional metadata to pass through to the LLM provider.
            model_preferences: Optional model selection preferences.
            tools: Optional list of tools the LLM can use during sampling.
                Requires client to have sampling.tools capability.
            tool_choice: Optional control over tool usage behavior.
                Requires client to have sampling.tools capability.
            related_request_id: Optional ID of a related request.

        Returns:
            The sampling result from the client.

        Raises:
            MCPError: If tools are provided but client doesn't support them.
            ValueError: If tool_use or tool_result message structure is invalid.
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        client_caps = self.client_params.capabilities if self.client_params else None
        validate_sampling_tools(client_caps, tools, tool_choice)
        validate_tool_use_result_messages(messages)

        request = types.CreateMessageRequest(
            params=types.CreateMessageRequestParams(
                messages=messages,
                system_prompt=system_prompt,
                include_context=include_context,
                temperature=temperature,
                max_tokens=max_tokens,
                stop_sequences=stop_sequences,
                metadata=metadata,
                model_preferences=model_preferences,
                tools=tools,
                tool_choice=tool_choice,
            ),
        )
        metadata_obj = ServerMessageMetadata(related_request_id=related_request_id)

        if tools is not None:
            return await self.send_request(
                request=request,
                result_type=types.CreateMessageResultWithTools,
                metadata=metadata_obj,
            )
        return await self.send_request(
            request=request,
            result_type=types.CreateMessageResult,
            metadata=metadata_obj,
        )

    @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def list_roots(self) -> types.ListRootsResult:
        """Send a roots/list request.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ListRootsRequest(),
            types.ListRootsResult,
        )

    async def elicit(
        self,
        message: str,
        requested_schema: types.ElicitRequestedSchema,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a form mode elicitation/create request.

        Args:
            message: The message to present to the user.
            requested_schema: Schema defining the expected response structure.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response.

        Note:
            This method is deprecated in favor of elicit_form(). It remains for
            backward compatibility but new code should use elicit_form().
        """
        return await self.elicit_form(message, requested_schema, related_request_id)

    async def elicit_form(
        self,
        message: str,
        requested_schema: types.ElicitRequestedSchema,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a form mode elicitation/create request.

        Args:
            message: The message to present to the user.
            requested_schema: Schema defining the expected response structure.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response with form data.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ElicitRequest(
                params=types.ElicitRequestFormParams(
                    message=message,
                    requested_schema=requested_schema,
                ),
            ),
            types.ElicitResult,
            metadata=ServerMessageMetadata(related_request_id=related_request_id),
        )

    async def elicit_url(
        self,
        message: str,
        url: str,
        elicitation_id: str,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a URL mode elicitation/create request.

        This directs the user to an external URL for out-of-band interactions
        like OAuth flows, credential collection, or payment processing.

        Args:
            message: Human-readable explanation of why the interaction is needed.
            url: The URL the user should navigate to.
            elicitation_id: Unique identifier for tracking this elicitation.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response indicating acceptance, decline, or cancellation.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ElicitRequest(
                params=types.ElicitRequestURLParams(
                    message=message,
                    url=url,
                    elicitation_id=elicitation_id,
                ),
            ),
            types.ElicitResult,
            metadata=ServerMessageMetadata(related_request_id=related_request_id),
        )

    async def send_ping(self) -> types.EmptyResult:
        """Send a ping request."""
        return await self.send_request(
            types.PingRequest(),
            types.EmptyResult,
        )

    async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
        """Report progress for the inbound request this session is scoped to.

        A no-op when the caller did not request progress. Dispatcher-agnostic:
        on JSON-RPC the held `DispatchContext` emits ``notifications/progress``
        against the caller's token; on the in-process direct dispatcher it
        invokes the caller's callback directly.
        """
        await self._request_outbound.progress(progress, total, message)

    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
        related_request_id: str | 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,
                ),
            ),
            related_request_id,
        )

    async def send_resource_list_changed(self) -> None:
        """Send a resource list changed notification."""
        await self.send_notification(types.ResourceListChangedNotification())

    async def send_tool_list_changed(self) -> None:
        """Send a tool list changed notification."""
        await self.send_notification(types.ToolListChangedNotification())

    async def send_prompt_list_changed(self) -> None:
        """Send a prompt list changed notification."""
        await self.send_notification(types.PromptListChangedNotification())

    async def send_elicit_complete(
        self,
        elicitation_id: str,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send an elicitation completion notification.

        This should be sent when a URL mode elicitation has been completed
        out-of-band to inform the client that it may retry any requests
        that were waiting for this elicitation.

        Args:
            elicitation_id: The unique identifier of the completed elicitation
            related_request_id: Optional ID of the request that triggered this notification
        """
        await self.send_notification(
            types.ElicitCompleteNotification(
                params=types.ElicitCompleteNotificationParams(elicitation_id=elicitation_id)
            ),
            related_request_id,
        )

client_params property

client_params: InitializeRequestParams | None

The client's initialize request params; None when no client info was supplied.

protocol_version property

protocol_version: str

The protocol version this connection speaks.

Populated at Connection construction and overwritten once the handshake commits on the loop path; never None.

send_request async

send_request(
    request: ServerRequest,
    result_type: type[ResultT],
    request_read_timeout_seconds: float | None = None,
    metadata: ServerMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ResultT

Send a typed server-to-client request and validate the result.

Raises:

Type	Description
MCPError	The peer responded with an error.
NoBackChannelError	The connection has no back-channel for server-initiated requests (raised by the held Outbound).
ValidationError	The peer's result does not match result_type.

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

async def send_request(
    self,
    request: types.ServerRequest,
    result_type: type[ResultT],
    request_read_timeout_seconds: float | None = None,
    metadata: ServerMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ResultT:
    """Send a typed server-to-client request and validate the result.

    Raises:
        MCPError: The peer responded with an error.
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests (raised by the held `Outbound`).
        pydantic.ValidationError: The peer's result does not match `result_type`.
    """
    related = metadata.related_request_id if metadata is not None else None
    channel = self._request_outbound if related is not None else self._connection.outbound
    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
    opts: CallOptions = {}
    if request_read_timeout_seconds is not None:
        opts["timeout"] = request_read_timeout_seconds
    if progress_callback is not None:
        opts["on_progress"] = progress_callback
    result = await channel.send_raw_request(data["method"], data.get("params"), opts or None)
    try:
        _methods.validate_client_result(request.method, self.protocol_version, result)
    except KeyError:
        pass
    return result_type.model_validate(result, by_name=False)

send_notification async

send_notification(
    notification: ServerNotification,
    related_request_id: RequestId | None = None,
) -> None

Send a typed server-to-client notification.

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

async def send_notification(
    self,
    notification: types.ServerNotification,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send a typed server-to-client notification."""
    channel = self._request_outbound if related_request_id is not None else self._connection.outbound
    data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
    await channel.notify(data["method"], data.get("params"))

check_client_capability

check_client_capability(
    capability: ClientCapabilities,
) -> bool

Check if the client supports a specific capability.

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

def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
    """Check if the client supports a specific capability."""
    return self._connection.check_capability(capability)

send_log_message async

send_log_message(
    level: LoggingLevel,
    data: Any,
    logger: str | None = None,
    related_request_id: RequestId | None = None,
) -> None

Send a log message notification.

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

@deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def send_log_message(
    self,
    level: types.LoggingLevel,
    data: Any,
    logger: str | None = None,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send a log message notification."""
    await self.send_notification(
        types.LoggingMessageNotification(
            params=types.LoggingMessageNotificationParams(
                level=level,
                data=data,
                logger=logger,
            ),
        ),
        related_request_id,
    )

send_resource_updated async

send_resource_updated(uri: str | AnyUrl) -> None

Send a resource updated notification.

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

async def send_resource_updated(self, uri: str | AnyUrl) -> None:
    """Send a resource updated notification."""
    await self.send_notification(
        types.ResourceUpdatedNotification(
            params=types.ResourceUpdatedNotificationParams(uri=str(uri)),
        )
    )

create_message async

create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: None = None,
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResult

create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: list[Tool],
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResultWithTools

create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: list[Tool] | None = None,
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResult | CreateMessageResultWithTools

Send a sampling/create_message request.

Parameters:

Name	Type	Description	Default
messages	list[SamplingMessage]	The conversation messages to send.	required
max_tokens	int	Maximum number of tokens to generate.	required
system_prompt	str | None	Optional system prompt.	None
include_context	IncludeContext | None	Optional context inclusion setting. Should only be set to "thisServer" or "allServers" if the client has sampling.context capability.	None
temperature	float | None	Optional sampling temperature.	None
stop_sequences	list[str] | None	Optional stop sequences.	None
metadata	dict[str, Any] | None	Optional metadata to pass through to the LLM provider.	None
model_preferences	ModelPreferences | None	Optional model selection preferences.	None
tools	list[Tool] | None	Optional list of tools the LLM can use during sampling. Requires client to have sampling.tools capability.	None
tool_choice	ToolChoice | None	Optional control over tool usage behavior. Requires client to have sampling.tools capability.	None
related_request_id	RequestId | None	Optional ID of a related request.	None

Returns:

Type	Description
CreateMessageResult | CreateMessageResultWithTools	The sampling result from the client.

Raises:

Type	Description
MCPError	If tools are provided but client doesn't support them.
ValueError	If tool_use or tool_result message structure is invalid.
NoBackChannelError	The connection has no back-channel for server-initiated requests.

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

@deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def create_message(
    self,
    messages: list[types.SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: types.IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: types.ModelPreferences | None = None,
    tools: list[types.Tool] | None = None,
    tool_choice: types.ToolChoice | None = None,
    related_request_id: types.RequestId | None = None,
) -> types.CreateMessageResult | types.CreateMessageResultWithTools:
    """Send a sampling/create_message request.

    Args:
        messages: The conversation messages to send.
        max_tokens: Maximum number of tokens to generate.
        system_prompt: Optional system prompt.
        include_context: Optional context inclusion setting.
            Should only be set to "thisServer" or "allServers"
            if the client has sampling.context capability.
        temperature: Optional sampling temperature.
        stop_sequences: Optional stop sequences.
        metadata: Optional metadata to pass through to the LLM provider.
        model_preferences: Optional model selection preferences.
        tools: Optional list of tools the LLM can use during sampling.
            Requires client to have sampling.tools capability.
        tool_choice: Optional control over tool usage behavior.
            Requires client to have sampling.tools capability.
        related_request_id: Optional ID of a related request.

    Returns:
        The sampling result from the client.

    Raises:
        MCPError: If tools are provided but client doesn't support them.
        ValueError: If tool_use or tool_result message structure is invalid.
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    client_caps = self.client_params.capabilities if self.client_params else None
    validate_sampling_tools(client_caps, tools, tool_choice)
    validate_tool_use_result_messages(messages)

    request = types.CreateMessageRequest(
        params=types.CreateMessageRequestParams(
            messages=messages,
            system_prompt=system_prompt,
            include_context=include_context,
            temperature=temperature,
            max_tokens=max_tokens,
            stop_sequences=stop_sequences,
            metadata=metadata,
            model_preferences=model_preferences,
            tools=tools,
            tool_choice=tool_choice,
        ),
    )
    metadata_obj = ServerMessageMetadata(related_request_id=related_request_id)

    if tools is not None:
        return await self.send_request(
            request=request,
            result_type=types.CreateMessageResultWithTools,
            metadata=metadata_obj,
        )
    return await self.send_request(
        request=request,
        result_type=types.CreateMessageResult,
        metadata=metadata_obj,
    )

list_roots async

list_roots() -> ListRootsResult

Send a roots/list request.

Raises:

Type	Description
NoBackChannelError	The connection has no back-channel for server-initiated requests.

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

@deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def list_roots(self) -> types.ListRootsResult:
    """Send a roots/list request.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ListRootsRequest(),
        types.ListRootsResult,
    )

elicit async

elicit(
    message: str,
    requested_schema: ElicitRequestedSchema,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a form mode elicitation/create request.

Parameters:

Name	Type	Description	Default
message	str	The message to present to the user.	required
requested_schema	ElicitRequestedSchema	Schema defining the expected response structure.	required
related_request_id	RequestId | None	Optional ID of the request that triggered this elicitation.	None

Returns:

Type	Description
ElicitResult	The client's response.

Note

This method is deprecated in favor of elicit_form(). It remains for backward compatibility but new code should use elicit_form().

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

async def elicit(
    self,
    message: str,
    requested_schema: types.ElicitRequestedSchema,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a form mode elicitation/create request.

    Args:
        message: The message to present to the user.
        requested_schema: Schema defining the expected response structure.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response.

    Note:
        This method is deprecated in favor of elicit_form(). It remains for
        backward compatibility but new code should use elicit_form().
    """
    return await self.elicit_form(message, requested_schema, related_request_id)

elicit_form async

elicit_form(
    message: str,
    requested_schema: ElicitRequestedSchema,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a form mode elicitation/create request.

Parameters:

Name	Type	Description	Default
message	str	The message to present to the user.	required
requested_schema	ElicitRequestedSchema	Schema defining the expected response structure.	required
related_request_id	RequestId | None	Optional ID of the request that triggered this elicitation.	None

Returns:

Type	Description
ElicitResult	The client's response with form data.

Raises:

Type	Description
NoBackChannelError	The connection has no back-channel for server-initiated requests.

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

async def elicit_form(
    self,
    message: str,
    requested_schema: types.ElicitRequestedSchema,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a form mode elicitation/create request.

    Args:
        message: The message to present to the user.
        requested_schema: Schema defining the expected response structure.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response with form data.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ElicitRequest(
            params=types.ElicitRequestFormParams(
                message=message,
                requested_schema=requested_schema,
            ),
        ),
        types.ElicitResult,
        metadata=ServerMessageMetadata(related_request_id=related_request_id),
    )

elicit_url async

elicit_url(
    message: str,
    url: str,
    elicitation_id: str,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a URL mode elicitation/create request.

This directs the user to an external URL for out-of-band interactions like OAuth flows, credential collection, or payment processing.

Parameters:

Name	Type	Description	Default
message	str	Human-readable explanation of why the interaction is needed.	required
url	str	The URL the user should navigate to.	required
elicitation_id	str	Unique identifier for tracking this elicitation.	required
related_request_id	RequestId | None	Optional ID of the request that triggered this elicitation.	None

Returns:

Type	Description
ElicitResult	The client's response indicating acceptance, decline, or cancellation.

Raises:

Type	Description
NoBackChannelError	The connection has no back-channel for server-initiated requests.

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

async def elicit_url(
    self,
    message: str,
    url: str,
    elicitation_id: str,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a URL mode elicitation/create request.

    This directs the user to an external URL for out-of-band interactions
    like OAuth flows, credential collection, or payment processing.

    Args:
        message: Human-readable explanation of why the interaction is needed.
        url: The URL the user should navigate to.
        elicitation_id: Unique identifier for tracking this elicitation.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response indicating acceptance, decline, or cancellation.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ElicitRequest(
            params=types.ElicitRequestURLParams(
                message=message,
                url=url,
                elicitation_id=elicitation_id,
            ),
        ),
        types.ElicitResult,
        metadata=ServerMessageMetadata(related_request_id=related_request_id),
    )

send_ping async

send_ping() -> EmptyResult

Send a ping request.

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

async def send_ping(self) -> types.EmptyResult:
    """Send a ping request."""
    return await self.send_request(
        types.PingRequest(),
        types.EmptyResult,
    )

report_progress async

report_progress(
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None

Report progress for the inbound request this session is scoped to.

A no-op when the caller did not request progress. Dispatcher-agnostic: on JSON-RPC the held DispatchContext emits notifications/progress against the caller's token; on the in-process direct dispatcher it invokes the caller's callback directly.

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

async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
    """Report progress for the inbound request this session is scoped to.

    A no-op when the caller did not request progress. Dispatcher-agnostic:
    on JSON-RPC the held `DispatchContext` emits ``notifications/progress``
    against the caller's token; on the in-process direct dispatcher it
    invokes the caller's callback directly.
    """
    await self._request_outbound.progress(progress, total, message)

send_progress_notification async

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

Send a progress notification.

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

async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    related_request_id: str | 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,
            ),
        ),
        related_request_id,
    )

send_resource_list_changed async

send_resource_list_changed() -> None

Send a resource list changed notification.

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

async def send_resource_list_changed(self) -> None:
    """Send a resource list changed notification."""
    await self.send_notification(types.ResourceListChangedNotification())

send_tool_list_changed async

send_tool_list_changed() -> None

Send a tool list changed notification.

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

async def send_tool_list_changed(self) -> None:
    """Send a tool list changed notification."""
    await self.send_notification(types.ToolListChangedNotification())

send_prompt_list_changed async

send_prompt_list_changed() -> None

Send a prompt list changed notification.

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

async def send_prompt_list_changed(self) -> None:
    """Send a prompt list changed notification."""
    await self.send_notification(types.PromptListChangedNotification())

send_elicit_complete async

send_elicit_complete(
    elicitation_id: str,
    related_request_id: RequestId | None = None,
) -> None

Send an elicitation completion notification.

This should be sent when a URL mode elicitation has been completed out-of-band to inform the client that it may retry any requests that were waiting for this elicitation.

Parameters:

Name	Type	Description	Default
elicitation_id	str	The unique identifier of the completed elicitation	required
related_request_id	RequestId | None	Optional ID of the request that triggered this notification	None

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

async def send_elicit_complete(
    self,
    elicitation_id: str,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send an elicitation completion notification.

    This should be sent when a URL mode elicitation has been completed
    out-of-band to inform the client that it may retry any requests
    that were waiting for this elicitation.

    Args:
        elicitation_id: The unique identifier of the completed elicitation
        related_request_id: Optional ID of the request that triggered this notification
    """
    await self.send_notification(
        types.ElicitCompleteNotification(
            params=types.ElicitCompleteNotificationParams(elicitation_id=elicitation_id)
        ),
        related_request_id,
    )

stdio_server async

stdio_server(
    stdin: AsyncFile[str] | None = None,
    stdout: AsyncFile[str] | None = None,
)

Server transport for stdio: this communicates with an MCP client by reading from the current process' stdin and writing to stdout.

Source code in src/mcp/server/stdio.py

@asynccontextmanager
async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.AsyncFile[str] | None = None):
    """Server transport for stdio: this communicates with an MCP client by reading
    from the current process' stdin and writing to stdout.
    """
    # Purposely not using context managers for these, as we don't want to close
    # standard process handles. Encoding of stdin/stdout as text streams on
    # python is platform-dependent (Windows is particularly problematic), so we
    # re-wrap the underlying binary stream to ensure UTF-8.
    if not stdin:
        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace"))
    if not stdout:
        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))

    read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
    write_stream, write_stream_reader = create_context_streams[SessionMessage](0)

    async def stdin_reader():
        try:
            async with read_stream_writer:
                async for line in stdin:
                    try:
                        message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
                    except Exception as exc:
                        await read_stream_writer.send(exc)
                        continue

                    session_message = SessionMessage(message)
                    await read_stream_writer.send(session_message)
        except anyio.ClosedResourceError:  # pragma: no cover
            await anyio.lowlevel.checkpoint()

    async def stdout_writer():
        try:
            async with write_stream_reader:
                async for session_message in write_stream_reader:
                    json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
                    await stdout.write(json + "\n")
                    await stdout.flush()
        except anyio.ClosedResourceError:  # pragma: no cover
            await anyio.lowlevel.checkpoint()

    async with anyio.create_task_group() as tg:
        tg.start_soon(stdin_reader)
        tg.start_soon(stdout_writer)
        yield read_stream, write_stream

MCPDeprecationWarning

Bases: UserWarning

A custom deprecation warning for the MCP SDK.

Unlike the built-in DeprecationWarning, this inherits from UserWarning so it is shown by default, helping users discover deprecated features without enabling warnings explicitly.

Reference: https://sethmlarson.dev/deprecations-via-warnings-dont-work-for-python-libraries

Source code in src/mcp/shared/exceptions.py

class MCPDeprecationWarning(UserWarning):
    """A custom deprecation warning for the MCP SDK.

    Unlike the built-in `DeprecationWarning`, this inherits from `UserWarning` so
    it is shown by default, helping users discover deprecated features without
    enabling warnings explicitly.

    Reference: https://sethmlarson.dev/deprecations-via-warnings-dont-work-for-python-libraries
    """

MCPError

Bases: Exception

Exception type raised when an error arrives over an MCP connection.

Source code in src/mcp/shared/exceptions.py

class MCPError(Exception):
    """Exception type raised when an error arrives over an MCP connection."""

    error: ErrorData

    def __init__(self, code: int, message: str, data: Any = None):
        super().__init__(code, message, data)
        if data is not None:
            self.error = ErrorData(code=code, message=message, data=data)
        else:
            self.error = ErrorData(code=code, message=message)

    @property
    def code(self) -> int:
        return self.error.code

    @property
    def message(self) -> str:
        return self.error.message

    @property
    def data(self) -> Any:
        return self.error.data

    @classmethod
    def from_jsonrpc_error(cls, error: JSONRPCError) -> MCPError:
        return cls.from_error_data(error.error)

    @classmethod
    def from_error_data(cls, error: ErrorData) -> MCPError:
        return cls(code=error.code, message=error.message, data=error.data)

    def __str__(self) -> str:
        return self.message

UrlElicitationRequiredError

Bases: MCPError

Specialized error for when a tool requires URL mode elicitation(s) before proceeding.

Servers can raise this error from tool handlers to indicate that the client must complete one or more URL elicitations before the request can be processed.

Example

raise UrlElicitationRequiredError([
    ElicitRequestURLParams(
        message="Authorization required for your files",
        url="https://example.com/oauth/authorize",
        elicitation_id="auth-001"
    )
])

Source code in src/mcp/shared/exceptions.py

class UrlElicitationRequiredError(MCPError):
    """Specialized error for when a tool requires URL mode elicitation(s) before proceeding.

    Servers can raise this error from tool handlers to indicate that the client
    must complete one or more URL elicitations before the request can be processed.

    Example:
        ```python
        raise UrlElicitationRequiredError([
            ElicitRequestURLParams(
                message="Authorization required for your files",
                url="https://example.com/oauth/authorize",
                elicitation_id="auth-001"
            )
        ])
        ```
    """

    def __init__(self, elicitations: list[ElicitRequestURLParams], message: str | None = None):
        """Initialize UrlElicitationRequiredError."""
        if message is None:
            message = f"URL elicitation{'s' if len(elicitations) > 1 else ''} required"

        self._elicitations = elicitations

        super().__init__(
            code=URL_ELICITATION_REQUIRED,
            message=message,
            data={"elicitations": [e.model_dump(by_alias=True, exclude_none=True) for e in elicitations]},
        )

    @property
    def elicitations(self) -> list[ElicitRequestURLParams]:
        """The list of URL elicitations required before the request can proceed."""
        return self._elicitations

    @classmethod
    def from_error(cls, error: ErrorData) -> UrlElicitationRequiredError:
        """Reconstruct from an ErrorData received over the wire."""
        if error.code != URL_ELICITATION_REQUIRED:
            raise ValueError(f"Expected error code {URL_ELICITATION_REQUIRED}, got {error.code}")

        data = cast(dict[str, Any], error.data or {})
        raw_elicitations = cast(list[dict[str, Any]], data.get("elicitations", []))
        elicitations = [ElicitRequestURLParams.model_validate(e) for e in raw_elicitations]
        return cls(elicitations, error.message)

__init__

__init__(
    elicitations: list[ElicitRequestURLParams],
    message: str | None = None,
)

Initialize UrlElicitationRequiredError.

Source code in src/mcp/shared/exceptions.py

def __init__(self, elicitations: list[ElicitRequestURLParams], message: str | None = None):
    """Initialize UrlElicitationRequiredError."""
    if message is None:
        message = f"URL elicitation{'s' if len(elicitations) > 1 else ''} required"

    self._elicitations = elicitations

    super().__init__(
        code=URL_ELICITATION_REQUIRED,
        message=message,
        data={"elicitations": [e.model_dump(by_alias=True, exclude_none=True) for e in elicitations]},
    )

elicitations property

elicitations: list[ElicitRequestURLParams]

The list of URL elicitations required before the request can proceed.

from_error classmethod

from_error(error: ErrorData) -> UrlElicitationRequiredError

Reconstruct from an ErrorData received over the wire.

Source code in src/mcp/shared/exceptions.py

@classmethod
def from_error(cls, error: ErrorData) -> UrlElicitationRequiredError:
    """Reconstruct from an ErrorData received over the wire."""
    if error.code != URL_ELICITATION_REQUIRED:
        raise ValueError(f"Expected error code {URL_ELICITATION_REQUIRED}, got {error.code}")

    data = cast(dict[str, Any], error.data or {})
    raw_elicitations = cast(list[dict[str, Any]], data.get("elicitations", []))
    elicitations = [ElicitRequestURLParams.model_validate(e) for e in raw_elicitations]
    return cls(elicitations, error.message)

InvalidUriTemplate

Bases: ValueError

Raised when a URI template string is malformed or unsupported.

Attributes:

Name	Type	Description
template		The template string that failed to parse.
position		Character offset where the error was detected, or None if the error is not tied to a specific position.

Source code in src/mcp/shared/uri_template.py

class InvalidUriTemplate(ValueError):
    """Raised when a URI template string is malformed or unsupported.

    Attributes:
        template: The template string that failed to parse.
        position: Character offset where the error was detected, or None
            if the error is not tied to a specific position.
    """

    def __init__(self, message: str, *, template: str, position: int | None = None) -> None:
        super().__init__(message)
        self.template = template
        self.position = position

UriTemplate dataclass

A parsed RFC 6570 URI template.

Construct via :meth:parse. Instances are immutable and hashable; equality is based on the template string alone.

Source code in src/mcp/shared/uri_template.py

@dataclass(frozen=True)
class UriTemplate:
    """A parsed RFC 6570 URI template.

    Construct via :meth:`parse`. Instances are immutable and hashable;
    equality is based on the template string alone.
    """

    template: str
    _parts: list[_Part] = field(repr=False, compare=False)
    _variables: list[Variable] = field(repr=False, compare=False)
    _prefix: list[_Atom] = field(repr=False, compare=False)
    _greedy: Variable | None = field(repr=False, compare=False)
    _suffix: list[_Atom] = field(repr=False, compare=False)
    _query_variables: list[Variable] = field(repr=False, compare=False)

    @staticmethod
    def is_template(value: str) -> bool:
        """Check whether a string contains URI template expressions.

        A cheap heuristic for distinguishing concrete URIs from templates
        without the cost of full parsing. Returns ``True`` if the string
        contains at least one ``{...}`` pair.

        Example::

            >>> UriTemplate.is_template("file://docs/{name}")
            True
            >>> UriTemplate.is_template("file://docs/readme.txt")
            False

        Note:
            This does not validate the template. A ``True`` result does
            not guarantee :meth:`parse` will succeed.
        """
        open_i = value.find("{")
        return open_i != -1 and value.find("}", open_i) != -1

    @classmethod
    def parse(
        cls,
        template: str,
        *,
        max_length: int = DEFAULT_MAX_TEMPLATE_LENGTH,
        max_variables: int = DEFAULT_MAX_VARIABLES,
    ) -> UriTemplate:
        """Parse a URI template string.

        Args:
            template: An RFC 6570 URI template.
            max_length: Maximum permitted length of the template string.
                Guards against resource exhaustion.
            max_variables: Maximum number of variables permitted across
                all expressions. Counting variables rather than
                ``{...}`` expressions closes the gap where a single
                ``{v0,v1,...,vN}`` expression packs arbitrarily many
                variables under one expression count.

        Raises:
            InvalidUriTemplate: If the template is malformed, exceeds the
                size limits, or uses unsupported RFC 6570 features.
        """
        if len(template) > max_length:
            raise InvalidUriTemplate(
                f"Template exceeds maximum length of {max_length}",
                template=template,
            )

        parts, variables = _parse(template, max_variables=max_variables)

        # Trailing {?...}/{&...} expressions are split off and matched as
        # a query string (order-agnostic, partial, extras ignored) rather
        # than via the linear scan.
        path_parts, query_vars = _split_query_tail(parts)
        atoms = _flatten(path_parts)
        prefix, greedy, suffix = _partition_greedy(atoms, template)

        return cls(
            template=template,
            _parts=parts,
            _variables=variables,
            _prefix=prefix,
            _greedy=greedy,
            _suffix=suffix,
            _query_variables=query_vars,
        )

    @property
    def variables(self) -> list[Variable]:
        """All variables in the template, in order of appearance."""
        return list(self._variables)

    @property
    def variable_names(self) -> list[str]:
        """All variable names in the template, in order of appearance."""
        return [v.name for v in self._variables]

    @property
    def query_variable_names(self) -> frozenset[str]:
        """Names of variables that :meth:`match` treats as optional query parameters.

        These are the variables in a trailing run of ``{?...}``/``{&...}``
        expressions, which are matched leniently: a URI that omits some
        (or all) of them still matches, and the omitted names are simply
        absent from the result. Any value bound to such a name therefore
        needs a fallback for the omitted case.

        Every other variable is bound on every successful :meth:`match`
        (possibly to an empty string) and is *not* in this set. That
        includes a ``{&...}`` expression with no preceding ``{?...}``: it
        never emits the ``?`` the lenient query split keys on, so it is
        matched strictly.
        """
        return frozenset(v.name for v in self._query_variables)

    def expand(self, variables: Mapping[str, str | Sequence[str]]) -> str:
        """Expand the template by substituting variable values.

        String values are percent-encoded according to their operator:
        simple ``{var}`` encodes reserved characters; ``{+var}`` and
        ``{#var}`` leave them intact. Sequence values are joined with
        commas for non-explode variables, or with the operator's
        separator for explode variables.

        Example::

            >>> t = UriTemplate.parse("file://docs/{name}")
            >>> t.expand({"name": "hello world.txt"})
            'file://docs/hello%20world.txt'

            >>> t = UriTemplate.parse("file://docs/{+path}")
            >>> t.expand({"path": "src/main.py"})
            'file://docs/src/main.py'

            >>> t = UriTemplate.parse("/search{?q,lang}")
            >>> t.expand({"q": "mcp", "lang": "en"})
            '/search?q=mcp&lang=en'

            >>> t = UriTemplate.parse("/files{/path*}")
            >>> t.expand({"path": ["a", "b", "c"]})
            '/files/a/b/c'

        Args:
            variables: Values for each template variable. Keys must be
                strings; values must be ``str`` or a sequence of ``str``.

        Returns:
            The expanded URI string.

        Note:
            Per RFC 6570, variables absent from the mapping are
            **silently omitted**. This is the correct behavior for
            optional query parameters (``{?page}`` with no page yields
            no ``?page=``), but for required path segments it produces
            a structurally incomplete URI. If you need all variables
            present, validate before calling::

                missing = set(t.variable_names) - variables.keys()
                if missing:
                    raise ValueError(f"Missing: {missing}")

        Raises:
            TypeError: If a value is neither ``str`` nor an iterable of
                ``str``. Non-string scalars (``int``, ``None``) are not
                coerced.
        """
        out: list[str] = []
        for part in self._parts:
            if isinstance(part, str):
                out.append(part)
            else:
                out.append(_expand_expression(part, variables))
        return "".join(out)

    def match(self, uri: str, *, max_uri_length: int = DEFAULT_MAX_URI_LENGTH) -> dict[str, str | list[str]] | None:
        """Match a concrete URI against this template and extract variables.

        This is the inverse of :meth:`expand`. The URI is matched via a
        linear scan of the template and captured values are
        percent-decoded. The round-trip ``match(expand({k: v})) == {k: v}``
        holds when ``v`` does not contain its operator's separator
        unencoded: ``{.ext}`` with ``ext="tar.gz"`` expands to
        ``.tar.gz`` but does not match — the scan stops ``ext`` at the
        first ``.`` and the trailing ``.gz`` has nothing to consume it.
        RFC 6570 §1.4 notes this is an inherent reversal limitation.

        Matching is structural at the URI level only: a simple ``{name}``
        will not match across a literal ``/`` in the URI (the scan stops
        there), but a percent-encoded ``%2F`` that decodes to ``/`` is
        accepted as part of the value. Path-safety validation belongs at
        a higher layer; see :mod:`mcp.shared.path_security`.

        Example::

            >>> t = UriTemplate.parse("file://docs/{name}")
            >>> t.match("file://docs/readme.txt")
            {'name': 'readme.txt'}
            >>> t.match("file://docs/hello%20world.txt")
            {'name': 'hello world.txt'}

            >>> t = UriTemplate.parse("file://docs/{+path}")
            >>> t.match("file://docs/src/main.py")
            {'path': 'src/main.py'}

            >>> t = UriTemplate.parse("/files{/path*}")
            >>> t.match("/files/a/b/c")
            {'path': ['a', 'b', 'c']}

        **Query parameters** (``{?q,lang}`` at the end of a template)
        are matched leniently: order-agnostic, partial, and unrecognized
        params are ignored. Absent params are omitted from the result so
        downstream function defaults can apply::

            >>> t = UriTemplate.parse("logs://{service}{?since,level}")
            >>> t.match("logs://api")
            {'service': 'api'}
            >>> t.match("logs://api?level=error")
            {'service': 'api', 'level': 'error'}
            >>> t.match("logs://api?level=error&since=5m&utm=x")
            {'service': 'api', 'since': '5m', 'level': 'error'}

        Args:
            uri: A concrete URI string.
            max_uri_length: Maximum permitted length of the input URI.
                Oversized inputs return ``None`` without scanning,
                guarding against resource exhaustion.

        Returns:
            A mapping from variable names to decoded values (``str`` for
            scalar variables, ``list[str]`` for explode variables), or
            ``None`` if the URI does not match the template or exceeds
            ``max_uri_length``.
        """
        if len(uri) > max_uri_length:
            return None

        if self._query_variables:
            # Two-phase: scan matches the path, the query is split and
            # decoded manually. Query params may be partial, reordered,
            # or include extras; absent params stay absent so downstream
            # defaults can apply. Fragment is stripped first since the
            # template's {?...} tail never describes a fragment.
            before_fragment, _, _ = uri.partition("#")
            path, _, query = before_fragment.partition("?")
            result = self._scan(path)
            if result is None:
                return None
            if query:
                parsed = _parse_query(query)
                for var in self._query_variables:
                    if var.name in parsed:
                        result[var.name] = parsed[var.name]
            return result

        return self._scan(uri)

    def _scan(self, uri: str) -> dict[str, str | list[str]] | None:
        """Run the two-ended linear scan against the path portion of a URI."""
        n = len(uri)

        if self._greedy is None:
            # No greedy var: the suffix IS the whole template, scanned
            # right-to-left and anchored so atoms[0] matches at position 0.
            suffix = _scan_suffix(self._suffix, uri, n, anchored=True)
            if suffix is None:
                return None
            suffix_result, suffix_start = suffix
            return suffix_result if suffix_start == 0 else None

        # Greedy var present. The parser rejects a capture adjacent to
        # the greedy slot, so a non-empty suffix begins with a _Lit whose
        # rfind-derived anchor does not depend on how far the prefix
        # scans. Scan the suffix first, then give the prefix that exact
        # position as its ceiling so it cannot consume past the anchor.
        suffix = _scan_suffix(self._suffix, uri, n, anchored=False)
        if suffix is None:
            return None
        suffix_result, suffix_start = suffix
        prefix = _scan_prefix(self._prefix, uri, 0, suffix_start)
        if prefix is None:
            return None
        prefix_result, prefix_end = prefix

        # Prefix consumed [0, prefix_end); suffix consumed [suffix_start, n);
        # the greedy var takes the gap. The prefix scan is bounded by
        # suffix_start, so this holds by construction; guard explicitly
        # rather than asserting so a future regression surfaces as a
        # non-match, not an exception.
        if suffix_start < prefix_end:
            return None  # pragma: no cover - unreachable while bounds hold
        middle = uri[prefix_end:suffix_start]
        greedy_value = _extract_greedy(self._greedy, middle)
        if greedy_value is None:
            return None

        return {**prefix_result, self._greedy.name: greedy_value, **suffix_result}

    def __str__(self) -> str:
        return self.template

is_template staticmethod

is_template(value: str) -> bool

Check whether a string contains URI template expressions.

A cheap heuristic for distinguishing concrete URIs from templates without the cost of full parsing. Returns True if the string contains at least one {...} pair.

Example::

>>> UriTemplate.is_template("file://docs/{name}")
True
>>> UriTemplate.is_template("file://docs/readme.txt")
False

Note

This does not validate the template. A True result does not guarantee :meth:parse will succeed.

Source code in src/mcp/shared/uri_template.py

@staticmethod
def is_template(value: str) -> bool:
    """Check whether a string contains URI template expressions.

    A cheap heuristic for distinguishing concrete URIs from templates
    without the cost of full parsing. Returns ``True`` if the string
    contains at least one ``{...}`` pair.

    Example::

        >>> UriTemplate.is_template("file://docs/{name}")
        True
        >>> UriTemplate.is_template("file://docs/readme.txt")
        False

    Note:
        This does not validate the template. A ``True`` result does
        not guarantee :meth:`parse` will succeed.
    """
    open_i = value.find("{")
    return open_i != -1 and value.find("}", open_i) != -1

parse classmethod

parse(
    template: str,
    *,
    max_length: int = DEFAULT_MAX_TEMPLATE_LENGTH,
    max_variables: int = DEFAULT_MAX_VARIABLES
) -> UriTemplate

Parse a URI template string.

Parameters:

Name	Type	Description	Default
template	str	An RFC 6570 URI template.	required
max_length	int	Maximum permitted length of the template string. Guards against resource exhaustion.	DEFAULT_MAX_TEMPLATE_LENGTH
max_variables	int	Maximum number of variables permitted across all expressions. Counting variables rather than {...} expressions closes the gap where a single {v0,v1,...,vN} expression packs arbitrarily many variables under one expression count.	DEFAULT_MAX_VARIABLES

Raises:

Type	Description
InvalidUriTemplate	If the template is malformed, exceeds the size limits, or uses unsupported RFC 6570 features.

Source code in src/mcp/shared/uri_template.py

@classmethod
def parse(
    cls,
    template: str,
    *,
    max_length: int = DEFAULT_MAX_TEMPLATE_LENGTH,
    max_variables: int = DEFAULT_MAX_VARIABLES,
) -> UriTemplate:
    """Parse a URI template string.

    Args:
        template: An RFC 6570 URI template.
        max_length: Maximum permitted length of the template string.
            Guards against resource exhaustion.
        max_variables: Maximum number of variables permitted across
            all expressions. Counting variables rather than
            ``{...}`` expressions closes the gap where a single
            ``{v0,v1,...,vN}`` expression packs arbitrarily many
            variables under one expression count.

    Raises:
        InvalidUriTemplate: If the template is malformed, exceeds the
            size limits, or uses unsupported RFC 6570 features.
    """
    if len(template) > max_length:
        raise InvalidUriTemplate(
            f"Template exceeds maximum length of {max_length}",
            template=template,
        )

    parts, variables = _parse(template, max_variables=max_variables)

    # Trailing {?...}/{&...} expressions are split off and matched as
    # a query string (order-agnostic, partial, extras ignored) rather
    # than via the linear scan.
    path_parts, query_vars = _split_query_tail(parts)
    atoms = _flatten(path_parts)
    prefix, greedy, suffix = _partition_greedy(atoms, template)

    return cls(
        template=template,
        _parts=parts,
        _variables=variables,
        _prefix=prefix,
        _greedy=greedy,
        _suffix=suffix,
        _query_variables=query_vars,
    )

variables property

variables: list[Variable]

All variables in the template, in order of appearance.

variable_names property

variable_names: list[str]

All variable names in the template, in order of appearance.

query_variable_names property

query_variable_names: frozenset[str]

Names of variables that :meth:match treats as optional query parameters.

These are the variables in a trailing run of {?...}/{&...} expressions, which are matched leniently: a URI that omits some (or all) of them still matches, and the omitted names are simply absent from the result. Any value bound to such a name therefore needs a fallback for the omitted case.

Every other variable is bound on every successful :meth:match (possibly to an empty string) and is not in this set. That includes a {&...} expression with no preceding {?...}: it never emits the ? the lenient query split keys on, so it is matched strictly.

expand

expand(variables: Mapping[str, str | Sequence[str]]) -> str

Expand the template by substituting variable values.

String values are percent-encoded according to their operator: simple {var} encodes reserved characters; {+var} and {#var} leave them intact. Sequence values are joined with commas for non-explode variables, or with the operator's separator for explode variables.

Example::

>>> t = UriTemplate.parse("file://docs/{name}")
>>> t.expand({"name": "hello world.txt"})
'file://docs/hello%20world.txt'

>>> t = UriTemplate.parse("file://docs/{+path}")
>>> t.expand({"path": "src/main.py"})
'file://docs/src/main.py'

>>> t = UriTemplate.parse("/search{?q,lang}")
>>> t.expand({"q": "mcp", "lang": "en"})
'/search?q=mcp&lang=en'

>>> t = UriTemplate.parse("/files{/path*}")
>>> t.expand({"path": ["a", "b", "c"]})
'/files/a/b/c'

Parameters:

Name	Type	Description	Default
variables	Mapping[str, str | Sequence[str]]	Values for each template variable. Keys must be strings; values must be str or a sequence of str.	required

Returns:

Type	Description
str	The expanded URI string.

Note

Per RFC 6570, variables absent from the mapping are silently omitted. This is the correct behavior for optional query parameters ({?page} with no page yields no ?page=), but for required path segments it produces a structurally incomplete URI. If you need all variables present, validate before calling::

missing = set(t.variable_names) - variables.keys()
if missing:
    raise ValueError(f"Missing: {missing}")

Raises:

Type	Description
TypeError	If a value is neither str nor an iterable of str. Non-string scalars (int, None) are not coerced.

Source code in src/mcp/shared/uri_template.py

def expand(self, variables: Mapping[str, str | Sequence[str]]) -> str:
    """Expand the template by substituting variable values.

    String values are percent-encoded according to their operator:
    simple ``{var}`` encodes reserved characters; ``{+var}`` and
    ``{#var}`` leave them intact. Sequence values are joined with
    commas for non-explode variables, or with the operator's
    separator for explode variables.

    Example::

        >>> t = UriTemplate.parse("file://docs/{name}")
        >>> t.expand({"name": "hello world.txt"})
        'file://docs/hello%20world.txt'

        >>> t = UriTemplate.parse("file://docs/{+path}")
        >>> t.expand({"path": "src/main.py"})
        'file://docs/src/main.py'

        >>> t = UriTemplate.parse("/search{?q,lang}")
        >>> t.expand({"q": "mcp", "lang": "en"})
        '/search?q=mcp&lang=en'

        >>> t = UriTemplate.parse("/files{/path*}")
        >>> t.expand({"path": ["a", "b", "c"]})
        '/files/a/b/c'

    Args:
        variables: Values for each template variable. Keys must be
            strings; values must be ``str`` or a sequence of ``str``.

    Returns:
        The expanded URI string.

    Note:
        Per RFC 6570, variables absent from the mapping are
        **silently omitted**. This is the correct behavior for
        optional query parameters (``{?page}`` with no page yields
        no ``?page=``), but for required path segments it produces
        a structurally incomplete URI. If you need all variables
        present, validate before calling::

            missing = set(t.variable_names) - variables.keys()
            if missing:
                raise ValueError(f"Missing: {missing}")

    Raises:
        TypeError: If a value is neither ``str`` nor an iterable of
            ``str``. Non-string scalars (``int``, ``None``) are not
            coerced.
    """
    out: list[str] = []
    for part in self._parts:
        if isinstance(part, str):
            out.append(part)
        else:
            out.append(_expand_expression(part, variables))
    return "".join(out)

match

match(
    uri: str,
    *,
    max_uri_length: int = DEFAULT_MAX_URI_LENGTH
) -> dict[str, str | list[str]] | None

Match a concrete URI against this template and extract variables.

This is the inverse of :meth:expand. The URI is matched via a linear scan of the template and captured values are percent-decoded. The round-trip match(expand({k: v})) == {k: v} holds when v does not contain its operator's separator unencoded: {.ext} with ext="tar.gz" expands to .tar.gz but does not match — the scan stops ext at the first . and the trailing .gz has nothing to consume it. RFC 6570 §1.4 notes this is an inherent reversal limitation.

Matching is structural at the URI level only: a simple {name} will not match across a literal / in the URI (the scan stops there), but a percent-encoded %2F that decodes to / is accepted as part of the value. Path-safety validation belongs at a higher layer; see :mod:mcp.shared.path_security.

Example::

>>> t = UriTemplate.parse("file://docs/{name}")
>>> t.match("file://docs/readme.txt")
{'name': 'readme.txt'}
>>> t.match("file://docs/hello%20world.txt")
{'name': 'hello world.txt'}

>>> t = UriTemplate.parse("file://docs/{+path}")
>>> t.match("file://docs/src/main.py")
{'path': 'src/main.py'}

>>> t = UriTemplate.parse("/files{/path*}")
>>> t.match("/files/a/b/c")
{'path': ['a', 'b', 'c']}

Query parameters ({?q,lang} at the end of a template) are matched leniently: order-agnostic, partial, and unrecognized params are ignored. Absent params are omitted from the result so downstream function defaults can apply::

>>> t = UriTemplate.parse("logs://{service}{?since,level}")
>>> t.match("logs://api")
{'service': 'api'}
>>> t.match("logs://api?level=error")
{'service': 'api', 'level': 'error'}
>>> t.match("logs://api?level=error&since=5m&utm=x")
{'service': 'api', 'since': '5m', 'level': 'error'}

Parameters:

Name	Type	Description	Default
uri	str	A concrete URI string.	required
max_uri_length	int	Maximum permitted length of the input URI. Oversized inputs return None without scanning, guarding against resource exhaustion.	DEFAULT_MAX_URI_LENGTH

Returns:

Type	Description
dict[str, str | list[str]] | None	A mapping from variable names to decoded values (str for
dict[str, str | list[str]] | None	scalar variables, list[str] for explode variables), or
dict[str, str | list[str]] | None	None if the URI does not match the template or exceeds
dict[str, str | list[str]] | None	max_uri_length.

Source code in src/mcp/shared/uri_template.py

def match(self, uri: str, *, max_uri_length: int = DEFAULT_MAX_URI_LENGTH) -> dict[str, str | list[str]] | None:
    """Match a concrete URI against this template and extract variables.

    This is the inverse of :meth:`expand`. The URI is matched via a
    linear scan of the template and captured values are
    percent-decoded. The round-trip ``match(expand({k: v})) == {k: v}``
    holds when ``v`` does not contain its operator's separator
    unencoded: ``{.ext}`` with ``ext="tar.gz"`` expands to
    ``.tar.gz`` but does not match — the scan stops ``ext`` at the
    first ``.`` and the trailing ``.gz`` has nothing to consume it.
    RFC 6570 §1.4 notes this is an inherent reversal limitation.

    Matching is structural at the URI level only: a simple ``{name}``
    will not match across a literal ``/`` in the URI (the scan stops
    there), but a percent-encoded ``%2F`` that decodes to ``/`` is
    accepted as part of the value. Path-safety validation belongs at
    a higher layer; see :mod:`mcp.shared.path_security`.

    Example::

        >>> t = UriTemplate.parse("file://docs/{name}")
        >>> t.match("file://docs/readme.txt")
        {'name': 'readme.txt'}
        >>> t.match("file://docs/hello%20world.txt")
        {'name': 'hello world.txt'}

        >>> t = UriTemplate.parse("file://docs/{+path}")
        >>> t.match("file://docs/src/main.py")
        {'path': 'src/main.py'}

        >>> t = UriTemplate.parse("/files{/path*}")
        >>> t.match("/files/a/b/c")
        {'path': ['a', 'b', 'c']}

    **Query parameters** (``{?q,lang}`` at the end of a template)
    are matched leniently: order-agnostic, partial, and unrecognized
    params are ignored. Absent params are omitted from the result so
    downstream function defaults can apply::

        >>> t = UriTemplate.parse("logs://{service}{?since,level}")
        >>> t.match("logs://api")
        {'service': 'api'}
        >>> t.match("logs://api?level=error")
        {'service': 'api', 'level': 'error'}
        >>> t.match("logs://api?level=error&since=5m&utm=x")
        {'service': 'api', 'since': '5m', 'level': 'error'}

    Args:
        uri: A concrete URI string.
        max_uri_length: Maximum permitted length of the input URI.
            Oversized inputs return ``None`` without scanning,
            guarding against resource exhaustion.

    Returns:
        A mapping from variable names to decoded values (``str`` for
        scalar variables, ``list[str]`` for explode variables), or
        ``None`` if the URI does not match the template or exceeds
        ``max_uri_length``.
    """
    if len(uri) > max_uri_length:
        return None

    if self._query_variables:
        # Two-phase: scan matches the path, the query is split and
        # decoded manually. Query params may be partial, reordered,
        # or include extras; absent params stay absent so downstream
        # defaults can apply. Fragment is stripped first since the
        # template's {?...} tail never describes a fragment.
        before_fragment, _, _ = uri.partition("#")
        path, _, query = before_fragment.partition("?")
        result = self._scan(path)
        if result is None:
            return None
        if query:
            parsed = _parse_query(query)
            for var in self._query_variables:
                if var.name in parsed:
                    result[var.name] = parsed[var.name]
        return result

    return self._scan(uri)