Module 6: Advanced Hooks and Headless SDK

4. Headless SDK — Python: Claude Code as an Invocable Service

4. Headless SDK — Python: Claude Code as an Invocable Service

Description

So far, Claude Code is something you open in the terminal, type into, and wait on. Functional, but limited: it requires your presence, your manual input, your interpretation of the output. The headless SDK changes that completely. Claude Code becomes a function you can call from a Python script — you pass it a prompt, give it allowed tools, and receive a structured JSON result you can parse, analyze, and use as input for the next action.

This opens a world of possibilities: changelog scripts that generate themselves, code review pipelines that analyze PRs automatically, fixers that repair lint errors without intervention, and orchestrators that chain multiple Claude Code invocations for complex tasks. All from Python — the language you probably already use for automation scripts.

By the end of this capsule you'll know how to run Claude Code in headless mode with claude -p, parse the JSON result, handle errors, and build real automation scripts. You'll also understand --allowedTools as a security mechanism to control what Claude can do in programmatic mode.


Headless Mode: The Basics

The -p flag

The -p (prompt) flag is what activates headless mode. Instead of opening an interactive session, Claude Code runs the prompt and finishes:

# Interactive (open session)
claude

# Headless (one execution, one result)
claude -p "How many Python files are in src/?"

Output formats

FlagFormatUse
--output-format textPlain textSimple scripts, human reading
--output-format jsonStructured JSONProgrammatic parsing
--output-format stream-jsonStreaming JSONReal-time monitoring

text — The default. The output is exactly what Claude would respond in the terminal. Simple but hard to parse programmatically.

json — The output is a JSON object with Claude's response, metadata, and cost. It's what you'll use 90% of the time in scripts.

stream-json — The output is a sequence of JSON objects, one per event (tool used, partial response, etc.). Useful for real-time monitoring of long executions.

--allowedTools: Security in headless mode

In headless mode there's no human approving each action. --allowedTools defines which tools Claude can use:

# Read-only — safe for analysis
claude -p "Analyze the code in src/" \
  --allowedTools "Read,Grep,Glob" \
  --output-format json

# Read + write — for implementation
claude -p "Fix the lint errors in src/api/" \
  --allowedTools "Read,Write,Edit,Grep,Glob" \
  --output-format json

# With shell — to run commands
claude -p "Run the tests and report failures" \
  --allowedTools "Read,Grep,Glob,Bash" \
  --output-format json

Security rule: Use the minimum tools necessary. If the script only needs to analyze code, don't give it Write or Bash.


Running Claude Code from Python

The basic pattern with subprocess

import subprocess
import json

def run_claude(prompt, allowed_tools=None, output_format="json"):
    cmd = ["claude", "-p", prompt, "--output-format", output_format]

    if allowed_tools:
        cmd.extend(["--allowedTools", ",".join(allowed_tools)])

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=300
    )

    if result.returncode != 0:
        raise RuntimeError(f"Claude failed: {result.stderr}")

    if output_format == "json":
        return json.loads(result.stdout)

    return result.stdout

output = run_claude(
    "How many Python files are in src/?",
    allowed_tools=["Read", "Glob", "Grep"]
)

print(output)

The JSON result

When you use --output-format json, the result has this structure:

{
  "type": "result",
  "subtype": "success",
  "is_error": false,
  "result": "I found 23 Python files in src/...",
  "session_id": "abc123",
  "cost_usd": 0.0042,
  "duration_ms": 8500,
  "num_turns": 3
}

Key fields:

  • result — Claude's response (the useful text)
  • is_error — Whether the execution had an error
  • cost_usd — Cost in dollars
  • duration_ms — Duration in milliseconds
  • num_turns — How many "turns" of tools Claude used

Parsing the result

import subprocess
import json

def run_claude(prompt, tools=None):
    cmd = ["claude", "-p", prompt, "--output-format", "json"]
    if tools:
        cmd.extend(["--allowedTools", ",".join(tools)])

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)

    if result.returncode != 0:
        return {"error": result.stderr, "is_error": True}

    try:
        parsed = json.loads(result.stdout)
    except json.JSONDecodeError:
        return {"error": "Invalid JSON output", "is_error": True}

    return parsed

output = run_claude(
    "List the endpoints defined in src/api/",
    tools=["Read", "Glob", "Grep"]
)

if output.get("is_error"):
    print(f"Error: {output.get('error', output.get('result'))}")
else:
    print(f"Result: {output['result']}")
    print(f"Cost: ${output.get('cost_usd', 0):.4f}")
    print(f"Duration: {output.get('duration_ms', 0)}ms")

Real Automation Scripts

Script 1: Automatic Changelog Generator

#!/usr/bin/env python3
"""Generates a changelog based on recent commits."""

import subprocess
import json
import sys
from datetime import datetime

def run_claude(prompt, tools):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    if result.returncode != 0:
        print(f"Error: {result.stderr}", file=sys.stderr)
        sys.exit(1)
    return json.loads(result.stdout)

def get_recent_commits(n=20):
    result = subprocess.run(
        ["git", "log", f"-{n}", "--oneline", "--no-merges"],
        capture_output=True, text=True
    )
    return result.stdout.strip()

def main():
    commits = get_recent_commits()

    if not commits:
        print("No recent commits")
        sys.exit(0)

    prompt = f"""Analyze these git commits and generate a professional changelog
in English with the following sections:
- New Features
- Bug Fixes
- Improvements
- Internal Changes

Commits:
{commits}

Read the files modified in the most relevant commits to better understand
the context of each change. Generate the changelog in Markdown format.
Don't include commit hashes."""

    output = run_claude(prompt, tools=["Read", "Grep", "Glob"])

    if output.get("is_error"):
        print(f"Error: {output['result']}", file=sys.stderr)
        sys.exit(1)

    date_str = datetime.now().strftime("%Y-%m-%d")
    changelog_entry = f"## {date_str}\n\n{output['result']}\n"

    changelog_path = "CHANGELOG.md"
    try:
        with open(changelog_path, "r") as f:
            existing = f.read()
    except FileNotFoundError:
        existing = "# Changelog\n\n"

    header = "# Changelog\n\n"
    body = existing.replace(header, "")

    with open(changelog_path, "w") as f:
        f.write(f"{header}{changelog_entry}\n{body}")

    print(f"Changelog updated: {changelog_path}")
    print(f"Cost: ${output.get('cost_usd', 0):.4f}")

if __name__ == "__main__":
    main()

Script 2: Automatic Code Review

#!/usr/bin/env python3
"""Runs an automatic code review on modified files."""

import subprocess
import json
import sys

def run_claude(prompt, tools):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    if result.returncode != 0:
        return {"is_error": True, "result": result.stderr}
    return json.loads(result.stdout)

def get_changed_files():
    result = subprocess.run(
        ["git", "diff", "--name-only", "HEAD~1"],
        capture_output=True, text=True
    )
    files = result.stdout.strip().split("\n")
    return [f for f in files if f.endswith((".py", ".ts", ".js", ".tsx"))]

def review_file(filepath):
    prompt = f"""Review the file {filepath} and produce a code review focused on:

1. **Potential bugs** — incorrect logic, unhandled edge cases
2. **Security** — SQL injection, XSS, exposed secrets, unsanitized input
3. **Performance** — N+1 queries, unnecessary loops, memory leaks
4. **Maintainability** — duplicated code, overly long functions, poor naming

For each issue found, report:
- Approximate line
- Severity (CRITICAL, WARNING, SUGGESTION)
- Description of the problem
- Fix suggestion

If the file is fine, report "No issues found."
Be concise. Only report real issues, not stylistic ones."""

    return run_claude(prompt, tools=["Read", "Grep", "Glob"])

def main():
    files = get_changed_files()

    if not files:
        print("No modified files to review")
        sys.exit(0)

    print(f"Reviewing {len(files)} files...")

    reviews = []
    total_cost = 0

    for filepath in files:
        print(f"  Reviewing: {filepath}")
        result = review_file(filepath)

        if result.get("is_error"):
            print(f"  Error reviewing {filepath}: {result['result']}")
            continue

        reviews.append({
            "file": filepath,
            "review": result["result"],
            "cost": result.get("cost_usd", 0)
        })
        total_cost += result.get("cost_usd", 0)

    print(f"\n{'='*60}")
    print("CODE REVIEW RESULTS")
    print(f"{'='*60}\n")

    for review in reviews:
        print(f"## {review['file']}")
        print(review["review"])
        print()

    print(f"Total files: {len(reviews)}")
    print(f"Total cost: ${total_cost:.4f}")

if __name__ == "__main__":
    main()

Script 3: Automatic Test Fixing

#!/usr/bin/env python3
"""Detects failing tests and asks Claude to fix them."""

import subprocess
import json
import sys

def run_tests():
    result = subprocess.run(
        ["python", "-m", "pytest", "--tb=short", "-q"],
        capture_output=True, text=True
    )
    return result.returncode, result.stdout + result.stderr

def run_claude(prompt, tools):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    if result.returncode != 0:
        return {"is_error": True, "result": result.stderr}
    return json.loads(result.stdout)

def main():
    max_attempts = 3

    for attempt in range(1, max_attempts + 1):
        print(f"\n--- Attempt {attempt}/{max_attempts} ---")

        exit_code, test_output = run_tests()

        if exit_code == 0:
            print("All tests pass!")
            sys.exit(0)

        print(f"Tests failing. Output:\n{test_output[:500]}")

        prompt = f"""The tests are failing. Here's the pytest output:

{test_output}

Analyze the errors, read the test files and the relevant source code files,
and fix the problems.

Rules:
- Prefer fixing the source code, not the tests (unless the test is
  clearly incorrect)
- If a test expects a specific value and the code returns another, verify
  which is the correct behavior
- Don't change the business logic unless it's a clear bug"""

        result = run_claude(
            prompt,
            tools=["Read", "Write", "Edit", "Grep", "Glob", "Bash"]
        )

        if result.get("is_error"):
            print(f"Claude error: {result['result']}")
            continue

        print(f"Claude fix applied (cost: ${result.get('cost_usd', 0):.4f})")

    exit_code, _ = run_tests()
    if exit_code == 0:
        print("All tests pass after fixes!")
    else:
        print(f"Tests still failing after {max_attempts} attempts")
        sys.exit(1)

if __name__ == "__main__":
    main()

Error Handling in Headless Mode

The possible errors

import subprocess
import json
import sys

def run_claude_safe(prompt, tools, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout
        )
    except subprocess.TimeoutExpired:
        return {"is_error": True, "error_type": "timeout",
                "result": f"Timeout after {timeout}s"}
    except FileNotFoundError:
        return {"is_error": True, "error_type": "not_found",
                "result": "claude CLI not found. Is it installed?"}

    if result.returncode != 0:
        return {"is_error": True, "error_type": "exit_code",
                "result": result.stderr or "Unknown error",
                "exit_code": result.returncode}

    try:
        parsed = json.loads(result.stdout)
    except json.JSONDecodeError:
        return {"is_error": True, "error_type": "json_parse",
                "result": f"Invalid JSON: {result.stdout[:200]}"}

    if parsed.get("is_error"):
        return {"is_error": True, "error_type": "claude_error",
                "result": parsed.get("result", "Unknown Claude error")}

    return parsed

output = run_claude_safe(
    "Analyze src/",
    tools=["Read", "Glob"],
    timeout=120
)

if output.get("is_error"):
    error_type = output.get("error_type", "unknown")
    print(f"Error ({error_type}): {output['result']}", file=sys.stderr)
else:
    print(output["result"])

Recommended timeouts

Task typeRecommended timeout
Analysis of one file60s
Analysis of a directory120s
Code review of multiple files300s
Feature implementation600s
Refactor of a complete module900s

Integration with CI/CD (Preview)

Example: GitHub Actions

name: Auto Code Review

on:
  pull_request:
    types: [opened, synchronize]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Install Claude Code
        run: npm install -g @anthropic-ai/claude-code

      - name: Run code review
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: python scripts/code-review.py > review-output.txt

      - name: Post review comment
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const review = fs.readFileSync('review-output.txt', 'utf8');
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: `## Automated Code Review\n\n${review}`
            });

This example is a preview — Module 10 (CI/CD Pipelines) covers the complete integration. Here we mention it so you can see where the SDK is headed.

The CI script

#!/usr/bin/env python3
"""Script for code review in CI/CD."""

import subprocess
import json
import os
import sys

def run_claude(prompt, tools, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)

    if result.returncode != 0:
        print(f"Error: {result.stderr}", file=sys.stderr)
        return None

    return json.loads(result.stdout)

def get_pr_diff():
    result = subprocess.run(
        ["git", "diff", "origin/main...HEAD", "--name-only"],
        capture_output=True, text=True
    )
    return result.stdout.strip()

def main():
    changed_files = get_pr_diff()

    if not changed_files:
        print("No files changed")
        sys.exit(0)

    prompt = f"""Review the changes in these files for a PR code review:

{changed_files}

Read each file and produce a concise review. For each file:
1. Summarize the changes
2. Identify bugs, security issues, or performance problems
3. Suggest improvements

Output in Markdown. Be direct and useful."""

    output = run_claude(prompt, tools=["Read", "Grep", "Glob"])

    if output and not output.get("is_error"):
        print(output["result"])
    else:
        print("Code review failed", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

Stream JSON: Real-Time Monitoring

When to use stream-json

For long executions where you want to see the progress:

import subprocess
import json

def run_claude_streaming(prompt, tools):
    cmd = ["claude", "-p", prompt, "--output-format", "stream-json",
           "--allowedTools", ",".join(tools)]

    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )

    for line in process.stdout:
        line = line.strip()
        if not line:
            continue

        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue

        event_type = event.get("type", "")

        if event_type == "assistant":
            content = event.get("message", {}).get("content", [])
            for block in content:
                if block.get("type") == "text":
                    print(f"Claude: {block['text'][:100]}...")

        elif event_type == "result":
            print(f"\nFinal: {event.get('result', '')[:200]}")
            print(f"Cost: ${event.get('cost_usd', 0):.4f}")

    process.wait()
    return process.returncode

run_claude_streaming(
    "Analyze all the files in src/ and generate a quality report",
    tools=["Read", "Grep", "Glob"]
)

Advanced Patterns

Chaining invocations

A script that runs Claude multiple times, using the output of one invocation as the input for the next:

import subprocess
import json

def run_claude(prompt, tools, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        return None
    return json.loads(result.stdout)

analysis = run_claude(
    "Analyze src/api/ and list all the endpoints with their response schemas",
    tools=["Read", "Grep", "Glob"]
)

if analysis and not analysis.get("is_error"):
    endpoints_info = analysis["result"]

    docs = run_claude(
        f"""Generate API documentation in OpenAPI format for these endpoints:

{endpoints_info}

Read the code files to get exact details of schemas,
parameters, and error responses.""",
        tools=["Read", "Grep", "Glob"]
    )

    if docs and not docs.get("is_error"):
        with open("docs/api-reference.md", "w") as f:
            f.write(docs["result"])
        print("API docs generated!")

Parallel execution with ThreadPoolExecutor

import subprocess
import json
from concurrent.futures import ThreadPoolExecutor, as_completed

def run_claude(prompt, tools, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        return {"is_error": True, "result": result.stderr}
    return json.loads(result.stdout)

modules = ["src/auth/", "src/products/", "src/orders/"]

def analyze_module(module_path):
    return run_claude(
        f"Analyze {module_path} and report: files, dependencies, issues",
        tools=["Read", "Grep", "Glob"]
    )

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {
        executor.submit(analyze_module, mod): mod
        for mod in modules
    }

    results = {}
    for future in as_completed(futures):
        module = futures[future]
        results[module] = future.result()
        print(f"Completed: {module}")

for module, result in results.items():
    if not result.get("is_error"):
        print(f"\n{module}: {result['result'][:200]}...")

Troubleshooting

"claude: command not found"

Cause: Claude Code isn't installed globally or isn't in the PATH.

Solution:

npm install -g @anthropic-ai/claude-code
which claude

If you use a Python virtual environment, Claude may not be in the subprocess's PATH. Use the full path:

claude_path = subprocess.run(["which", "claude"], capture_output=True, text=True).stdout.strip()
cmd = [claude_path, "-p", prompt, ...]

"JSON parse error in the output"

Cause: Claude printed additional text before or after the JSON, or the output was truncated.

Solution: Use --output-format json explicitly. If it persists, filter the output:

stdout = result.stdout.strip()
json_start = stdout.find("{")
json_end = stdout.rfind("}") + 1
if json_start >= 0 and json_end > json_start:
    parsed = json.loads(stdout[json_start:json_end])

"Timeout on long executions"

Cause: The subprocess timeout is too short for the task.

Solution: Increase the timeout or use stream-json to monitor progress:

result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)

"Claude can't read the project files"

Cause: The subprocess runs from a directory different from the project's.

Solution: Specify the working directory:

result = subprocess.run(
    cmd,
    capture_output=True, text=True,
    cwd="/path/to/your/project"
)

"The cost is higher than expected"

Cause: Claude uses too many turns or reads unnecessary files.

Solution: Be specific in the prompt and limit the tools:

# BAD: vague prompt, many tools
run_claude("Improve the code", tools=["Read", "Write", "Edit", "Bash", "Grep", "Glob"])

# GOOD: specific prompt, minimal tools
run_claude(
    "Read src/api/routes.py and suggest 3 specific performance improvements",
    tools=["Read", "Grep"]
)

Comparison: Execution Modes

AspectInteractive (claude)Headless textHeadless JSONStream JSON
UseManual, explorationSimple scriptsAutomationMonitoring
InputYou type-p "prompt"-p "prompt"-p "prompt"
OutputTerminalPlain textParseable JSONJSON events
ParsingNot neededHardEasyModerate
FeedbackReal-timeAt the endAt the endReal-time
Best forDevelopmentSimple CIPython scriptsLong executions

Exercises

Exercise 1: Headless "Hello World" (Easy)

Write a Python script that runs claude -p "How many files are in the current directory?" in headless mode with JSON output, parses the result, and prints only Claude's response and the cost.

See solution
#!/usr/bin/env python3
import subprocess
import json

result = subprocess.run(
    ["claude", "-p", "How many files are in the current directory?",
     "--output-format", "json",
     "--allowedTools", "Glob"],
    capture_output=True, text=True, timeout=60
)

output = json.loads(result.stdout)
print(f"Response: {output['result']}")
print(f"Cost: ${output.get('cost_usd', 0):.4f}")

Exercise 2: Reusable function with error handling (Easy)

Create an ask_claude(prompt, tools, timeout) function that encapsulates the headless execution logic with error handling for: timeout, invalid JSON, Claude error, and CLI not found.

See solution
import subprocess
import json

def ask_claude(prompt, tools=None, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json"]
    if tools:
        cmd.extend(["--allowedTools", ",".join(tools)])

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return {"is_error": True, "error": "timeout"}
    except FileNotFoundError:
        return {"is_error": True, "error": "claude not found"}

    if result.returncode != 0:
        return {"is_error": True, "error": result.stderr}

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return {"is_error": True, "error": "invalid json"}

output = ask_claude("List the functions in src/api/routes.py", tools=["Read"])
if output.get("is_error"):
    print(f"Error: {output['error']}")
else:
    print(output["result"])

Exercise 3: Changelog generator (Medium)

Write a script that: (1) gets the last 10 commits with git log, (2) passes that information to Claude in headless mode to generate a changelog, and (3) saves the result to CHANGELOG.md.

See solution
#!/usr/bin/env python3
import subprocess
import json
from datetime import datetime

def get_commits():
    result = subprocess.run(
        ["git", "log", "-10", "--oneline", "--no-merges"],
        capture_output=True, text=True
    )
    return result.stdout.strip()

def ask_claude(prompt, tools, timeout=300):
    cmd = ["claude", "-p", prompt, "--output-format", "json",
           "--allowedTools", ",".join(tools)]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        return None
    return json.loads(result.stdout)

commits = get_commits()
if not commits:
    print("No commits")
    exit()

output = ask_claude(
    f"Generate a changelog in Markdown for these commits:\n{commits}\n"
    "Categorize into: Features, Fixes, Improvements. In English.",
    tools=["Read", "Grep", "Glob"]
)

if output and not output.get("is_error"):
    date = datetime.now().strftime("%Y-%m-%d")
    with open("CHANGELOG.md", "w") as f:
        f.write(f"# Changelog\n\n## {date}\n\n{output['result']}\n")
    print(f"Changelog generated (${output.get('cost_usd', 0):.4f})")

Exercise 4: Parallel module analysis (Medium)

Write a script that analyzes 3 directories in parallel using ThreadPoolExecutor, each with its own headless Claude invocation, and consolidates the results.

See solution
#!/usr/bin/env python3
import subprocess
import json
from concurrent.futures import ThreadPoolExecutor, as_completed

def analyze(path):
    cmd = ["claude", "-p",
           f"Analyze {path}: files, dependencies, issues. Be concise.",
           "--output-format", "json",
           "--allowedTools", "Read,Grep,Glob"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    if result.returncode != 0:
        return {"path": path, "error": result.stderr}
    parsed = json.loads(result.stdout)
    return {"path": path, "result": parsed.get("result", ""),
            "cost": parsed.get("cost_usd", 0)}

modules = ["src/api/", "src/models/", "src/services/"]

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(analyze, m): m for m in modules}
    results = []
    for future in as_completed(futures):
        results.append(future.result())
        print(f"Done: {futures[future]}")

total_cost = 0
for r in results:
    print(f"\n{'='*40}")
    print(f"Module: {r['path']}")
    if "error" in r:
        print(f"Error: {r['error']}")
    else:
        print(r["result"][:300])
        total_cost += r.get("cost", 0)

print(f"\nTotal cost: ${total_cost:.4f}")

Exercise 5: Complete auto-fix pipeline (Hard)

Write a script that: (1) runs the tests, (2) if they fail, asks Claude to fix the errors (headless mode with Write/Edit), (3) runs the tests again, (4) repeats up to 3 times or until they pass, (5) reports the final result with the accumulated cost.

See solution
#!/usr/bin/env python3
import subprocess
import json
import sys

def run_tests():
    result = subprocess.run(
        ["python", "-m", "pytest", "-x", "--tb=short", "-q"],
        capture_output=True, text=True, timeout=120
    )
    return result.returncode == 0, result.stdout + result.stderr

def fix_with_claude(test_output):
    cmd = ["claude", "-p",
           f"Tests failing. Fix the bugs:\n\n{test_output}\n\n"
           "Read the failing test and source files. Fix the source code, "
           "not the tests (unless the test is clearly wrong).",
           "--output-format", "json",
           "--allowedTools", "Read,Write,Edit,Grep,Glob"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    if result.returncode != 0:
        return None
    return json.loads(result.stdout)

total_cost = 0

for attempt in range(1, 4):
    print(f"\n--- Attempt {attempt}/3 ---")
    passed, output = run_tests()

    if passed:
        print(f"All tests pass! Total cost: ${total_cost:.4f}")
        sys.exit(0)

    print(f"Tests failing. Asking Claude to fix...")
    result = fix_with_claude(output)

    if result:
        cost = result.get("cost_usd", 0)
        total_cost += cost
        print(f"Fix applied (${cost:.4f})")
    else:
        print("Claude fix failed")

passed, _ = run_tests()
if passed:
    print(f"Fixed! Total cost: ${total_cost:.4f}")
else:
    print(f"Still failing after 3 attempts. Cost: ${total_cost:.4f}")
    sys.exit(1)

Summary

  • The -p flag activates headless mode — Claude Code runs a prompt and finishes
  • --output-format json produces parseable output with json.loads() — it includes result, is_error, cost_usd, duration_ms
  • --allowedTools controls which tools Claude can use — fundamental security in headless mode
  • subprocess.run() is the standard way to invoke Claude from Python — with capture_output=True, text=True, and timeout
  • Real use cases: automatic changelog, code review, auto-fix of tests, parallel module analysis
  • Error handling must cover: timeout, invalid JSON, Claude error, CLI not found
  • Stream JSON enables real-time monitoring of long executions
  • ThreadPoolExecutor enables parallel analysis of multiple modules
  • Chaining invocations enables pipelines where one invocation's output feeds the next
  • The integration with CI/CD (GitHub Actions) is a preview — covered in depth in the CI/CD Pipelines guide

Additional Resources

  1. Claude Code CLI Reference — Official documentation of -p, --output-format, --allowedTools
  2. Python subprocess Module — subprocess reference for invoking processes
  3. Python json Module — JSON parsing reference
  4. concurrent.futures — ThreadPoolExecutor for parallel execution
  5. Claude Code Best Practices — Automation best practices
  6. GitHub Actions — CI/CD reference for integration with Claude
  7. Claude Code Overview — General context of Claude Code
  8. Claude Code Hooks — Hooks that complement the SDK

Next capsule: In capsule 05 you'll do the same from TypeScript/Node.js. You'll see child_process for invocation via subprocess, the @anthropic-ai/claude-code package for native integration, and when to choose Python vs TypeScript for your automation scripts. If your stack is JavaScript, this capsule is where the SDK comes to life.