Multi-Agent Orchestration

A Python coordinator that runs Claude Code subprocesses as agents, with the team topology decided per task instead of hard-coded.

Why dynamic topology? Most multi-agent setups always spin up the same fixed hierarchy (a "CEO" plus workers), which burns tokens and latency on simple tasks and is too rigid for complex ones. Here a one-shot dispatcher reads each task first and picks the cheapest structure that can solve it — so "what is 2+2" stays ~free while a real build gets a coordinated team.

Task
  │
  ▼
┌─────────────┐   FAST_PATH ─► dispatcher answers directly (no agents spawn)
│ Dispatcher  │─► SOLO ──────► one dynamically-configured agent
└─────────────┘   TEAM ──────► a dynamic lead coordinates specialists
                                  ▲
                  Monitor (Haiku, every 90s): HEALTHY / ADVISORY / HARD_STOP

How it works

There is no permanent CEO. Each run begins with a one-shot dispatcher that reads the task and outputs ONE of:

• FAST_PATH: — for trivial tasks. The dispatcher answers directly, no agents spawn. ~free for "what is 2+2" type questions.
• SOLO: {...} — one dynamically-configured agent does the work end-to-end. No coordination overhead.
• TEAM: {...} — a custom-prompted lead coordinates specialists via SPAWN_REQUEST / SPAWN_BATCH. The lead's name and prompt are generated by the dispatcher, so it's "lead_engineer" for a coding task, "research_lead" for a research task, etc.

Alongside execution, a monitor (Haiku, cheap) wakes every 90s, reads recent events, and emits one of:

• HEALTHY: — nothing wrong
• ADVISORY: — concern detected (stuck / runaway / scope creep); delivered to the lead at the next spawn approval
• HARD_STOP: — kill the run; only honored after at least one prior advisory (tiered authority — the monitor has to warn first)

Solo agents that discover the task is too big can ESCALATE: to flag that a team would have been better. (Auto-promotion not yet wired — step 3.)

Files

File	What it does
store.py	SQLite schema, event log, task/agent records
roles.py	Static roles + dynamic lead/solo builders
agent.py	Wraps one Claude Code subprocess; parses protocol
coordinator.py	Dispatch → execute → monitor lifecycle
run.py	CLI entrypoint

Static roles

• dispatcher — runs once at start; outputs FAST_PATH / SOLO / TEAM
• monitor — runs every 90s; HEALTHY / ADVISORY / HARD_STOP
• planner — produces ordered sub-task lists
• thinker_inverter — divergent strategy via inversion
• thinker_analogist — divergent strategy via far-domain analogy
• researcher, coder, reviewer, writer — leaf workers

The lead role is dynamic — the dispatcher invents it per-task.

Prerequisites

• Python 3.9+ (standard library only — no pip install needed)
• The Claude Code CLI installed and on your PATH (the orchestrator shells out to claude)
• ANTHROPIC_API_KEY set in your environment

Run

python run.py "your task here"

Optional env vars:

TOKEN_BUDGET=1000000 AGENT_SLOTS=15 MAX_DEPTH=4 python run.py "..."

Example

$ python run.py "what is the capital of France?"
DISPATCH: FAST_PATH        # trivial → answered directly, zero agents spawned
> Paris.

$ python run.py "build a CLI todo app in Python with add/list/done and tests"
DISPATCH: TEAM             # dispatcher invents a 'lead_engineer' lead
  SPAWN coder      → writes todo.py
  SPAWN coder      → writes test_todo.py
  SPAWN reviewer   → checks edge cases
MONITOR: HEALTHY (t+90s)
> Done. 3 files written, 7 tests passing.

(Output abbreviated; the real run streams structured events — inspect them below.)

Inspecting a run

sqlite3 logs/orchestrator.db
> SELECT id, dispatch_decision, status, tokens_used FROM tasks;
> SELECT ts, kind, agent_id, payload FROM events
     WHERE task_id = '...' ORDER BY ts;
> SELECT role, depth, status, tokens_used FROM agents
     WHERE task_id = '...';

What still isn't built (step 3)

• Live web dashboard. Right now everything streams to the terminal. The event log is structured, so a small FastAPI + HTMX viewer is the next obvious thing.
• Auto-promotion on ESCALATE. When a solo agent escalates, currently we just stop and surface the partial output. A future version re-runs the dispatcher with the partial output as context to build a team.
• Advisories actually nudging the lead mid-stream. Right now they're delivered at the next spawn approval (prepended to the spawn's prompt). True mid-stream nudges require multi-turn Claude Code, not -p.
• Result piping back to the lead. When a child agent completes, its output is logged but not yet re-prompted into the lead. The lead currently has to infer success from the event log. Step 3 fixes this with a proper "child finished, here's their output" re-prompt.

Known sharp edges

• Dispatcher can guess wrong. If it picks SOLO for something that really needed a team, the solo agent will likely hit its budget and return a partial answer. Watch for this in early runs.
• Monitor false positives. Haiku-on-snapshot is cheap but imprecise. The advisory-then-hard-stop tiering exists to absorb this.
• claude binary path. If your install is in a non-standard location, edit agent.py _build_command to use the full path.

License

MIT