ractogateway.mcp.client

RactoMCPClient — connect to an MCP server and consume its tools.

Requires the mcp package:

pip install ractogateway[mcp]

Usage (async context manager — recommended for long-lived connections):

from ractogateway.mcp import RactoMCPClient, MCPClientConfig

config = MCPClientConfig(
    transport="stdio",
    command="python",
    args=["-m", "my_package.server"],
)

async with RactoMCPClient(config) as client:
    tools = await client.list_tools()
    result = await client.call_tool("add", {"a": 1, "b": 2})
    registry = await client.to_registry()   # use with any kit

Usage (sync, one-shot — for scripts / REPLs):

client = RactoMCPClient(config)
tools = client.list_tools_sync()
result = client.call_tool_sync("add", {"a": 1, "b": 2})

Note

The sync *_sync() helpers use asyncio.run() and cannot be called from within a running event loop (e.g. inside an async def or a Jupyter notebook with %autoawait). Use the async context manager interface in those environments.

class ractogateway.mcp.client.RactoMCPClient(config)

Bases: object

Connect to an MCP server and consume its tools as ToolSchema objects.

This is an async context manager. Keep it alive to reuse the underlying connection for multiple tool calls (O(1) per call after connection setup). For single calls from synchronous code, use the *_sync() convenience methods.

Parameters:

• config (MCPClientConfig) – Connection configuration (transport, command / URL, env, …).

• (recommended) (Example — async)

• ------------------------------

• :: –

  config = MCPClientConfig(transport=”stdio”, command=”python”,

  args=[“-m”, “my_server”])

  async with RactoMCPClient(config) as client:

  # Reuse this connection for all calls. tools = await client.list_tools() result = await client.call_tool(“search”, {“query”: “AI”}) registry = await client.to_registry()

• one-shot (Example — sync)

• -----------------------

• :: – client = RactoMCPClient(config) tools = client.list_tools_sync()

async list_tools()

List all tools exposed by the MCP server.

Return type:

list[ToolSchema]

Returns:

list[ToolSchema] – Provider-agnostic tool schemas — ready to be registered in any ToolRegistry or passed directly to a developer kit via ChatConfig(tools=…).

Raises:

RuntimeError – If called outside an async with block.

async call_tool(name, arguments=None)

Call a remote MCP tool.

Parameters:

• name (str) – Tool name (must exist on the server).

• arguments (dict[str, Any] | None) – Keyword arguments to pass to the tool. Pass None or {} for tools with no parameters.

Return type:

MCPToolResult

Returns:

MCPToolResult – content contains all text blocks joined by "\n". is_error is True when the server signals a tool error.

Raises:

RuntimeError – If called outside an async with block.

async to_registry()

Return a ToolRegistry populated with all server tools.

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

For high-throughput usage, hold the RactoMCPClient context manager alive and call call_tool() directly.

Return type:

ToolRegistry

Returns:

ToolRegistry – Registry compatible with all three developer kits via ChatConfig(tools=registry).

list_tools_sync()

Synchronous wrapper: connect, list tools, disconnect.

Return type:

list[ToolSchema]

Returns:

list[ToolSchema] – All tool schemas exposed by the server.

Raises:

RuntimeError – If called from within a running event loop.

call_tool_sync(name, arguments=None)

Synchronous wrapper: connect, call tool, disconnect.

Parameters:

• name (str) – Tool name.

• arguments (dict[str, Any] | None) – Tool arguments.

Return type:

MCPToolResult

Returns:

MCPToolResult – Tool output.

Raises:

RuntimeError – If called from within a running event loop.