MCP Server¶

RLM Code includes a full Model Context Protocol (MCP) implementation with both a server (exposing RLM capabilities to external clients) and a client manager (connecting to external MCP servers).

Module¶

rlm_code.mcp.server          -- MCP server and tool definitions
rlm_code.mcp.client_manager  -- Client connection manager
rlm_code.mcp.session_wrapper -- Session lifecycle wrapper
rlm_code.mcp.config          -- Configuration dataclasses
rlm_code.mcp.transports      -- Transport factory (stdio, SSE, WebSocket)
rlm_code.mcp.exceptions      -- MCP-specific error hierarchy

RLM MCP Server¶

The RLMServer exposes five tools over the MCP protocol, allowing Claude Desktop, VS Code, and other MCP clients to execute RLM tasks directly.

Architecture¶

graph LR
    Client["MCP Client<br>(Claude Desktop, VS Code)"] -->|JSON-RPC| Server["RLMServer"]
    Server -->|initialize| Init["Protocol handshake"]
    Server -->|tools/list| List["Return 5 tool schemas"]
    Server -->|tools/call| Dispatch["Route to handler"]
    Dispatch --> Execute["rlm_execute"]
    Dispatch --> Query["rlm_query"]
    Dispatch --> Compare["rlm_compare"]
    Dispatch --> Benchmark["rlm_benchmark"]
    Dispatch --> Trajectory["rlm_trajectory"]
    Execute --> Runner["RLM Runner"]
    Query --> Runner
    Compare --> Runner

Server Configuration¶

from rlm_code.mcp.server import ServerConfig, create_rlm_server

config = ServerConfig(
    name="rlm-code",
    version="1.0.0",
    transport="stdio",
    host="127.0.0.1",
    port=8765,
)

server = create_rlm_server(config)

Running the Server¶

python -m rlm_code.mcp.server.rlm_server

import asyncio
from rlm_code.mcp.server import create_rlm_server

async def main():
    server = create_rlm_server()
    await server.run()

asyncio.run(main())

Claude Desktop Integration¶

Add to your Claude Desktop claude_desktop_config.json:

{
  "mcpServers": {
    "rlm-code": {
      "command": "python",
      "args": ["-m", "rlm_code.mcp.server.rlm_server"]
    }
  }
}

Tools¶

The server exposes five tools via the RLMTools class. Each tool is defined as a ToolDefinition with typed ToolParameter entries.

rlm_execute¶

Execute a task using the RLM paradigm.

Response fields: run_id, completed, answer, steps, paradigm, total_reward

rlm_query¶

Query a large context efficiently using the RLM paradigm.

Response fields: question, answer, context_length, steps_used

rlm_compare¶

Compare different RLM paradigms on the same task.

Response fields: task, comparison_id, results[] (paradigm, success, answer, tokens, duration, iterations), summary

rlm_benchmark¶

Run benchmark presets to evaluate paradigm performance.

Available presets:

Response fields: preset, cases_run, cases_completed, completion_rate, avg_steps, avg_reward, total_time_seconds

rlm_trajectory¶

View or export an RLM execution trajectory.

Response fields: Varies by format -- summary returns stats, tree returns a tree string, json returns events array.

Tool Call Result¶

All tool handlers return a ToolCallResult:

from rlm_code.mcp.server import ToolCallResult

# Success
result = ToolCallResult(success=True, content={"answer": "42"})

# Error
result = ToolCallResult(success=False, content=None, error="Task is required")

The to_mcp_response() method converts to the MCP response format:

• Success: {"content": [{"type": "text", "text": "..."}]}
• Error: {"isError": true, "content": [{"type": "text", "text": "error message"}]}

MCP Client Manager¶

The MCPClientManager connects to external MCP servers and manages their lifecycle.

Constructor¶

from rlm_code.mcp.client_manager import MCPClientManager
from rlm_code.core.config import ConfigManager

manager = MCPClientManager(config_manager=ConfigManager())

Key Methods¶

Connection Lifecycle¶

import asyncio
from rlm_code.mcp.client_manager import MCPClientManager
from rlm_code.mcp.config import MCPServerConfig, MCPTransportConfig

async def example():
    manager = MCPClientManager(config_manager=config)

    # Add server
    server_config = MCPServerConfig(
        name="my-server",
        transport=MCPTransportConfig(
            type="stdio",
            command="npx",
            args=["@my-org/my-mcp-server"],
        ),
    )
    await manager.add_server(server_config)

    # Connect
    session = await manager.connect("my-server")

    # List and call tools
    tools = await manager.list_tools("my-server")
    result = await manager.call_tool("my-server", "my_tool", {"arg": "value"})

    # Cleanup
    await manager.cleanup()

Configuration¶

MCPTransportConfig¶

Transport-specific settings for MCP connections.

from rlm_code.mcp.config import MCPTransportConfig

# Stdio transport
stdio = MCPTransportConfig(
    type="stdio",
    command="npx",
    args=["@anthropic-ai/mcp-filesystem"],
    env={"HOME": "${HOME}"},
)

# SSE transport
sse = MCPTransportConfig(
    type="sse",
    url="https://api.example.com/mcp",
    headers={"Authorization": "Bearer ${API_TOKEN}"},
)

# WebSocket transport
ws = MCPTransportConfig(
    type="websocket",
    url="wss://api.example.com/mcp/ws",
)

MCPServerConfig¶

Server-level configuration wrapping transport settings.

from rlm_code.mcp.config import MCPServerConfig

config = MCPServerConfig(
    name="filesystem",
    description="File system access",
    transport=MCPTransportConfig(type="stdio", command="npx", args=["@anthropic-ai/mcp-filesystem"]),
    enabled=True,
    auto_connect=False,
    timeout_seconds=30,
    retry_attempts=3,
)

Session Wrapper¶

MCPSessionWrapper wraps the MCP ClientSession with additional status tracking and operation delegation.

from rlm_code.mcp.session_wrapper import MCPSessionWrapper

Key capabilities:

• Status tracking -- get_status() returns connection state and metadata
• Tool operations -- list_tools(), call_tool()
• Resource operations -- list_resources(), read_resource()
• Prompt operations -- list_prompts(), get_prompt()
• Lifecycle management -- initialize(), close()

Transport Factory¶

The transport factory creates the appropriate transport for a given configuration:

from rlm_code.mcp.transports.factory import MCPTransportFactory

transport_cm = MCPTransportFactory.create_transport(transport_config)

Exception Hierarchy¶

All MCP exceptions inherit from MCPError and include details dictionaries and troubleshooting messages.

MCPError (base)
 +-- MCPConnectionError      -- Connection failures (per transport type)
 +-- MCPConfigurationError   -- Invalid config, missing fields
 +-- MCPOperationError       -- Tool call or resource access failures
 +-- MCPTransportError       -- Transport-level errors
 +-- MCPTimeoutError         -- Operation timeouts

Each exception class provides a get_troubleshooting_message() method with context-specific guidance:

from rlm_code.mcp.exceptions import MCPConnectionError

try:
    await manager.connect("my-server")
except MCPConnectionError as e:
    print(e.get_troubleshooting_message())
    # Troubleshooting:
    #   1. Ensure the server command is installed and in your PATH
    #   2. Verify the command and arguments in your configuration
    #   ...

YAML Configuration¶

MCP servers can be configured in rlm_config.yaml:

mcp_servers:
  filesystem:
    name: filesystem
    description: "File system access"
    transport:
      type: stdio
      command: npx
      args: ["@anthropic-ai/mcp-filesystem"]
    enabled: true
    auto_connect: false
    timeout_seconds: 30
    retry_attempts: 3

  api-server:
    name: api-server
    description: "Remote API server"
    transport:
      type: sse
      url: "https://api.example.com/mcp"
      headers:
        Authorization: "Bearer ${API_TOKEN}"
    enabled: true
    timeout_seconds: 60

Class Reference¶

RLMServer¶

RLMTools¶