Python workflow generated in the above example looks like this:

#!/usr/bin/env python3
"""
Workflow: Commit Analysis
1. Get the 6 most recent git commit short IDs.
2. For each commit, run a 2-stage pipeline:
   Stage 1 (tools: git_query): Examine the commit details.
   Stage 2 (structured output): Return structured JSON with classification.
3. Reduce: Combine all JSON results into a markdown table (pure Python).
"""
import json
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))

from agent.workflow import (
    agent_call,
    map_step,
    reduce_step,
    run_workflow,
    finish_workflow,
    StepResult,
)
from agent.config import OutputFormat

COMMIT_SCHEMA = {
    "type": "object",
    "properties": {
        "short_id": {"type": "string"},
        "files_modified": {"type": "integer"},
        "classification": {"type": "string", "enum": ["bugfix", "feature", "other"]},
        "description": {"type": "string"},
    },
    "required": ["short_id", "files_modified", "classification", "description"],
}


def get_last_n_commits(n: int = 6) -> list[str]:
    """Get the last N git commit short IDs."""
    result = subprocess.run(
        ["git", "log", f"-{n}", "--format=%h"],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"git log failed: {result.stderr}")
    return result.stdout.strip().splitlines()


def analyze_commit(short_id: str) -> StepResult:
    """Two-stage pipeline: tools → structured output."""
    # Stage 1: gather data with tools
    gather = agent_call(
        f"Examine git commit {short_id}.\n"
        f"Use git_query with subcommand='show' and ref='{short_id}' to get the commit details.\n"
        f"Then use git_query with subcommand='show_full' and ref='{short_id}' to see the full diff.\n\n"
        f"Report:\n"
        f"1. The total number of files changed\n"
        f"2. The commit message subject line\n"
        f"3. A brief summary of what the code changes do",
        tools=["git_query"],
        step_name=f"examine_{short_id}",
    )
    if not gather.ok:
        return StepResult(text="", ok=False, error=f"Stage 1 failed: {gather.error}")

    # Stage 2: structured JSON (tools stripped by output_format)
    return agent_call(
        f"Convert this commit analysis into structured JSON.\n\n"
        f"Commit short ID: {short_id}\n"
        f"Analysis:\n{gather.text}\n\n"
        f"JSON fields:\n"
        f'- short_id: the git short hash exactly "{short_id}"\n'
        f"- files_modified: count of files changed as an integer\n"
        f'- classification: "bugfix" if the commit fixes a bug, "feature" if it adds new functionality, "other" otherwise\n'
        f"- description: a one-sentence terse description starting with a verb (e.g., 'Fixes login issue', 'Adds user profile', 'Refactors validation')",
        tools=[],
        output_format=OutputFormat(json_schema=COMMIT_SCHEMA),
        step_name=f"classify_{short_id}",
    )


def build_table(texts: list[str]) -> str:
    """Pure Python reduce: JSON results → markdown table."""
    rows = [json.loads(t) for t in texts]
    header = "| Short ID | Files | Classification | Description |"
    sep = "|----------|-------|----------------|-------------|"
    body = []
    for r in rows:
        badge = "🐛" if r["classification"] == "bugfix" else "✨" if r["classification"] == "feature" else "🔧"
        body.append(f"| `{r['short_id']}` | {r['files_modified']} | {r['classification']} | {r['description']} |")
    return "\n".join([header, sep] + body)


def main():
    run_id = run_workflow("commit_analysis", {})
    try:
        commits = get_last_n_commits(6)
        print(f"Analyzing {len(commits)} commits: {', '.join(commits)}\n")

        # Fan out: each worker runs a 2-stage pipeline in parallel
        results = map_step(
            commits,
            worker_fn=analyze_commit,
            concurrency=5,
            run_id=run_id,
            step_name="audit",
        )

        # Print individual results
        for r in results:
            if not r.ok:
                print(f"  ⚠️  failed: {r.error}")
                continue
            data = json.loads(r.text)
            badge = "🐛" if data["classification"] == "bugfix" else "✨" if data["classification"] == "feature" else "🔧"
            print(f"  {badge} {data['short_id']} — {data['description']}")

        # Reduce: markdown table (pure Python, no LLM call)
        table = reduce_step(results, python_fn=build_table, run_id=run_id)
        print(f"\n{table.text}")
        finish_workflow(run_id, summary=f"Analyzed {len(commits)} commits")

    except Exception as e:
        finish_workflow(run_id, error=str(e))
        raise


if __name__ == "__main__":
    main()

This is just an example to demonstrate the map-reduce patter, the ability for workers to make tool calls, chain steps, contstrain outputs to a JSON schema.

If registered the backend can monitor workers and detect failed workers to recover.