How to Use Claude Code Workflows in Python: AGENTS.md, Ultracode CLI, and AsyncAnthropic Orchestration

etl_pipeline/
├── AGENTS.md                    # agent instructions (shared by both tracks)
├── src/
│   ├── extractors/
│   │   └── api_extractor.py     # fetches data from external API
│   ├── transformers/
│   │   └── data_transformer.py  # normalises and validates raw records
│   ├── loaders/
│   │   └── db_loader.py         # writes to PostgreSQL via SQLAlchemy
│   └── validators/
│       └── schema_validator.py  # validates Pydantic schemas against data
└── tests/                       # empty — agents will populate this

# AGENTS.md — ETL Pipeline Code Review

## Context
Python ETL pipeline. Four modules in src/. Tests in tests/ are empty and must be populated.
The pipeline runs in production on AWS Lambda with a 15-second timeout per invocation.

## Review criteria (apply to every module)
1. Security: SQL injection, secrets in code, unsafe deserialization, unvalidated inputs
2. Error handling: all IO operations must have try/except with specific exception types
3. Type annotations: all public functions must have full signatures (PEP 484)
4. Testability: functions with side effects must be injectable (no hard-coded dependencies)

## Output format per module
For each file, produce three artefacts:
- ISSUES.md  — numbered list of issues with severity (critical / high / medium)
- tests/test_[module].py — pytest file that fails on the identified issues
- PATCH.diff  — unified diff that fixes all issues and makes the tests pass

## Forbidden
- Do NOT modify files outside src/ and tests/
- Do NOT add external dependencies not already in requirements.txt
- Do NOT rewrite a module from scratch — patch the existing code

## Concurrency note
Each module is independent. All four can be reviewed in parallel.
Assign one primary agent and one adversarial reviewer per module.

# 1. Navigate to the project root (where AGENTS.md lives)
cd etl_pipeline/

# 2. Start a Claude Code session
claude

# 3. Inside the session: switch to Opus 4.8
/model claude-opus-4-8

# 4. Set ultracode effort — activates xhigh reasoning + workflow engine
/effort ultracode

# 5. Trigger the workflow — include "workflow" anywhere in the prompt
Create a workflow to review all four Python modules in src/ following AGENTS.md,
producing ISSUES.md, a test file, and a PATCH.diff for each module.
Modules are independent — run reviews in parallel.

# Claude will now plan the orchestration, confirm the approach,
# then dispatch agents. Each wave runs up to 16 agents concurrently.
# Track spend in real time:
/usage

import asyncio
import anthropic
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

@dataclass
class ReviewResult:
    file_path: str
    issues: str = ""
    test_code: str = ""
    patch: str = ""
    tokens_used: int = 0
    error: Optional[str] = None


class AsyncWorkflowOrchestrator:
    """Parallel code review orchestrator with token budget enforcement."""

    SYSTEM_PROMPT = """You are a senior Python engineer performing a code review.
For the file provided, return a JSON object with three keys:
  "issues": markdown list of issues with severity labels
  "test_code": complete pytest file that fails on the issues you found
  "patch": unified diff that fixes all issues and makes the tests pass
Return ONLY the JSON — no prose before or after."""

    def __init__(
        self,
        model: str = "claude-opus-4-8",
        token_budget: int = 200_000,  # kill switch threshold
        max_concurrency: int = 5,       # respect API rate limits
    ):
        self.client = anthropic.AsyncAnthropic()
        self.model = model
        self.token_budget = token_budget
        self.tokens_spent = 0
        self._killed = False
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._lock = asyncio.Lock()   # guards tokens_spent counter

    async def _charge_tokens(self, tokens: int) -> bool:
        """Thread-safe token accounting. Returns False if budget exceeded."""
        async with self._lock:
            self.tokens_spent += tokens
            if self.tokens_spent > self.token_budget:
                self._killed = True
                return False
        return True

    async def review_file(self, path: Path) -> ReviewResult:
        result = ReviewResult(file_path=str(path))

        if self._killed:
            result.error = "Workflow killed: token budget exceeded"
            return result

        async with self._semaphore:
            try:
                content = path.read_text(encoding="utf-8")
                response = await self.client.messages.create(
                    model=self.model,
                    max_tokens=8192,
                    thinking={"type": "enabled", "budget_tokens": 10_000},
                    system=[{
                        "type": "text",
                        "text": self.SYSTEM_PROMPT,
                        "cache_control": {"type": "ephemeral", "ttl": "1h"}
                    }],
                    messages=[{
                        "role": "user",
                        "content": f"Review this file ({path.name}):\n\n```python\n{content}\n```"
                    }]
                )

                # Charge tokens — kill if over budget
                total = response.usage.input_tokens + response.usage.output_tokens
                result.tokens_used = total
                if not await self._charge_tokens(total):
                    result.error = f"Budget exceeded after {self.tokens_spent:,} tokens"
                    return result

                # Extract the text block (skip thinking blocks)
                text = next(
                    b.text for b in response.content if b.type == "text"
                )
                import json
                data = json.loads(text)
                result.issues    = data.get("issues", "")
                result.test_code = data.get("test_code", "")
                result.patch     = data.get("patch", "")

            except Exception as e:
                result.error = str(e)

        return result

    async def run(self, src_dir: Path) -> list[ReviewResult]:
        files = list(src_dir.rglob("*.py"))
        tasks = [self.review_file(f) for f in files]
        return await asyncio.gather(*tasks)

import asyncio
from pathlib import Path
from orchestrator import AsyncWorkflowOrchestrator

async def main():
    src = Path("etl_pipeline/src")
    out = Path("review_output")
    out.mkdir(exist_ok=True)

    orchestrator = AsyncWorkflowOrchestrator(
        model="claude-opus-4-8",
        token_budget=300_000,  # ~$1.50 at Standard tier — adjust to your comfort
        max_concurrency=4,      # 4 files in parallel, one API call each
    )

    print(f"Reviewing {src} — kill switch at {orchestrator.token_budget:,} tokens")
    results = await orchestrator.run(src)

    # Write artefacts and print summary
    errors, reviewed = [], []
    for r in results:
        name = Path(r.file_path).stem
        if r.error:
            errors.append(r)
            print(f"  ✗ {name}: {r.error}")
            continue

        (out / f"{name}_ISSUES.md").write_text(r.issues)
        (out / f"test_{name}.py").write_text(r.test_code)
        (out / f"{name}.patch").write_text(r.patch)
        reviewed.append(r)
        print(f"  ✓ {name}: {r.tokens_used:,} tokens")

    total = orchestrator.tokens_spent
    cost  = (total / 1_000_000) * 5.00   # Standard tier input rate
    print(f"\nTotal: {len(reviewed)} reviewed, {len(errors)} failed")
    print(f"Tokens: {total:,} | Estimated cost: ${cost:.4f}")
    print(f"Output: {out}/")

asyncio.run(main())

async def call_with_compaction(
    client: anthropic.AsyncAnthropic,
    messages: list,
    vector_store,
    session_id: str,
    **kwargs,
) -> anthropic.types.Message:
    """Wrap a beta messages call; handles compaction transparently."""
    response = await client.beta.messages.create(
        betas=["compact-2026-01-12"],
        context_management={
            "edits": [{
                "strategy": "compact_20260112",
                "trigger": {"type": "token_count", "threshold": 80_000},
                "instructions": "Preserve all file paths, issue IDs, and patch hunks verbatim.",
                "pause_after_compaction": True,
            }]
        },
        messages=messages,
        **kwargs,
    )

    if response.stop_reason == "compaction":
        block = next(b for b in response.content if b.type == "compaction")
        summary = block.content or ""           # guard against null (known edge case)
        if summary:
            await vector_store.upsert(session_id, summary)
        # Resume with empty messages — server has already replaced history
        return await call_with_compaction(client, [], vector_store, session_id, **kwargs)

    return response