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

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

ClientSession

Bases: BaseSession[ClientRequest, ClientNotification, ClientResult, ServerRequest, ServerNotification]

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

class ClientSession(
    BaseSession[
        types.ClientRequest,
        types.ClientNotification,
        types.ClientResult,
        types.ServerRequest,
        types.ServerNotification,
    ]
):
    def __init__(
        self,
        read_stream: ReadStream[SessionMessage | Exception],
        write_stream: WriteStream[SessionMessage],
        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,
        experimental_task_handlers: ExperimentalTaskHandlers | None = None,
    ) -> None:
        super().__init__(read_stream, write_stream, 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._experimental_features: ExperimentalClientFeatures | None = None

        # Experimental: Task handlers (use defaults if not provided)
        self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers()

    @property
    def _receive_request_adapter(self) -> TypeAdapter[types.ServerRequest]:
        return types.server_request_adapter

    @property
    def _receive_notification_adapter(self) -> TypeAdapter[types.ServerNotification]:
        return types.server_notification_adapter

    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,
                        tasks=self._task_handlers.build_capability(),
                    ),
                    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

    @property
    def experimental(self) -> ExperimentalClientFeatures:
        """Experimental APIs for tasks and other features.

        !!! warning
            These APIs are experimental and may change without notice.

        Example:
            ```python
            status = await session.experimental.get_task(task_id)
            result = await session.experimental.get_task_result(task_id, CallToolResult)
            ```
        """
        if self._experimental_features is None:
            self._experimental_features = ExperimentalClientFeatures(self)
        return self._experimental_features

    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"
                )  # pragma: no cover
            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:  # pragma: no cover
        """Send a roots/list_changed notification."""
        await self.send_notification(types.RootsListChangedNotification())

    async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
        ctx = RequestContext[ClientSession](request_id=responder.request_id, meta=responder.request_meta, session=self)

        # Delegate to experimental task handler if applicable
        if self._task_handlers.handles_request(responder.request):
            with responder:
                await self._task_handlers.handle_request(ctx, responder)
            return None

        # Core request handling
        match responder.request:
            case types.CreateMessageRequest(params=params):
                with responder:
                    # Check if this is a task-augmented request
                    if params.task is not None:
                        response = await self._task_handlers.augmented_sampling(ctx, params, params.task)
                    else:
                        response = await self._sampling_callback(ctx, params)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.ElicitRequest(params=params):
                with responder:
                    # Check if this is a task-augmented request
                    if params.task is not None:
                        response = await self._task_handlers.augmented_elicitation(ctx, params, params.task)
                    else:
                        response = await self._elicitation_callback(ctx, params)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.ListRootsRequest():
                with responder:
                    response = await self._list_roots_callback(ctx)
                    client_response = ClientResponse.validate_python(response)
                    await responder.respond(client_response)

            case types.PingRequest():  # pragma: no cover
                with responder:
                    return await responder.respond(types.EmptyResult())

            case _:  # pragma: no cover
                pass  # Task requests handled above by _task_handlers

        return None

    async def _handle_incoming(
        self,
        req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
    ) -> None:
        """Handle incoming messages by forwarding to the message handler."""
        await self._message_handler(req)

    async def _received_notification(self, notification: types.ServerNotification) -> None:
        """Handle notifications from the server."""
        # Process specific notification types
        match notification:
            case types.LoggingMessageNotification(params=params):
                await self._logging_callback(params)
            case types.ElicitCompleteNotification(params=params):
                # Handle elicitation completion notification
                # Clients MAY use this to retry requests or update UI
                # The notification contains the elicitationId of the completed elicitation
                pass
            case _:
                pass

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.

experimental property

experimental: ExperimentalClientFeatures

Experimental APIs for tasks and other features.

Warning

These APIs are experimental and may change without notice.

Example

status = await session.experimental.get_task(task_id)
result = await session.experimental.get_task_result(task_id, CallToolResult)

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:  # pragma: no cover
    """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
    """
    The environment to use when spawning the process.

    If not specified, the result of get_default_environment() will be used.
    """

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

    encoding: str = "utf-8"
    """
    The text encoding used when sending/receiving messages to the server.

    Defaults to utf-8.
    """

    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
    """
    The text encoding error handler.

    See https://docs.python.org/3/library/codecs.html#codec-base-classes for
    explanations of possible values.
    """

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

The environment to use when spawning the process.

If not specified, the result of get_default_environment() will be used.

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'

The text encoding used when sending/receiving messages to the server.

Defaults to utf-8.

encoding_error_handler class-attribute instance-attribute

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

The text encoding error handler.

See https://docs.python.org/3/library/codecs.html#codec-base-classes for explanations of possible values.

stdio_client async

stdio_client(
    server: StdioServerParameters, errlog: TextIO = stderr
)

Client transport for stdio: this will connect to a server by spawning a process and communicating with it over stdin/stdout.

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

@asynccontextmanager
async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stderr):
    """Client transport for stdio: this will connect to a server by spawning a
    process and communicating with it over stdin/stdout.
    """
    read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
    read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]

    write_stream: MemoryObjectSendStream[SessionMessage]
    write_stream_reader: MemoryObjectReceiveStream[SessionMessage]

    read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

    try:
        command = _get_executable_command(server.command)

        # Open process with stderr piped for capture
        process = await _create_platform_compatible_process(
            command=command,
            args=server.args,
            env=({**get_default_environment(), **server.env} if server.env is not None else get_default_environment()),
            errlog=errlog,
            cwd=server.cwd,
        )
    except OSError:
        # Clean up streams if process creation fails
        await read_stream.aclose()
        await write_stream.aclose()
        await read_stream_writer.aclose()
        await write_stream_reader.aclose()
        raise

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

        try:
            async with read_stream_writer:
                buffer = ""
                async for chunk in TextReceiveStream(
                    process.stdout,
                    encoding=server.encoding,
                    errors=server.encoding_error_handler,
                ):
                    lines = (buffer + chunk).split("\n")
                    buffer = lines.pop()

                    for line in lines:
                        try:
                            message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
                        except Exception as exc:  # pragma: no cover
                            logger.exception("Failed to parse JSONRPC message from server")
                            await read_stream_writer.send(exc)
                            continue

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

    async def stdin_writer():
        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)
                    await process.stdin.send(
                        (json + "\n").encode(
                            encoding=server.encoding,
                            errors=server.encoding_error_handler,
                        )
                    )
        except anyio.ClosedResourceError:  # pragma: no cover
            await anyio.lowlevel.checkpoint()

    async with anyio.create_task_group() as tg, process:
        tg.start_soon(stdout_reader)
        tg.start_soon(stdin_writer)
        try:
            yield read_stream, write_stream
        finally:
            # MCP spec: stdio shutdown sequence
            # 1. Close input stream to server
            # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
            # 3. Send SIGKILL if still not exited
            if process.stdin:  # pragma: no branch
                try:
                    await process.stdin.aclose()
                except Exception:  # pragma: no cover
                    # stdin might already be closed, which is fine
                    pass

            try:
                # Give the process time to exit gracefully after stdin closes
                with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):
                    await process.wait()
            except TimeoutError:
                # Process didn't exit from stdin closure, use platform-specific termination
                # which handles SIGTERM -> SIGKILL escalation
                await _terminate_process_tree(process)
            except ProcessLookupError:  # pragma: no cover
                # Process already exited, which is fine
                pass
            await read_stream.aclose()
            await write_stream.aclose()
            await read_stream_writer.aclose()
            await write_stream_reader.aclose()

ServerSession

Bases: BaseSession[ServerRequest, ServerNotification, ServerResult, ClientRequest, ClientNotification]

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

class ServerSession(
    BaseSession[
        types.ServerRequest,
        types.ServerNotification,
        types.ServerResult,
        types.ClientRequest,
        types.ClientNotification,
    ]
):
    _initialized: InitializationState = InitializationState.NotInitialized
    _client_params: types.InitializeRequestParams | None = None
    _experimental_features: ExperimentalServerSessionFeatures | None = None

    def __init__(
        self,
        read_stream: ReadStream[SessionMessage | Exception],
        write_stream: WriteStream[SessionMessage],
        init_options: InitializationOptions,
        stateless: bool = False,
    ) -> None:
        super().__init__(read_stream, write_stream)
        self._stateless = stateless
        self._initialization_state = (
            InitializationState.Initialized if stateless else InitializationState.NotInitialized
        )

        self._init_options = init_options
        self._incoming_message_stream_writer, self._incoming_message_stream_reader = anyio.create_memory_object_stream[
            ServerRequestResponder
        ](0)
        self._exit_stack.push_async_callback(lambda: self._incoming_message_stream_reader.aclose())

    @property
    def _receive_request_adapter(self) -> TypeAdapter[types.ClientRequest]:
        return types.client_request_adapter

    @property
    def _receive_notification_adapter(self) -> TypeAdapter[types.ClientNotification]:
        return types.client_notification_adapter

    @property
    def client_params(self) -> types.InitializeRequestParams | None:
        return self._client_params

    @property
    def experimental(self) -> ExperimentalServerSessionFeatures:
        """Experimental APIs for server→client task operations.

        WARNING: These APIs are experimental and may change without notice.
        """
        if self._experimental_features is None:
            self._experimental_features = ExperimentalServerSessionFeatures(self)
        return self._experimental_features

    def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
        """Check if the client supports a specific capability."""
        if self._client_params is None:  # pragma: lax no cover
            return False

        client_caps = self._client_params.capabilities

        if capability.roots is not None:  # pragma: lax no cover
            if client_caps.roots is None:
                return False
            if capability.roots.list_changed and not client_caps.roots.list_changed:
                return False

        if capability.sampling is not None:  # pragma: lax no cover
            if client_caps.sampling is None:
                return False
            if capability.sampling.context is not None and client_caps.sampling.context is None:
                return False
            if capability.sampling.tools is not None and client_caps.sampling.tools is None:
                return False

        if capability.elicitation is not None and client_caps.elicitation is None:  # pragma: lax no cover
            return False

        if capability.experimental is not None:  # pragma: lax no cover
            if client_caps.experimental is None:
                return False
            for exp_key, exp_value in capability.experimental.items():
                if exp_key not in client_caps.experimental or client_caps.experimental[exp_key] != exp_value:
                    return False

        if capability.tasks is not None:  # pragma: lax no cover
            if client_caps.tasks is None:
                return False
            if not check_tasks_capability(capability.tasks, client_caps.tasks):
                return False

        return True

    async def _receive_loop(self) -> None:
        async with self._incoming_message_stream_writer:
            await super()._receive_loop()

    async def _received_request(self, responder: RequestResponder[types.ClientRequest, types.ServerResult]):
        match responder.request:
            case types.InitializeRequest(params=params):
                requested_version = params.protocol_version
                self._initialization_state = InitializationState.Initializing
                self._client_params = params
                with responder:
                    await responder.respond(
                        types.InitializeResult(
                            protocol_version=requested_version
                            if requested_version in SUPPORTED_PROTOCOL_VERSIONS
                            else types.LATEST_PROTOCOL_VERSION,
                            capabilities=self._init_options.capabilities,
                            server_info=types.Implementation(
                                name=self._init_options.server_name,
                                title=self._init_options.title,
                                description=self._init_options.description,
                                version=self._init_options.server_version,
                                website_url=self._init_options.website_url,
                                icons=self._init_options.icons,
                            ),
                            instructions=self._init_options.instructions,
                        )
                    )
                self._initialization_state = InitializationState.Initialized
            case types.PingRequest():
                # Ping requests are allowed at any time
                pass
            case _:
                if self._initialization_state != InitializationState.Initialized:
                    raise RuntimeError("Received request before initialization was complete")

    async def _received_notification(self, notification: types.ClientNotification) -> None:
        # Need this to avoid ASYNC910
        await anyio.lowlevel.checkpoint()
        match notification:
            case types.InitializedNotification():
                self._initialization_state = InitializationState.Initialized
            case _:
                if self._initialization_state != InitializationState.Initialized:  # pragma: no cover
                    raise RuntimeError("Received notification before initialization was complete")

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

        # Use different result types based on whether tools are provided
        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:  # pragma: no cover
        """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:  # pragma: no cover
        """Send a tool list changed notification."""
        await self.send_notification(types.ToolListChangedNotification())

    async def send_prompt_list_changed(self) -> None:  # pragma: no cover
        """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,
        )

    def _build_elicit_form_request(
        self,
        message: str,
        requested_schema: types.ElicitRequestedSchema,
        related_task_id: str | None = None,
        task: types.TaskMetadata | None = None,
    ) -> types.JSONRPCRequest:
        """Build a form mode elicitation request without sending it.

        Args:
            message: The message to present to the user
            requested_schema: Schema defining the expected response structure
            related_task_id: If provided, adds io.modelcontextprotocol/related-task metadata
            task: If provided, makes this a task-augmented request

        Returns:
            A JSONRPCRequest ready to be sent or queued
        """
        params = types.ElicitRequestFormParams(
            message=message,
            requested_schema=requested_schema,
            task=task,
        )
        params_data = params.model_dump(by_alias=True, mode="json", exclude_none=True)

        # Add related-task metadata if associated with a parent task
        if related_task_id is not None:
            # Defensive: model_dump() never includes _meta, but guard against future changes
            if "_meta" not in params_data:  # pragma: no branch
                params_data["_meta"] = {}
            params_data["_meta"][RELATED_TASK_METADATA_KEY] = types.RelatedTaskMetadata(
                task_id=related_task_id
            ).model_dump(by_alias=True)

        request_id = f"task-{related_task_id}-{id(params)}" if related_task_id else self._request_id
        if related_task_id is None:
            self._request_id += 1

        return types.JSONRPCRequest(
            jsonrpc="2.0",
            id=request_id,
            method="elicitation/create",
            params=params_data,
        )

    def _build_elicit_url_request(
        self,
        message: str,
        url: str,
        elicitation_id: str,
        related_task_id: str | None = None,
    ) -> types.JSONRPCRequest:
        """Build a URL mode elicitation request without sending it.

        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_task_id: If provided, adds io.modelcontextprotocol/related-task metadata

        Returns:
            A JSONRPCRequest ready to be sent or queued
        """
        params = types.ElicitRequestURLParams(
            message=message,
            url=url,
            elicitation_id=elicitation_id,
        )
        params_data = params.model_dump(by_alias=True, mode="json", exclude_none=True)

        # Add related-task metadata if associated with a parent task
        if related_task_id is not None:
            # Defensive: model_dump() never includes _meta, but guard against future changes
            if "_meta" not in params_data:  # pragma: no branch
                params_data["_meta"] = {}
            params_data["_meta"][RELATED_TASK_METADATA_KEY] = types.RelatedTaskMetadata(
                task_id=related_task_id
            ).model_dump(by_alias=True)

        request_id = f"task-{related_task_id}-{id(params)}" if related_task_id else self._request_id
        if related_task_id is None:
            self._request_id += 1

        return types.JSONRPCRequest(
            jsonrpc="2.0",
            id=request_id,
            method="elicitation/create",
            params=params_data,
        )

    def _build_create_message_request(
        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_task_id: str | None = None,
        task: types.TaskMetadata | None = None,
    ) -> types.JSONRPCRequest:
        """Build a sampling/createMessage request without sending it.

        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
            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
            tool_choice: Optional control over tool usage behavior
            related_task_id: If provided, adds io.modelcontextprotocol/related-task metadata
            task: If provided, makes this a task-augmented request

        Returns:
            A JSONRPCRequest ready to be sent or queued
        """
        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,
            task=task,
        )
        params_data = params.model_dump(by_alias=True, mode="json", exclude_none=True)

        # Add related-task metadata if associated with a parent task
        if related_task_id is not None:
            # Defensive: model_dump() never includes _meta, but guard against future changes
            if "_meta" not in params_data:  # pragma: no branch
                params_data["_meta"] = {}
            params_data["_meta"][RELATED_TASK_METADATA_KEY] = types.RelatedTaskMetadata(
                task_id=related_task_id
            ).model_dump(by_alias=True)

        request_id = f"task-{related_task_id}-{id(params)}" if related_task_id else self._request_id
        if related_task_id is None:
            self._request_id += 1

        return types.JSONRPCRequest(
            jsonrpc="2.0",
            id=request_id,
            method="sampling/createMessage",
            params=params_data,
        )

    async def send_message(self, message: SessionMessage) -> None:
        """Send a raw session message.

        This is primarily used by TaskResultHandler to deliver queued messages
        (elicitation/sampling requests) to the client during task execution.

        WARNING: This is a low-level experimental method that may change without
        notice. Prefer using higher-level methods like send_notification() or
        send_request() for normal operations.

        Args:
            message: The session message to send
        """
        await self._write_stream.send(message)

    async def _handle_incoming(self, req: ServerRequestResponder) -> None:
        await self._incoming_message_stream_writer.send(req)

    @property
    def incoming_messages(self) -> MemoryObjectReceiveStream[ServerRequestResponder]:
        return self._incoming_message_stream_reader

experimental property

experimental: ExperimentalServerSessionFeatures

Experimental APIs for server→client task operations.

WARNING: These APIs are experimental and may change without notice.

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."""
    if self._client_params is None:  # pragma: lax no cover
        return False

    client_caps = self._client_params.capabilities

    if capability.roots is not None:  # pragma: lax no cover
        if client_caps.roots is None:
            return False
        if capability.roots.list_changed and not client_caps.roots.list_changed:
            return False

    if capability.sampling is not None:  # pragma: lax no cover
        if client_caps.sampling is None:
            return False
        if capability.sampling.context is not None and client_caps.sampling.context is None:
            return False
        if capability.sampling.tools is not None and client_caps.sampling.tools is None:
            return False

    if capability.elicitation is not None and client_caps.elicitation is None:  # pragma: lax no cover
        return False

    if capability.experimental is not None:  # pragma: lax no cover
        if client_caps.experimental is None:
            return False
        for exp_key, exp_value in capability.experimental.items():
            if exp_key not in client_caps.experimental or client_caps.experimental[exp_key] != exp_value:
                return False

    if capability.tasks is not None:  # pragma: lax no cover
        if client_caps.tasks is None:
            return False
        if not check_tasks_capability(capability.tasks, client_caps.tasks):
            return False

    return True

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

    # Use different result types based on whether tools are provided
    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:  # pragma: no cover
    """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:  # pragma: no cover
    """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:  # pragma: no cover
    """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,
    )

send_message async

send_message(message: SessionMessage) -> None

Send a raw session message.

This is primarily used by TaskResultHandler to deliver queued messages (elicitation/sampling requests) to the client during task execution.

WARNING: This is a low-level experimental method that may change without notice. Prefer using higher-level methods like send_notification() or send_request() for normal operations.

Parameters:

Name	Type	Description	Default
message	SessionMessage	The session message to send	required

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

async def send_message(self, message: SessionMessage) -> None:
    """Send a raw session message.

    This is primarily used by TaskResultHandler to deliver queued messages
    (elicitation/sampling requests) to the client during task execution.

    WARNING: This is a low-level experimental method that may change without
    notice. Prefer using higher-level methods like send_notification() or
    send_request() for normal operations.

    Args:
        message: The session message to send
    """
    await self._write_stream.send(message)

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."""
    tasks: ClientTasksCapability | None = None
    """Present if the client supports task-augmented requests."""

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.

tasks class-attribute instance-attribute

tasks: ClientTasksCapability | None = None

Present if the client supports task-augmented requests.

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

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

    tasks: ServerTasksCapability | None = None
    """Present if the server supports task-augmented requests."""

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.

tasks class-attribute instance-attribute

tasks: ServerTasksCapability | None = None

Present if the server supports task-augmented requests.

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

    execution: ToolExecution | None = None

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