athanore

Code-defined AI agent workflows over ACP.

A workflow is a graph declared in code: decorated functions as nodes, edges inferred from parameter names. Node bodies are arbitrary async Python — they may dispatch an ACP agent, or do purely deterministic work. A server runs the graph; a CLI submits work and watches it.

See DESIGN.md for the full design and the workflow/ package for a complete pipeline (one agent class per file, prompts inlined).

Install

uv sync

The interface — three rules

1. The signature is the graph. Positional parameters are edges; a keyword-only parameter after * is the optional payload. Exactly one node is start=True.
2. The return value is the routing. A plain return auto-transitions to a single successor; edge_ref(value) routes explicitly; returning from a node with no successors completes the run.
3. The exception is the failure policy. Raising triggers retry/dead-letter.

from athanore import AthanoreWorkflow
from athanore.testing import MockAgent   # stand-in for AthanoreACPAgent

wf = AthanoreWorkflow("feature_build")

@wf.node(start=True)
async def prompt(build, /, brief):
    result = await MockAgent(output=f"prompt for: {brief}").run(brief)
    return result.output

@wf.node()
async def build(review, /, spec):
    result = await MockAgent(output=f"built: {spec}").run(str(spec))
    return result.output

@wf.node()
async def review(build, ship, /, work):
    # branch: approve after one pass through build, else loop back
    if str(work).count("built") >= 2:
        return ship(work)
    return build(work)

@wf.node()
async def ship(final):
    return await MockAgent(output=f"shipped: {final}").run(str(final))

if __name__ == "__main__":
    wf.run(port=4002, workers=1)

Run it

uv run python -m workflow          # or your own workflow file

Use the CLI

athanore                               # bare command opens the live TUI
athanore feature_build "Build X" ["details"] # create a task for the intake agent
athanore gamedev "Neon snake" ["details"]     # ...or any registered workflow
athanore ls                            # list runs
athanore run <id>                      # inspect a run + its tasks
athanore logs <id>                     # show the run's event history
athanore workflows                     # show registered graphs
athanore tui                           # live TUI: runs list + run detail

--port/--host (or ARTIFICER_PORT) point the CLI at the server. Add --web to athanore or athanore tui to serve the TUI in a browser (--web-port picks the port, default 2424).

The server can host the browser TUI itself, so one command runs both:

uv run python -m workflow --web          # API on 4002, browser TUI on 2424
uv run python -m workflow --web --web-port 8080

In code that is wf.run(web=True) / server.run(web=True, web_port=...); the server spawns athanore tui --web as a child and stops it on shutdown.

Tests

uv run pytest -q

Layout

• athanore/graph.py — DSL: signature parsing, finalization, generations
• athanore/store.py — sqlite store (runs, tasks, submissions, events)
• athanore/scheduler.py — priority dispatch loop, retries, recovery
• athanore/agents.py — AthanoreACPAgent facade (ACP Python SDK)
• athanore/server.py — HTTP surface (CLI backend + agent submissions)
• athanore/cli.py — the athanore command
• athanore/testing.py — MockAgent for tests without a real ACP subprocess

Note

dispatch.py, dispatcher/, and main.py are the earlier athanore-based prototype and are no longer used by this package (its deps were dropped).