Leith Document Company

Plumbing

A typed composition language for agent pipelines.

You declare agents and their types. You wire them together with sequential composition and tensor product. The runtime enforces types at every channel boundary. Agents are isolated processes with no shared state.

let claude : !string -> !string = agent {
  provider: "anthropic",
  model: "claude-haiku-4-5",
  runtime_context: true
}

let main : !string -> !string = plumb(input, output) {
  input ; claude ; output
}

Six structural primitives. Two combining forms. Together with processes, that is the whole language.

The six structural primitives: id, copy, merge, discard, empty, and barrier.

The two combining forms: sequential composition and tensor product.

What it solves

Multi-agent systems built with imperative glue code are fragile. Control flow is implicit, types are unchecked, and adding an agent means rewriting the orchestration. Plumbing makes the wiring explicit and typed. You change the pipeline by changing the declaration, not the plumbing code — because the declaration is the plumbing code.

The compiler checks that types match at every boundary before anything runs. No tokens are spent on a pipeline that cannot type-check. At current API rates, the difference between a typed pipeline and an untyped loop is the difference between a viable product and an unsustainable one.

What it looks like at scale

A three-agent debate with structured verdict and feedback loop:

type Verdict = { resolved: bool, verdict: string, topic: string, heat: number }
type Control = { set_temp: number }

let advocate : (!string, !Control) -> !string = agent { ... }
let skeptic  : (!string, !Control) -> !string = agent { ... }
let judge    : !(string, string) -> !Verdict  = agent { ... }
let cool     : !Verdict -> !Control = map({set_temp: heat})

let main : !string -> !string = plumb(input, output) {
  input ; (advocate * skeptic) ; barrier ; judge
  judge ; filter(resolved = false).topic ; (advocate * skeptic)
  judge ; filter(resolved = true).verdict ; output
  judge ; cool ; (advocate@ctrl_in * skeptic@ctrl_in)
}

Graph diagram of the heated debate pipeline showing advocate, skeptic, and judge wired with fan-out, barrier, and feedback loop.

Three agents, four wiring statements. Fan-out and fan-in are inserted automatically. The judge's verdict type is checked at compile time; verdict values are validated at runtime.

Session types

Session type duals: send and recv map to map and filter with barrier synchronisation; select and offer map to coproduct injection and copy with label filtering. Barrier chains: the sequencing operator maps to a chain of barriers, each synchronising the outgoing message with the incoming acknowledgement.

Plumbing supports session types for describing protocols on control channels. A protocol is a sequence of typed send-and-receive steps:

protocol Compaction =
  send Pause . recv PauseAck .
  send GetMemory . recv MemoryDump .
  send SetMemory . recv SetMemoryAck .
  send Resume . recv ResumeAck . end

The compiler translates session types into barrier chains — the synchronisation primitive that ensures each step completes before the next begins. There is a total functor from session types into the plumbing category. Session types are a specification language; barrier chains are the executable form.

This is used in practice for memory management. The compaction protocol above manages an agent's context window: a supervisory homunculus watches token counts, pauses the agent when the context fills up, extracts and compresses memory, writes it back, and resumes.

Use it from Python

import persevere.plumbing as pb
from pathlib import Path

print(pb.call_sync(Path("mistral.plumb"), "Write a poem about category theory"))

Or use the Pydantic AI adapter if you already have a Pydantic AI codebase:

from pydantic_ai import Agent
from plumbing.provider import PlumbingModel

async with PlumbingModel(Path("mistral.plumb")) as model:
    agent = Agent(model)
    result = await agent.run("Write a poem about category theory")
    print(result.output)

Three API patterns: call_sync() for one-shot, run() for streaming, and PlumbingModel for Pydantic AI integration.

Reading

These ideas are explained in more detail in a pair of posts on the n-Category Café:

Can LLMs write plumbing? Yes — 6 out of 25 frontier models scored a perfect 40/40 on one-shot generation of typed pipelines from natural-language descriptions:

The broader research programme of which Plumbing is a part:

Install

pip install persevere_plumbing

Or download the binary directly:

After downloading, unpack and put the plumb binary somewhere on your PATH.

Licence

Plumbing is free for personal use, academic research, and education. All other use requires a commercial licence from Leith Document Company Limited.

For commercial licensing enquiries, contact licensing@leithdocs.com. We are happy to discuss terms.

Reference