M21C: Headless Agents

Learning Objectives

• Define headless execution and contrast it with an interactive REPL or chat loop
• Choose input channels for an unattended agent: CLI arguments, stdin, files, and environment variables
• Design a headless contract — structured JSON on stdout, logs on stderr, and meaningful process exit codes
• Schedule an agent with cron and run it inside a GitHub Actions CI pipeline
• Replace the missing human with hard guardrails: wall-clock timeout, max iterations, and a token budget cap
• Compose headless agents into Unix pipelines with tools like jq
• Know when to reach for the Message Batches API for large overnight jobs

Headed vs Headless: An Assistant You Chat With vs a Worker You Deploy

The Two Models Side by Side

The animation below runs the same agent twice. On the left, the interactive (headed) version: every cycle bounces back to a human who reads and decides whether to continue — the blue loop never ends on its own. On the right, the headless version: a trigger fires once, the agent runs to completion untouched, writes JSON, and exits with a status code. Watch where the human appears — and where they do not.

Anatomy of a Headless Run

A headless agent is a normal command-line program that happens to call Claude. It has three I/O surfaces, and getting them right is 80% of making an agent automation-friendly:

A Minimal Headless Agent

Here is a complete, runnable headless agent. It reads a task from --query or stdin, makes a single Claude call, and prints one JSON object to stdout. Notice every log line goes to stderr.

# triage_agent.py  —  run:  python triage_agent.py --query "disk full on web-03"
# WHAT: A headless agent that classifies an ops alert into a JSON verdict
# WHY:  No input() calls, no interactive prompts — safe to run from cron
# GOTCHA: ANTHROPIC_API_KEY must be in the environment; logs go to stderr only

import sys, json, argparse, time
from anthropic import Anthropic

client = Anthropic()                 # reads ANTHROPIC_API_KEY from the environment
MODEL = "claude-sonnet-4-6"

def log(msg: str) -> None:
    # Human-readable progress -> stderr, so stdout stays pure JSON
    print(f"[triage] {msg}", file=sys.stderr)

def read_task(args) -> str:
    # Priority: --query flag, else stdin (so you can pipe text in)
    if args.query:
        return args.query.strip()
    if not sys.stdin.isatty():            # data was piped in
        return sys.stdin.read().strip()
    raise SystemExit("no task: pass --query or pipe text on stdin")

def triage(task: str) -> dict:
    log(f"classifying: {task!r}")
    resp = client.messages.create(
        model=MODEL,
        max_tokens=512,
        temperature=0,
        system=(                          # force JSON-shaped output; we still validate
            "You are an ops triage agent. Reply with ONLY a JSON object: "
            '{"severity": "low|medium|high|critical", '
            '"category": str, "summary": str}. No prose.'
        ),
        messages=[{"role": "user", "content": task}],
    )
    raw = resp.content[0].text.strip()
    # Claude is reliable at JSON, but a stray ``` fence can still appear — strip it
    raw = raw.removeprefix("```json").removeprefix("```").removesuffix("```").strip()
    verdict = json.loads(raw)                       # raises if not valid JSON
    verdict["tokens"] = {
        "input": resp.usage.input_tokens,
        "output": resp.usage.output_tokens,
    }
    return verdict

def main() -> int:
    ap = argparse.ArgumentParser(description="Headless ops triage agent")
    ap.add_argument("--query", help="alert text; if omitted, read stdin")
    args = ap.parse_args()
    started = time.time()
    try:
        task = read_task(args)
        verdict = triage(task)
        verdict["latency_ms"] = int((time.time() - started) * 1000)
        # THE result -> stdout, as a single JSON line
        print(json.dumps(verdict))
        log(f"done in {verdict['latency_ms']}ms, severity={verdict.get('severity')}")
        return 0
    except json.JSONDecodeError as e:
        log(f"model did not return valid JSON: {e}")
        print(json.dumps({"error": "invalid_model_output"}))
        return 2                                    # distinct code: bad output
    except Exception as e:
        log(f"fatal: {type(e).__name__}: {e}")
        print(json.dumps({"error": str(e)}))
        return 1                                    # generic failure

if __name__ == "__main__":
    sys.exit(main())                                # exit code = the verdict flag

// triage_agent.js  —  run:  node triage_agent.js --query "disk full on web-03"
// WHAT: A headless agent that classifies an ops alert into a JSON verdict
// WHY:  No readline/prompt calls — safe to run unattended from cron or CI
// GOTCHA: console.error writes to stderr (logs); process.stdout is the result
import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();        // reads ANTHROPIC_API_KEY from the environment
const MODEL = "claude-sonnet-4-6";

const log = (msg) => console.error(`[triage] ${msg}`);  // -> stderr

async function readTask() {
  // Priority: --query flag, else stdin (so you can pipe text in)
  const flagIdx = process.argv.indexOf("--query");
  if (flagIdx !== -1 && process.argv[flagIdx + 1]) {
    return process.argv[flagIdx + 1].trim();
  }
  if (!process.stdin.isTTY) {               // data was piped in
    const chunks = [];
    for await (const c of process.stdin) chunks.push(c);
    return Buffer.concat(chunks).toString("utf8").trim();
  }
  throw new Error("no task: pass --query or pipe text on stdin");
}

async function triage(task) {
  log(`classifying: ${JSON.stringify(task)}`);
  const resp = await client.messages.create({
    model: MODEL,
    max_tokens: 512,
    temperature: 0,
    system:
      "You are an ops triage agent. Reply with ONLY a JSON object: " +
      '{"severity": "low|medium|high|critical", "category": str, "summary": str}. No prose.',
    messages: [{ role: "user", content: task }],
  });
  let raw = resp.content[0].text.trim();
  raw = raw.replace(/^```json/, "").replace(/^```/, "").replace(/```$/, "").trim();
  const verdict = JSON.parse(raw);          // throws if not valid JSON
  verdict.tokens = {
    input: resp.usage.input_tokens,
    output: resp.usage.output_tokens,
  };
  return verdict;
}

async function main() {
  const started = Date.now();
  try {
    const task = await readTask();
    const verdict = await triage(task);
    verdict.latency_ms = Date.now() - started;
    process.stdout.write(JSON.stringify(verdict) + "\n");   // THE result
    log(`done in ${verdict.latency_ms}ms, severity=${verdict.severity}`);
    return 0;
  } catch (e) {
    const code = e instanceof SyntaxError ? 2 : 1;          // 2 = bad output
    log(`fatal: ${e.name}: ${e.message}`);
    process.stdout.write(JSON.stringify({ error: e.message }) + "\n");
    return code;
  }
}

main().then((code) => process.exit(code));   // exit code = the verdict flag

That is a lot of code at once, so let's read it the way the program runs — in four parts:

1. The client (lines 8–9). Anthropic() reads ANTHROPIC_API_KEY from the environment — never hard-coded, which matters doubly when this runs unattended in cron or CI. From here the call is the ordinary Messages API you have used all course; nothing about being headless changes the model call itself.
2. Getting the task in: read_task. This is the headless part. It checks --query first, then falls back to reading stdin only if data was piped in (not sys.stdin.isatty()). That isatty() check is the safety catch: if nobody piped anything and no flag was given, it raises immediately instead of doing what an interactive program would do — block forever waiting for a human to type. A blocked read_task is the classic "my cron job hangs" bug.
3. Doing the work: triage. A single messages.create with a system prompt that demands JSON-only output, temperature=0 for repeatability, then defense before we trust the result: strip a stray ``` fence, json.loads it (which throws if the model added prose), and record input_tokens/output_tokens from resp.usage. Claude is strong at instruction-following, but a headless agent still never assumes the text is valid JSON — it verifies.
4. Delivering the answer: main. Notice the shape of the return values. Success prints the JSON to stdout and returns 0. A JSON failure returns 2; any other error returns 1. Those integers are not cosmetic — sys.exit(main()) hands them to the operating system, and the next section shows why a caller cares about the difference between 1 and 2.

Every log(...) call routes to stderr, so the four parts above produce a program whose stdout is only the JSON verdict. The terminal session below shows exactly that split.

The Headless Contract

When a human runs your agent, they tolerate variety — they can read whatever comes out. A machine cannot. The consumer of a headless agent (a cron wrapper, a CI step, the next stage of a pipeline) needs two guarantees: the output always has the same shape, and the exit code always means the same thing. Together these are the headless contract.

Exit Codes Are Your Status API

The exit code is a one-integer status channel that every shell, scheduler, and CI system already understands — you do not have to invent or parse anything. Pick a small, documented set and stick to it:

A Reusable Contract Wrapper

Rather than scatter exit codes through your logic, define the contract once. This wrapper takes any agent function, guarantees a fixed JSON envelope on stdout, and maps outcomes to the codes above.

# contract.py — a stable envelope + exit-code mapping for any headless agent
import sys, json, time

# Custom exceptions let the agent declare *which* failure happened
class BadOutput(Exception):  code = 2     # model output failed schema
class NeedsReview(Exception): code = 3    # ran fine, but low confidence
# anything else -> code 1 (transient/operational), 0 -> success

def run_headless(agent_fn, payload: dict) -> int:
    """Run agent_fn(payload), print a fixed JSON envelope, return an exit code."""
    started = time.time()
    envelope = {"ok": False, "data": None, "error": None, "meta": {}}
    try:
        envelope["data"] = agent_fn(payload)        # your agent logic
        envelope["ok"] = True
        code = 0
    except NeedsReview as e:
        envelope["error"] = {"type": "needs_review", "message": str(e)}
        code = NeedsReview.code
    except BadOutput as e:
        envelope["error"] = {"type": "bad_output", "message": str(e)}
        code = BadOutput.code
    except Exception as e:                            # transient / unexpected
        envelope["error"] = {"type": type(e).__name__, "message": str(e)}
        code = 1
    envelope["meta"] = {
        "exit_code": code,
        "latency_ms": int((time.time() - started) * 1000),
    }
    print(json.dumps(envelope))                       # stdout: the one true line
    return code

# --- usage ---
def my_agent(payload):
    verdict = {"severity": "high"}                    # ... real work here ...
    if verdict["severity"] == "critical":
        raise NeedsReview("critical alerts are never auto-actioned")
    return verdict

if __name__ == "__main__":
    raw = sys.stdin.read() or "{}"
    sys.exit(run_headless(my_agent, json.loads(raw)))

// contract.js — a stable envelope + exit-code mapping for any headless agent
class BadOutput extends Error  { code = 2; }   // model output failed schema
class NeedsReview extends Error { code = 3; }  // ran fine, but low confidence
// anything else -> code 1 (transient); success -> code 0

async function runHeadless(agentFn, payload) {
  const started = Date.now();
  const envelope = { ok: false, data: null, error: null, meta: {} };
  let code;
  try {
    envelope.data = await agentFn(payload);     // your agent logic
    envelope.ok = true;
    code = 0;
  } catch (e) {
    code = e.code ?? 1;                          // exception carries the code
    const type = e instanceof NeedsReview ? "needs_review"
               : e instanceof BadOutput   ? "bad_output"
               : e.name;
    envelope.error = { type, message: e.message };
  }
  envelope.meta = { exit_code: code, latency_ms: Date.now() - started };
  process.stdout.write(JSON.stringify(envelope) + "\n");   // the one true line
  return code;
}

// --- usage ---
async function myAgent(payload) {
  const verdict = { severity: "high" };          // ... real work here ...
  if (verdict.severity === "critical") {
    throw new NeedsReview("critical alerts are never auto-actioned");
  }
  return verdict;
}

const chunks = [];
for await (const c of process.stdin) chunks.push(c);
const payload = JSON.parse(Buffer.concat(chunks).toString() || "{}");
runHeadless(myAgent, payload).then((code) => process.exit(code));

Reading it top to bottom, the wrapper does three jobs and nothing else:

1. It lets the agent name its own failure. The two custom exceptions, NeedsReview and BadOutput, each carry a code attribute. Your business logic raises the one that fits ("this is critical, a human must see it" vs. "the model gave me garbage"), and the wrapper trusts that choice. The wrapper does not guess the severity of a failure — the code that understands the task does.
2. It guarantees one fixed envelope. Whatever happens — success, review, bad output, or an unexpected crash — exactly one {ok, data, error, meta} object is printed to stdout. A consumer can therefore write json.loads(line) once and never special-case "but what if it failed?", because failure is just ok: false with a populated error.
3. It maps each outcome to a stable exit code. The except ladder is ordered most-specific first: NeedsReview → 3, BadOutput → 2, and the bare except Exception at the bottom catches everything else — including an Anthropic APIError — as a transient 1. Success falls through to 0. The meta.exit_code field echoes that same number into the JSON so logs and the process status always agree.

Write this wrapper once and every headless agent you build speaks the same dialect — which is what makes the cron and CI drivers in the next section completely generic.

Automation: Cron & CI

"Headless" only pays off when something other than you launches the agent. The animation shows the shape every automation shares: a trigger on the left fires the agent in the middle, which produces structured output that flows to a downstream sink — a database, a Slack channel, the next CI step. The agent in the middle never changes; only the trigger and the sink do.

Scheduling with Cron

cronA time-based job scheduler on Unix systems. Each line in a "crontab" has a five-field time spec (minute, hour, day-of-month, month, day-of-week) followed by a command to run. "0 6 * * *" means 06:00 every day. is the simplest possible driver. A crontab line is a schedule plus a command. The discipline that makes it production-grade is in the wrapper, not the cron line: capture both streams, branch on the exit code, and never let the job hang.

# /etc/cron.d/triage  — run the triage agent every day at 06:00
# m h dom mon dow   user      command
0 6 * * *   deploy   /opt/agents/run_triage.sh >> /var/log/triage.log 2>&1

# ----- /opt/agents/run_triage.sh -----
#!/usr/bin/env bash
set -euo pipefail
export PATH=/usr/local/bin:/usr/bin:/bin        # cron's PATH is minimal — be explicit
source /opt/agents/.env                         # load ANTHROPIC_API_KEY (chmod 600)

cd /opt/agents
# Feed the day's alerts in on stdin; capture stdout (JSON) and the exit code
OUT="$(cat alerts.txt | python3 triage_agent.py)" && CODE=0 || CODE=$?

case "$CODE" in
  0) echo "$OUT" | jq -c . >> results.jsonl ;;            # success: append result
  2) echo "BAD OUTPUT, escalating" | mail -s triage oncall@co ;;  # never retry
  3) echo "$OUT" >> review_queue.jsonl ;;                 # needs a human
  *) echo "transient failure (code $CODE), will retry next run" ;;  # 1 etc.
esac
exit 0    # the wrapper itself always succeeds; cron should not "fail"

# Cron's five time fields:  minute hour day-of-month month day-of-week
#
#   *  *  *  *  *
#   |  |  |  |  +--- day of week  (0-6, Sun=0)
#   |  |  |  +------ month        (1-12)
#   |  |  +--------- day of month (1-31)
#   |  +------------ hour         (0-23)
#   +--------------- minute       (0-59)
#
# 0 6 * * *      every day at 06:00
# */15 * * * *   every 15 minutes
# 0 */4 * * *    every 4 hours, on the hour
# 0 9 * * 1-5    09:00 Mon-Fri only
#
# Test a crontab spec without waiting:  run the wrapper by hand first.
#   /opt/agents/run_triage.sh ; echo "exit=$?"

Running Inside CI (GitHub Actions)

The second great driver is CI. Here the agent runs on every push — for example, an agent that reviews the changelog or summarizes a diff. The key trick: provide the API key as an encrypted secret, then run the agent and let its exit code gate the pipeline.

# .github/workflows/agent-review.yml
name: Headless agent review
on: [push]

jobs:
  review:
    runs-on: ubuntu-latest
    env:
      # Stored once under Settings -> Secrets -> Actions. Never in the YAML.
      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
    steps:
      - uses: actions/checkout@v4

      - name: Install deps
        run: pip install anthropic

      - name: Run headless agent (exit code gates the build)
        run: |
          # The agent reads the changelog and emits a JSON verdict.
          # If it exits non-zero, this step fails and the build goes red.
          git log -1 --pretty=%B | python triage_agent.py | tee verdict.json
          jq -e '.severity != "critical"' verdict.json   # fail build on critical

      - name: Upload verdict artifact
        if: always()                       # keep the JSON even when the gate fails
        uses: actions/upload-artifact@v4
        with:
          name: agent-verdict
          path: verdict.json

Guardrails Without a Human

The animation traces a guarded headless run. Input enters, passes a guard layer (timeout + budget armed), and runs the agent loop. Two endings are possible: the loop finishes within limits and exits 0, or it trips a guard and exits non-zero — deterministically, with no human involved.

Encoding the Three Limits

This wraps a Claude tool-use loop in all three guards at once. The timeout is the outermost net; max-iterations bounds the loop; the token budget aborts mid-run if the conversation balloons.

# guarded_loop.py — three hard limits replace the human circuit-breaker
import signal, sys, json
from anthropic import Anthropic

client = Anthropic()
TOOLS = [...]   # your tool schemas (see M05 / M12)

class GuardTripped(Exception): pass

def _timeout_handler(signum, frame):
    raise GuardTripped("wall-clock timeout exceeded")

def run_guarded(task: str, *, max_seconds=60, max_iters=8, max_tokens=8000) -> dict:
    # GUARD 1: wall-clock timeout (Unix). SIGALRM fires after max_seconds.
    signal.signal(signal.SIGALRM, _timeout_handler)
    signal.alarm(max_seconds)

    messages = [{"role": "user", "content": task}]
    tokens_used = 0
    try:
        for i in range(max_iters):                  # GUARD 2: hard iteration cap
            resp = client.messages.create(
                model="claude-sonnet-4-6", max_tokens=1024,
                tools=TOOLS, messages=messages,
            )
            tokens_used += resp.usage.input_tokens + resp.usage.output_tokens
            print(f"[guard] iter={i} tokens={tokens_used}", file=sys.stderr)

            if tokens_used > max_tokens:             # GUARD 3: token budget
                raise GuardTripped(f"token budget {max_tokens} exceeded")

            if resp.stop_reason != "tool_use":       # loop ends: final answer
                text = "".join(b.text for b in resp.content if b.type == "text")
                return {"result": text, "iters": i + 1, "tokens": tokens_used}

            # Append the assistant turn, run each tool_use block, append tool_result
            messages.append({"role": "assistant", "content": resp.content})
            # ... execute tools, append {"role":"user","content":[tool_result, ...]} ...
        raise GuardTripped(f"max_iters {max_iters} reached without finishing")
    finally:
        signal.alarm(0)                              # always disarm the timer

if __name__ == "__main__":
    try:
        out = run_guarded(sys.stdin.read().strip())
        print(json.dumps({"ok": True, **out}))
        sys.exit(0)
    except GuardTripped as e:
        print(json.dumps({"ok": False, "error": str(e)}))
        print(f"[guard] TRIPPED: {e}", file=sys.stderr)
        sys.exit(1)                                  # deterministic non-zero exit

// guarded_loop.js — three hard limits replace the human circuit-breaker
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
const TOOLS = [/* your tool schemas (see M05 / M12) */];

class GuardTripped extends Error {}

async function runGuarded(task, { maxSeconds = 60, maxIters = 8, maxTokens = 8000 } = {}) {
  // GUARD 1: wall-clock timeout via an AbortController + a race
  const ac = new AbortController();
  const timer = setTimeout(() => ac.abort(), maxSeconds * 1000);

  const messages = [{ role: "user", content: task }];
  let tokensUsed = 0;
  try {
    for (let i = 0; i < maxIters; i++) {            // GUARD 2: hard iteration cap
      const resp = await client.messages.create(
        { model: "claude-sonnet-4-6", max_tokens: 1024, tools: TOOLS, messages },
        { signal: ac.signal }                       // abort kills the in-flight call
      );
      tokensUsed += resp.usage.input_tokens + resp.usage.output_tokens;
      console.error(`[guard] iter=${i} tokens=${tokensUsed}`);

      if (tokensUsed > maxTokens)                   // GUARD 3: token budget
        throw new GuardTripped(`token budget ${maxTokens} exceeded`);

      if (resp.stop_reason !== "tool_use") {        // loop ends: final answer
        const text = resp.content.filter((b) => b.type === "text").map((b) => b.text).join("");
        return { result: text, iters: i + 1, tokens: tokensUsed };
      }
      // Append the assistant turn, run each tool_use block, append tool_result
      messages.push({ role: "assistant", content: resp.content });
      // ... execute tools, push { role: "user", content: [toolResult, ...] } ...
    }
    throw new GuardTripped(`maxIters ${maxIters} reached without finishing`);
  } finally {
    clearTimeout(timer);                            // always disarm the timer
  }
}

const chunks = [];
for await (const c of process.stdin) chunks.push(c);
try {
  const out = await runGuarded(Buffer.concat(chunks).toString().trim());
  process.stdout.write(JSON.stringify({ ok: true, ...out }) + "\n");
  process.exit(0);
} catch (e) {
  process.stdout.write(JSON.stringify({ ok: false, error: e.message }) + "\n");
  console.error(`[guard] TRIPPED: ${e.message}`);
  process.exit(1);                                  // deterministic non-zero exit
}

Each guard maps to one specific way an unattended run goes wrong, so it helps to see what trips which:

1. Guard 1 — the timeout (signal.alarm / AbortController). This is the outermost net, armed before the loop and disarmed in finally so it never leaks into the next run. It catches the failure mode you cannot predict from the loop body: one API call that simply hangs. After max_seconds, SIGALRM fires (or the request aborts) and raises GuardTripped mid-call.
2. Guard 2 — the iteration cap (for i in range(max_iters)). A tool-use agent loops "reason → call a tool → observe" until stop_reason is no longer "tool_use". A confused model can decide it is never done. Bounding the loop with a plain for range means the worst case is max_iters calls, not infinity — the loop literally cannot spin forever.
3. Guard 3 — the token budget (tokens_used > max_tokens). Iterations can be few but enormous — a tool result that balloons the conversation, or a model that starts emitting a wall of text. We accumulate input_tokens + output_tokens every turn and abort the moment the running total crosses the cap, before the next billable call.

All three funnel into the same place: a GuardTripped exception, a JSON line on stdout describing what happened, a log on stderr, and sys.exit(1). Tripping a guard is a normal, expected, deterministic outcome — the agent fails loudly and on purpose instead of hanging silently. That is exactly the behavior the missing human used to provide.

Piping & Composition

A headless agent that follows the contract is a well-behaved Unix citizen: text in on stdin, JSON out on stdout, status in the exit code. That is exactly the interface grep, jq, and curl use — so your agent slots into a pipeline beside them. This is the Unix philosophy applied to AI: small programs that do one thing and connect through pipes.

# 1) Classify one alert and pull a single field with jq
echo "db replica lag 45s on orders-db" | python triage_agent.py 2>/dev/null \
  | jq -r '.severity'
# -> high

# 2) Triage a whole file, keep only the critical ones, count them
cat overnight_alerts.txt \
  | while read -r line; do echo "$line" | python triage_agent.py 2>/dev/null; done \
  | jq -c 'select(.severity=="critical")' \
  | tee critical.jsonl | wc -l
# -> 3

# 3) Chain TWO agents: triage, then for high-sev ones, draft a response
echo "$ALERT" | python triage_agent.py 2>/dev/null \
  | jq -r 'select(.severity=="high") | .summary' \
  | python draft_response_agent.py 2>/dev/null \
  | jq -r '.draft'

# 4) Fan an event from a webhook straight into the agent and post to Slack
curl -s "$EVENT_URL" \
  | python triage_agent.py 2>/dev/null \
  | jq '{text: ("Triage: " + .severity + " — " + .summary)}' \
  | curl -s -X POST -H 'Content-type: application/json' -d @- "$SLACK_WEBHOOK"

Lab: A Nightly Log-Triage Agent

Build a headless agent that a cron job runs every night: it reads a log file, asks Claude to flag anomalies, emits a JSON report, exits with a meaningful code, and the wrapper routes the result.

Knowledge Check

Test your understanding of headless execution, the headless contract, automation, and guardrails.

Summary

You learned to turn a Claude agent into an unattended worker:

• Headless vs headed: a program drives the agent and a program consumes its output — no UI, no human approving steps mid-run.
• The three I/O surfaces: the work order arrives via stdin/args/env, the JSON result goes to stdout and only stdout, and every log goes to stderr.
• The contract: a fixed {ok, data, error, meta} envelope plus a documented exit-code set (0/1/2/3) so callers branch on one stable signal.
• Drivers & guardrails: cron and CI launch the same program; a timeout, iteration cap, and token budget replace the human who used to hit Ctrl-C — and the Message Batches API handles the bulk overnight work at half the cost.

Next, M23: Capstone Project Series puts deployment, cost control, and headless operation together into a full production agent.