ractogateway.mcp

RactoGateway MCP — Model Context Protocol server, client, and agentic loop.

This package provides four primary classes:

RactoMCPServer: Expose any ToolRegistry as an MCP server (stdio or SSE transport). Works with tools built for any of the three provider developer kits.
RactoMCPClient: Connect to an MCP server (stdio / SSE / streamable-HTTP) and import its tools as a ToolRegistry compatible with every kit.
MCPMultiClient: Connect to N MCP servers simultaneously and present their tools as a single merged ToolRegistry.
MCPAgent: Automatic agentic loop (LLM → tool call → execute → continue) that works with OpenAI, Google Gemini, and Anthropic Claude developer kits.

Configuration models:

MCPServerConfig: Server name / version metadata.
MCPClientConfig: Transport, command / URL, env vars.
MCPToolResult: Content + error flag returned by tool calls.

Quick start

Server (expose tools via MCP, works with Claude Desktop / Cursor / …):

from ractogateway import ToolRegistry
from ractogateway.mcp import RactoMCPServer

registry = ToolRegistry()

@registry.register
def get_weather(city: str) -> str:
    '''Return current weather for a city.'''
    return f"Sunny in {city}"

server = RactoMCPServer.from_registry(registry, name="weather-tools")
server.run()  # stdio, blocking — add to claude_desktop_config.json

Client + any developer kit:

from ractogateway.mcp import RactoMCPClient, MCPClientConfig, MCPAgent
from ractogateway.openai_developer_kit import OpenAIDeveloperKit
from ractogateway._models.chat import ChatConfig

cfg      = MCPClientConfig(transport="stdio", command="python",
                           args=["-m", "my_server"])
registry = RactoMCPClient(cfg).list_tools_sync()

# Works for OpenAI, Google Gemini, and Anthropic Claude:
kit    = OpenAIDeveloperKit(model="gpt-4o")
agent  = MCPAgent(kit, registry, max_turns=6)
result = agent.run(ChatConfig(user_message="What is the weather in Tokyo?"))
print(result.content)

Multi-server aggregation:

from ractogateway.mcp import MCPMultiClient, MCPClientConfig

configs = [
    MCPClientConfig(transport="stdio", command="python",
                    args=["-m", "math_server"]),
    MCPClientConfig(transport="sse", url="http://localhost:8001/sse"),
]

async with MCPMultiClient(configs) as multi:
    registry = await multi.to_registry()

Installation

pip install ractogateway[mcp]          # core (stdio + SSE client)
pip install ractogateway[mcp-sse]      # server SSE transport (+ starlette/uvicorn)

class ractogateway.mcp.MCPClientConfig(**data)[source]

Bases: BaseModel

Configuration for connecting to an MCP server.

Parameters:

transport (Literal['stdio', 'sse', 'streamable-http']) – MCP transport to use. "stdio" is standard for subprocess-based servers (e.g. Claude Desktop). "sse" / "streamable-http" are for HTTP-based servers.
command (str | None) – Executable to launch the server process — required for stdio.
args (list[str]) – Command-line arguments passed to command (stdio only).
env (dict[str, str]) – Extra environment variables injected into the server process (stdio only). Merged on top of the inherited environment.
url (str | None) – Server URL — required for sse / streamable-http. Example: "http://localhost:8000/sse".

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

transport: Literal['stdio', 'sse', 'streamable-http']

command: str | None

args: list[str]

env: dict[str, str]

url: str | None

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ractogateway.mcp.MCPServerConfig(**data)[source]

Bases: BaseModel

Configuration for a RactoMCPServer instance.

Parameters:

name (str) – Server name shown to MCP clients (e.g. Claude Desktop).
description (str) – Optional human-readable description of the server.
version (str) – Server version string (SemVer recommended).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

name: str

description: str

version: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ractogateway.mcp.MCPToolResult(**data)[source]

Bases: BaseModel

Result returned from calling a remote MCP tool.

Parameters:

content (str) – Text content of the tool response. Multiple content blocks (e.g. from parallel calls) are joined with "\n".
is_error (bool) – True when the tool itself reported an error condition.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

content: str

is_error: bool

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ractogateway.mcp.RactoMCPServer(name, *, description='', version='0.1.0')[source]

Bases: object

Expose a ToolRegistry (or individual functions) as a Model Context Protocol server.

Supported transports

stdio (default) — standard I/O; ideal for Claude Desktop and any subprocess-based MCP client.
sse — HTTP Server-Sent Events; for remote / browser-based clients. Requires pip install ractogateway[mcp-sse].

type name:: str
param name:: Server name visible to MCP clients.
type description:: str
param description:: Optional human-readable description.
type version:: str
param version:: Server version string.

Example

from ractogateway import ToolRegistry
from ractogateway.mcp import RactoMCPServer

registry = ToolRegistry()

@registry.register
def add(a: int, b: int) -> int:
    '''Add two integers.'''
    return a + b

server = RactoMCPServer.from_registry(registry, name="math-tools")
server.run()  # stdio, blocking

classmethod from_registry(registry, *, name='ractogateway-server', description='RactoGateway MCP Server', version='0.1.0')[source]

Build a server from a populated ToolRegistry.

Every registered callable becomes an MCP tool. Tools registered only as Pydantic models (no backing callable) are silently skipped.

Parameters:

registry (ToolRegistry) – A ToolRegistry with one or more registered tools.
name (str) – MCP server name shown to clients.
description (str) – Optional server description.
version (str) – Server version string.

Return type:

RactoMCPServer

Returns:

RactoMCPServer – Ready-to-run server instance.

add_tool(fn, *, name=None, description=None)[source]

Register a single function as an MCP tool.

The function’s type annotations drive the JSON Schema; its docstring provides the description (overridable via description).

Parameters:

fn (Callable[..., Any]) – The callable to expose. Both sync and async functions are supported.
name (str | None) – Override the tool name (defaults to fn.__name__).
description (str | None) – Override the tool description (defaults to the docstring).

Return type:

None

run(transport='stdio', *, host='0.0.0.0', port=8000)[source]

Start the MCP server (blocking).

Parameters:

transport (Literal['stdio', 'sse']) – "stdio" — standard I/O (default; integrates with Claude Desktop and subprocess clients). "sse" — HTTP Server-Sent Events (requires pip install ractogateway[mcp-sse]).
host (str) – Bind host for SSE transport.
port (int) – Bind port for SSE transport.

Return type:

None

get_asgi_app()[source]

Return a Starlette ASGI app for SSE transport.

Use this to mount the MCP server into an existing web application rather than starting a standalone server with run().

Requires pip install ractogateway[mcp-sse].

Return type:: Any

Example

import uvicorn
from ractogateway.mcp import RactoMCPServer

server = RactoMCPServer.from_registry(registry, name="tools")
app = server.get_asgi_app()
uvicorn.run(app, host="0.0.0.0", port=8000)

property tool_names: list[str]: Sorted list of registered tool names.

class ractogateway.mcp.RactoMCPClient(config)[source]

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()[source]

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)[source]

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()[source]

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()[source]

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)[source]

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.

class ractogateway.mcp.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.

class ractogateway.mcp.MCPAgent(kit, registry, *, max_turns=10)[source]

Bases: object

Agentic tool-execution loop compatible with all three developer kits.

Runs the LLM → tool-call → execute → continue loop automatically, returning the final LLMResponse once the LLM produces a non-tool response or max_turns is reached.

Parameters:

kit (Any) – Any developer kit with chat() / achat() methods: OpenAIDeveloperKit, GoogleDeveloperKit, or AnthropicDeveloperKit.
registry (ToolRegistry) – Tool registry containing callables for each tool the LLM can call. Typically populated via RactoMCPClient.to_registry() or MCPMultiClient.to_registry().
max_turns (int) – Maximum number of tool-call rounds before the loop stops and returns the last response. Prevents infinite recursion.

Example

from ractogateway.openai_developer_kit import OpenAIDeveloperKit
from ractogateway.mcp import MCPAgent, RactoMCPClient, MCPClientConfig
from ractogateway._models.chat import ChatConfig

cfg    = MCPClientConfig(transport="stdio", command="python",
                         args=["-m", "my_server"])
reg    = RactoMCPClient(cfg).list_tools_sync()
kit    = OpenAIDeveloperKit(model="gpt-4o")
agent  = MCPAgent(kit, reg, max_turns=6)
result = agent.run(ChatConfig(user_message="Search for recent AI papers"))
print(result.content)

classmethod from_mcp(kit, configs, *, max_turns=10)[source]

Build an agent by fetching tools from one or more MCP servers.

Opens a one-shot connection per config, fetches tools, closes the connection, and constructs the agent with the merged registry.

Note

This is a sync classmethod; it uses asyncio.run() and therefore cannot be called from within a running event loop. In async code, build the registry yourself:

async with MCPMultiClient(configs) as multi:
    registry = await multi.to_registry()
agent = MCPAgent(kit, registry)

Parameters:

kit (Any) – Any RactoGateway developer kit.
configs (list[MCPClientConfig]) – MCP server connection configs.
max_turns (int) – Maximum tool-call rounds.

Return type:

MCPAgent

Returns:

MCPAgent – Ready to call run() or arun().

run(config)[source]

Run the agentic loop synchronously.

Injects the tool registry from this agent into config (overriding config.tools if already set).

Parameters:: config (ChatConfig) – Initial chat config. prompt must be set here or on the kit.
Return type:: LLMResponse
Returns:: LLMResponse – Final response after tool calls are resolved.

async arun(config)[source]

Run the agentic loop asynchronously.

Supports async tool callables (async def); sync callables are called directly.

Parameters:: config (ChatConfig) – Initial chat config.
Return type:: LLMResponse
Returns:: LLMResponse – Final response after tool calls are resolved.

property registry: ToolRegistry: The ToolRegistry used by this agent.

property max_turns: int: Maximum number of tool-call rounds per run() call.