strands.tools.mcp.mcp_client

Model Context Protocol (MCP) server connection management module.

This module provides the MCPClient class which handles connections to MCP servers. It manages the lifecycle of MCP connections, including initialization, tool discovery, tool invocation, and proper cleanup of resources. The connection runs in a background thread to avoid blocking the main application thread while maintaining communication with the MCP service.

CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE = 'the client session is not running. Ensure the agent is used within the MCP client context manager. For more information see: https://strandsagents.com/latest/user-guide/concepts/tools/mcp-tools/#mcpclientinitializationerror' module-attribute

ImageFormat = Literal['png', 'jpeg', 'gif', 'webp'] module-attribute

Supported image formats.

MCPTransport = AbstractAsyncContextManager[MessageStream | _MessageStreamWithGetSessionIdCallback] module-attribute

MIME_TO_FORMAT = {'image/jpeg': 'jpeg', 'image/jpg': 'jpeg', 'image/png': 'png', 'image/gif': 'gif', 'image/webp': 'webp'} module-attribute

T = TypeVar('T') module-attribute

ToolResultStatus = Literal['success', 'error'] module-attribute

Status of a tool execution result.

_NON_FATAL_ERROR_PATTERNS = ['unknown request id'] module-attribute

_ToolMatcher = str | Pattern[str] | _ToolFilterCallback module-attribute

logger = logging.getLogger(__name__) module-attribute

AgentTool

Bases: ABC

Abstract base class for all SDK tools.

This class defines the interface that all tool implementations must follow. Each tool must provide its name, specification, and implement a stream method that executes the tool's functionality.

Source code in strands/types/tools.py

class AgentTool(ABC):
    """Abstract base class for all SDK tools.

    This class defines the interface that all tool implementations must follow. Each tool must provide its name,
    specification, and implement a stream method that executes the tool's functionality.
    """

    _is_dynamic: bool

    def __init__(self) -> None:
        """Initialize the base agent tool with default dynamic state."""
        self._is_dynamic = False

    @property
    @abstractmethod
    # pragma: no cover
    def tool_name(self) -> str:
        """The unique name of the tool used for identification and invocation."""
        pass

    @property
    @abstractmethod
    # pragma: no cover
    def tool_spec(self) -> ToolSpec:
        """Tool specification that describes its functionality and parameters."""
        pass

    @property
    @abstractmethod
    # pragma: no cover
    def tool_type(self) -> str:
        """The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').

        Used for categorization and appropriate handling.
        """
        pass

    @property
    def supports_hot_reload(self) -> bool:
        """Whether the tool supports automatic reloading when modified.

        Returns:
            False by default.
        """
        return False

    @abstractmethod
    # pragma: no cover
    def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
        """Stream tool events and return the final result.

        Args:
            tool_use: The tool use request containing tool ID and parameters.
            invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
                              agent.invoke_async(), etc.).
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Tool events with the last being the tool result.
        """
        ...

    @property
    def is_dynamic(self) -> bool:
        """Whether the tool was dynamically loaded during runtime.

        Dynamic tools may have different lifecycle management.

        Returns:
            True if loaded dynamically, False otherwise.
        """
        return self._is_dynamic

    def mark_dynamic(self) -> None:
        """Mark this tool as dynamically loaded."""
        self._is_dynamic = True

    def get_display_properties(self) -> dict[str, str]:
        """Get properties to display in UI representations of this tool.

        Subclasses can extend this to include additional properties.

        Returns:
            Dictionary of property names and their string values.
        """
        return {
            "Name": self.tool_name,
            "Type": self.tool_type,
        }

is_dynamic property

Whether the tool was dynamically loaded during runtime.

Dynamic tools may have different lifecycle management.

Returns:

Type	Description
bool	True if loaded dynamically, False otherwise.

supports_hot_reload property

Whether the tool supports automatic reloading when modified.

Returns:

Type	Description
bool	False by default.

tool_name abstractmethod property

The unique name of the tool used for identification and invocation.

tool_spec abstractmethod property

Tool specification that describes its functionality and parameters.

tool_type abstractmethod property

The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').

Used for categorization and appropriate handling.

__init__()

Initialize the base agent tool with default dynamic state.

Source code in strands/types/tools.py

def __init__(self) -> None:
    """Initialize the base agent tool with default dynamic state."""
    self._is_dynamic = False

get_display_properties()

Get properties to display in UI representations of this tool.

Subclasses can extend this to include additional properties.

Returns:

Type	Description
dict[str, str]	Dictionary of property names and their string values.

Source code in strands/types/tools.py

def get_display_properties(self) -> dict[str, str]:
    """Get properties to display in UI representations of this tool.

    Subclasses can extend this to include additional properties.

    Returns:
        Dictionary of property names and their string values.
    """
    return {
        "Name": self.tool_name,
        "Type": self.tool_type,
    }

mark_dynamic()

Mark this tool as dynamically loaded.

Source code in strands/types/tools.py

def mark_dynamic(self) -> None:
    """Mark this tool as dynamically loaded."""
    self._is_dynamic = True

stream(tool_use, invocation_state, **kwargs) abstractmethod

Stream tool events and return the final result.

Parameters:

Name	Type	Description	Default
tool_use	ToolUse	The tool use request containing tool ID and parameters.	required
invocation_state	dict[str, Any]	Caller-provided kwargs that were passed to the agent when it was invoked (agent(), agent.invoke_async(), etc.).	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
ToolGenerator	Tool events with the last being the tool result.

Source code in strands/types/tools.py

@abstractmethod
# pragma: no cover
def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
    """Stream tool events and return the final result.

    Args:
        tool_use: The tool use request containing tool ID and parameters.
        invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
                          agent.invoke_async(), etc.).
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Tool events with the last being the tool result.
    """
    ...

MCPAgentTool

Bases: AgentTool

Adapter class that wraps an MCP tool and exposes it as an AgentTool.

This class bridges the gap between the MCP protocol's tool representation and the agent framework's tool interface, allowing MCP tools to be used seamlessly within the agent framework.

Source code in strands/tools/mcp/mcp_agent_tool.py

class MCPAgentTool(AgentTool):
    """Adapter class that wraps an MCP tool and exposes it as an AgentTool.

    This class bridges the gap between the MCP protocol's tool representation
    and the agent framework's tool interface, allowing MCP tools to be used
    seamlessly within the agent framework.
    """

    def __init__(
        self,
        mcp_tool: MCPTool,
        mcp_client: "MCPClient",
        name_override: str | None = None,
        timeout: timedelta | None = None,
    ) -> None:
        """Initialize a new MCPAgentTool instance.

        Args:
            mcp_tool: The MCP tool to adapt
            mcp_client: The MCP server connection to use for tool invocation
            name_override: Optional name to use for the agent tool (for disambiguation)
                           If None, uses the original MCP tool name
            timeout: Optional timeout duration for tool execution
        """
        super().__init__()
        logger.debug("tool_name=<%s> | creating mcp agent tool", mcp_tool.name)
        self.mcp_tool = mcp_tool
        self.mcp_client = mcp_client
        self._agent_tool_name = name_override or mcp_tool.name
        self.timeout = timeout

    @property
    def tool_name(self) -> str:
        """Get the name of the tool.

        Returns:
            str: The agent-facing name of the tool (may be disambiguated)
        """
        return self._agent_tool_name

    @property
    def tool_spec(self) -> ToolSpec:
        """Get the specification of the tool.

        This method converts the MCP tool specification to the agent framework's
        ToolSpec format, including the input schema, description, and optional output schema.

        Returns:
            ToolSpec: The tool specification in the agent framework format
        """
        description: str = self.mcp_tool.description or f"Tool which performs {self.mcp_tool.name}"

        spec: ToolSpec = {
            "inputSchema": {"json": self.mcp_tool.inputSchema},
            "name": self.tool_name,  # Use agent-facing name in spec
            "description": description,
        }

        if self.mcp_tool.outputSchema:
            spec["outputSchema"] = {"json": self.mcp_tool.outputSchema}

        return spec

    @property
    def tool_type(self) -> str:
        """Get the type of the tool.

        Returns:
            str: The type of the tool, always "python" for MCP tools
        """
        return "python"

    @override
    async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
        """Stream the MCP tool.

        This method delegates the tool stream to the MCP server connection, passing the tool use ID, tool name, and
        input arguments.

        Args:
            tool_use: The tool use request containing tool ID and parameters.
            invocation_state: Context for the tool invocation, including agent state.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Tool events with the last being the tool result.
        """
        logger.debug("tool_name=<%s>, tool_use_id=<%s> | streaming", self.tool_name, tool_use["toolUseId"])

        result = await self.mcp_client.call_tool_async(
            tool_use_id=tool_use["toolUseId"],
            name=self.mcp_tool.name,  # Use original MCP name for server communication
            arguments=tool_use["input"],
            read_timeout_seconds=self.timeout,
        )
        yield ToolResultEvent(result)

tool_name property

Get the name of the tool.

Returns:

Name	Type	Description
str	str	The agent-facing name of the tool (may be disambiguated)

tool_spec property

Get the specification of the tool.

This method converts the MCP tool specification to the agent framework's ToolSpec format, including the input schema, description, and optional output schema.

Returns:

Name	Type	Description
ToolSpec	ToolSpec	The tool specification in the agent framework format

tool_type property

Get the type of the tool.

Returns:

Name	Type	Description
str	str	The type of the tool, always "python" for MCP tools

__init__(mcp_tool, mcp_client, name_override=None, timeout=None)

Initialize a new MCPAgentTool instance.

Parameters:

Name	Type	Description	Default
mcp_tool	Tool	The MCP tool to adapt	required
mcp_client	MCPClient	The MCP server connection to use for tool invocation	required
name_override	str | None	Optional name to use for the agent tool (for disambiguation) If None, uses the original MCP tool name	None
timeout	timedelta | None	Optional timeout duration for tool execution	None

Source code in strands/tools/mcp/mcp_agent_tool.py

def __init__(
    self,
    mcp_tool: MCPTool,
    mcp_client: "MCPClient",
    name_override: str | None = None,
    timeout: timedelta | None = None,
) -> None:
    """Initialize a new MCPAgentTool instance.

    Args:
        mcp_tool: The MCP tool to adapt
        mcp_client: The MCP server connection to use for tool invocation
        name_override: Optional name to use for the agent tool (for disambiguation)
                       If None, uses the original MCP tool name
        timeout: Optional timeout duration for tool execution
    """
    super().__init__()
    logger.debug("tool_name=<%s> | creating mcp agent tool", mcp_tool.name)
    self.mcp_tool = mcp_tool
    self.mcp_client = mcp_client
    self._agent_tool_name = name_override or mcp_tool.name
    self.timeout = timeout

stream(tool_use, invocation_state, **kwargs) async

Stream the MCP tool.

This method delegates the tool stream to the MCP server connection, passing the tool use ID, tool name, and input arguments.

Parameters:

Name	Type	Description	Default
tool_use	ToolUse	The tool use request containing tool ID and parameters.	required
invocation_state	dict[str, Any]	Context for the tool invocation, including agent state.	required
**kwargs	Any	Additional keyword arguments for future extensibility.	{}

Yields:

Type	Description
ToolGenerator	Tool events with the last being the tool result.

Source code in strands/tools/mcp/mcp_agent_tool.py

@override
async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
    """Stream the MCP tool.

    This method delegates the tool stream to the MCP server connection, passing the tool use ID, tool name, and
    input arguments.

    Args:
        tool_use: The tool use request containing tool ID and parameters.
        invocation_state: Context for the tool invocation, including agent state.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Tool events with the last being the tool result.
    """
    logger.debug("tool_name=<%s>, tool_use_id=<%s> | streaming", self.tool_name, tool_use["toolUseId"])

    result = await self.mcp_client.call_tool_async(
        tool_use_id=tool_use["toolUseId"],
        name=self.mcp_tool.name,  # Use original MCP name for server communication
        arguments=tool_use["input"],
        read_timeout_seconds=self.timeout,
    )
    yield ToolResultEvent(result)

MCPClient

Bases: ToolProvider

Represents a connection to a Model Context Protocol (MCP) server.

This class implements a context manager pattern for efficient connection management, allowing reuse of the same connection for multiple tool calls to reduce latency. It handles the creation, initialization, and cleanup of MCP connections.

The connection runs in a background thread to avoid blocking the main application thread while maintaining communication with the MCP service. When structured content is available from MCP tools, it will be returned as the last item in the content array of the ToolResult.

Warning

This class implements the experimental ToolProvider interface and its methods are subject to change.

Source code in strands/tools/mcp/mcp_client.py

class MCPClient(ToolProvider):
    """Represents a connection to a Model Context Protocol (MCP) server.

    This class implements a context manager pattern for efficient connection management,
    allowing reuse of the same connection for multiple tool calls to reduce latency.
    It handles the creation, initialization, and cleanup of MCP connections.

    The connection runs in a background thread to avoid blocking the main application thread
    while maintaining communication with the MCP service. When structured content is available
    from MCP tools, it will be returned as the last item in the content array of the ToolResult.

    Warning:
        This class implements the experimental ToolProvider interface and its methods
        are subject to change.
    """

    def __init__(
        self,
        transport_callable: Callable[[], MCPTransport],
        *,
        startup_timeout: int = 30,
        tool_filters: ToolFilters | None = None,
        prefix: str | None = None,
        elicitation_callback: Optional[ElicitationFnT] = None,
    ) -> None:
        """Initialize a new MCP Server connection.

        Args:
            transport_callable: A callable that returns an MCPTransport (read_stream, write_stream) tuple.
            startup_timeout: Timeout after which MCP server initialization should be cancelled.
                Defaults to 30.
            tool_filters: Optional filters to apply to tools.
            prefix: Optional prefix for tool names.
            elicitation_callback: Optional callback function to handle elicitation requests from the MCP server.
        """
        self._startup_timeout = startup_timeout
        self._tool_filters = tool_filters
        self._prefix = prefix
        self._elicitation_callback = elicitation_callback

        mcp_instrumentation()
        self._session_id = uuid.uuid4()
        self._log_debug_with_thread("initializing MCPClient connection")
        # Main thread blocks until future completes
        self._init_future: futures.Future[None] = futures.Future()
        # Set within the inner loop as it needs the asyncio loop
        self._close_future: asyncio.futures.Future[None] | None = None
        self._close_exception: None | Exception = None
        # Do not want to block other threads while close event is false
        self._transport_callable = transport_callable

        self._background_thread: threading.Thread | None = None
        self._background_thread_session: ClientSession | None = None
        self._background_thread_event_loop: AbstractEventLoop | None = None
        self._loaded_tools: list[MCPAgentTool] | None = None
        self._tool_provider_started = False
        self._consumers: set[Any] = set()

    def __enter__(self) -> "MCPClient":
        """Context manager entry point which initializes the MCP server connection.

        TODO: Refactor to lazy initialization pattern following idiomatic Python.
        Heavy work in __enter__ is non-idiomatic - should move connection logic to first method call instead.
        """
        return self.start()

    def __exit__(self, exc_type: BaseException, exc_val: BaseException, exc_tb: TracebackType) -> None:
        """Context manager exit point that cleans up resources."""
        self.stop(exc_type, exc_val, exc_tb)

    def start(self) -> "MCPClient":
        """Starts the background thread and waits for initialization.

        This method starts the background thread that manages the MCP connection
        and blocks until the connection is ready or times out.

        Returns:
            self: The MCPClient instance

        Raises:
            Exception: If the MCP connection fails to initialize within the timeout period
        """
        if self._is_session_active():
            raise MCPClientInitializationError("the client session is currently running")

        self._log_debug_with_thread("entering MCPClient context")
        self._background_thread = threading.Thread(target=self._background_task, args=[], daemon=True)
        self._background_thread.start()
        self._log_debug_with_thread("background thread started, waiting for ready event")
        try:
            # Blocking main thread until session is initialized in other thread or if the thread stops
            self._init_future.result(timeout=self._startup_timeout)
            self._log_debug_with_thread("the client initialization was successful")
        except futures.TimeoutError as e:
            logger.exception("client initialization timed out")
            # Pass None for exc_type, exc_val, exc_tb since this isn't a context manager exit
            self.stop(None, None, None)
            raise MCPClientInitializationError(
                f"background thread did not start in {self._startup_timeout} seconds"
            ) from e
        except Exception as e:
            logger.exception("client failed to initialize")
            # Pass None for exc_type, exc_val, exc_tb since this isn't a context manager exit
            self.stop(None, None, None)
            raise MCPClientInitializationError("the client initialization failed") from e
        return self

    # ToolProvider interface methods (experimental, as ToolProvider is experimental)
    async def load_tools(self, **kwargs: Any) -> Sequence[AgentTool]:
        """Load and return tools from the MCP server.

        This method implements the ToolProvider interface by loading tools
        from the MCP server and caching them for reuse.

        Args:
            **kwargs: Additional arguments for future compatibility.

        Returns:
            List of AgentTool instances from the MCP server.
        """
        logger.debug(
            "started=<%s>, cached_tools=<%s> | loading tools",
            self._tool_provider_started,
            self._loaded_tools is not None,
        )

        if not self._tool_provider_started:
            try:
                logger.debug("starting MCP client")
                self.start()
                self._tool_provider_started = True
                logger.debug("MCP client started successfully")
            except Exception as e:
                logger.error("error=<%s> | failed to start MCP client", e)
                raise ToolProviderException(f"Failed to start MCP client: {e}") from e

        if self._loaded_tools is None:
            logger.debug("loading tools from MCP server")
            self._loaded_tools = []
            pagination_token = None
            page_count = 0

            while True:
                logger.debug("page=<%d>, token=<%s> | fetching tools page", page_count, pagination_token)
                # Use constructor defaults for prefix and filters in load_tools
                paginated_tools = self.list_tools_sync(
                    pagination_token, prefix=self._prefix, tool_filters=self._tool_filters
                )

                # Tools are already filtered by list_tools_sync, so add them all
                for tool in paginated_tools:
                    self._loaded_tools.append(tool)

                logger.debug(
                    "page=<%d>, page_tools=<%d>, total_filtered=<%d> | processed page",
                    page_count,
                    len(paginated_tools),
                    len(self._loaded_tools),
                )

                pagination_token = paginated_tools.pagination_token
                page_count += 1

                if pagination_token is None:
                    break

            logger.debug("final_tools=<%d> | loading complete", len(self._loaded_tools))

        return self._loaded_tools

    def add_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
        """Add a consumer to this tool provider.

        Synchronous to prevent GC deadlocks when called from Agent finalizers.
        """
        self._consumers.add(consumer_id)
        logger.debug("added provider consumer, count=%d", len(self._consumers))

    def remove_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
        """Remove a consumer from this tool provider.

        This method is idempotent - calling it multiple times with the same ID
        has no additional effect after the first call.

        Synchronous to prevent GC deadlocks when called from Agent finalizers.
        Uses existing synchronous stop() method for safe cleanup.
        """
        self._consumers.discard(consumer_id)
        logger.debug("removed provider consumer, count=%d", len(self._consumers))

        if not self._consumers and self._tool_provider_started:
            logger.debug("no consumers remaining, cleaning up")
            try:
                self.stop(None, None, None)  # Existing sync method - safe for finalizers
                self._tool_provider_started = False
                self._loaded_tools = None
            except Exception as e:
                logger.error("error=<%s> | failed to cleanup MCP client", e)
                raise ToolProviderException(f"Failed to cleanup MCP client: {e}") from e

    # MCP-specific methods

    def stop(
        self, exc_type: Optional[BaseException], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
    ) -> None:
        """Signals the background thread to stop and waits for it to complete, ensuring proper cleanup of all resources.

        This method is defensive and can handle partial initialization states that may occur
        if start() fails partway through initialization.

        Resources to cleanup:
        - _background_thread: Thread running the async event loop
        - _background_thread_session: MCP ClientSession (auto-closed by context manager)
        - _background_thread_event_loop: AsyncIO event loop in background thread
        - _close_future: AsyncIO future to signal thread shutdown
        - _close_exception: Exception that caused the background thread shutdown; None if a normal shutdown occurred.
        - _init_future: Future for initialization synchronization

        Cleanup order:
        1. Signal close future to background thread (if session initialized)
        2. Wait for background thread to complete
        3. Reset all state for reuse

        Args:
            exc_type: Exception type if an exception was raised in the context
            exc_val: Exception value if an exception was raised in the context
            exc_tb: Exception traceback if an exception was raised in the context
        """
        self._log_debug_with_thread("exiting MCPClient context")

        # Only try to signal close future if we have a background thread
        if self._background_thread is not None:
            # Signal close future if event loop exists
            if self._background_thread_event_loop is not None:

                async def _set_close_event() -> None:
                    if self._close_future and not self._close_future.done():
                        self._close_future.set_result(None)

                # Not calling _invoke_on_background_thread since the session does not need to exist
                # we only need the thread and event loop to exist.
                asyncio.run_coroutine_threadsafe(coro=_set_close_event(), loop=self._background_thread_event_loop)

            self._log_debug_with_thread("waiting for background thread to join")
            self._background_thread.join()

        if self._background_thread_event_loop is not None:
            self._background_thread_event_loop.close()

        self._log_debug_with_thread("background thread is closed, MCPClient context exited")

        # Reset fields to allow instance reuse
        self._init_future = futures.Future()
        self._background_thread = None
        self._background_thread_session = None
        self._background_thread_event_loop = None
        self._session_id = uuid.uuid4()
        self._loaded_tools = None
        self._tool_provider_started = False
        self._consumers = set()

        if self._close_exception:
            exception = self._close_exception
            self._close_exception = None
            raise RuntimeError("Connection to the MCP server was closed") from exception

    def list_tools_sync(
        self,
        pagination_token: str | None = None,
        prefix: str | None = None,
        tool_filters: ToolFilters | None = None,
    ) -> PaginatedList[MCPAgentTool]:
        """Synchronously retrieves the list of available tools from the MCP server.

        This method calls the asynchronous list_tools method on the MCP session
        and adapts the returned tools to the AgentTool interface.

        Args:
            pagination_token: Optional token for pagination
            prefix: Optional prefix to apply to tool names. If None, uses constructor default.
                   If explicitly provided (including empty string), overrides constructor default.
            tool_filters: Optional filters to apply to tools. If None, uses constructor default.
                         If explicitly provided (including empty dict), overrides constructor default.

        Returns:
            List[AgentTool]: A list of available tools adapted to the AgentTool interface
        """
        self._log_debug_with_thread("listing MCP tools synchronously")
        if not self._is_session_active():
            raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

        effective_prefix = self._prefix if prefix is None else prefix
        effective_filters = self._tool_filters if tool_filters is None else tool_filters

        async def _list_tools_async() -> ListToolsResult:
            return await cast(ClientSession, self._background_thread_session).list_tools(cursor=pagination_token)

        list_tools_response: ListToolsResult = self._invoke_on_background_thread(_list_tools_async()).result()
        self._log_debug_with_thread("received %d tools from MCP server", len(list_tools_response.tools))

        mcp_tools = []
        for tool in list_tools_response.tools:
            # Apply prefix if specified
            if effective_prefix:
                prefixed_name = f"{effective_prefix}_{tool.name}"
                mcp_tool = MCPAgentTool(tool, self, name_override=prefixed_name)
                logger.debug("tool_rename=<%s->%s> | renamed tool", tool.name, prefixed_name)
            else:
                mcp_tool = MCPAgentTool(tool, self)

            # Apply filters if specified
            if self._should_include_tool_with_filters(mcp_tool, effective_filters):
                mcp_tools.append(mcp_tool)

        self._log_debug_with_thread("successfully adapted %d MCP tools", len(mcp_tools))
        return PaginatedList[MCPAgentTool](mcp_tools, token=list_tools_response.nextCursor)

    def list_prompts_sync(self, pagination_token: Optional[str] = None) -> ListPromptsResult:
        """Synchronously retrieves the list of available prompts from the MCP server.

        This method calls the asynchronous list_prompts method on the MCP session
        and returns the raw ListPromptsResult with pagination support.

        Args:
            pagination_token: Optional token for pagination

        Returns:
            ListPromptsResult: The raw MCP response containing prompts and pagination info
        """
        self._log_debug_with_thread("listing MCP prompts synchronously")
        if not self._is_session_active():
            raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

        async def _list_prompts_async() -> ListPromptsResult:
            return await cast(ClientSession, self._background_thread_session).list_prompts(cursor=pagination_token)

        list_prompts_result: ListPromptsResult = self._invoke_on_background_thread(_list_prompts_async()).result()
        self._log_debug_with_thread("received %d prompts from MCP server", len(list_prompts_result.prompts))
        for prompt in list_prompts_result.prompts:
            self._log_debug_with_thread(prompt.name)

        return list_prompts_result

    def get_prompt_sync(self, prompt_id: str, args: dict[str, Any]) -> GetPromptResult:
        """Synchronously retrieves a prompt from the MCP server.

        Args:
            prompt_id: The ID of the prompt to retrieve
            args: Optional arguments to pass to the prompt

        Returns:
            GetPromptResult: The prompt response from the MCP server
        """
        self._log_debug_with_thread("getting MCP prompt synchronously")
        if not self._is_session_active():
            raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

        async def _get_prompt_async() -> GetPromptResult:
            return await cast(ClientSession, self._background_thread_session).get_prompt(prompt_id, arguments=args)

        get_prompt_result: GetPromptResult = self._invoke_on_background_thread(_get_prompt_async()).result()
        self._log_debug_with_thread("received prompt from MCP server")

        return get_prompt_result

    def call_tool_sync(
        self,
        tool_use_id: str,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: timedelta | None = None,
    ) -> MCPToolResult:
        """Synchronously calls a tool on the MCP server.

        This method calls the asynchronous call_tool method on the MCP session
        and converts the result to the ToolResult format. If the MCP tool returns
        structured content, it will be included as the last item in the content array
        of the returned ToolResult.

        Args:
            tool_use_id: Unique identifier for this tool use
            name: Name of the tool to call
            arguments: Optional arguments to pass to the tool
            read_timeout_seconds: Optional timeout for the tool call

        Returns:
            MCPToolResult: The result of the tool call
        """
        self._log_debug_with_thread("calling MCP tool '%s' synchronously with tool_use_id=%s", name, tool_use_id)
        if not self._is_session_active():
            raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

        async def _call_tool_async() -> MCPCallToolResult:
            return await cast(ClientSession, self._background_thread_session).call_tool(
                name, arguments, read_timeout_seconds
            )

        try:
            call_tool_result: MCPCallToolResult = self._invoke_on_background_thread(_call_tool_async()).result()
            return self._handle_tool_result(tool_use_id, call_tool_result)
        except Exception as e:
            logger.exception("tool execution failed")
            return self._handle_tool_execution_error(tool_use_id, e)

    async def call_tool_async(
        self,
        tool_use_id: str,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: timedelta | None = None,
    ) -> MCPToolResult:
        """Asynchronously calls a tool on the MCP server.

        This method calls the asynchronous call_tool method on the MCP session
        and converts the result to the MCPToolResult format.

        Args:
            tool_use_id: Unique identifier for this tool use
            name: Name of the tool to call
            arguments: Optional arguments to pass to the tool
            read_timeout_seconds: Optional timeout for the tool call

        Returns:
            MCPToolResult: The result of the tool call
        """
        self._log_debug_with_thread("calling MCP tool '%s' asynchronously with tool_use_id=%s", name, tool_use_id)
        if not self._is_session_active():
            raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

        async def _call_tool_async() -> MCPCallToolResult:
            return await cast(ClientSession, self._background_thread_session).call_tool(
                name, arguments, read_timeout_seconds
            )

        try:
            future = self._invoke_on_background_thread(_call_tool_async())
            call_tool_result: MCPCallToolResult = await asyncio.wrap_future(future)
            return self._handle_tool_result(tool_use_id, call_tool_result)
        except Exception as e:
            logger.exception("tool execution failed")
            return self._handle_tool_execution_error(tool_use_id, e)

    def _handle_tool_execution_error(self, tool_use_id: str, exception: Exception) -> MCPToolResult:
        """Create error ToolResult with consistent logging."""
        return MCPToolResult(
            status="error",
            toolUseId=tool_use_id,
            content=[{"text": f"Tool execution failed: {str(exception)}"}],
        )

    def _handle_tool_result(self, tool_use_id: str, call_tool_result: MCPCallToolResult) -> MCPToolResult:
        """Maps MCP tool result to the agent's MCPToolResult format.

        This method processes the content from the MCP tool call result and converts it to the format
        expected by the framework.

        Args:
            tool_use_id: Unique identifier for this tool use
            call_tool_result: The result from the MCP tool call

        Returns:
            MCPToolResult: The converted tool result
        """
        self._log_debug_with_thread("received tool result with %d content items", len(call_tool_result.content))

        # Build a typed list of ToolResultContent.
        mapped_contents: list[ToolResultContent] = [
            mc
            for content in call_tool_result.content
            if (mc := self._map_mcp_content_to_tool_result_content(content)) is not None
        ]

        status: ToolResultStatus = "error" if call_tool_result.isError else "success"
        self._log_debug_with_thread("tool execution completed with status: %s", status)
        result = MCPToolResult(
            status=status,
            toolUseId=tool_use_id,
            content=mapped_contents,
        )

        if call_tool_result.structuredContent:
            result["structuredContent"] = call_tool_result.structuredContent
        if call_tool_result.meta:
            result["metadata"] = call_tool_result.meta

        return result

    async def _async_background_thread(self) -> None:
        """Asynchronous method that runs in the background thread to manage the MCP connection.

        This method establishes the transport connection, creates and initializes the MCP session,
        signals readiness to the main thread, and waits for a close signal.
        """
        self._log_debug_with_thread("starting async background thread for MCP connection")

        # Initialized here so that it has the asyncio loop
        self._close_future = asyncio.Future()

        try:
            async with self._transport_callable() as (read_stream, write_stream, *_):
                self._log_debug_with_thread("transport connection established")
                async with ClientSession(
                    read_stream,
                    write_stream,
                    message_handler=self._handle_error_message,
                    elicitation_callback=self._elicitation_callback,
                ) as session:
                    self._log_debug_with_thread("initializing MCP session")
                    await session.initialize()

                    self._log_debug_with_thread("session initialized successfully")
                    # Store the session for use while we await the close event
                    self._background_thread_session = session
                    # Signal that the session has been created and is ready for use
                    self._init_future.set_result(None)

                    self._log_debug_with_thread("waiting for close signal")
                    # Keep background thread running until signaled to close.
                    # Thread is not blocked as this a future
                    await self._close_future

                    self._log_debug_with_thread("close signal received")
        except Exception as e:
            # If we encounter an exception and the future is still running,
            # it means it was encountered during the initialization phase.
            if not self._init_future.done():
                self._init_future.set_exception(e)
            else:
                # _close_future is automatically cancelled by the framework which doesn't provide us with the useful
                # exception, so instead we store the exception in a different field where stop() can read it
                self._close_exception = e
                if self._close_future and not self._close_future.done():
                    self._close_future.set_result(None)

                self._log_debug_with_thread(
                    "encountered exception on background thread after initialization %s", str(e)
                )

    # Raise an exception if the underlying client raises an exception in a message
    # This happens when the underlying client has an http timeout error
    async def _handle_error_message(self, message: Exception | Any) -> None:
        if isinstance(message, Exception):
            error_msg = str(message).lower()
            if any(pattern in error_msg for pattern in _NON_FATAL_ERROR_PATTERNS):
                self._log_debug_with_thread("ignoring non-fatal MCP session error", message)
            else:
                raise message
        await anyio.lowlevel.checkpoint()

    def _background_task(self) -> None:
        """Sets up and runs the event loop in the background thread.

        This method creates a new event loop for the background thread,
        sets it as the current event loop, and runs the async_background_thread
        coroutine until completion. In this case "until completion" means until the _close_future is resolved.
        This allows for a long-running event loop.
        """
        self._log_debug_with_thread("setting up background task event loop")
        self._background_thread_event_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self._background_thread_event_loop)
        self._background_thread_event_loop.run_until_complete(self._async_background_thread())

    def _map_mcp_content_to_tool_result_content(
        self,
        content: MCPTextContent | MCPImageContent | MCPEmbeddedResource | Any,
    ) -> Union[ToolResultContent, None]:
        """Maps MCP content types to tool result content types.

        This method converts MCP-specific content types to the generic
        ToolResultContent format used by the agent framework.

        Args:
            content: The MCP content to convert

        Returns:
            ToolResultContent or None: The converted content, or None if the content type is not supported
        """
        if isinstance(content, MCPTextContent):
            self._log_debug_with_thread("mapping MCP text content")
            return {"text": content.text}
        elif isinstance(content, MCPImageContent):
            self._log_debug_with_thread("mapping MCP image content with mime type: %s", content.mimeType)
            return {
                "image": {
                    "format": MIME_TO_FORMAT[content.mimeType],
                    "source": {"bytes": base64.b64decode(content.data)},
                }
            }
        elif isinstance(content, MCPEmbeddedResource):
            """
            TODO: Include URI information in results.
                Models may find it useful to be aware not only of the information,
                but the location of the information too.

                This may be difficult without taking an opinionated position. For example,
                a content block may need to indicate that the following Image content block
                is of particular URI.
            """

            self._log_debug_with_thread("mapping MCP embedded resource content")

            resource = content.resource
            if isinstance(resource, TextResourceContents):
                return {"text": resource.text}
            elif isinstance(resource, BlobResourceContents):
                try:
                    raw_bytes = base64.b64decode(resource.blob)
                except Exception:
                    self._log_debug_with_thread("embedded resource blob could not be decoded - dropping")
                    return None

                if resource.mimeType and (
                    resource.mimeType.startswith("text/")
                    or resource.mimeType
                    in (
                        "application/json",
                        "application/xml",
                        "application/javascript",
                        "application/yaml",
                        "application/x-yaml",
                    )
                    or resource.mimeType.endswith(("+json", "+xml"))
                ):
                    try:
                        return {"text": raw_bytes.decode("utf-8", errors="replace")}
                    except Exception:
                        pass

                if resource.mimeType in MIME_TO_FORMAT:
                    return {
                        "image": {
                            "format": MIME_TO_FORMAT[resource.mimeType],
                            "source": {"bytes": raw_bytes},
                        }
                    }

                self._log_debug_with_thread("embedded resource blob with non-textual/unknown mimeType - dropping")
                return None

            return None  # type: ignore[unreachable]  # Defensive: future MCP resource types
        else:
            self._log_debug_with_thread("unhandled content type: %s - dropping content", content.__class__.__name__)
            return None

    def _log_debug_with_thread(self, msg: str, *args: Any, **kwargs: Any) -> None:
        """Logger helper to help differentiate logs coming from MCPClient background thread."""
        formatted_msg = msg % args if args else msg
        logger.debug(
            "[Thread: %s, Session: %s] %s", threading.current_thread().name, self._session_id, formatted_msg, **kwargs
        )

    def _invoke_on_background_thread(self, coro: Coroutine[Any, Any, T]) -> futures.Future[T]:
        # save a reference to this so that even if it's reset we have the original
        close_future = self._close_future

        if (
            self._background_thread_session is None
            or self._background_thread_event_loop is None
            or close_future is None
        ):
            raise MCPClientInitializationError("the client session was not initialized")

        async def run_async() -> T:
            # Fix for strands-agents/sdk-python/issues/995 - cancel all pending invocations if/when the session closes
            invoke_event = asyncio.create_task(coro)
            tasks: list[asyncio.Task | asyncio.Future] = [
                invoke_event,
                close_future,
            ]

            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

            if done.pop() == close_future:
                self._log_debug_with_thread("event loop for the server closed before the invoke completed")
                raise RuntimeError("Connection to the MCP server was closed")
            else:
                return await invoke_event

        invoke_future = asyncio.run_coroutine_threadsafe(coro=run_async(), loop=self._background_thread_event_loop)
        return invoke_future

    def _should_include_tool(self, tool: MCPAgentTool) -> bool:
        """Check if a tool should be included based on constructor filters."""
        return self._should_include_tool_with_filters(tool, self._tool_filters)

    def _should_include_tool_with_filters(self, tool: MCPAgentTool, filters: Optional[ToolFilters]) -> bool:
        """Check if a tool should be included based on provided filters."""
        if not filters:
            return True

        # Apply allowed filter
        if "allowed" in filters:
            if not self._matches_patterns(tool, filters["allowed"]):
                return False

        # Apply rejected filter
        if "rejected" in filters:
            if self._matches_patterns(tool, filters["rejected"]):
                return False

        return True

    def _matches_patterns(self, tool: MCPAgentTool, patterns: list[_ToolMatcher]) -> bool:
        """Check if tool matches any of the given patterns."""
        for pattern in patterns:
            if callable(pattern):
                if pattern(tool):
                    return True
            elif isinstance(pattern, Pattern):
                if pattern.match(tool.mcp_tool.name):
                    return True
            elif isinstance(pattern, str):
                if pattern == tool.mcp_tool.name:
                    return True
        return False

    def _is_session_active(self) -> bool:
        return self._background_thread is not None and self._background_thread.is_alive()

__enter__()

Context manager entry point which initializes the MCP server connection.

TODO: Refactor to lazy initialization pattern following idiomatic Python. Heavy work in enter is non-idiomatic - should move connection logic to first method call instead.

Source code in strands/tools/mcp/mcp_client.py

def __enter__(self) -> "MCPClient":
    """Context manager entry point which initializes the MCP server connection.

    TODO: Refactor to lazy initialization pattern following idiomatic Python.
    Heavy work in __enter__ is non-idiomatic - should move connection logic to first method call instead.
    """
    return self.start()

__exit__(exc_type, exc_val, exc_tb)

Context manager exit point that cleans up resources.

Source code in strands/tools/mcp/mcp_client.py

def __exit__(self, exc_type: BaseException, exc_val: BaseException, exc_tb: TracebackType) -> None:
    """Context manager exit point that cleans up resources."""
    self.stop(exc_type, exc_val, exc_tb)

__init__(transport_callable, *, startup_timeout=30, tool_filters=None, prefix=None, elicitation_callback=None)

Initialize a new MCP Server connection.

Parameters:

Name	Type	Description	Default
transport_callable	Callable[[], MCPTransport]	A callable that returns an MCPTransport (read_stream, write_stream) tuple.	required
startup_timeout	int	Timeout after which MCP server initialization should be cancelled. Defaults to 30.	30
tool_filters	ToolFilters | None	Optional filters to apply to tools.	None
prefix	str | None	Optional prefix for tool names.	None
elicitation_callback	Optional[ElicitationFnT]	Optional callback function to handle elicitation requests from the MCP server.	None

Source code in strands/tools/mcp/mcp_client.py

def __init__(
    self,
    transport_callable: Callable[[], MCPTransport],
    *,
    startup_timeout: int = 30,
    tool_filters: ToolFilters | None = None,
    prefix: str | None = None,
    elicitation_callback: Optional[ElicitationFnT] = None,
) -> None:
    """Initialize a new MCP Server connection.

    Args:
        transport_callable: A callable that returns an MCPTransport (read_stream, write_stream) tuple.
        startup_timeout: Timeout after which MCP server initialization should be cancelled.
            Defaults to 30.
        tool_filters: Optional filters to apply to tools.
        prefix: Optional prefix for tool names.
        elicitation_callback: Optional callback function to handle elicitation requests from the MCP server.
    """
    self._startup_timeout = startup_timeout
    self._tool_filters = tool_filters
    self._prefix = prefix
    self._elicitation_callback = elicitation_callback

    mcp_instrumentation()
    self._session_id = uuid.uuid4()
    self._log_debug_with_thread("initializing MCPClient connection")
    # Main thread blocks until future completes
    self._init_future: futures.Future[None] = futures.Future()
    # Set within the inner loop as it needs the asyncio loop
    self._close_future: asyncio.futures.Future[None] | None = None
    self._close_exception: None | Exception = None
    # Do not want to block other threads while close event is false
    self._transport_callable = transport_callable

    self._background_thread: threading.Thread | None = None
    self._background_thread_session: ClientSession | None = None
    self._background_thread_event_loop: AbstractEventLoop | None = None
    self._loaded_tools: list[MCPAgentTool] | None = None
    self._tool_provider_started = False
    self._consumers: set[Any] = set()

add_consumer(consumer_id, **kwargs)

Add a consumer to this tool provider.

Synchronous to prevent GC deadlocks when called from Agent finalizers.

Source code in strands/tools/mcp/mcp_client.py

def add_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
    """Add a consumer to this tool provider.

    Synchronous to prevent GC deadlocks when called from Agent finalizers.
    """
    self._consumers.add(consumer_id)
    logger.debug("added provider consumer, count=%d", len(self._consumers))

call_tool_async(tool_use_id, name, arguments=None, read_timeout_seconds=None) async

Asynchronously calls a tool on the MCP server.

This method calls the asynchronous call_tool method on the MCP session and converts the result to the MCPToolResult format.

Parameters:

Name	Type	Description	Default
tool_use_id	str	Unique identifier for this tool use	required
name	str	Name of the tool to call	required
arguments	dict[str, Any] | None	Optional arguments to pass to the tool	None
read_timeout_seconds	timedelta | None	Optional timeout for the tool call	None

Returns:

Name	Type	Description
MCPToolResult	MCPToolResult	The result of the tool call

Source code in strands/tools/mcp/mcp_client.py

async def call_tool_async(
    self,
    tool_use_id: str,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: timedelta | None = None,
) -> MCPToolResult:
    """Asynchronously calls a tool on the MCP server.

    This method calls the asynchronous call_tool method on the MCP session
    and converts the result to the MCPToolResult format.

    Args:
        tool_use_id: Unique identifier for this tool use
        name: Name of the tool to call
        arguments: Optional arguments to pass to the tool
        read_timeout_seconds: Optional timeout for the tool call

    Returns:
        MCPToolResult: The result of the tool call
    """
    self._log_debug_with_thread("calling MCP tool '%s' asynchronously with tool_use_id=%s", name, tool_use_id)
    if not self._is_session_active():
        raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

    async def _call_tool_async() -> MCPCallToolResult:
        return await cast(ClientSession, self._background_thread_session).call_tool(
            name, arguments, read_timeout_seconds
        )

    try:
        future = self._invoke_on_background_thread(_call_tool_async())
        call_tool_result: MCPCallToolResult = await asyncio.wrap_future(future)
        return self._handle_tool_result(tool_use_id, call_tool_result)
    except Exception as e:
        logger.exception("tool execution failed")
        return self._handle_tool_execution_error(tool_use_id, e)

call_tool_sync(tool_use_id, name, arguments=None, read_timeout_seconds=None)

Synchronously calls a tool on the MCP server.

This method calls the asynchronous call_tool method on the MCP session and converts the result to the ToolResult format. If the MCP tool returns structured content, it will be included as the last item in the content array of the returned ToolResult.

Parameters:

Name	Type	Description	Default
tool_use_id	str	Unique identifier for this tool use	required
name	str	Name of the tool to call	required
arguments	dict[str, Any] | None	Optional arguments to pass to the tool	None
read_timeout_seconds	timedelta | None	Optional timeout for the tool call	None

Returns:

Name	Type	Description
MCPToolResult	MCPToolResult	The result of the tool call

Source code in strands/tools/mcp/mcp_client.py

def call_tool_sync(
    self,
    tool_use_id: str,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: timedelta | None = None,
) -> MCPToolResult:
    """Synchronously calls a tool on the MCP server.

    This method calls the asynchronous call_tool method on the MCP session
    and converts the result to the ToolResult format. If the MCP tool returns
    structured content, it will be included as the last item in the content array
    of the returned ToolResult.

    Args:
        tool_use_id: Unique identifier for this tool use
        name: Name of the tool to call
        arguments: Optional arguments to pass to the tool
        read_timeout_seconds: Optional timeout for the tool call

    Returns:
        MCPToolResult: The result of the tool call
    """
    self._log_debug_with_thread("calling MCP tool '%s' synchronously with tool_use_id=%s", name, tool_use_id)
    if not self._is_session_active():
        raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

    async def _call_tool_async() -> MCPCallToolResult:
        return await cast(ClientSession, self._background_thread_session).call_tool(
            name, arguments, read_timeout_seconds
        )

    try:
        call_tool_result: MCPCallToolResult = self._invoke_on_background_thread(_call_tool_async()).result()
        return self._handle_tool_result(tool_use_id, call_tool_result)
    except Exception as e:
        logger.exception("tool execution failed")
        return self._handle_tool_execution_error(tool_use_id, e)

get_prompt_sync(prompt_id, args)

Synchronously retrieves a prompt from the MCP server.

Parameters:

Name	Type	Description	Default
prompt_id	str	The ID of the prompt to retrieve	required
args	dict[str, Any]	Optional arguments to pass to the prompt	required

Returns:

Name	Type	Description
GetPromptResult	GetPromptResult	The prompt response from the MCP server

Source code in strands/tools/mcp/mcp_client.py

def get_prompt_sync(self, prompt_id: str, args: dict[str, Any]) -> GetPromptResult:
    """Synchronously retrieves a prompt from the MCP server.

    Args:
        prompt_id: The ID of the prompt to retrieve
        args: Optional arguments to pass to the prompt

    Returns:
        GetPromptResult: The prompt response from the MCP server
    """
    self._log_debug_with_thread("getting MCP prompt synchronously")
    if not self._is_session_active():
        raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

    async def _get_prompt_async() -> GetPromptResult:
        return await cast(ClientSession, self._background_thread_session).get_prompt(prompt_id, arguments=args)

    get_prompt_result: GetPromptResult = self._invoke_on_background_thread(_get_prompt_async()).result()
    self._log_debug_with_thread("received prompt from MCP server")

    return get_prompt_result

list_prompts_sync(pagination_token=None)

Synchronously retrieves the list of available prompts from the MCP server.

This method calls the asynchronous list_prompts method on the MCP session and returns the raw ListPromptsResult with pagination support.

Parameters:

Name	Type	Description	Default
pagination_token	Optional[str]	Optional token for pagination	None

Returns:

Name	Type	Description
ListPromptsResult	ListPromptsResult	The raw MCP response containing prompts and pagination info

Source code in strands/tools/mcp/mcp_client.py

def list_prompts_sync(self, pagination_token: Optional[str] = None) -> ListPromptsResult:
    """Synchronously retrieves the list of available prompts from the MCP server.

    This method calls the asynchronous list_prompts method on the MCP session
    and returns the raw ListPromptsResult with pagination support.

    Args:
        pagination_token: Optional token for pagination

    Returns:
        ListPromptsResult: The raw MCP response containing prompts and pagination info
    """
    self._log_debug_with_thread("listing MCP prompts synchronously")
    if not self._is_session_active():
        raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

    async def _list_prompts_async() -> ListPromptsResult:
        return await cast(ClientSession, self._background_thread_session).list_prompts(cursor=pagination_token)

    list_prompts_result: ListPromptsResult = self._invoke_on_background_thread(_list_prompts_async()).result()
    self._log_debug_with_thread("received %d prompts from MCP server", len(list_prompts_result.prompts))
    for prompt in list_prompts_result.prompts:
        self._log_debug_with_thread(prompt.name)

    return list_prompts_result

list_tools_sync(pagination_token=None, prefix=None, tool_filters=None)

Synchronously retrieves the list of available tools from the MCP server.

This method calls the asynchronous list_tools method on the MCP session and adapts the returned tools to the AgentTool interface.

Parameters:

Name	Type	Description	Default
pagination_token	str | None	Optional token for pagination	None
prefix	str | None	Optional prefix to apply to tool names. If None, uses constructor default. If explicitly provided (including empty string), overrides constructor default.	None
tool_filters	ToolFilters | None	Optional filters to apply to tools. If None, uses constructor default. If explicitly provided (including empty dict), overrides constructor default.	None

Returns:

Type	Description
PaginatedList[MCPAgentTool]	List[AgentTool]: A list of available tools adapted to the AgentTool interface

Source code in strands/tools/mcp/mcp_client.py

def list_tools_sync(
    self,
    pagination_token: str | None = None,
    prefix: str | None = None,
    tool_filters: ToolFilters | None = None,
) -> PaginatedList[MCPAgentTool]:
    """Synchronously retrieves the list of available tools from the MCP server.

    This method calls the asynchronous list_tools method on the MCP session
    and adapts the returned tools to the AgentTool interface.

    Args:
        pagination_token: Optional token for pagination
        prefix: Optional prefix to apply to tool names. If None, uses constructor default.
               If explicitly provided (including empty string), overrides constructor default.
        tool_filters: Optional filters to apply to tools. If None, uses constructor default.
                     If explicitly provided (including empty dict), overrides constructor default.

    Returns:
        List[AgentTool]: A list of available tools adapted to the AgentTool interface
    """
    self._log_debug_with_thread("listing MCP tools synchronously")
    if not self._is_session_active():
        raise MCPClientInitializationError(CLIENT_SESSION_NOT_RUNNING_ERROR_MESSAGE)

    effective_prefix = self._prefix if prefix is None else prefix
    effective_filters = self._tool_filters if tool_filters is None else tool_filters

    async def _list_tools_async() -> ListToolsResult:
        return await cast(ClientSession, self._background_thread_session).list_tools(cursor=pagination_token)

    list_tools_response: ListToolsResult = self._invoke_on_background_thread(_list_tools_async()).result()
    self._log_debug_with_thread("received %d tools from MCP server", len(list_tools_response.tools))

    mcp_tools = []
    for tool in list_tools_response.tools:
        # Apply prefix if specified
        if effective_prefix:
            prefixed_name = f"{effective_prefix}_{tool.name}"
            mcp_tool = MCPAgentTool(tool, self, name_override=prefixed_name)
            logger.debug("tool_rename=<%s->%s> | renamed tool", tool.name, prefixed_name)
        else:
            mcp_tool = MCPAgentTool(tool, self)

        # Apply filters if specified
        if self._should_include_tool_with_filters(mcp_tool, effective_filters):
            mcp_tools.append(mcp_tool)

    self._log_debug_with_thread("successfully adapted %d MCP tools", len(mcp_tools))
    return PaginatedList[MCPAgentTool](mcp_tools, token=list_tools_response.nextCursor)

load_tools(**kwargs) async

Load and return tools from the MCP server.

This method implements the ToolProvider interface by loading tools from the MCP server and caching them for reuse.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional arguments for future compatibility.	{}

Returns:

Type	Description
Sequence[AgentTool]	List of AgentTool instances from the MCP server.

Source code in strands/tools/mcp/mcp_client.py

async def load_tools(self, **kwargs: Any) -> Sequence[AgentTool]:
    """Load and return tools from the MCP server.

    This method implements the ToolProvider interface by loading tools
    from the MCP server and caching them for reuse.

    Args:
        **kwargs: Additional arguments for future compatibility.

    Returns:
        List of AgentTool instances from the MCP server.
    """
    logger.debug(
        "started=<%s>, cached_tools=<%s> | loading tools",
        self._tool_provider_started,
        self._loaded_tools is not None,
    )

    if not self._tool_provider_started:
        try:
            logger.debug("starting MCP client")
            self.start()
            self._tool_provider_started = True
            logger.debug("MCP client started successfully")
        except Exception as e:
            logger.error("error=<%s> | failed to start MCP client", e)
            raise ToolProviderException(f"Failed to start MCP client: {e}") from e

    if self._loaded_tools is None:
        logger.debug("loading tools from MCP server")
        self._loaded_tools = []
        pagination_token = None
        page_count = 0

        while True:
            logger.debug("page=<%d>, token=<%s> | fetching tools page", page_count, pagination_token)
            # Use constructor defaults for prefix and filters in load_tools
            paginated_tools = self.list_tools_sync(
                pagination_token, prefix=self._prefix, tool_filters=self._tool_filters
            )

            # Tools are already filtered by list_tools_sync, so add them all
            for tool in paginated_tools:
                self._loaded_tools.append(tool)

            logger.debug(
                "page=<%d>, page_tools=<%d>, total_filtered=<%d> | processed page",
                page_count,
                len(paginated_tools),
                len(self._loaded_tools),
            )

            pagination_token = paginated_tools.pagination_token
            page_count += 1

            if pagination_token is None:
                break

        logger.debug("final_tools=<%d> | loading complete", len(self._loaded_tools))

    return self._loaded_tools

remove_consumer(consumer_id, **kwargs)

Remove a consumer from this tool provider.

This method is idempotent - calling it multiple times with the same ID has no additional effect after the first call.

Synchronous to prevent GC deadlocks when called from Agent finalizers. Uses existing synchronous stop() method for safe cleanup.

Source code in strands/tools/mcp/mcp_client.py

def remove_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
    """Remove a consumer from this tool provider.

    This method is idempotent - calling it multiple times with the same ID
    has no additional effect after the first call.

    Synchronous to prevent GC deadlocks when called from Agent finalizers.
    Uses existing synchronous stop() method for safe cleanup.
    """
    self._consumers.discard(consumer_id)
    logger.debug("removed provider consumer, count=%d", len(self._consumers))

    if not self._consumers and self._tool_provider_started:
        logger.debug("no consumers remaining, cleaning up")
        try:
            self.stop(None, None, None)  # Existing sync method - safe for finalizers
            self._tool_provider_started = False
            self._loaded_tools = None
        except Exception as e:
            logger.error("error=<%s> | failed to cleanup MCP client", e)
            raise ToolProviderException(f"Failed to cleanup MCP client: {e}") from e

start()

Starts the background thread and waits for initialization.

This method starts the background thread that manages the MCP connection and blocks until the connection is ready or times out.

Returns:

Name	Type	Description
self	MCPClient	The MCPClient instance

Raises:

Type	Description
Exception	If the MCP connection fails to initialize within the timeout period

Source code in strands/tools/mcp/mcp_client.py

def start(self) -> "MCPClient":
    """Starts the background thread and waits for initialization.

    This method starts the background thread that manages the MCP connection
    and blocks until the connection is ready or times out.

    Returns:
        self: The MCPClient instance

    Raises:
        Exception: If the MCP connection fails to initialize within the timeout period
    """
    if self._is_session_active():
        raise MCPClientInitializationError("the client session is currently running")

    self._log_debug_with_thread("entering MCPClient context")
    self._background_thread = threading.Thread(target=self._background_task, args=[], daemon=True)
    self._background_thread.start()
    self._log_debug_with_thread("background thread started, waiting for ready event")
    try:
        # Blocking main thread until session is initialized in other thread or if the thread stops
        self._init_future.result(timeout=self._startup_timeout)
        self._log_debug_with_thread("the client initialization was successful")
    except futures.TimeoutError as e:
        logger.exception("client initialization timed out")
        # Pass None for exc_type, exc_val, exc_tb since this isn't a context manager exit
        self.stop(None, None, None)
        raise MCPClientInitializationError(
            f"background thread did not start in {self._startup_timeout} seconds"
        ) from e
    except Exception as e:
        logger.exception("client failed to initialize")
        # Pass None for exc_type, exc_val, exc_tb since this isn't a context manager exit
        self.stop(None, None, None)
        raise MCPClientInitializationError("the client initialization failed") from e
    return self

stop(exc_type, exc_val, exc_tb)

Signals the background thread to stop and waits for it to complete, ensuring proper cleanup of all resources.

This method is defensive and can handle partial initialization states that may occur if start() fails partway through initialization.

Resources to cleanup: - _background_thread: Thread running the async event loop - _background_thread_session: MCP ClientSession (auto-closed by context manager) - _background_thread_event_loop: AsyncIO event loop in background thread - _close_future: AsyncIO future to signal thread shutdown - _close_exception: Exception that caused the background thread shutdown; None if a normal shutdown occurred. - _init_future: Future for initialization synchronization

Cleanup order: 1. Signal close future to background thread (if session initialized) 2. Wait for background thread to complete 3. Reset all state for reuse

Parameters:

Name	Type	Description	Default
exc_type	Optional[BaseException]	Exception type if an exception was raised in the context	required
exc_val	Optional[BaseException]	Exception value if an exception was raised in the context	required
exc_tb	Optional[TracebackType]	Exception traceback if an exception was raised in the context	required

Source code in strands/tools/mcp/mcp_client.py

def stop(
    self, exc_type: Optional[BaseException], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
) -> None:
    """Signals the background thread to stop and waits for it to complete, ensuring proper cleanup of all resources.

    This method is defensive and can handle partial initialization states that may occur
    if start() fails partway through initialization.

    Resources to cleanup:
    - _background_thread: Thread running the async event loop
    - _background_thread_session: MCP ClientSession (auto-closed by context manager)
    - _background_thread_event_loop: AsyncIO event loop in background thread
    - _close_future: AsyncIO future to signal thread shutdown
    - _close_exception: Exception that caused the background thread shutdown; None if a normal shutdown occurred.
    - _init_future: Future for initialization synchronization

    Cleanup order:
    1. Signal close future to background thread (if session initialized)
    2. Wait for background thread to complete
    3. Reset all state for reuse

    Args:
        exc_type: Exception type if an exception was raised in the context
        exc_val: Exception value if an exception was raised in the context
        exc_tb: Exception traceback if an exception was raised in the context
    """
    self._log_debug_with_thread("exiting MCPClient context")

    # Only try to signal close future if we have a background thread
    if self._background_thread is not None:
        # Signal close future if event loop exists
        if self._background_thread_event_loop is not None:

            async def _set_close_event() -> None:
                if self._close_future and not self._close_future.done():
                    self._close_future.set_result(None)

            # Not calling _invoke_on_background_thread since the session does not need to exist
            # we only need the thread and event loop to exist.
            asyncio.run_coroutine_threadsafe(coro=_set_close_event(), loop=self._background_thread_event_loop)

        self._log_debug_with_thread("waiting for background thread to join")
        self._background_thread.join()

    if self._background_thread_event_loop is not None:
        self._background_thread_event_loop.close()

    self._log_debug_with_thread("background thread is closed, MCPClient context exited")

    # Reset fields to allow instance reuse
    self._init_future = futures.Future()
    self._background_thread = None
    self._background_thread_session = None
    self._background_thread_event_loop = None
    self._session_id = uuid.uuid4()
    self._loaded_tools = None
    self._tool_provider_started = False
    self._consumers = set()

    if self._close_exception:
        exception = self._close_exception
        self._close_exception = None
        raise RuntimeError("Connection to the MCP server was closed") from exception

MCPClientInitializationError

Bases: Exception

Raised when the MCP server fails to initialize properly.

Source code in strands/types/exceptions.py

class MCPClientInitializationError(Exception):
    """Raised when the MCP server fails to initialize properly."""

    pass

MCPToolResult

Bases: ToolResult

Result of an MCP tool execution.

Extends the base ToolResult with MCP-specific structured content support. The structuredContent field contains optional JSON data returned by MCP tools that provides structured results beyond the standard text/image/document content.

Attributes:

Name	Type	Description
structuredContent	NotRequired[dict[str, Any]]	Optional JSON object containing structured data returned by the MCP tool. This allows MCP tools to return complex data structures that can be processed programmatically by agents or other tools.
metadata	NotRequired[dict[str, Any]]	Optional arbitrary metadata returned by the MCP tool. This field allows MCP servers to attach custom metadata to tool results (e.g., token usage, performance metrics, or business-specific tracking information).

Source code in strands/tools/mcp/mcp_types.py

class MCPToolResult(ToolResult):
    """Result of an MCP tool execution.

    Extends the base ToolResult with MCP-specific structured content support.
    The structuredContent field contains optional JSON data returned by MCP tools
    that provides structured results beyond the standard text/image/document content.

    Attributes:
        structuredContent: Optional JSON object containing structured data returned
            by the MCP tool. This allows MCP tools to return complex data structures
            that can be processed programmatically by agents or other tools.
        metadata: Optional arbitrary metadata returned by the MCP tool. This field allows
            MCP servers to attach custom metadata to tool results (e.g., token usage,
            performance metrics, or business-specific tracking information).
    """

    structuredContent: NotRequired[dict[str, Any]]
    metadata: NotRequired[dict[str, Any]]

PaginatedList

Bases: list, Generic[T]

A generic list-like object that includes a pagination token.

This maintains backwards compatibility by inheriting from list, so existing code that expects List[T] will continue to work.

Source code in strands/types/collections.py

class PaginatedList(list, Generic[T]):
    """A generic list-like object that includes a pagination token.

    This maintains backwards compatibility by inheriting from list,
    so existing code that expects List[T] will continue to work.
    """

    def __init__(self, data: List[T], token: Optional[str] = None):
        """Initialize a PaginatedList with data and an optional pagination token.

        Args:
            data: The list of items to store.
            token: Optional pagination token for retrieving additional items.
        """
        super().__init__(data)
        self.pagination_token = token

__init__(data, token=None)

Initialize a PaginatedList with data and an optional pagination token.

Parameters:

Name	Type	Description	Default
data	List[T]	The list of items to store.	required
token	Optional[str]	Optional pagination token for retrieving additional items.	None

Source code in strands/types/collections.py

def __init__(self, data: List[T], token: Optional[str] = None):
    """Initialize a PaginatedList with data and an optional pagination token.

    Args:
        data: The list of items to store.
        token: Optional pagination token for retrieving additional items.
    """
    super().__init__(data)
    self.pagination_token = token

ToolFilters

Bases: TypedDict

Filters for controlling which MCP tools are loaded and available.

Tools are filtered in this order: 1. If 'allowed' is specified, only tools matching these patterns are included 2. Tools matching 'rejected' patterns are then excluded

Source code in strands/tools/mcp/mcp_client.py

class ToolFilters(TypedDict, total=False):
    """Filters for controlling which MCP tools are loaded and available.

    Tools are filtered in this order:
    1. If 'allowed' is specified, only tools matching these patterns are included
    2. Tools matching 'rejected' patterns are then excluded
    """

    allowed: list[_ToolMatcher]
    rejected: list[_ToolMatcher]

ToolProvider

Bases: ABC

Interface for providing tools with lifecycle management.

Provides a way to load a collection of tools and clean them up when done, with lifecycle managed by the agent.

Source code in strands/experimental/tools/tool_provider.py

class ToolProvider(ABC):
    """Interface for providing tools with lifecycle management.

    Provides a way to load a collection of tools and clean them up
    when done, with lifecycle managed by the agent.
    """

    @abstractmethod
    async def load_tools(self, **kwargs: Any) -> Sequence["AgentTool"]:
        """Load and return the tools in this provider.

        Args:
            **kwargs: Additional arguments for future compatibility.

        Returns:
            List of tools that are ready to use.
        """
        pass

    @abstractmethod
    def add_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
        """Add a consumer to this tool provider.

        Args:
            consumer_id: Unique identifier for the consumer.
            **kwargs: Additional arguments for future compatibility.
        """
        pass

    @abstractmethod
    def remove_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
        """Remove a consumer from this tool provider.

        This method must be idempotent - calling it multiple times with the same ID
        should have no additional effect after the first call.

        Provider may clean up resources when no consumers remain.

        Args:
            consumer_id: Unique identifier for the consumer.
            **kwargs: Additional arguments for future compatibility.
        """
        pass

add_consumer(consumer_id, **kwargs) abstractmethod

Add a consumer to this tool provider.

Parameters:

Name	Type	Description	Default
consumer_id	Any	Unique identifier for the consumer.	required
**kwargs	Any	Additional arguments for future compatibility.	{}

Source code in strands/experimental/tools/tool_provider.py

@abstractmethod
def add_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
    """Add a consumer to this tool provider.

    Args:
        consumer_id: Unique identifier for the consumer.
        **kwargs: Additional arguments for future compatibility.
    """
    pass

load_tools(**kwargs) abstractmethod async

Load and return the tools in this provider.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional arguments for future compatibility.	{}

Returns:

Type	Description
Sequence[AgentTool]	List of tools that are ready to use.

Source code in strands/experimental/tools/tool_provider.py

@abstractmethod
async def load_tools(self, **kwargs: Any) -> Sequence["AgentTool"]:
    """Load and return the tools in this provider.

    Args:
        **kwargs: Additional arguments for future compatibility.

    Returns:
        List of tools that are ready to use.
    """
    pass

remove_consumer(consumer_id, **kwargs) abstractmethod

Remove a consumer from this tool provider.

This method must be idempotent - calling it multiple times with the same ID should have no additional effect after the first call.

Provider may clean up resources when no consumers remain.

Parameters:

Name	Type	Description	Default
consumer_id	Any	Unique identifier for the consumer.	required
**kwargs	Any	Additional arguments for future compatibility.	{}

Source code in strands/experimental/tools/tool_provider.py

@abstractmethod
def remove_consumer(self, consumer_id: Any, **kwargs: Any) -> None:
    """Remove a consumer from this tool provider.

    This method must be idempotent - calling it multiple times with the same ID
    should have no additional effect after the first call.

    Provider may clean up resources when no consumers remain.

    Args:
        consumer_id: Unique identifier for the consumer.
        **kwargs: Additional arguments for future compatibility.
    """
    pass

ToolProviderException

Bases: Exception

Exception raised when a tool provider fails to load or cleanup tools.

Source code in strands/types/exceptions.py

class ToolProviderException(Exception):
    """Exception raised when a tool provider fails to load or cleanup tools."""

    pass

ToolResultContent

Bases: TypedDict

Content returned by a tool execution.

Attributes:

Name	Type	Description
document	DocumentContent	Document content returned by the tool.
image	ImageContent	Image content returned by the tool.
json	Any	JSON-serializable data returned by the tool.
text	str	Text content returned by the tool.

Source code in strands/types/tools.py

class ToolResultContent(TypedDict, total=False):
    """Content returned by a tool execution.

    Attributes:
        document: Document content returned by the tool.
        image: Image content returned by the tool.
        json: JSON-serializable data returned by the tool.
        text: Text content returned by the tool.
    """

    document: DocumentContent
    image: ImageContent
    json: Any
    text: str

_ToolFilterCallback

Bases: Protocol

Source code in strands/tools/mcp/mcp_client.py

class _ToolFilterCallback(Protocol):
    def __call__(self, tool: AgentTool, **kwargs: Any) -> bool: ...

mcp_instrumentation()

Apply OpenTelemetry instrumentation patches to MCP components.

This function instruments three key areas of MCP communication: 1. Client-side: Injects tracing context into tool call requests 2. Transport-level: Extracts context from incoming messages 3. Session-level: Manages bidirectional context flow

The patches enable distributed tracing by: - Adding OpenTelemetry context to the _meta field of MCP requests - Extracting and activating context on the server side - Preserving context across async message processing boundaries

This function is idempotent - multiple calls will not accumulate wrappers.

Source code in strands/tools/mcp/mcp_instrumentation.py

def mcp_instrumentation() -> None:
    """Apply OpenTelemetry instrumentation patches to MCP components.

    This function instruments three key areas of MCP communication:
    1. Client-side: Injects tracing context into tool call requests
    2. Transport-level: Extracts context from incoming messages
    3. Session-level: Manages bidirectional context flow

    The patches enable distributed tracing by:
    - Adding OpenTelemetry context to the _meta field of MCP requests
    - Extracting and activating context on the server side
    - Preserving context across async message processing boundaries

    This function is idempotent - multiple calls will not accumulate wrappers.
    """
    global _instrumentation_applied

    # Return early if instrumentation has already been applied
    if _instrumentation_applied:
        return

    def patch_mcp_client(wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any) -> Any:
        """Patch MCP client to inject OpenTelemetry context into tool calls.

        Intercepts outgoing MCP requests and injects the current OpenTelemetry
        context into the request's _meta field for tools/call methods. This
        enables server-side context extraction and trace continuation.

        Args:
            wrapped: The original function being wrapped
            instance: The instance the method is being called on
            args: Positional arguments to the wrapped function
            kwargs: Keyword arguments to the wrapped function

        Returns:
            Result of the wrapped function call
        """
        if len(args) < 1:
            return wrapped(*args, **kwargs)

        request = args[0]
        method = getattr(request.root, "method", None)

        if method != "tools/call":
            return wrapped(*args, **kwargs)

        try:
            if hasattr(request.root, "params") and request.root.params:
                # Handle Pydantic models
                if hasattr(request.root.params, "model_dump") and hasattr(request.root.params, "model_validate"):
                    params_dict = request.root.params.model_dump()
                    # Add _meta with tracing context
                    meta = params_dict.setdefault("_meta", {})
                    propagate.get_global_textmap().inject(meta)

                    # Recreate the Pydantic model with the updated data
                    # This preserves the original model type and avoids serialization warnings
                    params_class = type(request.root.params)
                    try:
                        request.root.params = params_class.model_validate(params_dict)
                    except Exception:
                        # Fallback to dict if model recreation fails
                        request.root.params = params_dict

                elif isinstance(request.root.params, dict):
                    # Handle dict params directly
                    meta = request.root.params.setdefault("_meta", {})
                    propagate.get_global_textmap().inject(meta)

            return wrapped(*args, **kwargs)

        except Exception:
            return wrapped(*args, **kwargs)

    def transport_wrapper() -> Callable[
        [Callable[..., Any], Any, Any, Any], _AsyncGeneratorContextManager[tuple[Any, Any]]
    ]:
        """Create a wrapper for MCP transport connections.

        Returns a context manager that wraps transport read/write streams
        with context extraction capabilities. The wrapped reader will
        automatically extract OpenTelemetry context from incoming messages.

        Returns:
            An async context manager that yields wrapped transport streams
        """

        @asynccontextmanager
        async def traced_method(
            wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any
        ) -> AsyncGenerator[Tuple[Any, Any], None]:
            async with wrapped(*args, **kwargs) as result:
                try:
                    read_stream, write_stream = result
                except ValueError:
                    read_stream, write_stream, _ = result
                yield TransportContextExtractingReader(read_stream), write_stream

        return traced_method

    def session_init_wrapper() -> Callable[[Any, Any, Tuple[Any, ...], dict[str, Any]], None]:
        """Create a wrapper for MCP session initialization.

        Wraps session message streams to enable bidirectional context flow.
        The reader extracts and activates context, while the writer preserves
        context for async processing.

        Returns:
            A function that wraps session initialization
        """

        def traced_method(
            wrapped: Callable[..., Any], instance: Any, args: Tuple[Any, ...], kwargs: dict[str, Any]
        ) -> None:
            wrapped(*args, **kwargs)
            reader = getattr(instance, "_incoming_message_stream_reader", None)
            writer = getattr(instance, "_incoming_message_stream_writer", None)
            if reader and writer:
                instance._incoming_message_stream_reader = SessionContextAttachingReader(reader)
                instance._incoming_message_stream_writer = SessionContextSavingWriter(writer)

        return traced_method

    # Apply patches
    wrap_function_wrapper("mcp.shared.session", "BaseSession.send_request", patch_mcp_client)

    register_post_import_hook(
        lambda _: wrap_function_wrapper(
            "mcp.server.streamable_http", "StreamableHTTPServerTransport.connect", transport_wrapper()
        ),
        "mcp.server.streamable_http",
    )

    register_post_import_hook(
        lambda _: wrap_function_wrapper("mcp.server.session", "ServerSession.__init__", session_init_wrapper()),
        "mcp.server.session",
    )

    # Mark instrumentation as applied
    _instrumentation_applied = True