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.

ToolFilters

class ToolFilters(TypedDict)

Defined in: src/strands/tools/mcp/mcp_client.py:65

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

MCPClient

class MCPClient(ToolProvider)

Defined in: src/strands/tools/mcp/mcp_client.py:100

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.

__init__

def __init__(transport_callable: Callable[[], MCPTransport],
             *,
             startup_timeout: int = 30,
             tool_filters: ToolFilters | None = None,
             prefix: str | None = None,
             elicitation_callback: ElicitationFnT | None = None,
             tasks_config: TasksConfig | None = None) -> None

Defined in: src/strands/tools/mcp/mcp_client.py:112

Initialize a new MCP Server connection.

Arguments:

• 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.
• tasks_config - Configuration for MCP task-augmented execution for long-running tools. If provided (not None), enables task-augmented execution for tools that support it. See TasksConfig for details. This feature is experimental and subject to change.

__enter__

def __enter__() -> "MCPClient"

Defined in: src/strands/tools/mcp/mcp_client.py:169

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.

__exit__

def __exit__(exc_type: BaseException, exc_val: BaseException,
             exc_tb: TracebackType) -> None

Defined in: src/strands/tools/mcp/mcp_client.py:177

Context manager exit point that cleans up resources.

start

def start() -> "MCPClient"

Defined in: src/strands/tools/mcp/mcp_client.py:181

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

load_tools

async def load_tools(**kwargs: Any) -> Sequence[AgentTool]

Defined in: src/strands/tools/mcp/mcp_client.py:223

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.

Arguments:

• **kwargs - Additional arguments for future compatibility.

Returns:

List of AgentTool instances from the MCP server.

add_consumer

def add_consumer(consumer_id: Any, **kwargs: Any) -> None

Defined in: src/strands/tools/mcp/mcp_client.py:285

Add a consumer to this tool provider.

Synchronous to prevent GC deadlocks when called from Agent finalizers.

remove_consumer

def remove_consumer(consumer_id: Any, **kwargs: Any) -> None

Defined in: src/strands/tools/mcp/mcp_client.py:293

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.

stop

def stop(exc_type: BaseException | None, exc_val: BaseException | None,
         exc_tb: TracebackType | None) -> None

Defined in: src/strands/tools/mcp/mcp_client.py:317

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

Arguments:

• 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

list_tools_sync

def list_tools_sync(
        pagination_token: str | None = None,
        prefix: str | None = None,
        tool_filters: ToolFilters | None = None
) -> PaginatedList[MCPAgentTool]

Defined in: src/strands/tools/mcp/mcp_client.py:381

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.

Arguments:

• 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

list_prompts_sync

def list_prompts_sync(
        pagination_token: str | None = None) -> ListPromptsResult

Defined in: src/strands/tools/mcp/mcp_client.py:439

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.

Arguments:

• pagination_token - Optional token for pagination

Returns:

• ListPromptsResult - The raw MCP response containing prompts and pagination info

get_prompt_sync

def get_prompt_sync(prompt_id: str, args: dict[str, Any]) -> GetPromptResult

Defined in: src/strands/tools/mcp/mcp_client.py:465

Synchronously retrieves a prompt from the MCP server.

Arguments:

• 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

list_resources_sync

def list_resources_sync(
        pagination_token: str | None = None) -> ListResourcesResult

Defined in: src/strands/tools/mcp/mcp_client.py:487

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

This method calls the asynchronous list_resources method on the MCP session and returns the raw ListResourcesResult with pagination support.

Arguments:

• pagination_token - Optional token for pagination

Returns:

• ListResourcesResult - The raw MCP response containing resources and pagination info

read_resource_sync

def read_resource_sync(uri: AnyUrl | str) -> ReadResourceResult

Defined in: src/strands/tools/mcp/mcp_client.py:511

Synchronously reads a resource from the MCP server.

Arguments:

• uri - The URI of the resource to read

Returns:

• ReadResourceResult - The resource content from the MCP server

list_resource_templates_sync

def list_resource_templates_sync(
        pagination_token: str | None = None) -> ListResourceTemplatesResult

Defined in: src/strands/tools/mcp/mcp_client.py:534

Synchronously retrieves the list of available resource templates from the MCP server.

Resource templates define URI patterns that can be used to access resources dynamically.

Arguments:

• pagination_token - Optional token for pagination

Returns:

• ListResourceTemplatesResult - The raw MCP response containing resource templates and pagination info

call_tool_sync

def call_tool_sync(
        tool_use_id: str,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: timedelta | None = None) -> MCPToolResult

Defined in: src/strands/tools/mcp/mcp_client.py:603

Synchronously calls a tool on the MCP server.

This method automatically uses task-augmented execution when appropriate, based on server capabilities and tool-level taskSupport settings.

Arguments:

• 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

call_tool_async

async def call_tool_async(
        tool_use_id: str,
        name: str,
        arguments: dict[str, Any] | None = None,
        read_timeout_seconds: timedelta | None = None) -> MCPToolResult

Defined in: src/strands/tools/mcp/mcp_client.py:636

Asynchronously calls a tool on the MCP server.

This method automatically uses task-augmented execution when appropriate, based on server capabilities and tool-level taskSupport settings.

Arguments:

• 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