Athena Loops — backend-agnostic AI agent orchestration loop for Python

agentloop is a lightweight Python framework for multi-agent orchestration. It implements the orchestrator → worker → reviewer pattern (the AI agent orchestration loop) as a deterministic harness with a closed feedback loop: a goal is decomposed into subtasks, fanned out to worker subagents, aggregated, and run through a review gate that loops until the work meets its success criteria. One loop drives any LLM backend — Anthropic Claude, Claude Code, Codex, opencode, or aider — through a single Agent interface, and it ships as both an MCP server and a plain CLI so any coding agent can call it.

The design principle: the loop is a harness (deterministic code), not a skill. A prompt can describe "decompose, review, loop until done" but can't guarantee it. So the control flow lives in code, and the model-facing judgement (how to decompose, the review rubric) lives in swappable prompts. One harness drives any backend through a single Agent interface.

The tweet that started it all — Peter Steinberger (@steipete): "You shouldn't be prompting coding agents anymore. You should be designing loops that prompt your agents." agentloop is that idea as a reusable harness.

            ┌──────────── harness (this package) ────────────┐
goal ─▶ decompose ─▶ fan-out to subagents ─▶ aggregate ─▶ review gate ─▶ done?
            ▲                                                          │ no
            └──────────────── feedback: refine plan ◀──────────────────┘

Quick start

python3 -m examples.run_demo        # zero-dependency MockAgent
python3 -m pytest                   # full test suite, no deps

from agentloop import Orchestrator, Budget
from agentloop.adapters import MockAgent

orch = Orchestrator(MockAgent(), budget=Budget(max_iterations=4))
result = orch.run(
    goal="Write a briefing on the orchestrator-worker pattern.",
    success_criteria="Covers decomposition, execution, review, and the feedback loop.",
)
print(result.completed, result.iterations, result.stop_reason)
print(result.final_output)

Use a real model

Claude via Anthropic:

pip install -e ".[claude]"
export ANTHROPIC_API_KEY=sk-...
python3 -m examples.run_demo --claude

Grok via xAI's OpenAI-compatible API:

pip install -e ".[grok]"
export XAI_API_KEY=xai-...
python3 -m examples.run_demo --grok

GrokAgent defaults to grok-build-0.1 for coding-oriented loop work; pass model="grok-4.3" or --model grok-4.3 for general chat-style work.

Grok Build CLI as the worker:

curl -fsSL https://x.ai/cli/install.sh | bash
grok login
agentloop run --goal "Add a /health endpoint + test" --criteria "test passes" \
  --backend grok_build --cwd /path/to/repo --skip-permissions

Use backend="grok_build" for the local Grok Build CLI (grok -p), or backend="grok_api" for direct xAI API calls.

Plug into any coding agent

The loop is pluggable in two directions, both thin wrappers over the Agent seam:

Inward — coding agents are the workers. CliAgent runs each role (decomposer / subagent / reviewer) through a headless coding-agent CLI, so the workers get that agent's tools, file access, and repo context:

from agentloop import Orchestrator
from agentloop.adapters import CliAgent

# Point the worker at a repo and let it actually edit files headlessly:
agent = CliAgent.claude_code(cwd="/path/to/repo", skip_permissions=True)
orch = Orchestrator(agent)                     # or .codex() / .opencode() / .aider() / .grok_build()
result = orch.run(goal="Add a /health endpoint + test", success_criteria="test passes")

Knobs for autonomous coding workers:

• cwd=... — run the worker inside a specific repo (works for every preset).
• skip_permissions=True — let the worker use tools without prompting (--dangerously-skip-permissions / --dangerously-bypass-approvals-and-sandbox / --yes-always). Needed for headless coding, but it bypasses all safety prompts — point it at a worktree or throwaway branch, not your main checkout.
• timeout=... — seconds to cap each worker CLI call. Default is None (no cap): a real coding worker is slow and unpredictable, so a short per-call timeout just kills it mid-task and throws the work away. Bound the run instead with Budget(max_seconds=...), which is checked between iterations.
• verify_commands=[...] — run real commands after each worker iteration and feed failures back into the next loop. Use this for deterministic checks like python3 -m pytest, npm test, or npx playwright test. Any failing verifier blocks completion even if the reviewer would otherwise accept the work.

For big builds, keep subgoals small (the worker has to finish one in a single call) and give the loop room with max_iterations; one cold worker can't build everything in one shot.

Auth piggybacks on the CLI's own login, so Claude.ai, ChatGPT, or Grok Build subscription/OAuth sessions work with no provider API key.

Isolated runs. The safe default for skip_permissions is to run inside a throwaway git worktree on its own branch — your main checkout is never touched:

from agentloop import Orchestrator, worktree
from agentloop.adapters import CliAgent

with worktree("/path/to/repo") as wt:               # new branch + checkout
    agent = CliAgent.claude_code(cwd=wt.path, skip_permissions=True)
    Orchestrator(agent).run(goal="...", success_criteria="...")
    print(wt.changed_files())   # what the run touched
    wt.commit("agentloop run")  # optional: persist on the branch

Cleanup mirrors the harness's own worktrees: cleanup="auto" (default) keeps the worktree iff the run changed something (so you can inspect/merge the branch) and removes it if it left nothing; "always" / "never" force the choice.

Partial work is never lost. When you orchestrate against a repo, the loop commits the worktree after every iteration (agentloop: iteration N). So if a later iteration fails or a budget guard stops the run, each completed iteration's work is preserved as a checkpoint commit on the branch — recoverable, not discarded. The result's worktree.checkpoints lists them. Workers are also told to inspect the working directory and continue from prior work rather than restart, so a retry builds on what's already there instead of clobbering it.

The example runner isolates automatically when given a repo:

python3 -m examples.run_with_cli_agent claude /path/to/repo

python3 -m examples.run_with_cli_agent claude   # codex | opencode | aider | grok

Custom CLI? It's just a command template ({prompt}, {system}, {combined}; no prompt placeholder ⇒ text is piped on stdin):

CliAgent(["my-agent", "--system", "{system}", "--ask", "{prompt}"])

Presets are starting points — CLI flags vary by version; confirm yours and tweak agentloop/adapters/cli.py. A non-zero exit or timeout becomes a FAILED task (with retries), not a silent wrong answer.

Outward — a coding agent calls the loop. An MCP server exposes the loop as a tool, so any MCP-aware agent (Claude Code, Cursor, Codex, opencode, Cline, Windsurf) can invoke it. By default backend="auto" uses the same agent family as the caller when detectable; pass backend explicitly to override it.

pip install -e ".[mcp]"                # installs the `mcp` SDK
python3 -m agentloop.mcp_server        # stdio transport

Tools:

• orchestrate(goal, success_criteria, backend, cwd, max_iterations, skip_permissions, isolate, model, verify_commands, verify_timeout, playwright, detach=true) — runs the loop. Detached by default: returns { status: "running", run_id, … } immediately and runs to completion in the background — you monitor it (see below). Pass detach=false to instead block and get the result (or { status: "needs_input", questions[], token }) back in a single call.
• orchestrate_resume(token, answers, detach) — continues a run that asked for input
• orchestrate_status(run_id) — light status of a detached run (phase, iteration, running)
• orchestrate_tail(run_id, cursor, limit) — the events a detached run produced since cursor
• orchestrate_result(run_id, wait, timeout) — the final result of a detached run
• orchestrate_list() — every detached run this server has started, with status
• list_backends() — the worker engines this server can drive
• doctor(cwd?) — non-invasive diagnostics for backend CLI availability, target directory access, and timeout interpretation

A completed result is structured: { completed, iterations, stop_reason, final_output, summary, history[], worktree? } — summary is a one-line human-readable digest for the calling agent to show. While it runs, orchestrate streams a notifications/progress update when it starts and after every iteration (e.g. iteration 2/4: 3/3 subgoals ok, gates pass, goal incomplete), so the caller sees live status instead of a bare spinner. When cwd is given it runs in an isolated worktree (see above) by default.

Fire and monitor — the default. orchestrate runs detached: it returns a run_id immediately and the loop runs to completion (or error) in the background. There is no run timeout — you watch it rather than race it against a clock:

orchestrate(goal=…, cwd="/repo")   # backend="auto", detach=true by default
  -> { status: "running", run_id, run_dir, events_path, tail_command }
orchestrate_status(run_id)        -> { phase, iteration, running, … }   # poll until running=false
orchestrate_tail(run_id, cursor)  -> { events[], cursor, running, more } # or strea