Write your first step

This page is a Python 3 worked example that uses only the standard library. Copy `examples/custom-steps/todo-check/` or type the excerpts as you go. The code on this page is that example. A [Go example](examples.md) that speaks the same protocol is linked from More examples.

This page is a Python 3 worked example that uses only the standard library. Copy examples/custom-steps/todo-check/ or type the excerpts as you go. The code on this page is that example. A Go example that speaks the same protocol is linked from More examples.

The step fails when the run's change adds a TODO line without an issue reference, and passes otherwise. A line such as TODO(#12) is fine.

1. The handshake

Before the first run, Kusaw starts the program with KUSAW_HANDSHAKE=1 and no request on stdin. The program prints one JSON line and exits 0. kusaw validate does the same for every listed step, so a broken handshake fails validate instead of a run.

Pythonexamples/custom-steps/todo-check/step.py

# Handshake: one JSON line, then exit 0. kusaw validate calls this.
def handshake():
    emit({"type": "handshake", "protocol_versions": [2], "name": "todo-check",
          "version": "1.0.0", "interactive": False,
          "description": "Fails when the change adds a TODO without an issue reference"})

protocol_versions must include 2. The handshake has five seconds and must be the only line on stdout.

Pythonexamples/custom-steps/todo-check/step.py

    if os.environ.get("KUSAW_HANDSHAKE") == "1":
        handshake()
        return 0

2. Reading the request

On a real attempt the request is JSON on stdin. In user terms it tells the step which worktree to inspect, which step this is, how long it has, what earlier steps already did, and, inside a loop, the previous iteration's feedback. Field names (worktree, repos, artifacts_dir, loop) are defined in the generated protocol reference; this page does not copy that list.

Pythonexamples/custom-steps/todo-check/step.py

    # Request: worktree, repos, artifacts_dir, and loop feedback. Field names:
    # generated protocol reference.
    req = json.load(sys.stdin)
    artifacts_dir = req.get("artifacts_dir") or "."
    repos = req.get("repos") or []
    trees = [(r.get("worktree"), r.get("base_ref") or "HEAD") for r in repos] or [
        (req.get("worktree"), "HEAD")]

The example lists added lines through git diff against each repo's base_ref.

Pythonexamples/custom-steps/todo-check/step.py

def added_lines(worktree, base_ref):
    if not worktree or not os.path.isdir(worktree):
        return {}
    proc = subprocess.run(
        ["git", "-C", worktree, "diff", "--unified=0", "--no-color", "--no-ext-diff", base_ref, "--"],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.strip() or "git diff failed")
    grouped, path, n = {}, None, 0
    for raw in proc.stdout.splitlines():
        if raw.startswith("+++ "):
            path = raw[4:][2:] if raw[4:].startswith("b/") else raw[4:]
            path = None if path == "/dev/null" else path
        elif raw.startswith("@@"):
            m = re.search(r"\+(\d+)", raw)
            n = int(m.group(1)) if m else 0
        elif path and raw.startswith("+") and not raw.startswith("+++"):
            grouped.setdefault(path, []).append((n, raw[1:]))
            n += 1
    return grouped

3. Progress and an artifact

Emit one progress event per file, write a Markdown report into artifacts_dir, and announce it with an artifact event. Events are one JSON object per stdout line. The generated protocol reference lists every event type.

Pythonexamples/custom-steps/todo-check/step.py

    try:
        found = []
        for worktree, base_ref in trees:
            grouped = added_lines(worktree, base_ref)
            for path, rows in grouped.items():
                emit({"type": "progress", "message": "checking {0}".format(path)})
                for n, text in rows:
                    if TODO_ANY.search(text) and not TODO_OK.search(text):
                        found.append((path, n, text))
    except RuntimeError as err:
        result("fail", "git_error", str(err), [], [])
        return 0
    report = write_report(artifacts_dir, found)
    art = {"path": report, "kind": "markdown", "title": "TODO report"}
    emit({"type": "artifact", "path": report, "kind": "markdown", "title": "TODO report"})

4. Ending with a result

The last stdout line is a result. Status is one of ok, fail, retry, or reject.

Status	What the pipeline does
`ok`	The step passed. The pipeline continues.
`fail`	Inside a loop, the iteration fails and the next iteration sees this step's feedback, so the agent can fix the lines. Outside a loop, the run fails.
`retry`	Kusaw runs the same step again, up to `retries` (default 2). Use this for a transient error, not for a finding the agent should fix.
`reject`	The run stops. Use this when the request or protocol cannot be used (wrong version, bad request).

A fail from this example carries one feedback item per bare TODO so the next iteration can fix it:

Pythonexamples/custom-steps/todo-check/step.py

MSG = 'Line {n} adds a TODO without an issue reference. Add one (for example "TODO(#12)") or remove it.'

Pythonexamples/custom-steps/todo-check/step.py

    if found:
        fb = [{"severity": "major", "file": f, "line": n, "message": MSG.format(n=n)}
              for f, n, _ in found]
        result("fail", "todo_without_issue", "Added TODO lines need an issue reference.", [art], fb)
    else:
        result("ok", "ok", "No bare TODO lines in the added diff.", [art], [])

reason_code is a lowercase token such as ok or todo_without_issue. Feedback fields (severity, file, line, message) are in the generated protocol reference.

5. Exit codes and timeouts

Situation	Exit	Status	What Kusaw does
Handshake success	0	handshake line	Validate or the run continues
Handshake slower than 5s, extra stdout, or non-zero exit	—	reject	The step never runs
Result `ok`, `fail`, `retry`, or `reject`	0	that status	As in the table above
Step timeout	—	fail	The attempt is timed out; the message names the timeout
Process exits non-zero with no result	non-zero	retry	Treated as a crash, then retried
Protocol version is not 2	0	reject	The run stops

Keep protocol JSON on stdout. Write logs to stderr.

6. Add it to `kusaw.yaml`

Declare the program under steps: with command:, then put its id in the pipeline. This example uses repo label demo and inserts todo-check after the default build step.

YAMLexamples/custom-steps/todo-check/kusaw.yaml

  todo-check:       { command: ["python3", "./step.py"], timeout: 1m }

YAMLexamples/custom-steps/todo-check/kusaw.yaml

pipeline:
  - spec-check
  - loop:
      name: build-review
      max_iterations: 3
      no_progress: true
      steps:
        - build
        - todo-check
        - parallel:
            name: review
            fail_fast: false
            steps:
              - review-standards
              - review-spec
        - verify
  - deliver

command: is relative to the factory working directory. From an empty folder, keep step.py next to kusaw.yaml as the example does. Then run kusaw validate. Validate handshakes every listed step, including this one.

7. Test it on its own, then in a run

Handshake by hand (one line, exit 0):

Terminal

KUSAW_HANDSHAKE=1 python3 step.py

Pipe a saved request. The sample has no worktree, so the step reports ok and writes /tmp/todo-check-artifacts/todo-report.md:

JSONexamples/custom-steps/todo-check/request.sample.json

{
  "protocol_version": 2,
  "mode": "pipeline",
  "run_id": "run-todo-check",
  "task_id": "task-todo-check",
  "step": {"id": "todo-check", "name": "todo-check", "stage": "build-review", "attempt": 1},
  "worktree": null,
  "repos": [],
  "artifacts_dir": "/tmp/todo-check-artifacts",
  "config": {},
  "limits": {"time_left_ms": 60000, "budget_left_usd": null},
  "context": {
    "task": {"id": "task-todo-check", "title": "Sample", "spec_path": "", "repos": ["demo"]},
    "steps": {}
  },
  "loop": {
    "name": "build-review",
    "iteration": 1,
    "max_iterations": 3,
    "previous_feedback": {}
  }
}

Terminal

python3 step.py < request.sample.json

Documented output of that pipe:

JSON

{"type":"artifact","path":"/tmp/todo-check-artifacts/todo-report.md","kind":"markdown","title":"TODO report"}
{"type":"result","protocol_version":2,"status":"ok","reason_code":"ok","message":"No bare TODO lines in the added diff.","artifacts":[{"path":"/tmp/todo-check-artifacts/todo-report.md","kind":"markdown","title":"TODO report"}],"feedback":[]}

In a run, a fake agent that adds a bare TODO should see fail with one feedback item per line, then remove it on the next iteration so the step returns ok and the run succeeds.

What else your step can do

Protocol features a custom step can use are documented here as they ship. Field lists stay in the generated protocol reference.

Progress, artifacts, feedback, results — this page.
Agent CLI argv — Use an agent inside your step. kusaw agent-args prints the same argv the built-in runner would use (kusaw claude-args is an alias for Claude Code).
Structured outputs — not published yet.
Evidence files — not published yet.
Notes while running — not published yet.
Ports and services — not published yet.
Outputs used by a select step — not published yet.
Planner step — not published yet.
Work orders — a custom step runs inside a work order the same way it runs in a pipeline. The work-order CLI is not in this build yet.
Atlas — context.atlas (repos, tools) plus env KUSAW_ATLAS_SOCKET, KUSAW_ATLAS_TOKEN, and KUSAW_ATLAS_OVERVIEW. Call kusaw atlas call read_map|search_code --json '<args>' inside the attempt. Handshake feature atlas_tools is optional. Off with atlas.agents: off or steps.<id>.atlas: false.

To swap a default for this program, see Replace a built-in step.

Custom steps

A [step](../glossary.md#step) is a program. Kusaw sends it a request, it reports progress, and it ends with a result. Every step, built-in or [custom](../glossary.md#custom-command-step), speaks the same [step protocol](../glossary.md#step-protocol). The [built-in steps](../glossary.md#built-in-step) are defaults you can replace.

Replace a built-in step

Point a pipeline entry at your `command:` instead of `builtin:`. Keep the same step id so gates, loops, and feedback keep working. The built-in steps stay available as defaults; replacing one is a config change, not a new pipeline.

On this page

1. The handshake 2. Reading the request 3. Progress and an artifact 4. Ending with a result 5. Exit codes and timeouts 6. Add it to kusaw.yaml7. Test it on its own, then in a run What else your step can do