Module 4. Claude API in Python

Chat tools get a team started. The API is how you build the pieces a client keeps: a document extractor, a triage service, a nightly job. This module takes you from one messages.create call to a validated, logged, costed pipeline you can wrap in a web service.

4.1 Messages API basics

Everything in the API is one call: you send a model ID, a list of messages and a max_tokens cap, and you get a message back. Set ANTHROPIC_API_KEY in the environment and the client finds it.

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment

# Check current model IDs at docs.claude.com
msg = client.messages.create(
    model="claude-haiku-5-5",
    max_tokens=300,                      # a hard cap on output tokens, always required
    system="Answer in the language of the question. Be brief.",
    messages=[{"role": "user", "content": "מה ההבדל בין חשבונית מס לקבלה?"}],
)
print(msg.content[0].text)
print(msg.stop_reason, msg.usage.input_tokens, msg.usage.output_tokens)

# The API is stateless: to continue a conversation, send the whole history again.
history = [
    {"role": "user", "content": "Name one risk of automating invoice approval."},
    {"role": "assistant", "content": msg.content[0].text},
    {"role": "user", "content": "And one control that reduces it?"},
]
follow_up = client.messages.create(model="claude-haiku-5-5", max_tokens=300, messages=history)

Five things to know:

• Roles. Messages alternate user and assistant. The system parameter is separate and holds standing instructions: role, rules, output format.
• Stateless. The API remembers nothing. Your code keeps the history and sends it each time.
• Content. A message's content is a string or a list of blocks (text, image, document, tool use, tool result). The reply is also a list of blocks, so read msg.content[0].text only after you know block 0 is text.
• max_tokens. It is a ceiling, not a target. If the answer is longer, it is cut off and the stop reason says so.
• usage. Every response reports input and output tokens. That is the raw material for cost control (4.8).

4.2 Streaming

Streaming sends text as it is generated. Use it when a person watches the answer appear (a chat box, a drafting tool). For back-office pipelines it adds code and no benefit, so use plain create there. The SDK's helper also handles long outputs that would otherwise risk a request timeout.

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-5-5",  # check current IDs
    max_tokens=800,
    messages=[{"role": "user", "content": "Summarise the benefits of e-invoicing for a 30-person firm."}],
) as stream:
    for text in stream.text_stream:       # print tokens as they arrive
        print(text, end="", flush=True)
    final = stream.get_final_message()    # the same Message object create() returns

print("\n", final.stop_reason, final.usage.output_tokens)

In a web service you forward each chunk to the browser (server-sent events or a websocket). That is a UI concern; the API side is just the loop above.

4.3 Vision and PDF input

Claude reads images and PDFs directly. For a PDF you send a document block; for a photo you send an image block. Both take base64 data (or a URL or a file reference in some cases; check the docs for what is current). The model sees the layout and any scanned text, so you often skip OCR entirely, which matters for Hebrew and mixed-language invoices.

import base64
from pathlib import Path


def file_block(path: str) -> dict:
    data = base64.standard_b64encode(Path(path).read_bytes()).decode("ascii")
    suffix = Path(path).suffix.lower()
    if suffix == ".pdf":
        return {"type": "document",
                "source": {"type": "base64", "media_type": "application/pdf", "data": data}}
    media = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}[suffix]
    return {"type": "image", "source": {"type": "base64", "media_type": media, "data": data}}


content = [file_block("receipt.jpg"), {"type": "text", "text": "What is the total and the currency?"}]
# messages=[{"role": "user", "content": content}]

Practical notes for client work:

• Put the file block before your text instruction in the same user message.
• Large or many-page PDFs cost many input tokens, because pages are processed as text and as images. Measure on a sample before you quote a price.
• There are limits on file size and page count. Read them on the API docs; do not rely on memory.
• Phone photos of receipts: ask the client to photograph flat and in good light. It improves accuracy more than any prompt change.

4.4 Structured outputs

Downstream code needs data, not prose. Three layers make JSON dependable, and you should use all three:

1. Schema. Describe the shape once. With pydantic, Invoice.model_json_schema() produces a JSON schema, and field descriptions double as instructions to the model.
2. Force the format. The portable way is a tool whose input_schema is your schema, plus tool_choice set to that tool. Claude must answer by calling it, and the arguments arrive as a parsed dict. Anthropic also documents native structured-output options; check the current docs for what exists, which models support it and the exact parameters before you switch to it.
3. Validate. Invoice.model_validate(block.input) rejects wrong types, bad enums and missing fields. Add business checks (does the sum of lines match the total? is the VAT number nine digits?) and route failures to a person.

On failure, retry once with the error message appended; if it fails again, put the document on a review list. Never drop it silently.

4.5 Tool use and the manual loop

A tool is a function you describe to Claude. Claude never runs it. It replies with a request to call it (stop_reason = "tool_use"), your code runs it, and you send the result back. Repeat until Claude answers in plain text.

import json
import anthropic

client = anthropic.Anthropic()
MODEL = "claude-sonnet-5-5"  # check current IDs

TOOLS = [{
    "name": "get_order_status",
    "description": "Look up the shipping status of one order by its ID. "
                   "Use it whenever the customer asks where an order is.",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "string", "description": "e.g. A-10442"}},
        "required": ["order_id"],
    },
}]


def get_order_status(order_id: str) -> dict:
    return {"order_id": order_id, "status": "shipped", "eta": "Thursday"}  # stub


HANDLERS = {"get_order_status": get_order_status}

messages = [{"role": "user", "content": "Where are orders A-10442 and A-10443?"}]
for _ in range(8):                                  # hard cap on loop turns
    resp = client.messages.create(model=MODEL, max_tokens=1000, tools=TOOLS, messages=messages)
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use":              # end_turn: Claude is done
        break
    results = []
    for block in resp.content:                      # may hold several tool_use blocks
        if block.type != "tool_use":
            continue
        try:
            out, is_error = json.dumps(HANDLERS[block.name](**block.input)), False
        except Exception as e:                      # tell Claude; it can recover
            out, is_error = f"{type(e).__name__}: {e}", True
        results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": out, "is_error": is_error})
    messages.append({"role": "user", "content": results})   # all results in ONE message

print("".join(b.text for b in resp.content if b.type == "text"))

Writing good tools

• The description is the most important field. Say what the tool does, when to use it and when not to.
• Keep schemas small: few parameters, enums where values are fixed, a description on each property.
• tool_choice can be left to Claude (auto), require some tool (any), or force one named tool, which is what the extractor uses.
• Return errors as a result with is_error set, in plain words. Claude often recovers by fixing its arguments.

Parallel tool calls

When Claude needs several independent lookups, one assistant message can contain several tool_use blocks (the example asks about two orders). Run all of them, optionally in a thread pool, and return all the results together in a single user message, each with the matching tool_use_id. Do not split results across several user messages, and do not forget one: every tool_use needs a tool_result.

Anthropic also offers server-side tools (such as web search) that run on Anthropic's side; see the tool use overview. Module 6 covers MCP, which standardises how tools are shared.

4.6 Prompt caching

Many calls repeat a large identical prefix: a long system prompt, a policy manual, the same contract asked ten questions. Mark the end of that stable prefix with cache_control. Later requests that start with exactly the same prefix read it from cache at a lower price and often faster. Writing the cache costs somewhat more than a normal read, so it only pays when the prefix is reused.

import anthropic

client = anthropic.Anthropic()
policy = open("supplier_policy.txt", encoding="utf-8").read()   # long, stable text


def ask(question: str):
    resp = client.messages.create(
        model="claude-haiku-5-5",  # check current IDs
        max_tokens=400,
        system=[
            {"type": "text", "text": "You answer questions about our supplier payment policy."},
            # everything up to and including this block becomes the cached prefix
            {"type": "text", "text": policy, "cache_control": {"type": "ephemeral"}},
        ],
        messages=[{"role": "user", "content": question}],
    )
    u = resp.usage
    print("written:", u.cache_creation_input_tokens, "read:", u.cache_read_input_tokens,
          "uncached input:", u.input_tokens)
    return resp


ask("When must invoices over the approval limit be signed off?")  # first call writes the cache
ask("Who approves foreign-currency suppliers?")                   # soon after: should read it

• Prefix match. The order is tools, then system, then messages. Any change before the marker, even one character, misses the cache. Put stable content first and the per-request part last.
• Minimum length. Prefixes below a model-specific minimum are not cached, with no error. That is why a short extraction prompt shows zero cache tokens.
• Lifetime. Cache entries expire after a short period unless refreshed by use. The docs list the lifetime options.
• Verify with usage. cache_creation_input_tokens and cache_read_input_tokens tell you what happened. Do not assume.

Rules and prices change, so read the prompt caching page before designing around it.

4.7 Message Batches API

If nobody is waiting for the answer, batch it. You submit many requests at once, they are processed asynchronously at a discount, and you collect results later. The flow is always the same:

1. Create: client.messages.batches.create(requests=[...]), each request with a custom_id and the same params you would pass to messages.create.
2. Poll: client.messages.batches.retrieve(id) until processing_status is ended.
3. Retrieve: client.messages.batches.results(id). Results are not in your submission order, so join on custom_id. Each is succeeded, errored, canceled or expired.

Good fits: nightly classification of tickets, back-catalogue extraction, evaluation runs over a test set. Bad fits: anything interactive. Batch size limits, turnaround window and the discount are in the batch processing docs. The lab uses it for 500 documents.

4.8 Errors, retries and cost logging

Production code fails in ordinary ways. Plan for each:

The SDK retries the retryable cases by itself (max_retries on the client, default is small). Tune max_retries and timeout there first. Write your own wrapper only when you need custom logging or limits:

import logging
import random
import time

import anthropic

log = logging.getLogger("llm")
RETRY_STATUS = {429, 500, 502, 503, 529}


def call_with_backoff(fn, tries: int = 5, base: float = 1.0):
    """The SDK retries by itself; use this when you want your own logging or limits."""
    for n in range(tries):
        last = n == tries - 1
        try:
            return fn()
        except anthropic.APIStatusError as e:
            if e.status_code not in RETRY_STATUS or last:
                raise                       # 400/401/403/404: retrying cannot help
            wait = float(e.response.headers.get("retry-after", base * 2 ** n))
        except anthropic.APIConnectionError:  # includes APITimeoutError
            if last:
                raise
            wait = base * 2 ** n
        wait += random.uniform(0, 1)         # jitter so parallel workers do not sync up
        log.warning("retry %d/%d in %.1fs", n + 1, tries, wait)
        time.sleep(wait)

Log tokens and cost per call

This is what turns a demo into something you can quote. For every call, log the request ID (needed if you ever open a support case), model, latency, stop reason and all four usage counters. Compute cost from those counters and a price table you keep in config, copied from the pricing page with the date. The lab does exactly this, and sums it into a cost-per-document figure. Clients ask for that number first.

4.9 A small service: project structure and FastAPI

Keep the pipeline in plain functions that know nothing about HTTP. Then a CLI, a batch script and a web endpoint are thin wrappers around the same code:

invoice-extractor/
  README.md
  requirements.txt      # anthropic, pydantic, fastapi, uvicorn, python-multipart
  .env.example          # ANTHROPIC_API_KEY=, PRICE_IN=, PRICE_OUT=, ...
  .gitignore            # .env, out/, real invoices
  models.py             # pydantic schema, review flags, cost helper
  extract.py            # prompt, tool, extract_bytes(), CLI for a folder
  batch_extract.py      # submit / collect for large jobs
  app.py                # FastAPI wrapper
  samples/              # synthetic PDFs only
  sample_output/        # CSVs and cost report from the samples

The endpoint, run with uvicorn app:app --reload:

# app.py - run with: uvicorn app:app --reload
from fastapi import FastAPI, HTTPException, UploadFile

from extract import ExtractionError, extract_bytes
from models import cost_usd

app = FastAPI(title="invoice-extractor")


@app.get("/health")
def health() -> dict:
    return {"ok": True}


@app.post("/extract")
def extract(file: UploadFile) -> dict:  # plain def: FastAPI runs it in a worker thread
    if file.content_type != "application/pdf":
        raise HTTPException(status_code=415, detail="send a PDF")
    data = file.file.read()
    try:
        invoice, usage = extract_bytes(data)
    except ExtractionError as e:
        raise HTTPException(status_code=422, detail=str(e)[:300])
    return {"invoice": invoice.model_dump(mode="json"),
            "review_flags": invoice.review_flags(),
            "usage": usage, "cost_usd": cost_usd(usage)}

• Declare it with plain def, not async def, because the SDK call here is blocking. FastAPI runs it in a worker thread. (The SDK also has an async client if you prefer.)
• Return 4xx for bad input and 422 for an extraction the model could not complete; do not return 200 with an error inside.
• Add authentication before any client can reach it. Uploaded invoices are personal and financial data.

mkdir invoice-extractor && cd invoice-extractor
git init
python -m venv .venv && source .venv/bin/activate
pip install anthropic pydantic fastapi uvicorn python-multipart
pip freeze | grep -iE "anthropic|pydantic|fastapi|uvicorn|multipart" > requirements.txt
mkdir samples out
# .env.example, copy the values yourself from the current pricing page (USD per million tokens)
export ANTHROPIC_API_KEY=sk-ant-...
export PRICE_IN=...  PRICE_OUT=...  PRICE_CACHE_WRITE=...  PRICE_CACHE_READ=...  BATCH_FACTOR=...

"""Pydantic models and cost helper for the invoice extractor."""
import os
from datetime import date
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, Field, field_validator


class Currency(str, Enum):
    ILS = "ILS"
    USD = "USD"
    EUR = "EUR"


class LineItem(BaseModel):
    description: str
    quantity: Decimal | None = None
    unit_price: Decimal | None = None
    amount: Decimal = Field(description="Line amount exactly as printed on the document")


class Invoice(BaseModel):
    supplier: str = Field(description="Legal name of the issuing supplier")
    invoice_number: str | None = None
    invoice_date: date = Field(description="Issue date as ISO YYYY-MM-DD")
    vat_number: str | None = Field(
        default=None,
        description="Supplier VAT / company / osek morshe number, digits only",
    )
    lines: list[LineItem] = Field(min_length=1)
    total: Decimal = Field(description="Grand total payable, as printed")
    currency: Currency

    @field_validator("vat_number")
    @classmethod
    def clean_vat(cls, v: str | None) -> str | None:
        if v is None:
            return None
        digits = "".join(ch for ch in v if ch.isdigit())
        return digits or None

    def lines_sum(self) -> Decimal:
        return sum((line.amount for line in self.lines), Decimal("0"))

    def review_flags(self) -> list[str]:
        """Soft checks: a flag means 'a person should look', not 'wrong'."""
        flags = []
        if abs(self.lines_sum() - self.total) > Decimal("0.05"):
            flags.append("lines_do_not_sum_to_total")  # may be normal if lines exclude VAT
        if self.vat_number is None or len(self.vat_number) != 9:
            flags.append("vat_number_missing_or_not_9_digits")
        return flags


def _rate(name: str, default: str = "0") -> float:
    return float(os.environ.get(name, default))


def cost_usd(usage: dict, batch: bool = False) -> float:
    """Prices live in env vars (USD per million tokens), copied from the pricing page.
    PRICE_IN, PRICE_OUT, PRICE_CACHE_WRITE, PRICE_CACHE_READ, and BATCH_FACTOR
    (the batch multiplier from the batch docs). Nothing is hard-coded here."""
    per_million = {
        "input_tokens": _rate("PRICE_IN"),
        "output_tokens": _rate("PRICE_OUT"),
        "cache_creation_input_tokens": _rate("PRICE_CACHE_WRITE"),
        "cache_read_input_tokens": _rate("PRICE_CACHE_READ"),
    }
    total = sum(usage.get(k, 0) * p for k, p in per_million.items()) / 1_000_000
    return total * _rate("BATCH_FACTOR", "1") if batch else total

"""Realtime extraction: folder of PDFs -> validated Invoice -> CSV.
Usage: python extract.py invoices/"""
import base64
import csv
import logging
import sys
import time
from pathlib import Path

import anthropic
from pydantic import ValidationError

from models import Invoice, cost_usd

# Check current model IDs and parameters at docs.claude.com before running
MODEL = "claude-haiku-5-5"
MAX_TOKENS = 2000

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("extract")

# The SDK already retries 429/5xx/connection errors with backoff; we set the budget.
client = anthropic.Anthropic(max_retries=5, timeout=120.0)

SYSTEM_PROMPT = """You extract data from supplier invoices and receipts for an Israeli SMB.
Documents may be Hebrew, Russian or English, and may be scans.
Rules:
- Always answer by calling the record_invoice tool. Never invent a value.
- Israeli documents write dates day first (DD/MM/YYYY). Output ISO YYYY-MM-DD.
- The symbol or word for shekel (NIS, ILS) means ILS. Use only ILS, USD or EUR.
- vat_number is the SUPPLIER's number (osek morshe / company number), not the customer's.
  If you cannot find it, use null.
- Amounts are plain numbers: no currency symbol, no thousands separator.
- A credit note has negative amounts.
- lines: one entry per printed line item. Do not add a line for VAT or totals."""

TOOL = {
    "name": "record_invoice",
    "description": "Record the fields extracted from one invoice or receipt.",
    "input_schema": Invoice.model_json_schema(),
}

INVOICE_FIELDS = ["file", "supplier", "invoice_number", "invoice_date", "vat_number",
                  "currency", "total", "lines_sum", "review_flags", "cost_usd"]
LINE_FIELDS = ["file", "description", "quantity", "unit_price", "amount"]
FAIL_FIELDS = ["file", "error"]


class ExtractionError(Exception):
    pass


def build_params(pdf_bytes: bytes, note: str = "") -> dict:
    """Request parameters, shared by the realtime and the batch scripts."""
    b64 = base64.standard_b64encode(pdf_bytes).decode("ascii")
    return {
        "model": MODEL,
        "max_tokens": MAX_TOKENS,
        # cache_control only helps once the prefix is long enough; check usage to see.
        "system": [{"type": "text", "text": SYSTEM_PROMPT,
                    "cache_control": {"type": "ephemeral"}}],
        "tools": [TOOL],
        "tool_choice": {"type": "tool", "name": "record_invoice"},
        "messages": [{"role": "user", "content": [
            {"type": "document",
             "source": {"type": "base64", "media_type": "application/pdf", "data": b64}},
            {"type": "text", "text": "Extract this document." + note},
        ]}],
    }


def usage_dict(u) -> dict:
    keys = ("input_tokens", "output_tokens",
            "cache_creation_input_tokens", "cache_read_input_tokens")
    return {k: getattr(u, k, 0) or 0 for k in keys}


def parse_message(msg) -> Invoice:
    if msg.stop_reason == "max_tokens":
        raise ExtractionError("output cut off at max_tokens")
    block = next((b for b in msg.content if b.type == "tool_use"), None)
    if block is None:
        raise ExtractionError(f"no tool call, stop_reason={msg.stop_reason}")
    try:
        return Invoice.model_validate(block.input)
    except ValidationError as e:
        raise ExtractionError(f"validation failed: {e}") from e


def extract_bytes(pdf_bytes: bytes, attempts: int = 2) -> tuple[Invoice, dict]:
    note, totals, last = "", {}, None
    for attempt in range(1, attempts + 1):
        started = time.perf_counter()
        try:
            msg = client.messages.create(**build_params(pdf_bytes, note))
        except anthropic.APIStatusError as e:
            raise ExtractionError(f"API error {e.status_code}: {e.message}") from e
        except anthropic.APIConnectionError as e:  # includes timeouts
            raise ExtractionError(f"connection problem: {e}") from e
        usage = usage_dict(msg.usage)
        for k, v in usage.items():
            totals[k] = totals.get(k, 0) + v
        log.info("request_id=%s attempt=%d stop=%s seconds=%.1f usage=%s cost_usd=%.5f",
                 getattr(msg, "_request_id", None), attempt, msg.stop_reason,
                 time.perf_counter() - started, usage, cost_usd(usage))
        try:
            return parse_message(msg), totals
        except ExtractionError as e:
            last = e
            note = f"\nYour previous answer was rejected: {e}. Fix it."
    raise last


def invoice_row(name: str, inv: Invoice, cost: float) -> dict:
    return {"file": name, "supplier": inv.supplier, "invoice_number": inv.invoice_number,
            "invoice_date": inv.invoice_date.isoformat(), "vat_number": inv.vat_number,
            "currency": inv.currency.value, "total": str(inv.total),
            "lines_sum": str(inv.lines_sum()), "review_flags": ";".join(inv.review_flags()),
            "cost_usd": f"{cost:.5f}"}


def line_rows(name: str, inv: Invoice) -> list[dict]:
    return [{"file": name, "description": ln.description, "quantity": ln.quantity,
             "unit_price": ln.unit_price, "amount": ln.amount} for ln in inv.lines]


def write_csv(path: Path, rows: list[dict], fields: list[str]) -> None:
    # utf-8-sig so Excel opens Hebrew text correctly
    with path.open("w", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        w.writerows(rows)


def run(folder: Path, out_dir: Path) -> None:
    out_dir.mkdir(exist_ok=True)
    inv_rows, ln_rows, failures, total_cost = [], [], [], 0.0
    for pdf in sorted(folder.glob("*.pdf")):
        try:
            inv, usage = extract_bytes(pdf.read_bytes())
        except ExtractionError as e:
            log.error("%s failed: %s", pdf.name, e)
            failures.append({"file": pdf.name, "error": str(e)[:300]})
            continue
        cost = cost_usd(usage)
        total_cost += cost
        inv_rows.append(invoice_row(pdf.name, inv, cost))
        ln_rows.extend(line_rows(pdf.name, inv))
    write_csv(out_dir / "invoices.csv", inv_rows, INVOICE_FIELDS)
    write_csv(out_dir / "lines.csv", ln_rows, LINE_FIELDS)
    write_csv(out_dir / "failures.csv", failures, FAIL_FIELDS)
    done = len(inv_rows)
    flagged = sum(1 for r in inv_rows if r["review_flags"])
    log.info("done=%d failed=%d flagged_for_review=%d total_cost_usd=%.4f per_doc=%.5f",
             done, len(failures), flagged, total_cost, total_cost / max(done, 1))


if __name__ == "__main__":
    if len(sys.argv) != 2:
        sys.exit("usage: python extract.py <folder-with-pdfs>")
    run(Path(sys.argv[1]), Path("out"))

"""Batch extraction for many documents.
  python batch_extract.py submit invoices/    # create batches, save ids
  python batch_extract.py collect             # poll, download, write CSVs and cost report"""
import json
import sys
import time
from pathlib import Path

from extract import (FAIL_FIELDS, INVOICE_FIELDS, LINE_FIELDS, ExtractionError, build_params,
                     client, invoice_row, line_rows, parse_message, usage_dict, write_csv)
from models import cost_usd

OUT = Path("out")
STATE = OUT / "batch_state.json"
CHUNK = 100  # documents per batch; check the docs for request size limits with base64 PDFs


def submit(folder: Path) -> None:
    OUT.mkdir(exist_ok=True)
    pdfs = sorted(folder.glob("*.pdf"))
    manifest, batch_ids = {}, []
    for start in range(0, len(pdfs), CHUNK):
        requests = []
        for i, pdf in enumerate(pdfs[start:start + CHUNK], start=start):
            custom_id = f"doc-{i:05d}"  # results come back unordered; this is the join key
            manifest[custom_id] = pdf.name
            requests.append({"custom_id": custom_id, "params": build_params(pdf.read_bytes())})
        batch = client.messages.batches.create(requests=requests)
        batch_ids.append(batch.id)
        print(f"submitted {batch.id} with {len(requests)} documents")
    STATE.write_text(json.dumps({"batch_ids": batch_ids, "manifest": manifest}, indent=2))


def wait(batch_id: str) -> None:
    while True:
        b = client.messages.batches.retrieve(batch_id)
        c = b.request_counts
        print(f"{batch_id}: {b.processing_status} processing={c.processing} "
              f"succeeded={c.succeeded} errored={c.errored}")
        if b.processing_status == "ended":
            return
        time.sleep(60)


def collect() -> None:
    state = json.loads(STATE.read_text())
    manifest = state["manifest"]
    inv_rows, ln_rows, failures = [], [], []
    cost_batch = cost_realtime = 0.0
    tokens = {"input_tokens": 0, "output_tokens": 0,
              "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}
    billed_docs = 0
    for batch_id in state["batch_ids"]:
        wait(batch_id)
        for entry in client.messages.batches.results(batch_id):
            name = manifest[entry.custom_id]
            if entry.result.type != "succeeded":  # errored, canceled or expired
                failures.append({"file": name, "error": entry.result.type})
                continue
            msg = entry.result.message
            usage = usage_dict(msg.usage)
            for k, v in usage.items():
                tokens[k] += v
            billed_docs += 1
            c_batch = cost_usd(usage, batch=True)
            cost_batch += c_batch
            cost_realtime += cost_usd(usage, batch=False)
            try:
                inv = parse_message(msg)
            except ExtractionError as e:
                failures.append({"file": name, "error": str(e)[:300]})
                continue
            inv_rows.append(invoice_row(name, inv, c_batch))
            ln_rows.extend(line_rows(name, inv))
    inv_rows.sort(key=lambda r: r["file"])
    write_csv(OUT / "invoices_batch.csv", inv_rows, INVOICE_FIELDS)
    write_csv(OUT / "lines_batch.csv", ln_rows, LINE_FIELDS)
    write_csv(OUT / "failures_batch.csv", failures, FAIL_FIELDS)
    n = max(billed_docs, 1)
    report = {
        "documents_total": len(manifest), "extracted_ok": len(inv_rows),
        "failed": len(failures),
        "flagged_for_review": sum(1 for r in inv_rows if r["review_flags"]),
        "tokens": tokens,
        "cost_usd_batch_total": round(cost_batch, 4),
        "cost_usd_per_doc_batch": round(cost_batch / n, 5),
        "cost_usd_per_doc_if_realtime": round(cost_realtime / n, 5),
        "note": "prices come from environment variables; see models.cost_usd",
    }
    (OUT / "cost_report.json").write_text(json.dumps(report, indent=2))
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    if len(sys.argv) == 3 and sys.argv[1] == "submit":
        submit(Path(sys.argv[2]))
    elif len(sys.argv) == 2 and sys.argv[1] == "collect":
        collect()
    else:
        sys.exit(__doc__)

Deliverable: GitHub repo "invoice-extractor"

A public or private repo you can show a prospect. It must contain:

1. Code: the files above, runnable from a clean clone using only the README.
2. README.md: what it does, setup, how to run realtime, batch and API modes, the schema, how review flags work, known limits, and the model and date you tested with.
3. Sample outputs: sample_output/ with the CSVs, cost_report.json and a short accuracy table from your test set. Synthetic documents only.
4. Results section in the README: field-level accuracy, share flagged for review, cost per document for realtime and batch, with the prices' source and the date.
5. Client-facing paragraph at the top of the README, in plain business language. Example to adapt:

Fill the brackets only with your measured numbers. Remove any claim you cannot back up.

Self-check

1. Your extractor sometimes returns truncated JSON that fails to parse. What do you check first, and what are two fixes?
   Model answer

   Check stop_reason: max_tokens means the output was cut off. Fixes: raise max_tokens, or reduce the output (fewer fields, shorter descriptions). Also force the format with a tool and validate, so a broken result is detected and retried instead of passed on.
2. In a tool loop, Claude returns three tool_use blocks in one reply. What exactly do you send back?
   Model answer

   Append the assistant message as it came, then send one user message whose content is a list of three tool_result blocks, each with the matching tool_use_id. Run the tools in any order or in parallel. Report failures with is_error rather than skipping a result.
3. A client asks Claude 30 questions an hour about the same 40-page manual. Would you use caching, batch, both or neither, and how do you confirm it worked?
   Model answer

   Caching, with the manual placed first in the system prompt and the marker after it, because the prefix repeats and calls are close together. Not batch: people are waiting. Confirm in usage: cache creation tokens on the first call, cache read tokens on later calls. If reads stay at zero, the prefix changed or is below the minimum length.
4. Which errors do you retry with backoff, which do you never retry, and who handles the retrying in the SDK?
   Model answer

   Retry 429 rate limits, 5xx and overload errors, and connection errors or timeouts, with exponential backoff and jitter, respecting retry-after. Never retry 400, 401, 403 or 404 style errors; fix the request, key or model ID. The SDK retries the retryable cases itself, controlled by max_retries and timeout on the client.
5. You need to extract 500 old invoices by tomorrow morning. Describe the Batch API steps and how you match results to files.
   Model answer

   Build one request per file with a unique custom_id (kept in a manifest mapping id to filename), create the batch (in chunks if size limits require), poll until processing_status is ended, then iterate the results. Results are unordered, so join on custom_id. Handle succeeded, errored, canceled and expired separately, validate each success with pydantic, and re-run the failures.

Further reading

• Claude API overview
• Tool use overview
• Prompt caching
• Batch processing
• Anthropic Cookbook (worked examples)