ai-infra / API Reference

MCPClient

from ai_infra.mcp import MCPClient

ai_infra.mcp

MCP Client for connecting to one or more MCP servers. Production-ready features: - Multi-server support with automatic discovery - All transports: stdio, sse, streamable_http - Auto-reconnect with configurable retry - Health checks - Timeout handling - Graceful shutdown via async context manager

python

# Simple usage
    mcp = MCPClient([
        {"command": "npx", "args": ["-y", "@anthropic/mcp-server-filesystem", "/tmp"]},
    ])
    await mcp.discover()
    tools = await mcp.list_tools()

    # With async context manager
    async with MCPClient(configs) as mcp:
        tools = await mcp.list_tools()
    # Automatic cleanup on exit

Constructor

MCPClient(config: list[dict] | list[McpServerConfig], callbacks: Callbacks | CallbackManager | None = None, interceptors: list[ToolCallInterceptor] | None = None, auto_reconnect: bool = False, reconnect_delay: float = 1.0, max_reconnect_attempts: int = 5, tool_timeout: float | None = 60.0, discover_timeout: float | None = 30.0, pool_size: int = 10)

Parameter	Type	Default	Description
`config`required	`list[dict] \|list[McpServerConfig]`	—	List of MCP server configurations.
`callbacks`	`Callbacks \| CallbackManager \|None`	None	Callback handler(s) for MCP events (progress, logging). Receives MCPProgressEvent and MCPLoggingEvent during tool execution. Can be a single Callbacks instance or a CallbackManager.
`interceptors`	`list[ToolCallInterceptor] \|None`	None	List of tool call interceptors for request/response modification.
`auto_reconnect`	`bool`	False	Whether to auto-reconnect on connection failure.
`reconnect_delay`	`float`	1.0	Delay between reconnect attempts in seconds.
`max_reconnect_attempts`	`int`	5	Maximum number of reconnect attempts.
`tool_timeout`	`float\|None`	60.0	Timeout for tool calls in seconds (default: 60.0). Set to None to disable timeout (not recommended).
`discover_timeout`	`float\|None`	30.0	Timeout for server discovery in seconds (default: 30.0). Set to None to disable timeout (not recommended).
`pool_size`	`int`	10	HTTP connection pool size (for future use).

Methods

# Simple usage mcp = MCPClient([ {"command": "npx", "args": ["-y", "@anthropic/mcp-server-filesystem", "/tmp"]}, ]) await mcp.discover() tools = await mcp.list_tools() # With async context manager async with MCPClient(configs) as mcp: tools = await mcp.list_tools() # Automatic cleanup on exit

Parameter

Type

Default

Description

configrequired

list[dict] |list[McpServerConfig]

—

List of MCP server configurations.

callbacks

Callbacks | CallbackManager |None

None

Callback handler(s) for MCP events (progress, logging). Receives MCPProgressEvent and MCPLoggingEvent during tool execution. Can be a single Callbacks instance or a CallbackManager.

interceptors

list[ToolCallInterceptor] |None

None

List of tool call interceptors for request/response modification.

auto_reconnect

bool

False

Whether to auto-reconnect on connection failure.

reconnect_delay

float

1.0

Delay between reconnect attempts in seconds.

max_reconnect_attempts

int

Maximum number of reconnect attempts.

tool_timeout

float|None

60.0

Timeout for tool calls in seconds (default: 60.0). Set to None to disable timeout (not recommended).

discover_timeout

float|None

30.0

Timeout for server discovery in seconds (default: 30.0). Set to None to disable timeout (not recommended).

pool_size

int

HTTP connection pool size (for future use).