MCP integration

Two operator shapes

Single call versus stateful session

Pick by how much server-side state your workflow needs across calls.

`MCPCallOp[In, Out]`

Single tool call per Run.

Each Run opens a session, completes the MCP handshake, calls one tool, tears the session down. tool_name, transport params, and timeouts are all set on the vertex. Define a named concrete type that embeds library.MCPCallOp[In, Out] and register it with operator.RegisterOp.

Default dispatch parses string, float64, int, bool, slices and maps thereof, and any struct decodable via json.Unmarshal.
Implement library.MCPArgsFormatter on *In to control argument marshaling.
Implement library.MCPResponseParser on *Out for full control of the response.
Transient errors retried up to max_retries times (default 3).

`MCPScriptOp[In, Out]`

Multiple tool calls share one live session.

One session opens, your user-supplied Script callback runs once, and it can call any number of tools in any order against the same subprocess. Use when calls depend on server state — browser pages, file cursors, open connections.

Script receives a library.MCPSession; CallTool(ctx, name, args) reuses the session.
Tool-level errors surface as *library.MCPToolError; use errors.As to recover.
Only session-start failures are retried; the script runs at most once per Run.
Register via operator.RegisterOpFactory so each vertex gets a fresh, script-bound instance.

Transports

stdio subprocesses or streamable HTTP

Set transport on the vertex to "stdio" (default) for local subprocesses or "http" for a remote MCP server. The other params follow.

stdio — local subprocess

params := map[string]string{
    "transport":       "stdio",
    "command":         "npx",
    "args":            "-y,@playwright/mcp@latest",
    "tool_name":       "browser_navigate",
    "init_timeout_ms": "120000",
    "call_timeout_ms": "90000",
}

http — remote MCP server

params := map[string]string{
    "transport":       "http",
    "url":             "https://docs.mcp.cloudflare.com/mcp",
    "tool_name":       "search_cloudflare_documentation",
    "init_timeout_ms": "30000",
    "call_timeout_ms": "60000",
    // optional auth header(s):
    // "headers": "Authorization=Bearer ${TOKEN}",
}

Warm-replenish pool

Keep cold-start off the critical path

Set pool_size on stdio vertices to maintain a target number of pre-started MCP subprocesses. Vertices that share the same canonical session spec (command, args, env, init_timeout) share warm slots — fan-out runs see no cold-start in steady state.

🔥

`pool_size`

Target capacity per spec. 0 (default) disables pooling — each Run opens and closes its own session.

⏱️

`pool_prewarm`

true (default) fills the pool during Setup so the first Run already has warm sessions. Set false for lazy fill on first use.

🛑

`ShutdownMCPPool`

Call library.ShutdownMCPPool(ctx) from main() (typically via defer) so pre-started subprocesses drain cleanly at exit.

Each borrow is a fresh session — the pool never reuses a session for a second logical call. Pooling is supported only for transport="stdio" in v1.

Custom marshaling

When the tool's wire shape isn't your wire shape

Two interfaces let you decouple your operator's In and Out types from the tool's exact JSON. Implement on the pointer receiver.

MCPArgsFormatter / MCPResponseParser

// Rewrite the args object before it goes on the wire.
func (in *MyInput) FormatMCPArgs() (any, error) {
    return map[string]any{
        "q":   in.Query,
        "top": in.Limit,
    }, nil
}

// Parse the tool's text + structured payload yourself.
func (out *MyOutput) ParseMCPResponse(text string, structured json.RawMessage) error {
    if len(structured) > 0 {
        return json.Unmarshal(structured, out)
    }
    out.Text = text
    return nil
}

Tool-level errors (servers that return IsError=true) surface as *library.MCPToolError with the tool name and the server's text. Use errors.As from inside an MCPScriptOp Script to recover from anticipated failures (element-not-found, page-not-loaded, etc.) without aborting the run.

See it in practice

Local MCP — Playwright fan-out

MCPScriptOp drives one playwright session to scrape Google, then MapOver spawns N parallel sessions to screenshot each URL. Pool size 8; warm-replenish.

Remote MCP — Cloudflare docs

MCPCallOp against the public Cloudflare docs MCP server over streamable HTTP. No subprocess, no API keys.