ractogateway.mcp.multi_client

MCPMultiClient — aggregate tools from multiple MCP servers into one ToolRegistry.

Connects to N servers in parallel, merges their tool schemas, and routes call_tool requests back to whichever server originally advertised the tool. The resulting ToolRegistry is compatible with all three developer kits (OpenAIDeveloperKit, GoogleDeveloperKit, AnthropicDeveloperKit).

Requires the mcp package:

pip install ractogateway[mcp]

Example

from ractogateway.mcp import MCPMultiClient, MCPClientConfig

configs = [
    MCPClientConfig(transport="stdio", command="python",
                    args=["-m", "pkg.math_server"]),
    MCPClientConfig(transport="stdio", command="python",
                    args=["-m", "pkg.search_server"]),
]

async with MCPMultiClient(configs) as multi:
    tools    = await multi.list_tools()
    registry = await multi.to_registry()   # use with any kit
    result   = await multi.call_tool("search", {"query": "AI"})

Sync (one-shot):

multi = MCPMultiClient(configs)
tools = multi.list_tools_sync()

class ractogateway.mcp.multi_client.MCPMultiClient(configs)[source]

Bases: object

Connect to multiple MCP servers and present them as a single tool surface.

Tools from all servers are merged into one flat namespace. If two servers advertise the same tool name, the later server’s definition wins (and a warning is embedded in the tool description noting the override).

Routing is O(1): an internal dict[tool_name → server_index] maps each tool back to its origin server for call_tool dispatch.

Parameters:: configs (list[MCPClientConfig]) – One MCPClientConfig per server. At least one config is required.

async list_tools()[source]

Return the merged list of tool schemas from all servers.

Return type:: list[ToolSchema]
Returns:: list[ToolSchema] – Deduplicated (last-server-wins) tool schemas sorted by name.

async call_tool(name, arguments=None)[source]

Call a tool on whichever server originally advertised it.

Routing is O(1) via the internal tool_name → server_index map.

Parameters:

name (str) – Tool name (must exist in the merged namespace).
arguments (dict[str, Any] | None) – Tool arguments; None or {} for parameterless tools.

Return type:

MCPToolResult

Returns:

MCPToolResult – Tool output.

Raises:

KeyError – If name is not in the merged tool namespace.
RuntimeError – If called outside an async with block.

async to_registry()[source]

Return a merged ToolRegistry with remote callables.

Each callable in the registry makes a fresh one-shot connection to the correct origin server when invoked. This keeps the registry self-contained and usable outside an async with block.

Return type:: ToolRegistry
Returns:: ToolRegistry – Merged registry compatible with all three developer kits.

list_tools_sync()[source]

Synchronous wrapper: connect all, list merged tools, disconnect all.

Raises:: RuntimeError – If called from within a running event loop.
Return type:: list[ToolSchema]

call_tool_sync(name, arguments=None)[source]

Synchronous wrapper: connect all, call tool, disconnect all.

Raises:: RuntimeError – If called from within a running event loop.
Return type:: MCPToolResult

property tool_names: list[str]: Sorted list of all tool names across all servers.

property server_count: int: Number of configured MCP servers.