MCP Tool Provider

The stirrup.tools.mcp module provides MCP (Model Context Protocol) integration.

Note

Requires pip install stirrup[mcp] (or: uv add stirrup[mcp])

MCPToolProvider

stirrup.tools.mcp.MCPToolProvider

MCPToolProvider(
    config: MCPConfig, server_names: list[str] | None = None
)

Bases: ToolProvider

MCP tool provider that manages connections to multiple MCP servers.

MCPToolProvider connects to MCP servers and exposes each server's tools as individual Tool objects.

Usage with Agent (preferred): from stirrup.clients.chat_completions_client import ChatCompletionsClient

client = ChatCompletionsClient(model="gpt-5")
agent = Agent(
    client=client,
    name="assistant",
    tools=[*DEFAULT_TOOLS, MCPToolProvider.from_config("mcp.json")],
)

async with agent.session(output_dir="./output") as session:
    await session.run("Use MCP tools")

Standalone usage with connect() context manager: provider = MCPToolProvider.from_config(Path("mcp.json")) async with provider.connect() as provider: tools = provider.get_all_tools() # Use tools...

Initialize the MCP manager.

Parameters:

Name	Type	Description	Default
config	MCPConfig	MCPConfig instance.	required
server_names	list[str] | None	Which servers to connect to. If None, connects to all servers in config.	None

Source code in src/stirrup/tools/mcp.py

def __init__(
    self,
    config: MCPConfig,
    server_names: list[str] | None = None,
) -> None:
    """Initialize the MCP manager.

    Args:
        config: MCPConfig instance.
        server_names: Which servers to connect to. If None, connects to all servers in config.
    """
    self._config = config
    self._server_names = server_names
    self._servers: dict[str, ClientSession] = {}
    self._tools: dict[str, list[dict[str, Any]]] = {}
    self._exit_stack: AsyncExitStack | None = None

from_config classmethod

from_config(
    config_path: Path | str,
    server_names: list[str] | None = None,
) -> Self

Create an MCPToolProvider from a config file.

Parameters:

Name	Type	Description	Default
config_path	Path | str	Path to the MCP config file.	required
server_names	list[str] | None	Which servers to connect to. If None, connects to all servers in config.	None

Returns:

Type	Description
Self	MCPToolProvider instance.

Source code in src/stirrup/tools/mcp.py

@classmethod
def from_config(cls, config_path: Path | str, server_names: list[str] | None = None) -> Self:
    """Create an MCPToolProvider from a config file.

    Args:
        config_path: Path to the MCP config file.
        server_names: Which servers to connect to. If None, connects to all servers in config.

    Returns:
        MCPToolProvider instance.
    """
    config = MCPConfig.model_validate_json(Path(config_path).read_text())

    return cls(config=config, server_names=server_names)

__aenter__ async

__aenter__() -> list[Tool[Any, ToolUseCountMetadata]]

Enter async context: connect to MCP servers and return all tools.

Returns:

Type	Description
list[Tool[Any, ToolUseCountMetadata]]	List of Tool objects, one for each tool available across all connected servers.

Source code in src/stirrup/tools/mcp.py

async def __aenter__(self) -> list[Tool[Any, ToolUseCountMetadata]]:
    """Enter async context: connect to MCP servers and return all tools.

    Returns:
        List of Tool objects, one for each tool available across all connected servers.
    """
    self._exit_stack = AsyncExitStack()
    await self._exit_stack.__aenter__()

    config = self._config
    servers_to_connect = self._server_names or list(config.mcp_servers.keys())

    for name in servers_to_connect:
        if name not in config.mcp_servers:
            raise KeyError(f"Server '{name}' not found in config. Available: {list(config.mcp_servers.keys())}")

        server_config = config.mcp_servers[name]

        # Connect to server based on transport type
        match server_config:
            case StdioServerConfig():
                server_params = StdioServerParameters(
                    command=server_config.command,
                    args=server_config.args,
                    env=server_config.env,
                    cwd=server_config.cwd,
                    encoding=server_config.encoding,
                )
                read, write = await self._exit_stack.enter_async_context(stdio_client(server_params))
            case SseServerConfig():
                read, write = await self._exit_stack.enter_async_context(
                    sse_client(
                        url=server_config.url,
                        headers=server_config.headers,
                        timeout=server_config.timeout,
                        sse_read_timeout=server_config.sse_read_timeout,
                    )
                )
            case StreamableHttpServerConfig():
                read, write, _ = await self._exit_stack.enter_async_context(
                    streamablehttp_client(
                        url=server_config.url,
                        headers=server_config.headers,
                        timeout=server_config.timeout,
                        sse_read_timeout=server_config.sse_read_timeout,
                        terminate_on_close=server_config.terminate_on_close,
                    )
                )
            case WebSocketServerConfig():
                if websocket_client is None:
                    raise ImportError(
                        f"WebSocket transport for server '{name}' requires the 'websockets' package. "
                        "Install with: pip install websockets"
                    )
                read, write = await self._exit_stack.enter_async_context(websocket_client(url=server_config.url))

        session = await self._exit_stack.enter_async_context(ClientSession(read, write))
        await session.initialize()

        # Cache session and available tools
        self._servers[name] = session
        response = await session.list_tools()
        self._tools[name] = [
            {"name": t.name, "description": t.description, "schema": t.inputSchema} for t in response.tools
        ]

    return self.get_all_tools()

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit async context: disconnect from MCP servers.

Source code in src/stirrup/tools/mcp.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context: disconnect from MCP servers."""
    self._servers.clear()
    self._tools.clear()
    if self._exit_stack:
        await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
        self._exit_stack = None

Configuration Models

MCPConfig

stirrup.tools.mcp.MCPConfig

Bases: BaseModel

Root configuration matching mcp.json format.

mcp_servers class-attribute instance-attribute

mcp_servers: dict[str, MCPServerConfig] = Field(
    alias="mcpServers"
)

Map of server names to their configurations.

_infer_transport_types classmethod

_infer_transport_types(
    data: dict[str, Any],
) -> dict[str, Any]

Convert raw server configs to appropriate typed instances.

Source code in src/stirrup/tools/mcp.py

@model_validator(mode="before")
@classmethod
def _infer_transport_types(cls, data: dict[str, Any]) -> dict[str, Any]:
    """Convert raw server configs to appropriate typed instances."""
    if "mcpServers" in data:
        data["mcpServers"] = {
            name: _infer_server_config(config) if isinstance(config, dict) else config
            for name, config in data["mcpServers"].items()
        }
    return data

Server Configurations

StdioServerConfig

For local command-based MCP servers:

stirrup.tools.mcp.StdioServerConfig

Bases: BaseModel

Configuration for stdio-based MCP servers (local process).

command instance-attribute

command: str

Command to run the MCP server (e.g., "npx", "python").

args class-attribute instance-attribute

args: list[str] = Field(default_factory=list)

Arguments to pass to the command.

env class-attribute instance-attribute

env: dict[str, str] | None = None

Environment variables to set for the server process.

cwd class-attribute instance-attribute

cwd: str | None = None

Working directory for the server process.

encoding class-attribute instance-attribute

encoding: str = 'utf-8'

Text encoding for messages.

SseServerConfig

For SSE (Server-Sent Events) based MCP servers:

stirrup.tools.mcp.SseServerConfig

Bases: BaseModel

Configuration for SSE-based MCP servers (HTTP GET with Server-Sent Events).

url instance-attribute

url: str

The SSE endpoint URL (must end with /sse).

headers class-attribute instance-attribute

headers: dict[str, str] | None = None

Optional HTTP headers.

timeout class-attribute instance-attribute

timeout: float = 5.0

HTTP timeout for regular operations (seconds).

sse_read_timeout class-attribute instance-attribute

sse_read_timeout: float = 300.0

Timeout for SSE read operations (seconds).

StreamableHttpServerConfig

For streamable HTTP MCP servers:

stirrup.tools.mcp.StreamableHttpServerConfig

Bases: BaseModel

Configuration for Streamable HTTP MCP servers (HTTP POST with optional SSE responses).

url instance-attribute

url: str

The endpoint URL.

headers class-attribute instance-attribute

headers: dict[str, str] | None = None

Optional HTTP headers.

timeout class-attribute instance-attribute

timeout: float = 30.0

HTTP timeout (seconds).

sse_read_timeout class-attribute instance-attribute

sse_read_timeout: float = 300.0

SSE read timeout (seconds).

terminate_on_close class-attribute instance-attribute

terminate_on_close: bool = True

Close session when transport closes.

WebSocketServerConfig

For WebSocket-based MCP servers:

stirrup.tools.mcp.WebSocketServerConfig

Bases: BaseModel

Configuration for WebSocket-based MCP servers.

url instance-attribute

url: str

The WebSocket URL (must start with ws:// or wss://).