strands.tools.mcp.mcp_agent_tool

MCP Agent Tool module for adapting Model Context Protocol tools to the agent framework.

This module provides the MCPAgentTool class which serves as an adapter between MCP (Model Context Protocol) tools and the agent framework's tool interface. It allows MCP tools to be seamlessly integrated and used within the agent ecosystem.

ToolGenerator = AsyncGenerator[Any, None] module-attribute

Generator of tool events with the last being the tool result.

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

ToolResultEvent

Bases: TypedEvent

Event emitted when a tool execution completes.

Source code in strands/types/_events.py

class ToolResultEvent(TypedEvent):
    """Event emitted when a tool execution completes."""

    def __init__(self, tool_result: ToolResult) -> None:
        """Initialize with the completed tool result.

        Args:
            tool_result: Final result from the tool execution
        """
        super().__init__({"type": "tool_result", "tool_result": tool_result})

    @property
    def tool_use_id(self) -> str:
        """The toolUseId associated with this result."""
        return cast(ToolResult, self.get("tool_result"))["toolUseId"]

    @property
    def tool_result(self) -> ToolResult:
        """Final result from the completed tool execution."""
        return cast(ToolResult, self.get("tool_result"))

    @property
    @override
    def is_callback_event(self) -> bool:
        return False

tool_result property

Final result from the completed tool execution.

tool_use_id property

The toolUseId associated with this result.

__init__(tool_result)

Initialize with the completed tool result.

Parameters:

Name	Type	Description	Default
tool_result	ToolResult	Final result from the tool execution	required

Source code in strands/types/_events.py

def __init__(self, tool_result: ToolResult) -> None:
    """Initialize with the completed tool result.

    Args:
        tool_result: Final result from the tool execution
    """
    super().__init__({"type": "tool_result", "tool_result": tool_result})

ToolSpec

Bases: TypedDict

Specification for a tool that can be used by an agent.

Attributes:

Name	Type	Description
description	str	A human-readable description of what the tool does.
inputSchema	JSONSchema	JSON Schema defining the expected input parameters.
name	str	The unique name of the tool.
outputSchema	NotRequired[JSONSchema]	Optional JSON Schema defining the expected output format. Note: Not all model providers support this field. Providers that don't support it should filter it out before sending to their API.

Source code in strands/types/tools.py

class ToolSpec(TypedDict):
    """Specification for a tool that can be used by an agent.

    Attributes:
        description: A human-readable description of what the tool does.
        inputSchema: JSON Schema defining the expected input parameters.
        name: The unique name of the tool.
        outputSchema: Optional JSON Schema defining the expected output format.
            Note: Not all model providers support this field. Providers that don't
            support it should filter it out before sending to their API.
    """

    description: str
    inputSchema: JSONSchema
    name: str
    outputSchema: NotRequired[JSONSchema]

ToolUse

Bases: TypedDict

A request from the model to use a specific tool with the provided input.

Attributes:

Name	Type	Description
input	Any	The input parameters for the tool. Can be any JSON-serializable type.
name	str	The name of the tool to invoke.
toolUseId	str	A unique identifier for this specific tool use request.

Source code in strands/types/tools.py

class ToolUse(TypedDict):
    """A request from the model to use a specific tool with the provided input.

    Attributes:
        input: The input parameters for the tool.
            Can be any JSON-serializable type.
        name: The name of the tool to invoke.
        toolUseId: A unique identifier for this specific tool use request.
    """

    input: Any
    name: str
    toolUseId: str