4. SDK Headless — Python: Claude Code como Servicio Invocable

Descripción

Hasta ahora, Claude Code es algo que abres en la terminal, le escribes, y esperas. Funcional, pero limitado: requiere tu presencia, tu input manual, tu interpretación del output. El SDK headless cambia eso completamente. Claude Code se convierte en una función que puedes llamar desde un script de Python — le pasas un prompt, le das herramientas permitidas, y recibes un resultado JSON estructurado que puedes parsear, analizar, y usar como input para la siguiente acción.

Esto abre un mundo de posibilidades: scripts de changelog que se generan solos, pipelines de code review que analizan PRs automáticamente, correctors que arreglan errores de lint sin intervención, y orquestadores que encadenan múltiples invocaciones de Claude Code para tareas complejas. Todo desde Python — el lenguaje que probablemente ya usas para scripts de automatización.

Al terminar esta cápsula sabrás ejecutar Claude Code en modo headless con claude -p, parsear el resultado JSON, manejar errores, y construir scripts de automatización reales. También entenderás --allowedTools como mecanismo de seguridad para controlar qué puede hacer Claude en modo programático.

Modo Headless: Lo Básico

El flag -p

El flag -p (prompt) es lo que activa el modo headless. En lugar de abrir una sesión interactiva, Claude Code ejecuta el prompt y termina:

# Interactivo (sesión abierta)
claude

# Headless (una ejecución, un resultado)
claude -p "¿Cuántos archivos Python hay en src/?"

Formatos de output

text — El default. El output es exactamente lo que Claude respondería en la terminal. Simple pero difícil de parsear programáticamente.

json — El output es un objeto JSON con la respuesta de Claude, metadata, y costo. Es lo que usarás el 90% del tiempo en scripts.

stream-json — El output es una secuencia de objetos JSON, uno por cada evento (herramienta usada, respuesta parcial, etc.). Útil para monitoring en tiempo real de ejecuciones largas.

--allowedTools: Seguridad en modo headless

En modo headless no hay un humano aprobando cada acción. --allowedTools define qué herramientas puede usar Claude:

# Solo lectura — seguro para análisis
claude -p "Analiza el código en src/" \
  --allowedTools "Read,Grep,Glob" \
  --output-format json

# Lectura + escritura — para implementación
claude -p "Corrige los errores de lint en src/api/" \
  --allowedTools "Read,Write,Edit,Grep,Glob" \
  --output-format json

# Con shell — para ejecutar comandos
claude -p "Corre los tests y reporta fallos" \
  --allowedTools "Read,Grep,Glob,Bash" \
  --output-format json

Regla de seguridad: Usa el mínimo de herramientas necesarias. Si el script solo necesita analizar código, no le des Write ni Bash.

Ejecutar Claude Code desde Python

El patrón básico con 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(
    "¿Cuántos archivos Python hay en src/?",
    allowed_tools=["Read", "Glob", "Grep"]
)

print(output)

El resultado JSON

Cuando usas --output-format json, el resultado tiene esta estructura:

{
  "type": "result",
  "subtype": "success",
  "is_error": false,
  "result": "Encontré 23 archivos Python en src/...",
  "session_id": "abc123",
  "cost_usd": 0.0042,
  "duration_ms": 8500,
  "num_turns": 3
}

Campos clave:

• result — La respuesta de Claude (el texto útil)
• is_error — Si la ejecución tuvo un error
• cost_usd — Costo en dólares
• duration_ms — Duración en milisegundos
• num_turns — Cuántos "turnos" de herramientas usó Claude

Parsing del resultado

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(
    "Lista los endpoints definidos en src/api/",
    tools=["Read", "Glob", "Grep"]
)

if output.get("is_error"):
    print(f"Error: {output.get('error', output.get('result'))}")
else:
    print(f"Resultado: {output['result']}")
    print(f"Costo: ${output.get('cost_usd', 0):.4f}")
    print(f"Duración: {output.get('duration_ms', 0)}ms")

Scripts de Automatización Reales

Script 1: Generador de Changelog Automático

#!/usr/bin/env python3
"""Genera un changelog basado en los commits recientes."""

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 hay commits recientes")
        sys.exit(0)

    prompt = f"""Analiza estos commits de git y genera un changelog profesional 
en español con las siguientes secciones:
- Nuevas Features
- Correcciones de Bugs
- Mejoras
- Cambios Internos

Commits:
{commits}

Lee los archivos modificados en los commits más relevantes para entender 
mejor el contexto de cada cambio. Genera el changelog en formato Markdown.
No incluyas hashes de commit."""

    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 actualizado: {changelog_path}")
    print(f"Costo: ${output.get('cost_usd', 0):.4f}")

if __name__ == "__main__":
    main()

Script 2: Code Review Automático

#!/usr/bin/env python3
"""Ejecuta code review automático en archivos modificados."""

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"""Revisa el archivo {filepath} y produce un code review enfocado en:

1. **Bugs potenciales** — lógica incorrecta, edge cases no manejados
2. **Seguridad** — inyección SQL, XSS, secrets expuestos, input sin sanitizar
3. **Performance** — N+1 queries, loops innecesarios, memory leaks
4. **Mantenibilidad** — código duplicado, funciones muy largas, naming pobre

Para cada issue encontrado, reporta:
- Línea aproximada
- Severidad (CRITICAL, WARNING, SUGGESTION)
- Descripción del problema
- Sugerencia de fix

Si el archivo está bien, reporta "No issues found."
Sé conciso. Solo reporta issues reales, no estilísticos."""

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

def main():
    files = get_changed_files()

    if not files:
        print("No hay archivos modificados para review")
        sys.exit(0)

    print(f"Revisando {len(files)} archivos...")

    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: Fix de Tests Automático

#!/usr/bin/env python3
"""Detecta tests fallidos y pide a Claude que los corrija."""

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"""Los tests están fallando. Aquí está el output de pytest:

{test_output}

Analiza los errores, lee los archivos de test y los archivos de código fuente 
relevantes, y corrige los problemas. 

Reglas:
- Prefiere corregir el código fuente, no los tests (a menos que el test sea 
  claramente incorrecto)
- Si un test espera un valor específico y el código retorna otro, verifica 
  cuál es el comportamiento correcto
- No cambies la lógica de negocio a menos que sea un bug claro"""

        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 en Modo Headless

Los errores posibles

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(
    "Analiza 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"])

Timeouts recomendados

Integración con CI/CD (Preview)

Ejemplo: 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}`
            });

Este ejemplo es un preview — el Módulo 10 (CI/CD Pipelines) cubre la integración completa. Aquí lo mencionamos para que veas hacia dónde va el SDK.

El script de CI

#!/usr/bin/env python3
"""Script para code review en 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"""Revisa los cambios en estos archivos para un code review de PR:

{changed_files}

Lee cada archivo y produce un review conciso. Para cada archivo:
1. Resume los cambios
2. Identifica bugs, issues de seguridad, o problemas de performance
3. Sugiere mejoras

Output en Markdown. Sé directo y útil."""

    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: Monitoring en Tiempo Real

Cuándo usar stream-json

Para ejecuciones largas donde quieres ver el progreso:

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(
    "Analiza todos los archivos en src/ y genera un reporte de calidad",
    tools=["Read", "Grep", "Glob"]
)

Patrones Avanzados

Encadenamiento de invocaciones

Un script que ejecuta Claude múltiples veces, usando el output de una invocación como input de la siguiente:

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(
    "Analiza src/api/ y lista todos los endpoints con sus schemas de response",
    tools=["Read", "Grep", "Glob"]
)

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

    docs = run_claude(
        f"""Genera documentación API en formato OpenAPI para estos endpoints:

{endpoints_info}

Lee los archivos de código para obtener detalles exactos de schemas,
parámetros, y respuestas de error.""",
        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!")

Ejecución paralela con 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"Analiza {module_path} y reporta: archivos, dependencias, 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"

Causa: Claude Code no está instalado globalmente o no está en el PATH.

Solución:

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

Si usas un virtual environment de Python, Claude puede no estar en el PATH del subprocess. Usa la ruta completa:

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

"JSON parse error en el output"

Causa: Claude imprimió texto adicional antes o después del JSON, o el output se truncó.

Solución: Usa --output-format json explícitamente. Si persiste, filtra el 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 en ejecuciones largas"

Causa: El timeout de subprocess es demasiado corto para la tarea.

Solución: Incrementa el timeout o usa stream-json para monitorear progreso:

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

"Claude no puede leer archivos del proyecto"

Causa: El subprocess se ejecuta desde un directorio diferente al del proyecto.

Solución: Especifica el working directory:

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

"El costo es más alto de lo esperado"

Causa: Claude usa demasiados turnos o lee archivos innecesarios.

Solución: Sé específico en el prompt y limita las herramientas:

# MAL: prompt vago, muchas herramientas
run_claude("Mejora el código", tools=["Read", "Write", "Edit", "Bash", "Grep", "Glob"])

# BIEN: prompt específico, herramientas mínimas
run_claude(
    "Lee src/api/routes.py y sugiere 3 mejoras de performance específicas",
    tools=["Read", "Grep"]
)

Comparación: Modos de Ejecución

Ejercicios

Ejercicio 1: "Hello World" headless (Fácil)

Escribe un script Python que ejecute claude -p "¿Cuántos archivos hay en el directorio actual?" en modo headless con output JSON, parsee el resultado, e imprima solo la respuesta de Claude y el costo.

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

result = subprocess.run(
    ["claude", "-p", "¿Cuántos archivos hay en el directorio actual?",
     "--output-format", "json",
     "--allowedTools", "Glob"],
    capture_output=True, text=True, timeout=60
)

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

Ejercicio 2: Función reutilizable con error handling (Fácil)

Crea una función ask_claude(prompt, tools, timeout) que encapsule la lógica de ejecución headless con manejo de errores para: timeout, JSON inválido, error de Claude, y CLI no encontrado.

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("Lista las funciones en src/api/routes.py", tools=["Read"])
if output.get("is_error"):
    print(f"Error: {output['error']}")
else:
    print(output["result"])

Ejercicio 3: Changelog generator (Medio)

Escribe un script que: (1) obtenga los últimos 10 commits con git log, (2) pase esa información a Claude en modo headless para generar un changelog, y (3) guarde el resultado en CHANGELOG.md.

#!/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"Genera un changelog en Markdown para estos commits:\n{commits}\n"
    "Categoriza en: Features, Fixes, Improvements. En español.",
    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 generado (${output.get('cost_usd', 0):.4f})")

Ejercicio 4: Análisis paralelo de módulos (Medio)

Escribe un script que analice 3 directorios en paralelo usando ThreadPoolExecutor, cada uno con su propia invocación headless de Claude, y consolide los resultados.

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

def analyze(path):
    cmd = ["claude", "-p",
           f"Analiza {path}: archivos, dependencias, issues. Sé conciso.",
           "--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}")

Ejercicio 5: Auto-fix pipeline completo (Difícil)

Escribe un script que: (1) corre los tests, (2) si fallan, pide a Claude que corrija los errores (modo headless con Write/Edit), (3) corre los tests de nuevo, (4) repite hasta 3 veces o hasta que pasen, (5) reporta el resultado final con costo acumulado.

#!/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)

Resumen

• El flag -p activa el modo headless — Claude Code ejecuta un prompt y termina
• --output-format json produce output parseable con json.loads() — incluye result, is_error, cost_usd, duration_ms
• --allowedTools controla qué herramientas puede usar Claude — seguridad fundamental en modo headless
• subprocess.run() es la forma estándar de invocar Claude desde Python — con capture_output=True, text=True, y timeout
• Casos de uso reales: changelog automático, code review, auto-fix de tests, análisis paralelo de módulos
• El error handling debe cubrir: timeout, JSON inválido, error de Claude, CLI no encontrada
• Stream JSON permite monitoring en tiempo real de ejecuciones largas
• ThreadPoolExecutor habilita análisis paralelo de múltiples módulos
• El encadenamiento de invocaciones permite pipelines donde el output de una invocación alimenta la siguiente
• La integración con CI/CD (GitHub Actions) es un preview — se cubre a fondo en la guía de CI/CD Pipelines

Recursos Adicionales

1. Claude Code CLI Reference — Documentación oficial de -p, --output-format, --allowedTools
2. Python subprocess Module — Referencia de subprocess para invocar procesos
3. Python json Module — Referencia de parseo JSON
4. concurrent.futures — ThreadPoolExecutor para ejecución paralela
5. Claude Code Best Practices — Buenas prácticas de automatización
6. GitHub Actions — Referencia de CI/CD para integración con Claude
7. Claude Code Overview — Contexto general de Claude Code
8. Claude Code Hooks — Hooks que complementan el SDK

Siguiente cápsula: En la cápsula 05 harás lo mismo desde TypeScript/Node.js. Verás child_process para invocación via subprocess, el paquete @anthropic-ai/claude-code para integración nativa, y cuándo elegir Python vs TypeScript para tus scripts de automatización. Si tu stack es JavaScript, esta cápsula es donde el SDK cobra vida.