Skip to content
contraagentcontraagentcontraagent

MIT | Python 3.12 or later | Linux, macOS, Windows

Run coding agents as typed, checked, supervised tasks.

Each task is a typed Python function whose body is a coding agent. Its output is a file checked against a pydantic model.

Get startedRead the docs
uv add 'contraagent[monitor]'
The same tasks, plugin and config on five coding agents
contraagent run tasks.py:docs --kwargs '{"units": ["alpha"]}' --agent followed by one of claude, codex, opencode, omp, pi
  • claudeClaude Codeclaude -p
  • codexCodex CLIcodex exec
  • opencodeopencodeopencode run
  • ompoh-my-piomp -p
  • pipipi -p
The demo run, replayed

A run of the demo pipeline, replayed

examples/demo/tasks.pyquoted from the tested example
summarize = agent_task(
"summarize",
plugin="plugins/docs",
role="summarizer",
output=Summary,
validators=[three_points, its_own_unit],
max_turns=10,
timeout_s=300,
retry_backoff_base=0.5,
)
def review_loop(unit: str):
try:
summary = yield summarize(unit, prepared=prepare(unit))
except TaskFailed:
return None
while True:
figures = yield score(unit, summary=summary)
verdict = yield review(unit, summary=summary, score=figures)
if verdict.approved:
break
summary = yield revise(unit, summary=summary, verdict=verdict)
yield write_note(unit, summary=summary)
return summary
def docs(units: list[str]):
summaries = yield [review_loop(u) for u in units]
return (yield report(summaries=[s for s in summaries if s is not None]))
Pipeline graph<ca-pipeline-graph>
The run in numbers<ca-kpis>
Transcript summarize · delta<ca-transcript>
What happened in the run 5 moments
  1. 0:02summarize · beta was rate limited on primary-model. A model switch ran it on fallback-model.
  2. 0:03summarize · delta failed the check three_points. The same agent session corrected it.
  3. 0:05The review of gamma did not approve its summary. The loop ran a second pass for gamma.
  4. 0:08write_note · alpha failed the checks heading and names-its-unit. The same agent session corrected it.
  5. 0:2144 of 44 tasks done. 44 of 44 outputs passed their checks.

The flow is the demo pipeline of the package's examples. Its run on the demo agent plays again from its records in<ca-pipeline-graph>, <ca-kpis> and <ca-transcript>, the elements of the monitor. A highlighted line of code has running tasks that it created.

Your coding agent writes the pipeline. You run it.

contraagent is made for a coding agent to build with. Its documentation is installed with it, contraagent check names every problem and where it is explained, and a stub agent tests the flow with nothing spent. Nothing runs on a real coding agent until you run it.

  1. write
  2. check
  3. fix
  4. stub run
  5. done

  1. You

    Install contraagent

    Linux, Python 3.12 or later, and one or more coding agents on PATH.

    Install

    $ uv add 'contraagent[monitor]'

    $ contraagent --version

    contraagent 0.2.0

  2. You

    Point your coding agent at the docs

    The documentation is installed with the package and matches its version. contraagent docs --pointer prints a block that points a coding agent at it; add it to the instructions your agent reads.

    $ contraagent docs --pointer

    ## contraagent

    This project uses contraagent to run coding-agent tasks as a Python pipeline. Its documentation is installed with the package and matches the installed version. The documentation, the docstrings of the public API and the messages of `contraagent check` hold every fact that a pipeline needs, so contraagent's source code is not needed.

    - `contraagent docs` prints the index, with the order in which to read the topics. `contraagent docs <topic>` prints one topic. Read `guide`, then `tutorial`. `contraagent docs --path` prints where the documentation and its examples are. - contraagent's API is newer than your training data. Read the topic before you write or change the code. - Check your work in this order, which launches no coding agent and spends nothing. First, `contraagent check <file>:<flow> --config pipeline.toml --kwargs '{...}'` until it prints ✓ (no `--config` before the project has a config). Second, a stub run, `contraagent run <file>:<flow> --config pipeline.toml --stub --kwargs '{...}'`, in which each task's output is its output model's example. Give each output model an example that passes its validators, and the stub run goes through the whole flow. Third, tests of the flow on `contraagent.testing.StubAgent` in pytest, for correction rounds, failures and failover (`contraagent docs testing`). - A run on the coding agents (`contraagent run <file>:<flow> --config pipeline.toml`) spends tokens. The person who asked for the pipeline starts it, unless they ask you to, and watches it in `contraagent monitor`.

  3. You

    Ask for the pipeline in plain words

    To Claude Code, Codex CLI, opencode, oh-my-pi or pi.

    You, to your coding agent

    Use contraagent to write a pipeline that summarizes each document in inputs/, has a reviewer approve each summary or send it back for a revision, then writes a note for each document and one report. Check it with contraagent check and test it on a stub agent.

  4. The agent

    Reads the docs and writes the pipeline

    The topics of contraagent docs, then tasks.py: the models, the agent tasks and the flow, and a plugin with a role for each kind.

    Tasks and flows

    $ contraagent docs

    • Tutorial: your first pipeline
    • Guide
    • Concepts
    • Tasks and flows
    • Plugins
    • Contracts and validators
    • Routing and credentials
    • Coding agents
    • Security model
    • Checking and testing
    • The monitor
    • Many hosts: the hub
    • Operating a hub
    • Compatibility

    tasks.py

    36class Summary(BaseModel):
    37 model_config = ConfigDict(
    38 json_schema_extra={"examples": [{"unit": "alpha", "points": ["parses a config file", "validates it", "reports errors"]}]}
    39 )
    40
    41 unit: str = Field(description="the unit this summary is for, copied from the call file")
    42 points: list[str] = Field(description="the main points, one short sentence each")
    ⋮
    93summarize = agent_task(
    94 "summarize",
    95 plugin="plugins/docs",
    96 role="summarizer",
    97 output=Summary,
    98 validators=[three_points, its_own_unit],
    99 max_turns=10,
    100 timeout_s=300,
    101 retry_backoff_base=0.5,
    102)
    ⋮
    164def docs(units: list[str]):
    165 """Every unit's review loop at once (a flow call each), then one report over the approved summaries."""
    166 summaries = yield [review_loop(u) for u in units]
    167 return (yield report(summaries=[s for s in summaries if s is not None]))

  5. The agent

    Checks it and fixes what check names

    contraagent check launches nothing and reports every problem at once, each with the topic that explains it.

    Checking and testing

    $ contraagent check tasks.py:docs --config pipeline.toml --stub --kwargs '{"units": "alpha"}'

    ✗ 8 tasks, 2 kinds: prepare 4, summarize 4; later kinds: report, score, review, revise, write_note
  review: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300
  revise: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300
  summarize: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300; on a 429: stub/fallback-model
  write_note: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300
  error: flow call 0.5 (tasks.review_loop): tasks.py:151: ValueError: task key 'python:prepare-a' is created by flow calls 0.1 and 0.5; pass key= to tell them apart
  warning: routing: [tasks.sumarize] is not a kind of this run (kinds: review, revise, summarize, write_note); ignored (did you mean summarize?)
  read: (contraagent docs tasks)

    The agent's fix

    - --kwargs '{"units": "alpha"}'
+ --kwargs '{"units": ["alpha", "beta"]}'
- [tasks.sumarize]
+ [tasks.summarize]

    $ contraagent check tasks.py:docs --config pipeline.toml --stub --kwargs '{"units": ["alpha", "beta"]}'

    ✓ 4 tasks, 2 kinds: prepare 2, summarize 2; later kinds: report, score, review, revise, write_note
  review: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300
  revise: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300
  summarize: stub/primary-model effort=medium max_turns=10 timeout_s=300; on a 429: stub/fallback-model
  write_note: stub/(the agent default model) effort=medium max_turns=10 timeout_s=300

  6. The agent

    Tests the flow on a stub agent

    The stub agent plays the coding agent: the checks, corrections and failover are the real ones, and nothing is spent.

    $ pytest test_demo.py -v

    test_demo.py::test_the_flow_checks_clean PASSED
test_demo.py::test_the_flow_runs_on_the_stub PASSED
test_demo.py::test_the_report_covers_the_approved_summaries PASSED
test_demo.py::test_the_flow_without_waiting_refuses_what_a_failure_feeds PASSED
4 passed

  7. You

    Run it and watch it

    On the coding agent and model that pipeline.toml names for each kind. contraagent monitor shows the run live, as in the replay above.

    The replay above

    $ contraagent run tasks.py:docs --config pipeline.toml --kwargs '{"units": ["alpha", "beta"]}'

    Tasks done
    of 44
    Corrections
    Model switches

    $ contraagent monitor --runs-dir runs

What the runner does

Section titled “What the runner does”

Outputs are contracts

Each output file is checked against its pydantic model and its validators. A file that fails goes back to the same agent session with the errors.

Failover and quotas

Another model, coding agent or credential after a rate limit or a failed launch, and caps on a subscription’s 5-hour and weekly usage.

Deadlines and resume

A deadline defers the work that cannot finish, with its cause. A resumed run repeats only the tasks whose inputs or definition changed.

Checked before it spends

contraagent check reports every problem with nothing launched. A stub agent and recorded replays test a flow with nothing spent.

A clean context per task

A task’s agent sees its role, its plugin, its skills and its call file, not the user’s or the project’s settings, nor another task’s conversation.

A live monitor

The pipeline graph, each task’s transcript, inputs, outputs and checks, the history of attempts, and trends over a period. A hub serves the runs of many hosts.

Where to start

Section titled “Where to start”

Python package

Build and run pipelines.

Tutorial: your first pipelineFrom an empty directory: models, an agent task and its plugin, a flow, a stub run, a run on a coding agent, a test.
GuideBuilding a pipeline: the steps, the decisions, the pitfalls.
ConceptsThe words contraagent uses, each with one meaning.
The monitorThe runs, live, in a browser.

Web packages

The monitor in a page of your own.

@contraagent/uiThe monitor and each of its views as web components.
@contraagent/elementsGeneric web components: tables, graphs, tiles, charts, transcripts, and the themes.
@contraagent/apiA typed client for the monitor's HTTP API, its live feeds and a store.
Element galleryEvery element of the web packages, live.

Reference

Generated from the code.

Monitor HTTP APIThe monitor's endpoints and live feeds.
Elements of @contraagent/uiEach view's properties, attributes and events.
Elements of @contraagent/elementsEach generic element's properties, attributes, events and parts.
Theme propertiesEvery --ca-* custom property and its value in each built-in theme.