oats.mcp

MCP (Model Context Protocol) server management, tool resolution, and ranking.

oats.mcp.models

Pydantic models for the MCP tool calling protocol.

Defines the data structures for MCP server configuration, tool definitions, call tracking, ranking, and orchestration state.

class oats.mcp.models.MCPTransport(*values)[source]

Bases: str, Enum

Supported MCP transport types.

Per MCP spec 2025-06-18: SSE is deprecated, replaced by Streamable HTTP. STREAMABLE_HTTP is the recommended transport for remote servers. STDIO remains for local per-user integrations.

STDIO = 'stdio'

STREAMABLE_HTTP = 'streamable-http'

HTTP = 'http'

class oats.mcp.models.MCPServerConfig(**data)[source]

Bases: BaseModel

Configuration for a single MCP server.

name: str

description: str

transport: MCPTransport

url: str | None

command: str | None

args: list[str]

env: dict[str, str]

headers: dict[str, str]

enabled: bool

tags: list[str]

max_concurrent: int

timeout_seconds: int

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

class oats.mcp.models.MCPServersFile(**data)[source]

Bases: BaseModel

Root schema for mcp_servers.json config file.

version: str

servers: dict[str, MCPServerConfig]

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

class oats.mcp.models.ToolParameter(**data)[source]

Bases: BaseModel

A single parameter for a tool.

name: str

type: str

description: str

required: bool

enum: list[str] | None

default: Any | None

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

class oats.mcp.models.MCPToolDefinition(**data)[source]

Bases: BaseModel

Definition of a tool exposed by an MCP server.

name: str

description: str

server_name: str

parameters: dict[str, Any]

tags: list[str]

mcp_function_name: str

call_endpoint: str

list_endpoint: str

call_count: int

success_count: int

avg_latency_ms: float

last_used: float

property success_rate: float: Fraction of calls that succeeded (0.0 if no calls yet).

to_litellm_format()[source]

Convert to LiteLLM/OpenAI function calling format.

Return type:: dict[str, Any]

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

class oats.mcp.models.ToolCallStatus(*values)[source]

Bases: str, Enum

Status of a tool call.

PENDING = 'pending'

RUNNING = 'running'

SUCCESS = 'success'

ERROR = 'error'

STUCK = 'stuck'

RESOLVED = 'resolved'

CIRCUIT_OPEN = 'circuit_open'

DEGRADED = 'degraded'

class oats.mcp.models.ErrorCategory(*values)[source]

Bases: str, Enum

Classification of errors for retry strategy selection.

TRANSIENT = 'transient'

SERVER = 'server'

CLIENT = 'client'

UNKNOWN = 'unknown'

class oats.mcp.models.ToolCallRecord(**data)[source]

Bases: BaseModel

Record of a single tool call for tracking.

call_id: str

tool_name: str

server_name: str

arguments: dict[str, Any]

status: ToolCallStatus

result: str | None

error: str | None

error_category: ErrorCategory

started_at: float

completed_at: float | None

latency_ms: float | None

resolution_chain: list[str]

depth: int

parent_call_id: str | None

idempotency_key: str | None

attempt: int

max_attempts: int

mark_complete(result)[source]

Mark this call as successful, recording latency.

Return type:: None

mark_error(error, category=ErrorCategory.UNKNOWN)[source]

Mark this call as failed, recording the error and category.

Return type:: None

mark_stuck()[source]

Mark this call as stuck (circuit breaker or repeated failures).

Return type:: None

mark_circuit_open(server_name)[source]

Mark this call as blocked by an open circuit breaker.

Return type:: None

mark_degraded(result)[source]

Mark as degraded — partial/cached result returned.

Return type:: None

compute_idempotency_key()[source]

Compute idempotency key from tool name + arguments.

Return type:: str

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

class oats.mcp.models.CircuitState(*values)[source]

Bases: str, Enum

Circuit breaker states per distributed systems pattern.

CLOSED = 'closed'

OPEN = 'open'

HALF_OPEN = 'half_open'

class oats.mcp.models.ToolRankEntry(**data)[source]

Bases: BaseModel

Entry in the tool ranking index.

tool_name: str

server_name: str

score: float

relevance_score: float

reliability_score: float

latency_score: float

inertia_score: float

tags: list[str]

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

class oats.mcp.models.RankingIndex(**data)[source]

Bases: BaseModel

The full ranking index for tool selection.

entries: list[ToolRankEntry]

last_updated: float

top_k(k=10)[source]

Get top-k tools by score.

Return type:: list[ToolRankEntry]

for_query(query, k=10)[source]

Get top-k tools relevant to a query (simple keyword match).

Return type:: list[ToolRankEntry]

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

class oats.mcp.models.OrchestrationSession(**data)[source]

Bases: BaseModel

State for a hub-and-spoke orchestration session.

session_id: str

call_records: list[ToolCallRecord]

total_calls: int

max_depth: int

started_at: float

ranking_index: RankingIndex

timeout_seconds: float

add_record(record)[source]

Append a call record and update aggregate stats (total_calls, max_depth).

Return type:: None

property is_timed_out: bool: Check if session has exceeded wall-clock timeout.

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

class oats.mcp.models.LiteLLMEndpoint(**data)[source]

Bases: BaseModel

Reduced representation of a LiteLLM API endpoint.

path: str

method: str

summary: str

description: str

parameters: dict[str, Any]

request_body: dict[str, Any]

response_schema: dict[str, Any]

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

class oats.mcp.models.LiteLLMFilteredSpec(**data)[source]

Bases: BaseModel

Filtered/reduced LiteLLM API spec - only what we need for tool calling.

version: str

base_url: str

endpoints: list[LiteLLMEndpoint]

schemas: dict[str, Any]

total_original_size_bytes: int

filtered_size_bytes: int

property reduction_ratio: float: Fraction of the original spec size that was removed by filtering.

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

oats.mcp.config

MCP server configuration loader.

Loads MCP server definitions from a JSON config file that can be placed in: 1. .coder/mcp_servers.json (project-level) 2. ~/.config/coder/mcp_servers.json (user-level) 3. Path from MCP_SERVERS_CONFIG env var

Eventually this could be managed via a web app, but for now it’s file-based.

Environment Variable API Key Resolution

For security, API keys in header values can reference environment variables instead of containing plaintext tokens. The pattern is:

"headers": {
    "Authorization": "Bearer MCP_SERVER_API_KEY_PYTHON_SOFTWARE_ENGINEER"
}

At load time, the config resolver detects values matching the pattern Bearer MCP_SERVER_API_KEY_<SERVER_NAME> and replaces them with the actual environment variable value. The env var name is derived from the server name:

Server: python_software_engineer

Env var: MCP_SERVER_API_KEY_PYTHON_SOFTWARE_ENGINEER

If the env var is not set, a warning is logged and the original value is retained (backward compatibility).

This allows the JSON config file to be safely committed to version control.

oats.mcp.config.load_mcp_config(project_dir=None)[source]

Load MCP server configuration from the first available source.

Priority: 1. MCP_SERVERS_CONFIG env var 2. .coder/mcp_servers.json (project) 3. ~/.config/coder/mcp_servers.json (user)

After loading, resolves environment variable placeholders in server headers so that API keys are never stored as plaintext in the config file.

Return type:: MCPServersFile

oats.mcp.config.save_mcp_config(config, path)[source]

Save MCP server configuration to a JSON file.

Return type:: None

oats.mcp.config.add_server_to_config(name, url=None, command=None, description='', transport='http', tags=None, project_dir=None)[source]

Add a new MCP server to the configuration.

Return type:: MCPServersFile

oats.mcp.config.create_default_mcp_config(project_dir)[source]

Create a default mcp_servers.json with example structure.

Return type:: Path

oats.mcp.registry

MCP Server Registry.

Manages discovery, registration, and health checking of MCP servers. This is the central authority for knowing what tools are available across all connected MCP servers.

Key design: LiteLLM (and similar) uses per-function MCP endpoints (e.g. GET /{mcp_function_name}/tools/list and POST /{mcp_function_name}/tools/call), not a single /mcp-rest/tools/call for everything.

The registry discovers available MCP function names, then probes each one for its tools, and stores the correct call_endpoint per tool.

class oats.mcp.registry.MCPServerRegistry(project_dir=None)[source]

Bases: object

Registry of MCP servers and their tools.

Handles: - Loading server configs from mcp_servers.json - Discovering MCP function names from each server - Probing each function for tools via /{function_name}/tools/list - Building a route table: tool_name -> call_endpoint - Health checking servers - Providing a unified tool catalog

__init__(project_dir=None)[source]: Initialize the registry, loading server configs from disk.

async initialize()[source]

Load config and discover all tools.

Return type:: None

add_server(name, config)[source]

Register a new MCP server in the registry.

Return type:: None

remove_server(name)[source]

Remove a server and all its tools from the registry.

Return type:: None

async discover_all()[source]

Discover tools from all registered servers concurrently.

Return type:: dict[str, list[MCPToolDefinition]]

get_tool(name)[source]

Look up a tool by its qualified name.

Return type:: MCPToolDefinition | None

get_route(tool_name)[source]

Get the routing info for a tool (call_endpoint, mcp_function_name).

Return type:: dict[str, str] | None

list_tools(server_name=None)[source]

List all tools, optionally filtered by server name.

Return type:: list[MCPToolDefinition]

list_servers()[source]

Return a copy of all registered server configs.

Return type:: dict[str, MCPServerConfig]

search_tools(query)[source]

Search tools by keyword match against name, description, tags, and function name.

Return type:: list[MCPToolDefinition]

get_server_health()[source]

Return a copy of the current server health status map.

Return type:: dict[str, bool]

async health_check(server_name)[source]

Probe a server’s health endpoint and update its health status.

Return type:: bool

get_tools_for_litellm(server_name=None, max_tools=50)[source]

Return tools in LiteLLM/OpenAI function-calling format.

Return type:: list[dict[str, Any]]

property needs_rediscovery: bool: Check if the tool catalog is stale and needs re-discovery.

oats.mcp.registry.get_mcp_registry(project_dir=None)[source]

Return the process-wide MCP server registry, creating it on first use.

Return type:: MCPServerRegistry

async oats.mcp.registry.init_mcp_registry(project_dir=None)[source]

Get the global registry and initialize it (discover all tools).

Return type:: MCPServerRegistry

oats.mcp.tools

MCP tool implementations that plug into the existing coder tool registry.

These tools expose the MCP orchestration capabilities to the AI agent, allowing it to discover, call, and manage MCP tools during sessions.

class oats.mcp.tools.MCPDiscoverTool[source]

Bases: Tool

Discover available tools across all connected MCP servers.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Discover and list tools across MCP servers, optionally filtered by server or query.

Return type:: ToolResult

class oats.mcp.tools.MCPCallTool[source]

Bases: Tool

Call a tool on a connected MCP server.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Call a specific MCP tool by name with the given arguments.

Return type:: ToolResult

class oats.mcp.tools.MCPCallChainTool[source]

Bases: Tool

Execute a chain of MCP tool calls sequentially.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Execute a chain of tool calls sequentially, passing output between steps.

Return type:: ToolResult

class oats.mcp.tools.MCPFanOutTool[source]

Bases: Tool

Execute multiple MCP tool calls concurrently (fan-out).

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Execute multiple tool calls concurrently (fan-out) and aggregate results.

Return type:: ToolResult

class oats.mcp.tools.MCPRankTool[source]

Bases: Tool

Rank available MCP tools for a specific task.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Rank MCP tools by relevance to a task description using BM25 + inertia.

Return type:: ToolResult

class oats.mcp.tools.MCPSessionSummaryTool[source]

Bases: Tool

Get a summary of the current MCP orchestration session.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Return a summary of an MCP orchestration session (calls, errors, success rate).

Return type:: ToolResult

class oats.mcp.tools.MCPServerManageTool[source]

Bases: Tool

Add or remove MCP servers at runtime.

property name: str: Unique identifier for the tool.

property description: str: Description of what the tool does.

property parameters: dict[str, Any]: JSON Schema for the tool parameters.

async execute(args, ctx)[source]

Manage MCP servers: list, add, remove, or check health.

Return type:: ToolResult

oats.mcp.orchestrator

Hub-and-spoke tool calling orchestrator.

This is the core of the MCP tool calling protocol. It coordinates: - Routing tool calls to the right MCP server - Hub-and-spoke graph traversal (tools calling tools) - Depth-limited recursion (200-1000 calls) - Cross-referencing between MCP servers - Ranking-informed tool selection - Circuit breaker + backoff + stuck detection + resolution - Per-server bulkhead isolation (semaphores) - Watchdog timer + loop detection - Graceful degradation chain - Idempotency keys for safe retries - MD file tracking of all calls

Architecture:

#                    ┌─────────────┐
#                    │ Orchestrator│ (Hub)
#                    │   (Router)  │
#                    └──────┬──────┘
#                           │
#        ┌──────────┬───────┼───────┬──────────┐
#        ▼          ▼       ▼       ▼          ▼
#   ┌─────────┐ ┌──────┐ ┌─────┐ ┌──────┐ ┌───────┐
#   │ MCP Srv │ │ MCP  │ │ MCP │ │ MCP  │ │  MCP  │ (Spokes)
#   │    A    │ │  B   │ │  C  │ │  D   │ │  ...  │
#   └─────────┘ └──────┘ └─────┘ └──────┘ └───────┘

References: - Netflix Hystrix (circuit breaker + bulkhead) - LangGraph (error edges, checkpoint recovery) - AutoTool (tool inertia, sequential patterns)

class oats.mcp.orchestrator.MCPOrchestrator(registry, tracker=None, ranker=None)[source]

Bases: object

Hub-and-spoke tool calling orchestrator.

Manages the full lifecycle of tool calls across multiple MCP servers, including routing, execution, resilience, and tracking.

__init__(registry, tracker=None, ranker=None)[source]: Initialize the orchestrator with its registry, tracker, and ranker.

async initialize()[source]

Initialize the orchestrator and discover tools.

Return type:: None

create_session(session_id=None, timeout_seconds=1800.0)[source]

Create a new orchestration session.

Return type:: OrchestrationSession

get_session(session_id)[source]

Look up an orchestration session by ID.

Return type:: OrchestrationSession | None

async call_tool(tool_name, arguments, session_id, parent_call_id=None, depth=0, task_description='')[source]

Execute a tool call with full orchestration.

Flow: 1. Guard: depth limit, watchdog timeout, loop detection 2. Resolve tool definition 3. Check circuit breaker for target server 4. Acquire bulkhead semaphore 5. Execute with retry (backoff for transient errors) 6. On persistent failure: resolution via ranked alternatives 7. On exhaustion: graceful degradation chain 8. Track everything

Return type:: ToolCallRecord

async call_tool_chain(calls, session_id, task_description='')[source]

Execute a chain of tool calls sequentially.

Return type:: list[ToolCallRecord]

async fan_out(tool_calls, session_id, max_concurrent=10)[source]

Execute multiple tool calls concurrently (fan-out from hub).

Return type:: list[ToolCallRecord]

rank_tools_for_task(task_description, top_k=20)[source]

Rank available tools for a given task description.

Return type:: list[dict[str, Any]]

get_tools_for_call(task_description='', max_tools=20, server_name=None)[source]

Get a reduced set of tool definitions for a LiteLLM chat completion call.

Return type:: list[dict[str, Any]]

get_session_summary(session_id)[source]

Get a summary of an orchestration session.

Return type:: dict[str, Any]

oats.mcp.resolver

Resilience layer for the MCP tool calling protocol.

Implements production-grade failure handling based on distributed systems patterns (Hystrix, resilience4j) adapted for LLM tool calling:

Circuit Breaker (3-state: closed/open/half-open) per server
Exponential backoff with jitter, error-type-aware
Watchdog timer + loop detection
Graceful degradation chain (partial results > cache > structured error)
Iterative tool resolution (BM25-ranked alternatives)

References: - Netflix Hystrix circuit breaker pattern - Portkey: Retries, Fallbacks, and Circuit Breakers in LLM Apps - Self-Healing AI Agents: 7 Error Handling Patterns - AutoTool: Efficient Tool Selection for LLM Agents (2024)

oats.mcp.resolver.classify_error(error, status_code=None)[source]

Classify an error to determine retry strategy.

TRANSIENT: rate limits, temporary unavailability — use backoff
SERVER: persistent server errors — trigger circuit breaker
CLIENT: bad request, not found — don’t retry, fix the call

Return type:: ErrorCategory

class oats.mcp.resolver.CircuitBreaker(failure_threshold=3, cooldown_seconds=60.0, max_cooldown_seconds=300.0, window_seconds=120.0)[source]

Bases: object

Per-server circuit breaker following the Hystrix pattern.

States:

CLOSED: Normal. Count failures in a sliding window.
OPEN: Blocking all calls. Cooldown timer running.
HALF_OPEN: After cooldown, allow one probe request. If probe succeeds -> CLOSED. If fails -> OPEN with longer cooldown.

This replaces the simple StuckDetector. Key improvement: tools automatically RECOVER after a cooldown, instead of staying stuck forever.

__init__(failure_threshold=3, cooldown_seconds=60.0, max_cooldown_seconds=300.0, window_seconds=120.0)[source]: Initialize the circuit breaker with configurable thresholds and cooldowns.

can_call(server_name)[source]

Check if a call to this server is allowed.

Return type:: bool

record_success(server_name)[source]

Record a successful call. Resets circuit to CLOSED.

Return type:: None

record_failure(server_name)[source]

Record a failure. Returns True if circuit just opened.

Return type:: bool

get_state(server_name)[source]

Return the current circuit state for a server (CLOSED by default).

Return type:: CircuitState

get_all_states()[source]

Return a copy of all server circuit states.

Return type:: dict[str, CircuitState]

reset(server_name=None)[source]

Reset circuit state for a server or all servers.

Return type:: None

class oats.mcp.resolver.BackoffStrategy(base_delay=1.0, max_delay=30.0, jitter=0.5, max_retries=5)[source]

Bases: object

Exponential backoff with jitter for transient errors.

delay = min(base * 2^attempt + random(-jitter, +jitter), max_delay)

__init__(base_delay=1.0, max_delay=30.0, jitter=0.5, max_retries=5)[source]: Initialize the backoff strategy with configurable delay parameters.

should_retry(category, attempt)[source]

Determine if we should retry based on error type and attempt count.

Return type:: bool

get_delay(attempt)[source]

Calculate backoff delay for a given attempt number.

Return type:: float

async wait(attempt)[source]

Sleep for the calculated backoff delay.

Return type:: None

class oats.mcp.resolver.LoopDetector(max_repeats=3)[source]

Bases: object

Detects when the same tool+args combination is called repeatedly.

Prevents infinite loops where the agent keeps trying the same failing call with identical arguments.

__init__(max_repeats=3)[source]: Initialize the loop detector with a maximum repeat threshold.

check(tool_name, arguments)[source]

Check if this call is a loop. Returns True if loop detected.

Return type:: bool

reset()[source]

Clear all seen call signatures.

Return type:: None

class oats.mcp.resolver.DegradationChain[source]

Bases: object

When resolution exhausts all alternatives, provide the best possible fallback instead of a hard failure.

Chain: full result > partial result from resolution > cached result > structured error

__init__()[source]: Initialize the degradation chain with an empty cache.

cache_result(tool_name, arguments, result)[source]

Cache a successful result for potential degraded reuse.

Return type:: None

get_cached(tool_name, arguments)[source]

Get a cached result if available and not expired.

Return type:: str | None

best_partial_result(record)[source]

Extract the best partial result from a resolution chain.

Return type:: str | None

structured_error(record)[source]

Build a structured error message with full context.

Return type:: str

class oats.mcp.resolver.ToolResolver(ranker)[source]

Bases: object

Resolves stuck tool calls by discovering alternatives.

Combines circuit breaker, backoff, loop detection, degradation, and BM25-ranked alternative discovery into a single orchestrator.

The resolution protocol: 1. Classify the error (transient/server/client) 2. For transient: backoff + retry same tool 3. For server: record failure in circuit breaker 4. If circuit opens or retries exhausted: find alternatives via BM25 ranking 5. Try top alternative 6. If all alternatives exhausted: degradation chain (cache > partial > error)

__init__(ranker)[source]: Initialize the resolver with sub-components for circuit breaking, backoff, etc.

property circuit: CircuitBreaker: Return the circuit breaker instance.

property backoff: BackoffStrategy: Return the backoff strategy instance.

property loops: LoopDetector: Return the loop detector instance.

property degradation: DegradationChain: Return the degradation chain instance.

can_call_server(server_name)[source]

Check circuit breaker before calling a server.

Return type:: bool

on_call_result(record)[source]

Process a tool call result through the full resilience pipeline.

Returns the updated record (possibly marked as stuck/circuit_open).

Return type:: ToolCallRecord

should_retry(record)[source]

Check if this specific call should be retried with backoff.

Return type:: bool

async wait_for_retry(attempt)[source]

Wait for backoff delay before retrying.

Return type:: None

check_loop(tool_name, arguments)[source]

Check if this call would create a loop.

Return type:: bool

resolve(stuck_record, available_tools, task_description='')[source]

Find alternative tools when one is stuck.

Filters out tools on servers with open circuits.

Return type:: list[ToolRankEntry]

degrade(record)[source]

Apply graceful degradation when all resolution fails.

Returns the record with the best available result attached.

Return type:: ToolCallRecord

should_escalate(call_id, available_count)[source]

Check if resolution has tried enough alternatives to warrant escalation.

Return type:: bool

oats.mcp.ranking

Tool ranking index with inertia tracking.

Maintains a ranking of tools based on: 1. BM25 relevance (keyword match against tool descriptions) 2. Tool inertia (sequential usage patterns from AutoTool paper) 3. Reliability (success rate with Bayesian smoothing) 4. Latency (inverse normalized response time)

The ranking is persisted as an MD file for transparency and debugging.

References:

AutoTool: Efficient Tool Selection for LLM Agents (2024) CIPS = (1-alpha) * Scorefreq + alpha * Scorectx
Gorilla: LLM Connected with Massive APIs (NeurIPS 2024)

class oats.mcp.ranking.ToolRanker[source]

Bases: object

Builds and maintains a ranking index over registered MCP tools.

Scoring dimensions: 1. Relevance: BM25-style text matching against tool name/description/tags 2. Inertia: What tool typically follows the last-used tool (sequential patterns) 3. Reliability: Success rate weighted by recency 4. Latency: Inverse normalized average response time

__init__()[source]: Initialize the ranker with empty index, stats, and inertia graph.

build_index(tools)[source]

Build the ranking index from a list of tool definitions.

Return type:: RankingIndex

rank_for_query(query, tools, top_k=20)[source]

Rank tools for a specific query/task description.

Combines BM25 relevance, tool inertia, reliability, and latency.

Return type:: list[ToolRankEntry]

record_call(record)[source]

Update stats and inertia graph from a completed tool call.

Return type:: None

property index: RankingIndex: Return the current ranking index.

oats.mcp.intent

Intent-aware tool selector for the session processor.

This module bridges user prompts with the MCP tool calling protocol. Instead of sending ALL 32+ tools to the LLM on every request, it:

Queries the BM25 index first — if there’s a strong MCP match, that IS the intent
Falls back to keyword detection for explicit MCP signals
When MCP intent detected: includes MCP meta-tools + standard tools (so the LLM can compare and pick the best approach)
Enriches the system prompt with the pre-classified MCP resource

The index is built at startup from OpenAPI specs. At runtime, the user’s message is classified against the index to find the best MCP resource, and the system prompt is enriched with the exact tool name and endpoint.

oats.mcp.intent.get_coder_tool_index()[source]

Return the path to the coder tool uses index from the environment.

Return type:: str

oats.mcp.intent.select_tools_for_prompt(prompt, all_tools=None, mcp_tools=None, needs_local_tools=False, ranker=None, max_tools=30, project_dir=None, verbose=False)[source]

Select the most relevant tools for a user prompt.

Strategy:

Always include core tools (read, write, edit, bash, etc.)
Check the BM25 index — if there’s a match above threshold, the user wants an MCP resource (even if they didn’t say “mcp”)
Fall back to explicit keyword detection
When MCP detected: include MCP meta-tools AND standard tools (let the LLM decide the best approach)
When NOT MCP: include standard extras only

Return type:: SelectedToolsManifest

oats.mcp.intent.build_mcp_system_context(prompt, mcp_tools=None, project_dir=None)[source]

Build compact MCP context for the system prompt.

Only includes the best-match resource and a brief tool list. Usage instructions are omitted — tool schemas already describe arguments.

Return type:: str

oats.mcp.intent.enrich_mcp_tool_description(tool)[source]

Enrich an MCP tool’s description with routing info.

Return type:: str

oats.mcp.index

MCP Resource Index — built at startup, searched at runtime.

On first boot (or when stale), fetches OpenAPI specs from all configured MCP servers, extracts every endpoint/tool, and builds a searchable BM25 index. This index is persisted to disk so subsequent startups are instant.

The index supports queries like:

gg run -m 'search businesswire for investing news'

Which will:

Classifier detects “search businesswire”
Reranks MCP tools
Auto-selects the matching tool and calls it

Or directly search the index:

gg mcp search "investing"

Architecture

Startup:

Load mcp_servers.json
For each server: fetch /openapi.json
Extract all paths/operations into IndexEntry objects
Build BM25 corpus from names, descriptions, and tags
Persist index to .coder/mcp_index.json

Runtime:

Load index from disk (fast, no network)
User prompt into BM25 query into ranked results
Top result has call_endpoint and mcp_function_name, ready to call

class oats.mcp.index.IndexEntry(name, description, server_name, mcp_function_name, call_endpoint, method='POST', tags=None, parameters=None)[source]

Bases: object

A single searchable entry in the MCP index.

__init__(name, description, server_name, mcp_function_name, call_endpoint, method='POST', tags=None, parameters=None)[source]: Initialize an index entry with the given metadata and build its BM25 corpus.

to_dict()[source]

Serialize this entry to a plain dict for JSON persistence.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]

Deserialize an IndexEntry from a plain dict.

Return type:: IndexEntry

class oats.mcp.index.MCPIndex[source]

Bases: object

Searchable index of all MCP resources across all configured servers.

Built once at startup, persisted to disk, searched at runtime.

__init__()[source]: Initialize an empty MCP index with default BM25 parameters.

search(query, top_k=10)[source]

Search the index with BM25 ranking.

Returns list of (score, entry) tuples sorted by relevance.

Return type:: list[tuple[float, IndexEntry]]

classify(query)[source]

Classify a query to the single best matching MCP resource.

Returns None if no good match found (score too low).

Return type:: IndexEntry | None

to_dict()[source]

Serialize the full index to a plain dict for JSON persistence.

Return type:: dict[str, Any]

classmethod from_dict(data)[source]

Deserialize an MCPIndex from a plain dict, rebuilding BM25 stats.

Return type:: MCPIndex

property is_stale: bool: Check if the index has exceeded its TTL and needs rebuilding.

async oats.mcp.index.build_index(project_dir=None)[source]

Build the MCP index by fetching OpenAPI specs from all configured servers.

This is the main entry point called at startup.

Return type:: MCPIndex

oats.mcp.index.load_index(project_dir=None, verbose=False)[source]

Load the persisted index from disk. Returns None if not found or stale.

Return type:: MCPIndex | None

async oats.mcp.index.get_or_build_index(project_dir=None)[source]

Get the index from disk, or build it if missing/stale.

Return type:: MCPIndex

oats.mcp.fetch

mcp_fetch.py — Fetch tools from a running Playwright MCP HTTP server and emit them as an OpenAPI 3.0 JSON specification.

The Playwright MCP server exposes a Streamable HTTP transport at /mcp (MCP spec 2024-11-05) and a legacy SSE transport at /sse. This module uses the Streamable HTTP transport exclusively.

Protocol flow

POST /mcp — initialize (no session header) → get mcp-session-id from response
POST /mcp — notifications/initialized (with session header)
POST /mcp — tools/list (with session header) → emit OpenAPI JSON

Usage:: python mcp_fetch.py -u http://0.0.0.0:8931 python mcp_fetch.py -u http://0.0.0.0:8931 -o openapi.json -p

oats.mcp.fetch.fetch_tools(url)[source]

Connect to the Playwright MCP server at url and return the raw tool list.

Return type:: list[dict]
Parameters:: url – Base URL of the MCP HTTP server, e.g. http://0.0.0.0:8931.
Returns:: List of MCP tool objects as returned by tools/list.

oats.mcp.fetch.tools_to_openapi(tools, server_url)[source]

Convert a list of MCP tool definitions to an OpenAPI 3.0 specification.

Each tool becomes a POST /{tool_name} path entry. The request body schema is taken directly from the MCP inputSchema (which is already JSON Schema Draft-07 compatible).

Return type:

dict

Parameters:

tools – List of MCP tool objects from tools/list.
server_url – Base URL used to populate the servers block.

Returns:

OpenAPI 3.0 dict ready to be serialised as JSON.

oats.mcp.fetch.main(argv=None)[source]

Entry point — parse args, fetch tools, emit OpenAPI JSON.

Return type:: None

oats.mcp.tracker

MD file tracking system for tool call results.

Uses a single consolidated markdown log per session (append-mode) rather than individual files per call. This avoids filesystem bloat at 200-1000+ calls per session. Similar to how LiteLLM logs to a single structured destination per session rather than per-event files.

Structure:: .coder/mcp_tracking/ ├── {session_id}.md # Consolidated session log (appended per call) ├── {session_id}_ranking.md # Latest ranking snapshot (overwritten) └── _global_stats.md # Cross-session tool stats (overwritten)

Rotation: When a session log exceeds MAX_LOG_SIZE_BYTES, it is rotated to {session_id}.1.md and a fresh file is started. This keeps any single file from growing unbounded during very long sessions.

class oats.mcp.tracker.ToolCallTracker(tracking_dir=None)[source]

Bases: object

Tracks tool calls in a single consolidated markdown file per session.

Each call is appended as a section. The file is human-readable and grep-friendly. Ranking snapshots and global stats are separate files that get overwritten (not appended) since only the latest matters.

__init__(tracking_dir=None)[source]: Initialize the tracker with the given base directory for log files.

init_session(session)[source]

Create tracking dir and write session header.

Return type:: Path

record_call(session, record)[source]

Append a tool call record to the session log.

Return type:: Path

update_ranking(session, ranking)[source]

Write the current ranking index (overwrite, not append).

Return type:: Path

write_session_summary(session)[source]

Append a summary block to the end of the session log.

Return type:: Path

write_global_stats(stats)[source]

Write global cross-session statistics (overwrite).

Return type:: Path