API Reference

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: MemoryObjectReceiveStream[SessionMessage | Exception],
        write_stream: MemoryObjectSendStream[SessionMessage],
        read_timeout_seconds: timedelta | 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,
    ) -> None:
        super().__init__(
            read_stream,
            write_stream,
            types.ServerRequest,
            types.ServerNotification,
            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._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] = {}

    async def initialize(self) -> types.InitializeResult:
        sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
        elicitation = (
            types.ElicitationCapability() 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(listChanged=True)
            if self._list_roots_callback is not _default_list_roots_callback
            else None
        )

        result = await self.send_request(
            types.ClientRequest(
                types.InitializeRequest(
                    method="initialize",
                    params=types.InitializeRequestParams(
                        protocolVersion=types.LATEST_PROTOCOL_VERSION,
                        capabilities=types.ClientCapabilities(
                            sampling=sampling,
                            elicitation=elicitation,
                            experimental=None,
                            roots=roots,
                        ),
                        clientInfo=self._client_info,
                    ),
                )
            ),
            types.InitializeResult,
        )

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

        await self.send_notification(
            types.ClientNotification(types.InitializedNotification(method="notifications/initialized"))
        )

        return result

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

    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."""
        await self.send_notification(
            types.ClientNotification(
                types.ProgressNotification(
                    method="notifications/progress",
                    params=types.ProgressNotificationParams(
                        progressToken=progress_token,
                        progress=progress,
                        total=total,
                        message=message,
                    ),
                ),
            )
        )

    async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResult:
        """Send a logging/setLevel request."""
        return await self.send_request(
            types.ClientRequest(
                types.SetLevelRequest(
                    method="logging/setLevel",
                    params=types.SetLevelRequestParams(level=level),
                )
            ),
            types.EmptyResult,
        )

    async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult:
        """Send a resources/list request."""
        return await self.send_request(
            types.ClientRequest(
                types.ListResourcesRequest(
                    method="resources/list",
                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
                )
            ),
            types.ListResourcesResult,
        )

    async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult:
        """Send a resources/templates/list request."""
        return await self.send_request(
            types.ClientRequest(
                types.ListResourceTemplatesRequest(
                    method="resources/templates/list",
                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
                )
            ),
            types.ListResourceTemplatesResult,
        )

    async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
        """Send a resources/read request."""
        return await self.send_request(
            types.ClientRequest(
                types.ReadResourceRequest(
                    method="resources/read",
                    params=types.ReadResourceRequestParams(uri=uri),
                )
            ),
            types.ReadResourceResult,
        )

    async def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
        """Send a resources/subscribe request."""
        return await self.send_request(
            types.ClientRequest(
                types.SubscribeRequest(
                    method="resources/subscribe",
                    params=types.SubscribeRequestParams(uri=uri),
                )
            ),
            types.EmptyResult,
        )

    async def unsubscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
        """Send a resources/unsubscribe request."""
        return await self.send_request(
            types.ClientRequest(
                types.UnsubscribeRequest(
                    method="resources/unsubscribe",
                    params=types.UnsubscribeRequestParams(uri=uri),
                )
            ),
            types.EmptyResult,
        )

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

        result = await self.send_request(
            types.ClientRequest(
                types.CallToolRequest(
                    method="tools/call",
                    params=types.CallToolRequestParams(
                        name=name,
                        arguments=arguments,
                    ),
                )
            ),
            types.CallToolResult,
            request_read_timeout_seconds=read_timeout_seconds,
            progress_callback=progress_callback,
        )

        if not result.isError:
            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:
            if result.structuredContent is None:
                raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
            try:
                validate(result.structuredContent, output_schema)
            except ValidationError as e:
                raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
            except SchemaError as e:
                raise RuntimeError(f"Invalid schema for tool {name}: {e}")

    async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult:
        """Send a prompts/list request."""
        return await self.send_request(
            types.ClientRequest(
                types.ListPromptsRequest(
                    method="prompts/list",
                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
                )
            ),
            types.ListPromptsResult,
        )

    async def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult:
        """Send a prompts/get request."""
        return await self.send_request(
            types.ClientRequest(
                types.GetPromptRequest(
                    method="prompts/get",
                    params=types.GetPromptRequestParams(name=name, arguments=arguments),
                )
            ),
            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.ClientRequest(
                types.CompleteRequest(
                    method="completion/complete",
                    params=types.CompleteRequestParams(
                        ref=ref,
                        argument=types.CompletionArgument(**argument),
                        context=context,
                    ),
                )
            ),
            types.CompleteResult,
        )

    async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult:
        """Send a tools/list request."""
        result = await self.send_request(
            types.ClientRequest(
                types.ListToolsRequest(
                    method="tools/list",
                    params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
                )
            ),
            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.outputSchema

        return result

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

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

        match responder.request.root:
            case types.CreateMessageRequest(params=params):
                with responder:
                    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:
                    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():
                with responder:
                    return await responder.respond(types.ClientResult(root=types.EmptyResult()))

    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.root:
            case types.LoggingMessageNotification(params=params):
                await self._logging_callback(params)
            case _:
                pass

send_ping async

send_ping() -> EmptyResult

Send a ping request.

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

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

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.

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,
) -> None:
    """Send a progress notification."""
    await self.send_notification(
        types.ClientNotification(
            types.ProgressNotification(
                method="notifications/progress",
                params=types.ProgressNotificationParams(
                    progressToken=progress_token,
                    progress=progress,
                    total=total,
                    message=message,
                ),
            ),
        )
    )

set_logging_level async

set_logging_level(level: LoggingLevel) -> EmptyResult

Send a logging/setLevel request.

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

async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResult:
    """Send a logging/setLevel request."""
    return await self.send_request(
        types.ClientRequest(
            types.SetLevelRequest(
                method="logging/setLevel",
                params=types.SetLevelRequestParams(level=level),
            )
        ),
        types.EmptyResult,
    )

list_resources async

list_resources(
    cursor: str | None = None,
) -> ListResourcesResult

Send a resources/list request.

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

async def list_resources(self, cursor: str | None = None) -> types.ListResourcesResult:
    """Send a resources/list request."""
    return await self.send_request(
        types.ClientRequest(
            types.ListResourcesRequest(
                method="resources/list",
                params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
            )
        ),
        types.ListResourcesResult,
    )

list_resource_templates async

list_resource_templates(
    cursor: str | None = None,
) -> ListResourceTemplatesResult

Send a resources/templates/list request.

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

async def list_resource_templates(self, cursor: str | None = None) -> types.ListResourceTemplatesResult:
    """Send a resources/templates/list request."""
    return await self.send_request(
        types.ClientRequest(
            types.ListResourceTemplatesRequest(
                method="resources/templates/list",
                params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
            )
        ),
        types.ListResourceTemplatesResult,
    )

read_resource async

read_resource(uri: AnyUrl) -> ReadResourceResult

Send a resources/read request.

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

async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
    """Send a resources/read request."""
    return await self.send_request(
        types.ClientRequest(
            types.ReadResourceRequest(
                method="resources/read",
                params=types.ReadResourceRequestParams(uri=uri),
            )
        ),
        types.ReadResourceResult,
    )

subscribe_resource async

subscribe_resource(uri: AnyUrl) -> EmptyResult

Send a resources/subscribe request.

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

async def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
    """Send a resources/subscribe request."""
    return await self.send_request(
        types.ClientRequest(
            types.SubscribeRequest(
                method="resources/subscribe",
                params=types.SubscribeRequestParams(uri=uri),
            )
        ),
        types.EmptyResult,
    )

unsubscribe_resource async

unsubscribe_resource(uri: AnyUrl) -> EmptyResult

Send a resources/unsubscribe request.

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

async def unsubscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
    """Send a resources/unsubscribe request."""
    return await self.send_request(
        types.ClientRequest(
            types.UnsubscribeRequest(
                method="resources/unsubscribe",
                params=types.UnsubscribeRequestParams(uri=uri),
            )
        ),
        types.EmptyResult,
    )

call_tool async

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: timedelta | None = None,
    progress_callback: ProgressFnT | 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: timedelta | None = None,
    progress_callback: ProgressFnT | None = None,
) -> types.CallToolResult:
    """Send a tools/call request with optional progress callback support."""

    result = await self.send_request(
        types.ClientRequest(
            types.CallToolRequest(
                method="tools/call",
                params=types.CallToolRequestParams(
                    name=name,
                    arguments=arguments,
                ),
            )
        ),
        types.CallToolResult,
        request_read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
    )

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

    return result

list_prompts async

list_prompts(
    cursor: str | None = None,
) -> ListPromptsResult

Send a prompts/list request.

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

async def list_prompts(self, cursor: str | None = None) -> types.ListPromptsResult:
    """Send a prompts/list request."""
    return await self.send_request(
        types.ClientRequest(
            types.ListPromptsRequest(
                method="prompts/list",
                params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
            )
        ),
        types.ListPromptsResult,
    )

get_prompt async

get_prompt(
    name: str, arguments: dict[str, str] | 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) -> types.GetPromptResult:
    """Send a prompts/get request."""
    return await self.send_request(
        types.ClientRequest(
            types.GetPromptRequest(
                method="prompts/get",
                params=types.GetPromptRequestParams(name=name, arguments=arguments),
            )
        ),
        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.ClientRequest(
            types.CompleteRequest(
                method="completion/complete",
                params=types.CompleteRequestParams(
                    ref=ref,
                    argument=types.CompletionArgument(**argument),
                    context=context,
                ),
            )
        ),
        types.CompleteResult,
    )

list_tools async

list_tools(cursor: str | None = None) -> ListToolsResult

Send a tools/list request.

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

async def list_tools(self, cursor: str | None = None) -> types.ListToolsResult:
    """Send a tools/list request."""
    result = await self.send_request(
        types.ClientRequest(
            types.ListToolsRequest(
                method="tools/list",
                params=types.PaginatedRequestParams(cursor=cursor) if cursor is not None else None,
            )
        ),
        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.outputSchema

    return result

send_roots_list_changed async

send_roots_list_changed() -> None

Send a roots/list_changed notification.

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

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

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 Usage

name_fn = lambda name, server_info: f"{(server_info.name)}_{name}" async with ClientSessionGroup(component_name_hook=name_fn) as group: for server_params 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 Usage:
        name_fn = lambda name, server_info: f"{(server_info.name)}_{name}"
        async with ClientSessionGroup(component_name_hook=name_fn) as group:
            for server_params in server_params:
                await group.connect_to_server(server_param)
            ...

    """

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

        prompts: set[str] = set()
        resources: set[str] = set()
        tools: set[str] = 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, serverInfo) for custom names.
    # This is provide a means to mitigate naming conflicts across servers.
    # Example: (tool_name, serverInfo) => "{result.serverInfo.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:
        # 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:
        """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())

    @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, args: dict[str, Any]) -> 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, args)

    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(
                types.ErrorData(
                    code=types.INVALID_PARAMS,
                    message="Provided session is not managed or already disconnected.",
                )
            )

        if session_known_for_components:
            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:
                    del self._prompts[name]
            # Remove resources associated with the session.
            for name in component_names.resources:
                if name in self._resources:
                    del self._resources[name]
            # Remove tools associated with the session.
            for name in component_names.tools:
                if name in self._tools:
                    del self._tools[name]
                if name in self._tool_to_session:
                    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)
            await session_stack_to_close.aclose()

    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,
    ) -> mcp.ClientSession:
        """Connects to a single MCP server."""
        server_info, session = await self._establish_session(server_params)
        return await self.connect_with_session(server_info, session)

    async def _establish_session(
        self, server_params: ServerParameters
    ) -> 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:
                client = streamablehttp_client(
                    url=server_params.url,
                    headers=server_params.headers,
                    timeout=server_params.timeout,
                    sse_read_timeout=server_params.sse_read_timeout,
                    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))
            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.serverInfo, session
        except Exception:
            # 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:
            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:
            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:
            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]

        # Check for duplicates.
        matching_prompts = prompts_temp.keys() & self._prompts.keys()
        if matching_prompts:
            raise McpError(
                types.ErrorData(
                    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(
                types.ErrorData(
                    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(
                types.ErrorData(
                    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:
    """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, args: dict[str, Any]
) -> 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, args: dict[str, Any]) -> 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, args)

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(
            types.ErrorData(
                code=types.INVALID_PARAMS,
                message="Provided session is not managed or already disconnected.",
            )
        )

    if session_known_for_components:
        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:
                del self._prompts[name]
        # Remove resources associated with the session.
        for name in component_names.resources:
            if name in self._resources:
                del self._resources[name]
        # Remove tools associated with the session.
        for name in component_names.tools:
            if name in self._tools:
                del self._tools[name]
            if name in self._tool_to_session:
                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)
        await session_stack_to_close.aclose()

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,
) -> 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,
) -> mcp.ClientSession:
    """Connects to a single MCP server."""
    server_info, session = await self._establish_session(server_params)
    return await self.connect_with_session(server_info, session)

StdioServerParameters

Bases: BaseModel

Source code in src/mcp/client/stdio/__init__.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/__init__.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.JSONRPCMessage.model_validate_json(line)
                        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:
            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_none=True)
                    await process.stdin.send(
                        (json + "\n").encode(
                            encoding=server.encoding,
                            errors=server.encoding_error_handler,
                        )
                    )
        except anyio.ClosedResourceError:
            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:
                try:
                    await process.stdin.aclose()
                except Exception:
                    # 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:
                # 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

    def __init__(
        self,
        read_stream: MemoryObjectReceiveStream[SessionMessage | Exception],
        write_stream: MemoryObjectSendStream[SessionMessage],
        init_options: InitializationOptions,
        stateless: bool = False,
    ) -> None:
        super().__init__(read_stream, write_stream, types.ClientRequest, types.ClientNotification)
        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 client_params(self) -> types.InitializeRequestParams | None:
        return self._client_params

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

        # Get client capabilities from initialization params
        client_caps = self._client_params.capabilities

        # Check each specified capability in the passed in capability object
        if capability.roots is not None:
            if client_caps.roots is None:
                return False
            if capability.roots.listChanged and not client_caps.roots.listChanged:
                return False

        if capability.sampling is not None:
            if client_caps.sampling is None:
                return False

        if capability.elicitation is not None:
            if client_caps.elicitation is None:
                return False

        if capability.experimental is not None:
            if client_caps.experimental is None:
                return False
            # Check each experimental capability
            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

        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.root:
            case types.InitializeRequest(params=params):
                requested_version = params.protocolVersion
                self._initialization_state = InitializationState.Initializing
                self._client_params = params
                with responder:
                    await responder.respond(
                        types.ServerResult(
                            types.InitializeResult(
                                protocolVersion=requested_version
                                if requested_version in SUPPORTED_PROTOCOL_VERSIONS
                                else types.LATEST_PROTOCOL_VERSION,
                                capabilities=self._init_options.capabilities,
                                serverInfo=types.Implementation(
                                    name=self._init_options.server_name,
                                    version=self._init_options.server_version,
                                ),
                                instructions=self._init_options.instructions,
                            )
                        )
                    )
            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.root:
            case types.InitializedNotification():
                self._initialization_state = InitializationState.Initialized
            case _:
                if self._initialization_state != InitializationState.Initialized:
                    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.ServerNotification(
                types.LoggingMessageNotification(
                    method="notifications/message",
                    params=types.LoggingMessageNotificationParams(
                        level=level,
                        data=data,
                        logger=logger,
                    ),
                )
            ),
            related_request_id,
        )

    async def send_resource_updated(self, uri: AnyUrl) -> None:
        """Send a resource updated notification."""
        await self.send_notification(
            types.ServerNotification(
                types.ResourceUpdatedNotification(
                    method="notifications/resources/updated",
                    params=types.ResourceUpdatedNotificationParams(uri=uri),
                )
            )
        )

    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,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResult:
        """Send a sampling/create_message request."""
        return await self.send_request(
            request=types.ServerRequest(
                types.CreateMessageRequest(
                    method="sampling/createMessage",
                    params=types.CreateMessageRequestParams(
                        messages=messages,
                        systemPrompt=system_prompt,
                        includeContext=include_context,
                        temperature=temperature,
                        maxTokens=max_tokens,
                        stopSequences=stop_sequences,
                        metadata=metadata,
                        modelPreferences=model_preferences,
                    ),
                )
            ),
            result_type=types.CreateMessageResult,
            metadata=ServerMessageMetadata(
                related_request_id=related_request_id,
            ),
        )

    async def list_roots(self) -> types.ListRootsResult:
        """Send a roots/list request."""
        return await self.send_request(
            types.ServerRequest(
                types.ListRootsRequest(
                    method="roots/list",
                )
            ),
            types.ListRootsResult,
        )

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

        Args:
            message: The message to present to the user
            requestedSchema: Schema defining the expected response structure

        Returns:
            The client's response
        """
        return await self.send_request(
            types.ServerRequest(
                types.ElicitRequest(
                    method="elicitation/create",
                    params=types.ElicitRequestParams(
                        message=message,
                        requestedSchema=requestedSchema,
                    ),
                )
            ),
            types.ElicitResult,
            metadata=ServerMessageMetadata(related_request_id=related_request_id),
        )

    async def send_ping(self) -> types.EmptyResult:
        """Send a ping request."""
        return await self.send_request(
            types.ServerRequest(
                types.PingRequest(
                    method="ping",
                )
            ),
            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.ServerNotification(
                types.ProgressNotification(
                    method="notifications/progress",
                    params=types.ProgressNotificationParams(
                        progressToken=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.ServerNotification(
                types.ResourceListChangedNotification(
                    method="notifications/resources/list_changed",
                )
            )
        )

    async def send_tool_list_changed(self) -> None:
        """Send a tool list changed notification."""
        await self.send_notification(
            types.ServerNotification(
                types.ToolListChangedNotification(
                    method="notifications/tools/list_changed",
                )
            )
        )

    async def send_prompt_list_changed(self) -> None:
        """Send a prompt list changed notification."""
        await self.send_notification(
            types.ServerNotification(
                types.PromptListChangedNotification(
                    method="notifications/prompts/list_changed",
                )
            )
        )

    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

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:
        return False

    # Get client capabilities from initialization params
    client_caps = self._client_params.capabilities

    # Check each specified capability in the passed in capability object
    if capability.roots is not None:
        if client_caps.roots is None:
            return False
        if capability.roots.listChanged and not client_caps.roots.listChanged:
            return False

    if capability.sampling is not None:
        if client_caps.sampling is None:
            return False

    if capability.elicitation is not None:
        if client_caps.elicitation is None:
            return False

    if capability.experimental is not None:
        if client_caps.experimental is None:
            return False
        # Check each experimental capability
        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

    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.ServerNotification(
            types.LoggingMessageNotification(
                method="notifications/message",
                params=types.LoggingMessageNotificationParams(
                    level=level,
                    data=data,
                    logger=logger,
                ),
            )
        ),
        related_request_id,
    )

send_resource_updated async

send_resource_updated(uri: AnyUrl) -> None

Send a resource updated notification.

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

async def send_resource_updated(self, uri: AnyUrl) -> None:
    """Send a resource updated notification."""
    await self.send_notification(
        types.ServerNotification(
            types.ResourceUpdatedNotification(
                method="notifications/resources/updated",
                params=types.ResourceUpdatedNotificationParams(uri=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,
    related_request_id: RequestId | None = None
) -> CreateMessageResult

Send a sampling/create_message request.

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,
    related_request_id: types.RequestId | None = None,
) -> types.CreateMessageResult:
    """Send a sampling/create_message request."""
    return await self.send_request(
        request=types.ServerRequest(
            types.CreateMessageRequest(
                method="sampling/createMessage",
                params=types.CreateMessageRequestParams(
                    messages=messages,
                    systemPrompt=system_prompt,
                    includeContext=include_context,
                    temperature=temperature,
                    maxTokens=max_tokens,
                    stopSequences=stop_sequences,
                    metadata=metadata,
                    modelPreferences=model_preferences,
                ),
            )
        ),
        result_type=types.CreateMessageResult,
        metadata=ServerMessageMetadata(
            related_request_id=related_request_id,
        ),
    )

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."""
    return await self.send_request(
        types.ServerRequest(
            types.ListRootsRequest(
                method="roots/list",
            )
        ),
        types.ListRootsResult,
    )

elicit async

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

Send an elicitation/create request.

Parameters:

Name	Type	Description	Default
message	str	The message to present to the user	required
requestedSchema	ElicitRequestedSchema	Schema defining the expected response structure	required

Returns:

Type	Description
ElicitResult	The client's response

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

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

    Args:
        message: The message to present to the user
        requestedSchema: Schema defining the expected response structure

    Returns:
        The client's response
    """
    return await self.send_request(
        types.ServerRequest(
            types.ElicitRequest(
                method="elicitation/create",
                params=types.ElicitRequestParams(
                    message=message,
                    requestedSchema=requestedSchema,
                ),
            )
        ),
        types.ElicitResult,
        metadata=ServerMessageMetadata(related_request_id=related_request_id),
    )

send_ping async

send_ping() -> EmptyResult

Send a ping request.

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

async def send_ping(self) -> types.EmptyResult:
    """Send a ping request."""
    return await self.send_request(
        types.ServerRequest(
            types.PingRequest(
                method="ping",
            )
        ),
        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.ServerNotification(
            types.ProgressNotification(
                method="notifications/progress",
                params=types.ProgressNotificationParams(
                    progressToken=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.ServerNotification(
            types.ResourceListChangedNotification(
                method="notifications/resources/list_changed",
            )
        )
    )

send_tool_list_changed async

send_tool_list_changed() -> None

Send a tool list changed notification.

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

async def send_tool_list_changed(self) -> None:
    """Send a tool list changed notification."""
    await self.send_notification(
        types.ServerNotification(
            types.ToolListChangedNotification(
                method="notifications/tools/list_changed",
            )
        )
    )

send_prompt_list_changed async

send_prompt_list_changed() -> None

Send a prompt list changed notification.

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

async def send_prompt_list_changed(self) -> None:
    """Send a prompt list changed notification."""
    await self.send_notification(
        types.ServerNotification(
            types.PromptListChangedNotification(
                method="notifications/prompts/list_changed",
            )
        )
    )

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"))
    if not stdout:
        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))

    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)

    async def stdin_reader():
        try:
            async with read_stream_writer:
                async for line in stdin:
                    try:
                        message = types.JSONRPCMessage.model_validate_json(line)
                    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:
            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_none=True)
                    await stdout.write(json + "\n")
                    await stdout.flush()
        except anyio.ClosedResourceError:
            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, error: ErrorData):
        """Initialize McpError."""
        super().__init__(error.message)
        self.error = error

__init__

__init__(error: ErrorData)

Initialize McpError.

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

def __init__(self, error: ErrorData):
    """Initialize McpError."""
    super().__init__(error.message)
    self.error = error

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

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

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

ClientCapabilities

Bases: BaseModel

Capabilities a client may support.

Source code in src/mcp/types.py

class ClientCapabilities(BaseModel):
    """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."""
    elicitation: ElicitationCapability | None = None
    """Present if the client supports elicitation from the user."""
    roots: RootsCapability | None = None
    """Present if the client supports listing roots."""
    model_config = ConfigDict(extra="allow")

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.

elicitation class-attribute instance-attribute

elicitation: ElicitationCapability | None = None

Present if the client supports elicitation from the user.

roots class-attribute instance-attribute

roots: RootsCapability | None = None

Present if the client supports listing roots.

CompleteRequest

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

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

Source code in src/mcp/types.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"]
    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.py

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

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

CreateMessageResult

Bases: Result

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

Source code in src/mcp/types.py

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

    role: Role
    content: TextContent | ImageContent | AudioContent
    model: str
    """The name of the model that generated the message."""
    stopReason: StopReason | None = None
    """The reason why sampling stopped, if known."""

model instance-attribute

model: str

The name of the model that generated the message.

stopReason class-attribute instance-attribute

stopReason: StopReason | None = None

The reason why sampling stopped, if known.

ErrorData

Bases: BaseModel

Error information for JSON-RPC error responses.

Source code in src/mcp/types.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 = None
    """
    Additional information about the error. The value of this member is defined by the
    sender (e.g. detailed error information, nested errors etc.).
    """

    model_config = ConfigDict(extra="allow")

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 = 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.py

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

    method: Literal["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.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.py

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

    version: str
    model_config = ConfigDict(extra="allow")

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.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"]
    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.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"]
    params: InitializeRequestParams

InitializeResult

Bases: Result

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

Source code in src/mcp/types.py

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

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

protocolVersion instance-attribute

protocolVersion: 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.py

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

    jsonrpc: Literal["2.0"]
    id: str | int
    error: ErrorData
    model_config = ConfigDict(extra="allow")

JSONRPCRequest

Bases: Request[dict[str, Any] | None, str]

A request that expects a response.

Source code in src/mcp/types.py

class JSONRPCRequest(Request[dict[str, Any] | None, str]):
    """A 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.py

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

    jsonrpc: Literal["2.0"]
    id: RequestId
    result: dict[str, Any]
    model_config = ConfigDict(extra="allow")

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

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

    method: Literal["prompts/list"]

ListPromptsResult

Bases: PaginatedResult

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

Source code in src/mcp/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.py

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

    method: Literal["resources/list"]

ListResourcesResult

Bases: PaginatedResult

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

Source code in src/mcp/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.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.py

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

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

Notification

Bases: BaseModel, Generic[NotificationParamsT, MethodT]

Base class for JSON-RPC notifications.

Source code in src/mcp/types.py

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

    method: MethodT
    params: NotificationParamsT
    model_config = ConfigDict(extra="allow")

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.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"]
    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.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"]
    params: ProgressNotificationParams

PromptsCapability

Bases: BaseModel

Capability for prompts operations.

Source code in src/mcp/types.py

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

    listChanged: bool | None = None
    """Whether this server supports notifications for changes to the prompt list."""
    model_config = ConfigDict(extra="allow")

listChanged class-attribute instance-attribute

listChanged: 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.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"]
    params: ReadResourceRequestParams

ReadResourceResult

Bases: Result

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

Source code in src/mcp/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.py

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

    uri: Annotated[AnyUrl, UrlConstraints(host_required=False)]
    """The URI of this resource."""
    description: str | None = None
    """A description of what this resource represents."""
    mimeType: 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.
    """
    annotations: Annotations | None = None
    meta: dict[str, Any] | 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.
    """
    model_config = ConfigDict(extra="allow")

uri instance-attribute

uri: Annotated[AnyUrl, UrlConstraints(host_required=False)]

The URI of this resource.

description class-attribute instance-attribute

description: str | None = None

A description of what this resource represents.

mimeType class-attribute instance-attribute

mimeType: 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.

meta class-attribute instance-attribute

meta: dict[str, Any] | None = Field(
    alias="_meta", default=None
)

See MCP specification for notes on _meta usage.

ResourcesCapability

Bases: BaseModel

Capability for resources operations.

Source code in src/mcp/types.py

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

    subscribe: bool | None = None
    """Whether this server supports subscribing to resource updates."""
    listChanged: bool | None = None
    """Whether this server supports notifications for changes to the resource list."""
    model_config = ConfigDict(extra="allow")

subscribe class-attribute instance-attribute

subscribe: bool | None = None

Whether this server supports subscribing to resource updates.

listChanged class-attribute instance-attribute

listChanged: 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.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"]
    params: ResourceUpdatedNotificationParams

RootsCapability

Bases: BaseModel

Capability for root operations.

Source code in src/mcp/types.py

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

    listChanged: bool | None = None
    """Whether the client supports notifications for changes to the roots list."""
    model_config = ConfigDict(extra="allow")

listChanged class-attribute instance-attribute

listChanged: bool | None = None

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

SamplingMessage

Bases: BaseModel

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

Source code in src/mcp/types.py

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

    role: Role
    content: TextContent | ImageContent | AudioContent
    model_config = ConfigDict(extra="allow")

ServerCapabilities

Bases: BaseModel

Capabilities that a server may support.

Source code in src/mcp/types.py

class ServerCapabilities(BaseModel):
    """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."""
    model_config = ConfigDict(extra="allow")

experimental class-attribute instance-attribute

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

Experimental, non-standard capabilities that the server supports.

logging class-attribute instance-attribute

logging: LoggingCapability | None = None

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

prompts class-attribute instance-attribute

prompts: PromptsCapability | None = None

Present if the server offers any prompt templates.

resources class-attribute instance-attribute

resources: ResourcesCapability | None = None

Present if the server offers any resources to read.

tools class-attribute instance-attribute

tools: ToolsCapability | None = None

Present if the server offers any tools to call.

completions class-attribute instance-attribute

completions: CompletionsCapability | None = None

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

SetLevelRequest

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

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

Source code in src/mcp/types.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"]
    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.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"]
    params: SubscribeRequestParams

Tool

Bases: BaseMetadata

Definition for a tool the client can call.

Source code in src/mcp/types.py

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

    description: str | None = None
    """A human-readable description of the tool."""
    inputSchema: dict[str, Any]
    """A JSON Schema object defining the expected parameters for the tool."""
    outputSchema: dict[str, Any] | None = None
    """
    An optional JSON Schema object defining the structure of the tool's output
    returned in the structuredContent field of a CallToolResult.
    """
    annotations: ToolAnnotations | None = None
    """Optional additional tool information."""
    meta: dict[str, Any] | 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.
    """
    model_config = ConfigDict(extra="allow")

description class-attribute instance-attribute

description: str | None = None

A human-readable description of the tool.

inputSchema instance-attribute

inputSchema: dict[str, Any]

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

outputSchema class-attribute instance-attribute

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

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

annotations class-attribute instance-attribute

annotations: ToolAnnotations | None = None

Optional additional tool information.

meta class-attribute instance-attribute

meta: dict[str, Any] | None = Field(
    alias="_meta", default=None
)

See MCP specification for notes on _meta usage.

ToolsCapability

Bases: BaseModel

Capability for tools operations.

Source code in src/mcp/types.py

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

    listChanged: bool | None = None
    """Whether this server supports notifications for changes to the tool list."""
    model_config = ConfigDict(extra="allow")

listChanged class-attribute instance-attribute

listChanged: bool | None = None

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

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.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"]
    params: UnsubscribeRequestParams