AgentFlow

Orchestrate codex, claude, and kimi agents in dependency graphs with parallel fanout, iterative cycles, and remote execution on SSH/EC2/ECS.

94-node pipeline: plan → 64 workers → 8 batch merges → 16 reviews → 4 review merges → synthesis

Install

One line:

curl -fsSL https://raw.githubusercontent.com/shouc/agentflow/master/install.sh | bash

This installs agentflow, adds it to PATH, and installs the skill for Codex and Claude Code.

Or manually:

python3 -m venv .venv &&. .venv/bin/activate
pip install -e .[dev]

Quick Start

fromagentflowimportGraph, codex, claudewithGraph("my-pipeline", concurrency=3) asg:
plan=codex(task_id="plan", prompt="Inspect the repo and plan the work.", tools="read_only")
impl=claude(task_id="impl", prompt="Implement the plan:\n{{ nodes.plan.output }}", tools="read_write")
review=codex(task_id="review", prompt="Review:\n{{ nodes.impl.output }}")
plan>>impl>>reviewprint(g.to_json())

agentflow run pipeline.py --output summary

Or just ask Codex (the agentflow skill is auto-installed):

codex "Use agentflow to fan out 10 codex agents, each telling a unique joke, then merge their outputs and pick the funniest one. Write the pipeline and run it."

Parallel Fanout

Fan a node into many parallel copies with fanout():

fromagentflowimportGraph, codex, fanout, mergewithGraph("code-review", concurrency=8) asg:
scan=codex(task_id="scan", prompt="List the top 5 files to review.")
review=fanout(
codex(task_id="review", prompt="Review {{ item.file }}:\n{{ nodes.scan.output }}"),
[{"file": "api.py"}, {"file": "auth.py"}, {"file": "db.py"}],
)
summary=codex(task_id="summary", prompt=(
"Merge findings:\n{% for r in fanouts.review.nodes %}{{ r.output }}\n{% endfor %}"
))
scan>>review>>summaryprint(g.to_json())

fanout(node, source) dispatches on type:

• int -- N identical copies: fanout(node, 128)
• list -- one per item: fanout(node, [{"repo": "api"}, ...])
• dict -- cartesian product: fanout(node, {"axis1": [...], "axis2": [...]})

Reduce with merge(node, source, size=N) (batch) or merge(node, source, by=["field"]) (group).

Iterative Cycles

Loop until a stop condition with on_failure:

fromagentflowimportGraph, codex, claudewithGraph("iterative-impl", max_iterations=5) asg:
write=codex(
task_id="write",
prompt="Write a Python email validator.\n{% if nodes.review.output %}Fix: {{ nodes.review.output }}{% endif %}",
tools="read_write",
)
review=claude(
task_id="review",
prompt="Review:\n{{ nodes.write.output }}\nIf complete, say LGTM. Otherwise list issues.",
success_criteria=[{"kind": "output_contains", "value": "LGTM"}],
)
write>>reviewreview.on_failure>>write# loop until LGTM or max_iterationsprint(g.to_json())

Remote Execution

Run agents on remote machines -- zero config needed:

# EC2 (auto-discovers AMI, key pair, VPC)codex(task_id="remote", prompt="...", target={"kind": "ec2", "region": "us-east-1"})
# ECS Fargate (auto-discovers VPC, builds agent image)codex(task_id="remote", prompt="...", target={"kind": "ecs", "region": "us-east-1"})
# SSHcodex(task_id="remote", prompt="...", target={"kind": "ssh", "host": "server", "username": "deploy"})

Shared instances across nodes:

plan=codex(task_id="plan", prompt="...", target={"kind": "ec2", "shared": "dev-box"})
impl=codex(task_id="impl", prompt="...", target={"kind": "ec2", "shared": "dev-box"})
plan>>impl# same EC2 instance, files persist

Scratchboard

Shared memory file across all agents:

withGraph("campaign", scratchboard=True) asg:
shards=fanout(codex(task_id="fuzz", prompt="..."), 128)

Examples

Example	What it does
airflow_like.py	Basic pipeline: plan → implement → review → merge
code_review.py	Fan out code review across files, merge findings
dep_audit.py	Audit each dependency for security/license issues
test_gap.py	Find untested modules, suggest tests per module
multi_agent_debate.py	Codex vs Claude: independent solve + cross-critique
release_check.py	Parallel release gate: tests + security + changelog
iterative_impl.py	Write → review → fix cycle until LGTM
contract_audit.py	AgentFlow native contract audit for EVM + Solidity + Foundry with manifest-driven intake and report rendering
airflow_like_fuzz_batched.py	128-shard fanout with batch merge + periodic monitor
airflow_like_fuzz_grouped.py	Matrix fanout with grouped reducers
ec2_remote.py	Run codex on a remote EC2 instance
ecs_fargate.py	Run codex on ECS Fargate

CLI

agentflow run pipeline.py # run a pipeline
agentflow run pipeline.py --output summary
agentflow inspect pipeline.py # show expanded graph
agentflow validate pipeline.py # check without running
agentflow templates # list starter templates
agentflow init > pipeline.py # scaffold a starter

Acknowledgements

• gepa
• kiss-ai
• claude-code-telegram
• linux.do