Example: Local MCP — Playwright fan-out

The workflow

Pipeline structure

One stateful MCP session produces a list of URLs; MapOver turns that list into N parallel per-URL screenshot sub-vertices, each on its own playwright-mcp subprocess.

ContextValOp

SearchInput

→

MCPScriptOp

search + extract URLs

→

MapOver

3× MCPScriptOp · screenshot

→

CollectInto

[]ShotResult

The search subprocess maintains a live browser session — navigate, click consent, type the query, wait for results, extract URLs. All five tool calls run against the same playwright-mcp instance. The per-URL screenshot vertices get fresh subprocesses from the pool; each Run owns its session and tears it down when done.

Implementation

Walkthrough

Three patterns to study in main.go.

1. A multi-tool script over one session

The search op embeds library.MCPScriptOp[SearchInput, []string] and assigns its Script in the factory closure. Inside the script, every sess.CallTool invocation reuses the same playwright session — the browser page persists across browser_navigate, browser_evaluate, and browser_type calls.

          
          
          
        
main.go — MCPGoogleSearchURLsOp
type MCPGoogleSearchURLsOp struct {
    library.MCPScriptOp[SearchInput, []string]
}

func newMCPGoogleSearchURLsOp() operator.IOperator {
    op := &MCPGoogleSearchURLsOp{}
    op.Script = func(ctx context.Context, sess library.MCPSession,
                                   in *SearchInput, out *[]string) error {
        // Navigate, dismiss consent, type query, wait, extract URLs.
        if _, _, err := sess.CallTool(ctx, "browser_navigate",
            map[string]any{"url": "https://www.google.com/?hl=en"}); err != nil {
            return fmt.Errorf("browser_navigate: %w", err)
        }
        // ... browser_evaluate (consent), browser_type (query), browser_evaluate (extract URLs)
        *out = urls
        return nil
    }
    return op
}
main.py — MCPGoogleSearchURLsOp
@register_operator("MCPGoogleSearchURLsOp")
class MCPGoogleSearchURLsOp(Operator, BaseModel):
    query: Input = ""
    result: Output = []
    
    command: str = "npx"
    args: List[str] = ["-y", "@playwright/mcp@latest"]
    pool_size: int = 1

    async def run(self, ctx: Any) -> None:
        if not self.query:
            return
            
        client = await global_mcp_pool.acquire(
            command=self.command,
            args=self.args,
            pool_size=self.pool_size
        )
        try:
            await client.call_tool("browser_navigate", {"url": "https://www.google.com/?hl=en"})
            # ... dismiss consent, type query, extract URLs ...
        finally:
            if self.pool_size <= 0:
                await client.disconnect()
index.ts — googleSearchURLs script
const googleSearchURLs: MCPScriptCallback = async (sess, input) => {
  if (!input) throw new Error("empty query");
  await sess.callTool("browser_navigate", { url: "https://www.google.com/?hl=en" });
  try { await sess.callTool("browser_evaluate", { function: dismissConsentJS }); } catch {}
  await sess.callTool("browser_evaluate", { function: waitForSearchBoxJS });
  await sess.callTool("browser_type", { target: googleSearchBoxTarget, text: input, submit: true });
  try { await sess.callTool("browser_evaluate", { function: waitForResultsJS }); } catch {}
  const outcome = await sess.callTool("browser_evaluate", { function: extractURLsJS });
  
  // Decoding URLs from playwright-mcp response
  const decode = (v: any): string[] => {
      // ... parse JSON structure ...
      return [];
  };
  return decode(outcome.structured || outcome.text);
};

2. Extending `MCPScriptOp` with a custom param

The per-URL screenshot op needs an out_dir param that the base MCPScriptOp.Setup doesn't know about. Override Setup, call the embedded base first, then read your own params.

          
          
          
        
main.go — MCPScreenshotURLOp
type MCPScreenshotURLOp struct {
    library.MCPScriptOp[string, ShotResult]
    outDir string
}

func (op *MCPScreenshotURLOp) Setup(p *config.Params) error {
    if err := op.MCPScriptOp.Setup(p); err != nil {
        return err
    }
    op.outDir = p.GetString("out_dir", "")
    if op.outDir == "" || !filepath.IsAbs(op.outDir) {
        return fmt.Errorf("out_dir must be absolute")
    }
    return nil
}
main.py — MCPScreenshotURLOp
@register_operator("MCPScreenshotURLOp")
class MCPScreenshotURLOp(Operator, BaseModel):
    url: Input = ""
    out_dir: str = ""
    result: Output = None # Dict
    
    command: str = "npx"
    args: List[str] = ["-y", "@playwright/mcp@latest"]
    pool_size: int = 8

    async def run(self, ctx: Any) -> None:
        if not self.url:
            return
            
        client = await global_mcp_pool.acquire(
            command=self.command,
            args=self.args,
            pool_size=self.pool_size
        )
        try:
            filename = f"shot-{hashlib.sha1(self.url.encode()).hexdigest()[:16]}.png"
            path = os.path.join(self.out_dir, filename)
            # ... take screenshot ...
        finally:
            if self.pool_size <= 0:
                await client.disconnect()
index.ts — screenshot script
  const screenshotOpts = { 
      ...PLAYWRIGHT_OPTS, 
      poolSize: 8, 
      script: (async (sess, url) => {
          const path = join(outDir, "shot-" + createHash("sha1").update(url || "").digest("hex").slice(0, 16) + ".png");
          try {
              await sess.callTool("browser_navigate", { url });
              await sess.callTool("browser_take_screenshot", { filename: path });
              return { url, path };
          } catch (err) {
              return { url, error: String(err) };
          }
      }) as MCPScriptCallback
  };
  mcp.setupMCPScript(screenshotOpts);

3. MapOver + pool_size

The fan-out vertex declares its sub-graph with SubVertex, binds each item to the url wire, and collects per-URL results into a typed slice. The pool keeps 8 playwright-mcp subprocesses warm so a 3-way fan-out runs with zero cold-start in steady state.

          
          
          
        
main.go — buildGraph()
shootParams["out_dir"] = outDir
shootParams["pool_size"] = "8"

graph.NewBuilder("mcp_google_search_screenshot").
    Vertex("search_input").Op("search_input_const").Output("Result", "search_input").
    Vertex("find_results").Op("MCPGoogleSearchURLsOp").
        Params(playwrightParams).
        Input("Input", "search_input").
        Output("Result", "result_urls").
    Vertex("shoot_each").
        Input("Items", "result_urls").
        MapOver("url").
        SubVertex("shoot").
            Op("MCPScreenshotURLOp").
            Params(shootParams).
            Input("Input", "url").
            Output("Result", "shot_result").
        CollectInto("shot_result", "screenshot_results").
    Build()
main.py — build_graph()
def build_graph(out_dir: str):
    b = Builder("mcp_google_search_screenshot")
    
    b.vertex("search_input").op("ContextValOp").params({"key": "query"}).output("result", "query_wire")
    
    b.vertex("find_urls").op("MCPGoogleSearchURLsOp") \
        .input("query", "query_wire").output("result", "urls_wire")
        
    b.vertex("shoot_each").map_over("urls_wire", "url") \
        .vertex("shoot").op("MCPScreenshotURLOp").params({"out_dir": out_dir}) \
        .input("url", "url").output("result", "shot_result") \
        .collect_into("shot_result", "screenshot_results")
        
    return b.build()
index.ts — build()
  const resultUrls = wf.mcp.script(query, {
    ...PLAYWRIGHT_OPTS,
    script: googleSearchURLs,
    name: "find_results",
  });

  const shotResults = wf.map(resultUrls, (url, ctx) => mcp.mcpScript(url, screenshotOpts, ctx), { name: "shoot_each" });

  const result = wf.op({ shotResults }, ({ shotResults }) => {
      return shotResults.map(r => {
          if (r.path) {
              try { return { ...r, bytes: statSync(r.path).size }; } catch {}
          }
          return r;
      });
  }, { name: "final_result" });

Run it

First run downloads @playwright/mcp@latest plus a browser binary; subsequent runs reuse the cache. No CLAUDE_API_KEY needed. Don't forget the pool shutdown defer.

          
          
          
        
shell
# default: query "Shizuoka", screenshots under ./.playwright-mcp/
go run ./examples/local-mcp-server

# custom query, custom out dir
go run ./examples/local-mcp-server --query "Kyoto autumn" --out-dir /tmp/shots
shell
# default: query "Shizuoka", screenshots under ./.playwright-mcp/
python examples/local_mcp_server/main.py

# custom query, custom out dir
python examples/local_mcp_server/main.py --query "Kyoto autumn" --out-dir /tmp/shots
shell
# default: query "Shizuoka", screenshots under ./.playwright-mcp/
npx tsx examples/local_mcp_server/index.ts

# custom query, custom out dir
npx tsx examples/local_mcp_server/index.ts --query "Kyoto autumn" --out-dir /tmp/shots

In main(): defer library.ShutdownMCPPool(context.Background()) so warm subprocesses drain at exit.

Run it as an MCP server

The same binary is dual-mode. Pass --mcp and instead of running once and exiting it speaks the Model Context Protocol over stdin/stdout, exposing the entire workflow as a single MCP tool, search_and_screenshot. The result is an MCP server that is itself an MCP client: each tool call fans out over MapOver to one playwright-mcp subprocess per URL. An agent gets "search and screenshot N pages" as one composed tool call.

shell

# Speak MCP over stdin/stdout instead of running once
go run ./examples/local-mcp-server --mcp

shell

# Speak MCP over stdin/stdout instead of running once
python examples/local_mcp_server/main.py --mcp

shell

# Speak MCP over stdin/stdout instead of running once
npx tsx examples/local_mcp_server/index.ts --mcp

Register it with any MCP client by pointing the client at that command:

          
          
          
        
.mcp.json
{
  "mcpServers": {
    "local-mcp-server": {
      "command": "go",
      "args": ["run", "./examples/local-mcp-server", "--mcp"]
    }
  }
}
.mcp.json
{
  "mcpServers": {
    "local-mcp-server": {
      "command": "python",
      "args": ["examples/local_mcp_server/main.py", "--mcp"]
    }
  }
}
.mcp.json
{
  "mcpServers": {
    "local-mcp-server": {
      "command": "npx",
      "args": ["tsx", "examples/local_mcp_server/index.ts", "--mcp"]
    }
  }
}

← All Examples Next: Remote MCP →

Local MCP — Playwright fan-out

Pipeline structure

Walkthrough

1. A multi-tool script over one session

2. Extending MCPScriptOp with a custom param

3. MapOver + pool_size

Run it

Run it as an MCP server

2. Extending `MCPScriptOp` with a custom param