Index

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 wrapped in an `InMemoryTransport`.
    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."""

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

    _session: ClientSession | None = field(init=False, default=None)
    _exit_stack: AsyncExitStack | None = field(init=False, default=None)
    _transport: Transport = field(init=False)

    def __post_init__(self) -> None:
        if isinstance(self.server, Server | MCPServer):
            self._transport = InMemoryTransport(self.server, raise_exceptions=self.raise_exceptions)
        elif isinstance(self.server, str):
            self._transport = streamable_http_client(self.server)
        else:
            self._transport = self.server

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

        async with AsyncExitStack() as exit_stack:
            read_stream, write_stream = await exit_stack.enter_async_context(self._transport)

            self._session = await exit_stack.enter_async_context(
                ClientSession(
                    read_stream=read_stream,
                    write_stream=write_stream,
                    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=self.message_handler,
                    client_info=self.client_info,
                    elicitation_callback=self.elicitation_callback,
                )
            )

            await self._session.initialize()

            # Transfer ownership to self for __aexit__ to handle
            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

    @property
    def initialize_result(self) -> InitializeResult:
        """The server's InitializeResult.

        Contains server_info, capabilities, instructions, and the negotiated protocol_version.
        Raises RuntimeError if accessed outside the context manager.
        """
        result = self.session.initialize_result
        if result is None:  # pragma: no cover
            raise RuntimeError("Client must be used within an async context manager")
        return result

    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)

    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(
            progress_token=progress_token,
            progress=progress,
            total=total,
            message=message,
        )

    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)

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

    async def list_resource_templates(
        self,
        *,
        cursor: str | None = None,
        meta: RequestParamsMeta | None = None,
    ) -> ListResourceTemplatesResult:
        """List available resource templates from the server."""
        return await self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

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

        Args:
            uri: The URI of the resource to read.
            meta: Additional metadata for the request.

        Returns:
            The resource content.
        """
        return await self.session.read_resource(uri, meta=meta)

    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,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> CallToolResult:
        """Call a tool on the server.

        Args:
            name: The name of the tool to call
            arguments: Arguments to pass to the tool
            read_timeout_seconds: Timeout for the tool call
            progress_callback: Callback for progress updates
            meta: Additional metadata for the request

        Returns:
            The tool result.
        """
        return await self.session.call_tool(
            name=name,
            arguments=arguments,
            read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
            meta=meta,
        )

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

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

        Args:
            name: The name of the prompt
            arguments: Arguments to pass to the prompt
            meta: Additional metadata for the request

        Returns:
            The prompt content.
        """
        return await self.session.get_prompt(name=name, arguments=arguments, meta=meta)

    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) -> ListToolsResult:
        """List available tools from the server."""
        return await self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

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

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 wrapped in an InMemoryTransport. 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.

elicitation_callback class-attribute instance-attribute

elicitation_callback: ElicitationFnT | None = None

Callback for handling elicitation requests.

__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._session is not None:
        raise RuntimeError("Client is already entered; cannot reenter")

    async with AsyncExitStack() as exit_stack:
        read_stream, write_stream = await exit_stack.enter_async_context(self._transport)

        self._session = await exit_stack.enter_async_context(
            ClientSession(
                read_stream=read_stream,
                write_stream=write_stream,
                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=self.message_handler,
                client_info=self.client_info,
                elicitation_callback=self.elicitation_callback,
            )
        )

        await self._session.initialize()

        # Transfer ownership to self for __aexit__ to handle
        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.

initialize_result property

initialize_result: InitializeResult

The server's InitializeResult.

Contains server_info, capabilities, instructions, and the negotiated protocol_version. Raises RuntimeError if accessed outside the context manager.

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

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

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(
        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

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)

list_resources async

list_resources(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> 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,
) -> ListResourcesResult:
    """List available resources from the server."""
    return await 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
) -> 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,
) -> ListResourceTemplatesResult:
    """List available resource templates from the server."""
    return await self.session.list_resource_templates(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

read_resource async

read_resource(
    uri: str, *, meta: RequestParamsMeta | None = None
) -> ReadResourceResult

Read a resource from the server.

Parameters:

Name	Type	Description	Default
uri	str	The URI of the resource to read.	required
meta	RequestParamsMeta | None	Additional metadata for the request.	None

Returns:

Type	Description
ReadResourceResult	The resource content.

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

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

    Args:
        uri: The URI of the resource to read.
        meta: Additional metadata for the request.

    Returns:
        The resource content.
    """
    return await self.session.read_resource(uri, meta=meta)

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,
    *,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

Call a tool on the server.

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 the tool call	None
progress_callback	ProgressFnT | None	Callback for progress updates	None
meta	RequestParamsMeta | None	Additional metadata for the request	None

Returns:

Type	Description
CallToolResult	The tool result.

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,
    *,
    meta: RequestParamsMeta | None = None,
) -> CallToolResult:
    """Call a tool on the server.

    Args:
        name: The name of the tool to call
        arguments: Arguments to pass to the tool
        read_timeout_seconds: Timeout for the tool call
        progress_callback: Callback for progress updates
        meta: Additional metadata for the request

    Returns:
        The tool result.
    """
    return await self.session.call_tool(
        name=name,
        arguments=arguments,
        read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
        meta=meta,
    )

list_prompts async

list_prompts(
    *,
    cursor: str | None = None,
    meta: RequestParamsMeta | None = None
) -> 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,
) -> ListPromptsResult:
    """List available prompts from the server."""
    return await self.session.list_prompts(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

get_prompt async

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

Get a prompt from the server.

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
meta	RequestParamsMeta | None	Additional metadata for the request	None

Returns:

Type	Description
GetPromptResult	The prompt content.

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

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

    Args:
        name: The name of the prompt
        arguments: Arguments to pass to the prompt
        meta: Additional metadata for the request

    Returns:
        The prompt content.
    """
    return await self.session.get_prompt(name=name, arguments=arguments, meta=meta)

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
) -> 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) -> ListToolsResult:
    """List available tools from the server."""
    return await self.session.list_tools(params=PaginatedRequestParams(cursor=cursor, _meta=meta))

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

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

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.

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

    def __init__(
        self,
        read_stream: ReadStream[SessionMessage | Exception] | None = None,
        write_stream: WriteStream[SessionMessage] | None = None,
        read_timeout_seconds: float | None = None,
        sampling_callback: SamplingFnT | None = None,
        elicitation_callback: ElicitationFnT | None = None,
        list_roots_callback: ListRootsFnT | None = None,
        logging_callback: LoggingFnT | None = None,
        message_handler: MessageHandlerFnT | None = None,
        client_info: types.Implementation | None = None,
        *,
        sampling_capabilities: types.SamplingCapability | None = None,
        dispatcher: Dispatcher[Any] | None = None,
    ) -> None:
        self._session_read_timeout_seconds = read_timeout_seconds
        self._client_info = client_info or DEFAULT_CLIENT_INFO
        self._sampling_callback = sampling_callback or _default_sampling_callback
        self._sampling_capabilities = sampling_capabilities
        self._elicitation_callback = elicitation_callback or _default_elicitation_callback
        self._list_roots_callback = list_roots_callback or _default_list_roots_callback
        self._logging_callback = logging_callback or _default_logging_callback
        self._message_handler = message_handler or _default_message_handler
        self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
        self._initialize_result: types.InitializeResult | None = None
        self._task_group: anyio.abc.TaskGroup | None = None
        if dispatcher is not None:
            if read_stream is not None or write_stream is not None:
                raise ValueError("pass read_stream/write_stream or dispatcher, not both")
            self._dispatcher: Dispatcher[Any] = dispatcher
        else:
            if read_stream is None or write_stream is None:
                raise ValueError("read_stream and write_stream are required when no dispatcher is given")
            # Built 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:
            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
        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
            await task_group.__aexit__(None, None, None)
            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 and in-flight callbacks.
        assert self._task_group is not None
        self._task_group.cancel_scope.cancel()
        result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
        await resync_tracer()
        return result

    async def send_request(
        self,
        request: types.ClientRequest,
        result_type: type[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.
        """
        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
        method: str = data["method"]
        opts: CallOptions = {}
        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
        if method == "initialize":
            # The spec forbids cancelling initialize.
            opts["cancel_on_abandon"] = False
        raw = await self._dispatcher.send_raw_request(method, data.get("params"), opts)
        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)
        await self._dispatcher.notify(data["method"], data.get("params"))

    async def initialize(self) -> types.InitializeResult:
        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
        )

        result = await self.send_request(
            types.InitializeRequest(
                params=types.InitializeRequestParams(
                    protocol_version=types.LATEST_PROTOCOL_VERSION,
                    capabilities=types.ClientCapabilities(
                        sampling=sampling,
                        elicitation=elicitation,
                        experimental=None,
                        roots=roots,
                    ),
                    client_info=self._client_info,
                ),
            ),
            types.InitializeResult,
        )

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

        self._initialize_result = result

        await self.send_notification(types.InitializedNotification())

        return result

    @property
    def initialize_result(self) -> types.InitializeResult | None:
        """The server's InitializeResult. None until initialize() has been called.

        Contains server_info, capabilities, instructions, and the negotiated protocol_version.
        """
        return self._initialize_result

    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)

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

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

    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
        """Send a resources/read request."""
        return await self.send_request(
            types.ReadResourceRequest(params=types.ReadResourceRequestParams(uri=uri, _meta=meta)),
            types.ReadResourceResult,
        )

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

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

        result = await self.send_request(
            types.CallToolRequest(
                params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
            ),
            types.CallToolResult,
            request_read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
        )

        if not result.is_error:
            await self._validate_tool_result(name, result)

        return result

    async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
        """Validate the structured content of a tool result against its output 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)

    async def get_prompt(
        self,
        name: str,
        arguments: dict[str, str] | None = None,
        *,
        meta: RequestParamsMeta | None = None,
    ) -> types.GetPromptResult:
        """Send a prompts/get request."""
        return await self.send_request(
            types.GetPromptRequest(params=types.GetPromptRequestParams(name=name, arguments=arguments, _meta=meta)),
            types.GetPromptResult,
        )

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

        # Cache tool output schemas for future validation
        # Note: don't clear the cache, as we may be using a cursor
        for tool in result.tools:
            self._tool_output_schemas[tool.name] = tool.output_schema

        return result

    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."""
        if method not in _SERVER_REQUEST_METHODS:
            raise MCPError(code=types.METHOD_NOT_FOUND, message="Method not found", data=method)
        payload: dict[str, Any] = {"method": method}
        if params is not None:
            payload["params"] = dict(params)
        request = types.server_request_adapter.validate_python(payload, by_name=False)

        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
            )
            match request:
                case types.CreateMessageRequest(params=sampling_params):
                    response = await self._sampling_callback(ctx, sampling_params)
                case types.ElicitRequest(params=elicit_params):
                    response = await self._elicitation_callback(ctx, elicit_params)
                case types.ListRootsRequest():  # pragma: no branch
                    response = await self._list_roots_callback(ctx)
        client_response = ClientResponse.validate_python(response)
        if isinstance(client_response, types.ErrorData):
            raise MCPError.from_error_data(client_response)
        return client_response.model_dump(by_alias=True, mode="json", exclude_none=True)

    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."""
        payload: dict[str, Any] = {"method": method}
        if params is not None:
            payload["params"] = dict(params)
        try:
            notification = types.server_notification_adapter.validate_python(payload, by_name=False)
        except ValidationError:
            logger.warning("Failed to validate notification: %s", payload, 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,
    result_type: type[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.

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

async def send_request(
    self,
    request: types.ClientRequest,
    result_type: type[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.
    """
    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
    method: str = data["method"]
    opts: CallOptions = {}
    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
    if method == "initialize":
        # The spec forbids cancelling initialize.
        opts["cancel_on_abandon"] = False
    raw = await self._dispatcher.send_raw_request(method, data.get("params"), opts)
    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)
    await self._dispatcher.notify(data["method"], data.get("params"))

initialize_result property

initialize_result: InitializeResult | None

The server's InitializeResult. None until initialize() has been called.

Contains server_info, capabilities, instructions, and the negotiated protocol_version.

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

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

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, *, meta: RequestParamsMeta | None = None
) -> ReadResourceResult

Send a resources/read request.

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

async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
    """Send a resources/read request."""
    return await self.send_request(
        types.ReadResourceRequest(params=types.ReadResourceRequestParams(uri=uri, _meta=meta)),
        types.ReadResourceResult,
    )

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,
    *,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

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

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,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.CallToolResult:
    """Send a tools/call request with optional progress callback support."""

    result = await self.send_request(
        types.CallToolRequest(
            params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
        ),
        types.CallToolResult,
        request_read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
    )

    if not result.is_error:
        await self._validate_tool_result(name, result)

    return result

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,
    *,
    meta: RequestParamsMeta | None = None
) -> GetPromptResult

Send a prompts/get request.

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

async def get_prompt(
    self,
    name: str,
    arguments: dict[str, str] | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.GetPromptResult:
    """Send a prompts/get request."""
    return await self.send_request(
        types.GetPromptRequest(params=types.GetPromptRequestParams(name=name, arguments=arguments, _meta=meta)),
        types.GetPromptResult,
    )

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

    # Cache tool output schemas for future validation
    # Note: don't clear the cache, as we may be using a cursor
    for tool in result.tools:
        self._tool_output_schemas[tool.name] = tool.output_schema

    return result

send_roots_list_changed async

send_roots_list_changed() -> None

Send a roots/list_changed notification.

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

async def send_roots_list_changed(self) -> None:
    """Send a roots/list_changed notification."""
    await self.send_notification(types.RootsListChangedNotification())

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

    async def call_tool(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: float | None = None,
        progress_callback: ProgressFnT | None = None,
        *,
        meta: types.RequestParamsMeta | None = None,
    ) -> types.CallToolResult:
        """Executes a tool given its name and arguments."""
        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,
            meta=meta,
        )

    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,
    *,
    meta: RequestParamsMeta | None = None
) -> CallToolResult

Executes a tool given its name and arguments.

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

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

Connection-scoped proxy for server-to-client requests and notifications.

send_request / send_notification model-dump their argument and forward to the dispatcher; the typed helpers below are unchanged from the previous implementation and only call those two methods.

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

class ServerSession:
    """Connection-scoped proxy for server-to-client requests and notifications.

    `send_request` / `send_notification` model-dump their argument and forward
    to the dispatcher; the typed helpers below are unchanged from the previous
    implementation and only call those two methods.
    """

    def __init__(
        self,
        dispatcher: JSONRPCDispatcher[Any],
        connection: Connection,
        *,
        stateless: bool = False,
    ) -> None:
        self._dispatcher = dispatcher
        self._connection = connection
        self._stateless = stateless

    @property
    def client_params(self) -> types.InitializeRequestParams | None:
        """The client's `initialize` request params; `None` before initialization."""
        return self._connection.client_params

    @property
    def protocol_version(self) -> str | None:
        """The protocol version negotiated during `initialize`.

        `None` before initialization completes. Stateless connections don't
        require the handshake, so this is normally `None` there (on streamable
        HTTP the per-request version is the `MCP-Protocol-Version` header,
        available via `ctx.request.headers`).
        """
        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.

        `metadata.related_request_id` (when supplied) routes the outgoing
        message onto the originating request's response stream over
        streamable HTTP; it is the only metadata field honored here.

        Raises:
            MCPError: The peer responded with an error.
            NoBackChannelError: If there is no related request to ride on and
                the connection has no standalone channel (stateless HTTP), so
                a response could never arrive.
            pydantic.ValidationError: The peer's result does not match `result_type`.
        """
        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
        related = metadata.related_request_id if metadata is not None else None
        if related is None and not self._connection.has_standalone_channel:
            # Fail fast instead of parking forever on a response that cannot
            # arrive; matches `Connection.send_raw_request`.
            raise NoBackChannelError(data["method"])
        result = await self._dispatcher.send_raw_request(
            data["method"], data.get("params"), opts or None, _related_request_id=related
        )
        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."""
        data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
        await self._dispatcher.notify(data["method"], data.get("params"), _related_request_id=related_request_id)

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

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

    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.
            StatelessModeNotSupported: If called in stateless HTTP mode.
        """
        if self._stateless:
            raise StatelessModeNotSupported(method="sampling")
        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,
        )

    async def list_roots(self) -> types.ListRootsResult:
        """Send a roots/list request."""
        if self._stateless:
            raise StatelessModeNotSupported(method="list_roots")
        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:
            StatelessModeNotSupported: If called in stateless HTTP mode.
        """
        if self._stateless:
            raise StatelessModeNotSupported(method="elicitation")
        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:
            StatelessModeNotSupported: If called in stateless HTTP mode.
        """
        if self._stateless:
            raise StatelessModeNotSupported(method="elicitation")
        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 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 before initialization.

protocol_version property

protocol_version: str | None

The protocol version negotiated during initialize.

None before initialization completes. Stateless connections don't require the handshake, so this is normally None there (on streamable HTTP the per-request version is the MCP-Protocol-Version header, available via ctx.request.headers).

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.

metadata.related_request_id (when supplied) routes the outgoing message onto the originating request's response stream over streamable HTTP; it is the only metadata field honored here.

Raises:

Type	Description
MCPError	The peer responded with an error.
NoBackChannelError	If there is no related request to ride on and the connection has no standalone channel (stateless HTTP), so a response could never arrive.
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.

    `metadata.related_request_id` (when supplied) routes the outgoing
    message onto the originating request's response stream over
    streamable HTTP; it is the only metadata field honored here.

    Raises:
        MCPError: The peer responded with an error.
        NoBackChannelError: If there is no related request to ride on and
            the connection has no standalone channel (stateless HTTP), so
            a response could never arrive.
        pydantic.ValidationError: The peer's result does not match `result_type`.
    """
    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
    related = metadata.related_request_id if metadata is not None else None
    if related is None and not self._connection.has_standalone_channel:
        # Fail fast instead of parking forever on a response that cannot
        # arrive; matches `Connection.send_raw_request`.
        raise NoBackChannelError(data["method"])
    result = await self._dispatcher.send_raw_request(
        data["method"], data.get("params"), opts or None, _related_request_id=related
    )
    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."""
    data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
    await self._dispatcher.notify(data["method"], data.get("params"), _related_request_id=related_request_id)

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

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.
StatelessModeNotSupported	If called in stateless HTTP mode.

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

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.
        StatelessModeNotSupported: If called in stateless HTTP mode.
    """
    if self._stateless:
        raise StatelessModeNotSupported(method="sampling")
    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.

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

async def list_roots(self) -> types.ListRootsResult:
    """Send a roots/list request."""
    if self._stateless:
        raise StatelessModeNotSupported(method="list_roots")
    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
StatelessModeNotSupported	If called in stateless HTTP mode.

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:
        StatelessModeNotSupported: If called in stateless HTTP mode.
    """
    if self._stateless:
        raise StatelessModeNotSupported(method="elicitation")
    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
StatelessModeNotSupported	If called in stateless HTTP mode.

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:
        StatelessModeNotSupported: If called in stateless HTTP mode.
    """
    if self._stateless:
        raise StatelessModeNotSupported(method="elicitation")
    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,
    )

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

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  # pragma: no cover

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

CallToolRequest

Bases: Request[CallToolRequestParams, Literal['tools/call']]

Used by the client to invoke a tool provided by the server.

Source code in src/mcp/types/_types.py

class CallToolRequest(Request[CallToolRequestParams, Literal["tools/call"]]):
    """Used by the client to invoke a tool provided by the server."""

    method: Literal["tools/call"] = "tools/call"
    params: CallToolRequestParams

ClientCapabilities

Bases: MCPModel

Capabilities a client may support.

Source code in src/mcp/types/_types.py

class ClientCapabilities(MCPModel):
    """Capabilities a client may support."""

    experimental: dict[str, dict[str, Any]] | None = None
    """Experimental, non-standard capabilities that the client supports."""
    sampling: SamplingCapability | None = None
    """
    Present if the client supports sampling from an LLM.
    Can contain fine-grained capabilities like context and tools support.
    """
    elicitation: ElicitationCapability | None = None
    """Present if the client supports elicitation from the user."""
    roots: RootsCapability | None = None
    """Present if the client supports listing roots."""

experimental class-attribute instance-attribute

experimental: dict[str, dict[str, Any]] | None = None

Experimental, non-standard capabilities that the client supports.

sampling class-attribute instance-attribute

sampling: SamplingCapability | None = None

Present if the client supports sampling from an LLM. Can contain fine-grained capabilities like context and tools support.

elicitation class-attribute instance-attribute

elicitation: ElicitationCapability | None = None

Present if the client supports elicitation from the user.

roots class-attribute instance-attribute

roots: RootsCapability | None = None

Present if the client supports listing roots.

CompleteRequest

Bases: Request[CompleteRequestParams, Literal['completion/complete']]

A request from the client to the server, to ask for completion options.

Source code in src/mcp/types/_types.py

class CompleteRequest(Request[CompleteRequestParams, Literal["completion/complete"]]):
    """A request from the client to the server, to ask for completion options."""

    method: Literal["completion/complete"] = "completion/complete"
    params: CompleteRequestParams

CreateMessageRequest

Bases: Request[CreateMessageRequestParams, Literal['sampling/createMessage']]

A request from the server to sample an LLM via the client.

Source code in src/mcp/types/_types.py

class CreateMessageRequest(Request[CreateMessageRequestParams, Literal["sampling/createMessage"]]):
    """A request from the server to sample an LLM via the client."""

    method: Literal["sampling/createMessage"] = "sampling/createMessage"
    params: CreateMessageRequestParams

CreateMessageResult

Bases: Result

The client's response to a sampling/createMessage request from the server.

This is the backwards-compatible version that returns single content (no arrays). Used when the request does not include tools.

Source code in src/mcp/types/_types.py

class CreateMessageResult(Result):
    """The client's response to a sampling/createMessage request from the server.

    This is the backwards-compatible version that returns single content (no arrays).
    Used when the request does not include tools.
    """

    role: Role
    """The role of the message sender (typically 'assistant' for LLM responses)."""
    content: SamplingContent
    """Response content. Single content block (text, image, or audio)."""
    model: str
    """The name of the model that generated the message."""
    stop_reason: StopReason | None = None
    """The reason why sampling stopped, if known."""

role instance-attribute

role: Role

The role of the message sender (typically 'assistant' for LLM responses).

content instance-attribute

content: SamplingContent

Response content. Single content block (text, image, or audio).

model instance-attribute

model: str

The name of the model that generated the message.

stop_reason class-attribute instance-attribute

stop_reason: StopReason | None = None

The reason why sampling stopped, if known.

CreateMessageResultWithTools

Bases: Result

The client's response to a sampling/createMessage request when tools were provided.

This version supports array content for tool use flows.

Source code in src/mcp/types/_types.py

class CreateMessageResultWithTools(Result):
    """The client's response to a sampling/createMessage request when tools were provided.

    This version supports array content for tool use flows.
    """

    role: Role
    """The role of the message sender (typically 'assistant' for LLM responses)."""
    content: SamplingMessageContentBlock | list[SamplingMessageContentBlock]
    """
    Response content. May be a single content block or an array.
    May include ToolUseContent if stop_reason is 'toolUse'.
    """
    model: str
    """The name of the model that generated the message."""
    stop_reason: StopReason | None = None
    """
    The reason why sampling stopped, if known.
    'toolUse' indicates the model wants to use a tool.
    """

    @property
    def content_as_list(self) -> list[SamplingMessageContentBlock]:
        """Returns the content as a list of content blocks, regardless of whether
        it was originally a single block or a list."""
        return self.content if isinstance(self.content, list) else [self.content]

role instance-attribute

role: Role

The role of the message sender (typically 'assistant' for LLM responses).

content instance-attribute

content: (
    SamplingMessageContentBlock
    | list[SamplingMessageContentBlock]
)

Response content. May be a single content block or an array. May include ToolUseContent if stop_reason is 'toolUse'.

model instance-attribute

model: str

The name of the model that generated the message.

stop_reason class-attribute instance-attribute

stop_reason: StopReason | None = None

The reason why sampling stopped, if known. 'toolUse' indicates the model wants to use a tool.

content_as_list property

content_as_list: list[SamplingMessageContentBlock]

Returns the content as a list of content blocks, regardless of whether it was originally a single block or a list.

ErrorData

Bases: BaseModel

Error information for JSON-RPC error responses.

Source code in src/mcp/types/jsonrpc.py

class ErrorData(BaseModel):
    """Error information for JSON-RPC error responses."""

    code: int
    """The error type that occurred."""

    message: str
    """A short description of the error.

    The message SHOULD be limited to a concise single sentence.
    """

    data: Any = None
    """Additional information about the error.

    The value of this member is defined by the sender (e.g. detailed error information, nested errors, etc.).
    """

code instance-attribute

code: int

The error type that occurred.

message instance-attribute

message: str

A short description of the error.

The message SHOULD be limited to a concise single sentence.

data class-attribute instance-attribute

data: Any = None

Additional information about the error.

The value of this member is defined by the sender (e.g. detailed error information, nested errors, etc.).

GetPromptRequest

Bases: Request[GetPromptRequestParams, Literal['prompts/get']]

Used by the client to get a prompt provided by the server.

Source code in src/mcp/types/_types.py

class GetPromptRequest(Request[GetPromptRequestParams, Literal["prompts/get"]]):
    """Used by the client to get a prompt provided by the server."""

    method: Literal["prompts/get"] = "prompts/get"
    params: GetPromptRequestParams

GetPromptResult

Bases: Result

The server's response to a prompts/get request from the client.

Source code in src/mcp/types/_types.py

class GetPromptResult(Result):
    """The server's response to a prompts/get request from the client."""

    description: str | None = None
    """An optional description for the prompt."""
    messages: list[PromptMessage]

description class-attribute instance-attribute

description: str | None = None

An optional description for the prompt.

Implementation

Bases: BaseMetadata

Describes the name and version of an MCP implementation.

Source code in src/mcp/types/_types.py

class Implementation(BaseMetadata):
    """Describes the name and version of an MCP implementation."""

    version: str

    title: str | None = None
    """An optional human-readable title for this implementation."""

    description: str | None = None
    """An optional human-readable description of what this implementation does."""

    website_url: str | None = None
    """An optional URL of the website for this implementation."""

    icons: list[Icon] | None = None
    """An optional list of icons for this implementation."""

title class-attribute instance-attribute

title: str | None = None

An optional human-readable title for this implementation.

description class-attribute instance-attribute

description: str | None = None

An optional human-readable description of what this implementation does.

website_url class-attribute instance-attribute

website_url: str | None = None

An optional URL of the website for this implementation.

icons class-attribute instance-attribute

icons: list[Icon] | None = None

An optional list of icons for this implementation.

InitializedNotification

Bases: Notification[NotificationParams | None, Literal['notifications/initialized']]

This notification is sent from the client to the server after initialization has finished.

Source code in src/mcp/types/_types.py

class InitializedNotification(Notification[NotificationParams | None, Literal["notifications/initialized"]]):
    """This notification is sent from the client to the server after initialization has
    finished.
    """

    method: Literal["notifications/initialized"] = "notifications/initialized"
    params: NotificationParams | None = None

InitializeRequest

Bases: Request[InitializeRequestParams, Literal['initialize']]

This request is sent from the client to the server when it first connects, asking it to begin initialization.

Source code in src/mcp/types/_types.py

class InitializeRequest(Request[InitializeRequestParams, Literal["initialize"]]):
    """This request is sent from the client to the server when it first connects, asking it
    to begin initialization.
    """

    method: Literal["initialize"] = "initialize"
    params: InitializeRequestParams

InitializeResult

Bases: Result

After receiving an initialize request from the client, the server sends this.

Source code in src/mcp/types/_types.py

class InitializeResult(Result):
    """After receiving an initialize request from the client, the server sends this."""

    protocol_version: str
    """The version of the Model Context Protocol that the server wants to use."""
    capabilities: ServerCapabilities
    server_info: Implementation
    instructions: str | None = None
    """Instructions describing how to use the server and its features."""

protocol_version instance-attribute

protocol_version: str

The version of the Model Context Protocol that the server wants to use.

instructions class-attribute instance-attribute

instructions: str | None = None

Instructions describing how to use the server and its features.

JSONRPCError

Bases: BaseModel

A response to a request that indicates an error occurred.

Source code in src/mcp/types/jsonrpc.py

class JSONRPCError(BaseModel):
    """A response to a request that indicates an error occurred."""

    jsonrpc: Literal["2.0"]
    id: RequestId | None
    error: ErrorData

JSONRPCRequest

Bases: BaseModel

A JSON-RPC request that expects a response.

Source code in src/mcp/types/jsonrpc.py

class JSONRPCRequest(BaseModel):
    """A JSON-RPC request that expects a response."""

    jsonrpc: Literal["2.0"]
    id: RequestId
    method: str
    params: dict[str, Any] | None = None

JSONRPCResponse

Bases: BaseModel

A successful (non-error) response to a request.

Source code in src/mcp/types/jsonrpc.py

class JSONRPCResponse(BaseModel):
    """A successful (non-error) response to a request."""

    jsonrpc: Literal["2.0"]
    id: RequestId
    result: dict[str, Any]

ListPromptsRequest

Bases: PaginatedRequest[Literal['prompts/list']]

Sent from the client to request a list of prompts and prompt templates.

Source code in src/mcp/types/_types.py

class ListPromptsRequest(PaginatedRequest[Literal["prompts/list"]]):
    """Sent from the client to request a list of prompts and prompt templates."""

    method: Literal["prompts/list"] = "prompts/list"

ListPromptsResult

Bases: PaginatedResult

The server's response to a prompts/list request from the client.

Source code in src/mcp/types/_types.py

class ListPromptsResult(PaginatedResult):
    """The server's response to a prompts/list request from the client."""

    prompts: list[Prompt]

ListResourcesRequest

Bases: PaginatedRequest[Literal['resources/list']]

Sent from the client to request a list of resources the server has.

Source code in src/mcp/types/_types.py

class ListResourcesRequest(PaginatedRequest[Literal["resources/list"]]):
    """Sent from the client to request a list of resources the server has."""

    method: Literal["resources/list"] = "resources/list"

ListResourcesResult

Bases: PaginatedResult

The server's response to a resources/list request from the client.

Source code in src/mcp/types/_types.py

class ListResourcesResult(PaginatedResult):
    """The server's response to a resources/list request from the client."""

    resources: list[Resource]

ListToolsResult

Bases: PaginatedResult

The server's response to a tools/list request from the client.

Source code in src/mcp/types/_types.py

class ListToolsResult(PaginatedResult):
    """The server's response to a tools/list request from the client."""

    tools: list[Tool]

LoggingMessageNotification

Bases: Notification[LoggingMessageNotificationParams, Literal['notifications/message']]

Notification of a log message passed from server to client.

Source code in src/mcp/types/_types.py

class LoggingMessageNotification(Notification[LoggingMessageNotificationParams, Literal["notifications/message"]]):
    """Notification of a log message passed from server to client."""

    method: Literal["notifications/message"] = "notifications/message"
    params: LoggingMessageNotificationParams

Notification

Bases: MCPModel, Generic[NotificationParamsT, MethodT]

Base class for JSON-RPC notifications.

Source code in src/mcp/types/_types.py

class Notification(MCPModel, Generic[NotificationParamsT, MethodT]):
    """Base class for JSON-RPC notifications."""

    method: MethodT
    params: NotificationParamsT

PingRequest

Bases: Request[RequestParams | None, Literal['ping']]

A ping, issued by either the server or the client, to check that the other party is still alive.

Source code in src/mcp/types/_types.py

class PingRequest(Request[RequestParams | None, Literal["ping"]]):
    """A ping, issued by either the server or the client, to check that the other party is
    still alive.
    """

    method: Literal["ping"] = "ping"
    params: RequestParams | None = None

ProgressNotification

Bases: Notification[ProgressNotificationParams, Literal['notifications/progress']]

An out-of-band notification used to inform the receiver of a progress update for a long-running request.

Source code in src/mcp/types/_types.py

class ProgressNotification(Notification[ProgressNotificationParams, Literal["notifications/progress"]]):
    """An out-of-band notification used to inform the receiver of a progress update for a long-running request."""

    method: Literal["notifications/progress"] = "notifications/progress"
    params: ProgressNotificationParams

PromptsCapability

Bases: MCPModel

Capability for prompts operations.

Source code in src/mcp/types/_types.py

class PromptsCapability(MCPModel):
    """Capability for prompts operations."""

    list_changed: bool | None = None
    """Whether this server supports notifications for changes to the prompt list."""

list_changed class-attribute instance-attribute

list_changed: bool | None = None

Whether this server supports notifications for changes to the prompt list.

ReadResourceRequest

Bases: Request[ReadResourceRequestParams, Literal['resources/read']]

Sent from the client to the server, to read a specific resource URI.

Source code in src/mcp/types/_types.py

class ReadResourceRequest(Request[ReadResourceRequestParams, Literal["resources/read"]]):
    """Sent from the client to the server, to read a specific resource URI."""

    method: Literal["resources/read"] = "resources/read"
    params: ReadResourceRequestParams

ReadResourceResult

Bases: Result

The server's response to a resources/read request from the client.

Source code in src/mcp/types/_types.py

class ReadResourceResult(Result):
    """The server's response to a resources/read request from the client."""

    contents: list[TextResourceContents | BlobResourceContents]

Resource

Bases: BaseMetadata

A known resource that the server is capable of reading.

Source code in src/mcp/types/_types.py

class Resource(BaseMetadata):
    """A known resource that the server is capable of reading."""

    uri: str
    """The URI of this resource."""

    description: str | None = None
    """A description of what this resource represents."""

    mime_type: str | None = None
    """The MIME type of this resource, if known."""

    size: int | None = None
    """The size of the raw resource content, in bytes (i.e., before base64 encoding or any tokenization), if known.

    This can be used by Hosts to display file sizes and estimate context window usage.
    """

    icons: list[Icon] | None = None
    """An optional list of icons for this resource."""

    annotations: Annotations | None = None

    meta: Meta | None = Field(alias="_meta", default=None)
    """
    See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
    for notes on _meta usage.
    """

uri instance-attribute

uri: str

The URI of this resource.

description class-attribute instance-attribute

description: str | None = None

A description of what this resource represents.

mime_type class-attribute instance-attribute

mime_type: str | None = None

The MIME type of this resource, if known.

size class-attribute instance-attribute

size: int | None = None

The size of the raw resource content, in bytes (i.e., before base64 encoding or any tokenization), if known.

This can be used by Hosts to display file sizes and estimate context window usage.

icons class-attribute instance-attribute

icons: list[Icon] | None = None

An optional list of icons for this resource.

meta class-attribute instance-attribute

meta: Meta | None = Field(alias='_meta', default=None)

See MCP specification for notes on _meta usage.

ResourcesCapability

Bases: MCPModel

Capability for resources operations.

Source code in src/mcp/types/_types.py

class ResourcesCapability(MCPModel):
    """Capability for resources operations."""

    subscribe: bool | None = None
    """Whether this server supports subscribing to resource updates."""
    list_changed: bool | None = None
    """Whether this server supports notifications for changes to the resource list."""

subscribe class-attribute instance-attribute

subscribe: bool | None = None

Whether this server supports subscribing to resource updates.

list_changed class-attribute instance-attribute

list_changed: bool | None = None

Whether this server supports notifications for changes to the resource list.

ResourceUpdatedNotification

Bases: Notification[ResourceUpdatedNotificationParams, Literal['notifications/resources/updated']]

A notification from the server to the client, informing it that a resource has changed and may need to be read again.

Source code in src/mcp/types/_types.py

class ResourceUpdatedNotification(
    Notification[ResourceUpdatedNotificationParams, Literal["notifications/resources/updated"]]
):
    """A notification from the server to the client, informing it that a resource has
    changed and may need to be read again.
    """

    method: Literal["notifications/resources/updated"] = "notifications/resources/updated"
    params: ResourceUpdatedNotificationParams

RootsCapability

Bases: MCPModel

Capability for root operations.

Source code in src/mcp/types/_types.py

class RootsCapability(MCPModel):
    """Capability for root operations."""

    list_changed: bool | None = None
    """Whether the client supports notifications for changes to the roots list."""

list_changed class-attribute instance-attribute

list_changed: bool | None = None

Whether the client supports notifications for changes to the roots list.

SamplingCapability

Bases: MCPModel

Sampling capability structure, allowing fine-grained capability advertisement.

Source code in src/mcp/types/_types.py

class SamplingCapability(MCPModel):
    """Sampling capability structure, allowing fine-grained capability advertisement."""

    context: SamplingContextCapability | None = None
    """
    Present if the client supports non-'none' values for includeContext parameter.
    SOFT-DEPRECATED: New implementations should use tools parameter instead.
    """
    tools: SamplingToolsCapability | None = None
    """
    Present if the client supports tools and toolChoice parameters in sampling requests.
    Presence indicates full tool calling support during sampling.
    """

context class-attribute instance-attribute

context: SamplingContextCapability | None = None

Present if the client supports non-'none' values for includeContext parameter. SOFT-DEPRECATED: New implementations should use tools parameter instead.

tools class-attribute instance-attribute

tools: SamplingToolsCapability | None = None

Present if the client supports tools and toolChoice parameters in sampling requests. Presence indicates full tool calling support during sampling.

SamplingContent module-attribute

SamplingContent: TypeAlias = (
    TextContent | ImageContent | AudioContent
)

Basic content types for sampling responses (without tool use).

Used for backwards-compatible CreateMessageResult when tools are not used.

SamplingContextCapability

Bases: MCPModel

Capability for context inclusion during sampling.

Indicates support for non-'none' values in the includeContext parameter. SOFT-DEPRECATED: New implementations should use tools parameter instead.

Source code in src/mcp/types/_types.py

class SamplingContextCapability(MCPModel):
    """Capability for context inclusion during sampling.

    Indicates support for non-'none' values in the includeContext parameter.
    SOFT-DEPRECATED: New implementations should use tools parameter instead.
    """

SamplingMessage

Bases: MCPModel

Describes a message issued to or received from an LLM API.

Source code in src/mcp/types/_types.py

class SamplingMessage(MCPModel):
    """Describes a message issued to or received from an LLM API."""

    role: Role
    content: SamplingMessageContentBlock | list[SamplingMessageContentBlock]
    """
    Message content. Can be a single content block or an array of content blocks
    for multi-modal messages and tool interactions.
    """
    meta: Meta | None = Field(alias="_meta", default=None)
    """
    See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
    for notes on _meta usage.
    """

    @property
    def content_as_list(self) -> list[SamplingMessageContentBlock]:
        """Returns the content as a list of content blocks, regardless of whether
        it was originally a single block or a list."""
        return self.content if isinstance(self.content, list) else [self.content]

content instance-attribute

content: (
    SamplingMessageContentBlock
    | list[SamplingMessageContentBlock]
)

Message content. Can be a single content block or an array of content blocks for multi-modal messages and tool interactions.

meta class-attribute instance-attribute

meta: Meta | None = Field(alias='_meta', default=None)

See MCP specification for notes on _meta usage.

content_as_list property

content_as_list: list[SamplingMessageContentBlock]

Returns the content as a list of content blocks, regardless of whether it was originally a single block or a list.

SamplingMessageContentBlock module-attribute

SamplingMessageContentBlock: TypeAlias = (
    TextContent
    | ImageContent
    | AudioContent
    | ToolUseContent
    | ToolResultContent
)

Content block types allowed in sampling messages.

SamplingToolsCapability

Bases: MCPModel

Capability indicating support for tool calling during sampling.

When present in ClientCapabilities.sampling, indicates that the client supports the tools and toolChoice parameters in sampling requests.

Source code in src/mcp/types/_types.py

class SamplingToolsCapability(MCPModel):
    """Capability indicating support for tool calling during sampling.

    When present in ClientCapabilities.sampling, indicates that the client
    supports the tools and toolChoice parameters in sampling requests.
    """

ServerCapabilities

Bases: MCPModel

Capabilities that a server may support.

Source code in src/mcp/types/_types.py

class ServerCapabilities(MCPModel):
    """Capabilities that a server may support."""

    experimental: dict[str, dict[str, Any]] | None = None
    """Experimental, non-standard capabilities that the server supports."""

    logging: LoggingCapability | None = None
    """Present if the server supports sending log messages to the client."""

    prompts: PromptsCapability | None = None
    """Present if the server offers any prompt templates."""

    resources: ResourcesCapability | None = None
    """Present if the server offers any resources to read."""

    tools: ToolsCapability | None = None
    """Present if the server offers any tools to call."""

    completions: CompletionsCapability | None = None
    """Present if the server offers autocompletion suggestions for prompts and resources."""

experimental class-attribute instance-attribute

experimental: dict[str, dict[str, Any]] | None = None

Experimental, non-standard capabilities that the server supports.

logging class-attribute instance-attribute

logging: LoggingCapability | None = None

Present if the server supports sending log messages to the client.

prompts class-attribute instance-attribute

prompts: PromptsCapability | None = None

Present if the server offers any prompt templates.

resources class-attribute instance-attribute

resources: ResourcesCapability | None = None

Present if the server offers any resources to read.

tools class-attribute instance-attribute

tools: ToolsCapability | None = None

Present if the server offers any tools to call.

completions class-attribute instance-attribute

completions: CompletionsCapability | None = None

Present if the server offers autocompletion suggestions for prompts and resources.

SetLevelRequest

Bases: Request[SetLevelRequestParams, Literal['logging/setLevel']]

A request from the client to the server, to enable or adjust logging.

Source code in src/mcp/types/_types.py

class SetLevelRequest(Request[SetLevelRequestParams, Literal["logging/setLevel"]]):
    """A request from the client to the server, to enable or adjust logging."""

    method: Literal["logging/setLevel"] = "logging/setLevel"
    params: SetLevelRequestParams

SubscribeRequest

Bases: Request[SubscribeRequestParams, Literal['resources/subscribe']]

Sent from the client to request resources/updated notifications from the server whenever a particular resource changes.

Source code in src/mcp/types/_types.py

class SubscribeRequest(Request[SubscribeRequestParams, Literal["resources/subscribe"]]):
    """Sent from the client to request resources/updated notifications from the server
    whenever a particular resource changes.
    """

    method: Literal["resources/subscribe"] = "resources/subscribe"
    params: SubscribeRequestParams

Tool

Bases: BaseMetadata

Definition for a tool the client can call.

Source code in src/mcp/types/_types.py

class Tool(BaseMetadata):
    """Definition for a tool the client can call."""

    description: str | None = None
    """A human-readable description of the tool."""
    input_schema: dict[str, Any]
    """A JSON Schema object defining the expected parameters for the tool."""
    output_schema: dict[str, Any] | None = None
    """
    An optional JSON Schema object defining the structure of the tool's output
    returned in the structured_content field of a CallToolResult.
    """
    icons: list[Icon] | None = None
    """An optional list of icons for this tool."""
    annotations: ToolAnnotations | None = None
    """Optional additional tool information."""
    meta: Meta | None = Field(alias="_meta", default=None)
    """
    See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
    for notes on _meta usage.
    """

description class-attribute instance-attribute

description: str | None = None

A human-readable description of the tool.

input_schema instance-attribute

input_schema: dict[str, Any]

A JSON Schema object defining the expected parameters for the tool.

output_schema class-attribute instance-attribute

output_schema: dict[str, Any] | None = None

An optional JSON Schema object defining the structure of the tool's output returned in the structured_content field of a CallToolResult.

icons class-attribute instance-attribute

icons: list[Icon] | None = None

An optional list of icons for this tool.

annotations class-attribute instance-attribute

annotations: ToolAnnotations | None = None

Optional additional tool information.

meta class-attribute instance-attribute

meta: Meta | None = Field(alias='_meta', default=None)

See MCP specification for notes on _meta usage.

ToolChoice

Bases: MCPModel

Controls tool usage behavior during sampling.

Allows the server to specify whether and how the LLM should use tools in its response.

Source code in src/mcp/types/_types.py

class ToolChoice(MCPModel):
    """Controls tool usage behavior during sampling.

    Allows the server to specify whether and how the LLM should use tools
    in its response.
    """

    mode: Literal["auto", "required", "none"] | None = None
    """
    Controls when tools are used:
    - "auto": Model decides whether to use tools (default)
    - "required": Model MUST use at least one tool before completing
    - "none": Model should not use tools
    """

mode class-attribute instance-attribute

mode: Literal['auto', 'required', 'none'] | None = None

Controls when tools are used: - "auto": Model decides whether to use tools (default) - "required": Model MUST use at least one tool before completing - "none": Model should not use tools

ToolResultContent

Bases: MCPModel

Content representing the result of a tool execution.

This content type appears in user messages as a response to a ToolUseContent from the assistant. It contains the output of executing the requested tool.

Source code in src/mcp/types/_types.py

class ToolResultContent(MCPModel):
    """Content representing the result of a tool execution.

    This content type appears in user messages as a response to a ToolUseContent
    from the assistant. It contains the output of executing the requested tool.
    """

    type: Literal["tool_result"] = "tool_result"
    """Discriminator for tool result content."""

    tool_use_id: str
    """The unique identifier that corresponds to the tool call's id field."""

    content: list[ContentBlock] = []
    """
    A list of content objects representing the tool result.
    Defaults to empty list if not provided.
    """

    structured_content: dict[str, Any] | None = None
    """
    Optional structured tool output that matches the tool's outputSchema (if defined).
    """

    is_error: bool | None = None
    """Whether the tool execution resulted in an error."""

    meta: Meta | None = Field(alias="_meta", default=None)
    """
    See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
    for notes on _meta usage.
    """

type class-attribute instance-attribute

type: Literal['tool_result'] = 'tool_result'

Discriminator for tool result content.

tool_use_id instance-attribute

tool_use_id: str

The unique identifier that corresponds to the tool call's id field.

content class-attribute instance-attribute

content: list[ContentBlock] = []

A list of content objects representing the tool result. Defaults to empty list if not provided.

structured_content class-attribute instance-attribute

structured_content: dict[str, Any] | None = None

Optional structured tool output that matches the tool's outputSchema (if defined).

is_error class-attribute instance-attribute

is_error: bool | None = None

Whether the tool execution resulted in an error.

meta class-attribute instance-attribute

meta: Meta | None = Field(alias='_meta', default=None)

See MCP specification for notes on _meta usage.

ToolsCapability

Bases: MCPModel

Capability for tools operations.

Source code in src/mcp/types/_types.py

class ToolsCapability(MCPModel):
    """Capability for tools operations."""

    list_changed: bool | None = None
    """Whether this server supports notifications for changes to the tool list."""

list_changed class-attribute instance-attribute

list_changed: bool | None = None

Whether this server supports notifications for changes to the tool list.

ToolUseContent

Bases: MCPModel

Content representing an assistant's request to invoke a tool.

This content type appears in assistant messages when the LLM wants to call a tool during sampling. The server should execute the tool and return a ToolResultContent in the next user message.

Source code in src/mcp/types/_types.py

class ToolUseContent(MCPModel):
    """Content representing an assistant's request to invoke a tool.

    This content type appears in assistant messages when the LLM wants to call a tool
    during sampling. The server should execute the tool and return a ToolResultContent
    in the next user message.
    """

    type: Literal["tool_use"] = "tool_use"
    """Discriminator for tool use content."""

    name: str
    """The name of the tool to invoke. Must match a tool name from the request's tools array."""

    id: str
    """Unique identifier for this tool call, used to correlate with ToolResultContent."""

    input: dict[str, Any]
    """Arguments to pass to the tool. Must conform to the tool's inputSchema."""

    meta: Meta | None = Field(alias="_meta", default=None)
    """
    See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
    for notes on _meta usage.
    """

type class-attribute instance-attribute

type: Literal['tool_use'] = 'tool_use'

Discriminator for tool use content.

name instance-attribute

name: str

The name of the tool to invoke. Must match a tool name from the request's tools array.

id instance-attribute

id: str

Unique identifier for this tool call, used to correlate with ToolResultContent.

input instance-attribute

input: dict[str, Any]

Arguments to pass to the tool. Must conform to the tool's inputSchema.

meta class-attribute instance-attribute

meta: Meta | None = Field(alias='_meta', default=None)

See MCP specification for notes on _meta usage.

UnsubscribeRequest

Bases: Request[UnsubscribeRequestParams, Literal['resources/unsubscribe']]

Sent from the client to request cancellation of resources/updated notifications from the server.

Source code in src/mcp/types/_types.py

class UnsubscribeRequest(Request[UnsubscribeRequestParams, Literal["resources/unsubscribe"]]):
    """Sent from the client to request cancellation of resources/updated notifications from
    the server.
    """

    method: Literal["resources/unsubscribe"] = "resources/unsubscribe"
    params: UnsubscribeRequestParams