pi-agent

pi-agent (pi_*)

Delegate sub-tasks to the pi agent from a Monty plan. Backed by a pool of warm pi --mode rpc workers; every sub-agent run is traced to Logfire under one root span (the host reads LOGFIRE_WRITE_TOKEN; absent ⇒ untraced, no crash). The write token never crosses into the Monty sandbox.

FFI only — the RPC subprocess pool needs dart:io Process, absent on WASM.

Host setup (Dart)

import 'package:dart_monty/dart_monty.dart';
import 'package:pi_agent/pi_agent.dart';

// poolSize = fan-out width; tools = the sub-agents' `pi -t` allowlist.
final runtime = MontyRuntime(
  extensions: [PiExtension(poolSize: 3, tools: 'web_search,summarize')],
);

Capability check (always emit first)

requires([
    "pi_ask", "pi_start", "pi_poll", "pi_steer", "pi_abort", "pi_await",
])

Fan-out: one-shot prompts

pi_ask(prompt: str) -> str

Run one prompt through the pi agent and return its final assistant text. Dispatched as a deferred future, so independent calls run concurrently when awaited together — wall-clock collapses toward the slowest call, not the sum.

requires(["pi_ask"])
import asyncio

questions = ["summarize pi-tui's render model", "what does flterm provide?"]
answers = await asyncio.gather(*[pi_ask(q) for q in questions])
for q, a in zip(questions, answers):
    print(q, "->", a[:80])

Supervision: drive a running sub-agent

pi_start begins a turn non-blocking; the rest operate on its integer handle. These are synchronous — the supervisor loop is sequential on one handle.

pi_start(prompt: str) -> int

Start a sub-agent turn NON-BLOCKING; returns a handle.

pi_poll(handle: int) -> dict

Returns {events: [{type, tool, query, text}], done: bool, text: str}. Blocks briefly for the next event so the loop is event-paced. tool_execution_start events reveal the sub-agent's tool calls (e.g. web_search) so you can react.

pi_steer(handle: int, message: str) -> bool

Inject a steering message; obeyed at the next turn boundary (after the current turn's tool calls, before the next LLM call). Returns true if the handle was live.

pi_abort(handle: int) -> bool

Abort the running turn. Returns true if the handle was live.

pi_await(handle: int) -> str

Drain to the final answer, release the worker back to the pool, return the text.

requires(["pi_start", "pi_poll", "pi_steer", "pi_await"])

h = pi_start("research the pi-tui rendering model")
while True:
    s = pi_poll(h)
    for e in s["events"]:
        if e["type"] == "tool_execution_start" and e.get("tool") == "web_search":
            pi_steer(h, "focus on the npm package @earendil-works/pi-tui")
    if s["done"]:
        break
print(pi_await(h))

Combine with flmux: a pane per sub-agent

Stream a supervised sub-agent's output into its own live flmux pane (see the flmux skill):

requires(["fl_split", "fl_send", "pi_start", "pi_poll", "pi_await"])

sid = fl_split(direction="column")
h = pi_start("research the pi-tui rendering model")
while True:
    s = pi_poll(h)
    if s["text"]:
        fl_send(sid, s["text"])
    if s["done"]:
        break