graflag.mcp_server

GraFlag as an MCP server: the platform’s operations as tools for AI agents.

graflag mcp serves the Model Context Protocol over stdio, so an MCP client – Claude Code (claude mcp add graflag -- graflag mcp), Claude Desktop or any other – can list methods and datasets, launch and watch runs, evaluate and verify them, and read what they produced.

Three rules shape this module.

stdout is the protocol. Nothing else may write to it, and nothing may read the client’s requests off stdin. _claim_stdio() gives the protocol private copies of both pipes and points file descriptors 0 and 1 elsewhere, and every core call made here waits quietly (follow=False) instead of streaming logs.

Long operations do not hold a tool call. run_experiment and evaluate_experiment start a background job and return at once; wait_for_experiment blocks for a bounded time, so an agent can wait without polling in a loop.

What an LLM sends is screened like what a browser sends. Every name goes through graflag.utils.valid_name() and every parameter through _clean_params() before anything builds a remote command. Operations that delete data, set up the cluster or run arbitrary code (clear --apply, stop --rm, setup, register_metric) are not offered at all.

graflag.mcp_server.MAX_LIST = 200

Ceilings on what one tool call returns or waits for. An agent reads every byte of a reply, and a client gives up on a call that takes too long.

class graflag.mcp_server.Job(experiment: str, kind: str, state: str = 'running', error: str | None = None, started: float = <factory>, finished: float | None = None, thread: Thread | None = None)

Bases: object

A run or an evaluation this server started and has not forgotten.

experiment: str

kind: str

state: str = 'running'

error: str | None = None

started: float

finished: float | None = None

thread: Thread | None = None

to_dict() → dict

class graflag.mcp_server.JobRegistry

Bases: object

What this server has started, so a failure is reported even when it happened before the experiment wrote anything to disk.

start(experiment: str, kind: str, work: Callable[[], Any]) → Job

for_experiment(experiment: str) → List[Job]

graflag.mcp_server.build_server(factory: Callable[[], Any], guard_stdout: bool = False) → mcp.server.fastmcp.FastMCP

The MCP server, over a GraFlag made by factory on first use.

guard_stdout moves sys.stdout to stderr once the stdio transport holds the protocol pipe (see _claim_stdio()); tests that drive the server in memory leave it off.

graflag.mcp_server.serve(config_file: str | None = None)

Serve GraFlag’s tools over stdio until the client disconnects.