MCP — Model Context Protocol

Models

Pydantic models for MCP server and client configuration.

These models carry no optional-dependency imports: they are safe to import even without the mcp package installed.

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

Server

RactoMCPServer — expose a ToolRegistry as a Model Context Protocol server.

Uses the low-level mcp.server.Server API so that our own ToolSchema.to_json_schema() schemas are forwarded verbatim to every MCP client — no re-introspection via FastMCP, no drift.

Requires the mcp package:

pip install ractogateway[mcp]

SSE transport additionally requires starlette + uvicorn:

pip install ractogateway[mcp-sse]

Quick start (stdio — for Claude Desktop / subprocess):

from ractogateway import ToolRegistry
from ractogateway.mcp import RactoMCPServer

registry = ToolRegistry()

@registry.register
def search(query: str, limit: int = 5) -> str:
    '''Search the knowledge base.'''
    return f"top {limit} results for {query!r}"

server = RactoMCPServer.from_registry(registry, name="my-tools")
server.run()  # blocks; talks MCP via stdin/stdout

Claude Desktop claude_desktop_config.json:

{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["-m", "my_package.server"]
    }
  }
}

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

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)[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.

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.

Agent

MCPAgent — agentic tool-execution loop for OpenAI, Google, and Anthropic kits.

MCPAgent bridges the gap between an LLM developer kit and a ToolRegistry (populated from one or more MCP servers) by running the full agentic loop automatically:

LLM call
   │
   ├─ finish_reason == "tool_call"  ──►  execute all tool calls
   │                                         │
   │                                         └──► append results as
   │                                              user follow-up
   │
   └─ finish_reason != "tool_call"  ──►  return final LLMResponse

The loop repeats up to max_turns times to prevent infinite recursion.

Works with all three provider developer kits — the kit is duck-typed via a lightweight _ChatKitProtocol so no provider package is needed at import time.

Usage

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

# Build a ToolRegistry from an MCP server
config = MCPClientConfig(transport="stdio", command="python",
                         args=["-m", "my_server"])
registry = RactoMCPClient(config).list_tools_sync()   # one-shot fetch

# Or build the registry async:
# async with RactoMCPClient(config) as c:
#     registry = await c.to_registry()

kit   = OpenAIDeveloperKit(model="gpt-4o")
agent = MCPAgent(kit, registry, max_turns=8)

response = agent.run(
    ChatConfig(user_message="What is the weather in Tokyo and London?")
)
print(response.content)

Same code works for Google and Anthropic:

from ractogateway.google_developer_kit import GoogleDeveloperKit
kit = GoogleDeveloperKit(model="gemini-2.0-flash")
agent = MCPAgent(kit, registry)

from ractogateway.anthropic_developer_kit import AnthropicDeveloperKit
kit = AnthropicDeveloperKit(model="claude-opus-4-6")
agent = MCPAgent(kit, registry)

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