The Claude API and structured output

Now the code. This lesson gives you a wrapper that turns a prompt into a validated Python object, tracks token usage, and is tested with a fake client so you can run the tests offline.

What you will build

llm.py (the wrapper), classify_leads.py (CSV in, labelled CSV out) and guard.py (redaction and a cost estimate).

Figure 1. Treat the model’s output as untrusted input: validate it like you would a web form.

The official call, for reference

The documented quickstart for Python is this shape (the model id is the one used in the docs at the time of writing; always check the models page):

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment
message = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=1000,
    messages=[{"role": "user", "content": "Summarise this lead in one line."}],
)
for block in message.content:
    if block.type == "text":
        print(block.text)

The recommended way to get JSON

Anthropic documents structured outputs: you pass a schema and the response is guaranteed to match it. With the Python SDK, the documented pattern is to pass a Pydantic model to client.messages.parse() as output_format and read parsed_output:

from pydantic import BaseModel
import anthropic

class ContactInfo(BaseModel):
    name: str
    email: str
    plan_interest: str
    demo_requested: bool

client = anthropic.Anthropic()
response = client.messages.parse(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Extract the key information from this email: ..."}],
    output_format=ContactInfo,
)
print(response.parsed_output)

The SDK transforms unsupported schema features (for example length limits) into descriptions, so validate those yourself. When you use this method you can simplify the wrapper below; we keep a manual version because it works the same on any SDK version and teaches what the helper does for you.

Step 1: The guard

Anything that leaves your machine should have personal data removed first. This is a baseline, not a guarantee: it catches obvious emails and phone numbers, not names or addresses.

# guard.py
"""Pre-flight checks: remove personal data before text goes to a model, and estimate cost."""
import re

EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
PHONE = re.compile(r"(?<!\d)(?:\+?\d[\d\s().-]{7,}\d)(?!\d)")


def redact(text: str) -> str:
    """Replace emails and phone numbers with placeholders."""
    text = EMAIL.sub("[EMAIL]", text)
    return PHONE.sub("[PHONE]", text)


def estimate_cost(input_tokens: int, output_tokens: int, price_in: float, price_out: float) -> float:
    """USD cost for a call or batch, given per-million-token prices."""
    return round(input_tokens / 1e6 * price_in + output_tokens / 1e6 * price_out, 6)


def rough_tokens(text: str) -> int:
    """Very rough estimate (about 4 characters per token). Use real usage numbers when you have them."""
    return max(1, len(text) // 4)

Step 2: The wrapper

# llm.py
"""One small wrapper around the Anthropic SDK: structured output, retries, usage tracking."""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Type, TypeVar
from pydantic import BaseModel, ValidationError
import config

T = TypeVar("T", bound=BaseModel)


@dataclass
class Usage:
    input_tokens: int = 0
    output_tokens: int = 0
    calls: int = 0
    failures: int = 0
    log: list = field(default_factory=list)


def ask_json(client, system: str, user: str, schema: Type[T], usage: Usage | None = None) -> T:
    """Ask for JSON that matches `schema`. Retries on bad JSON or schema errors, then raises."""
    usage = usage or Usage()
    instruction = (
        system
        + "\n\nReturn ONLY a JSON object that matches this schema, with no extra text:\n"
        + json.dumps(schema.model_json_schema())
    )
    last_error = None
    for attempt in range(config.MAX_RETRIES + 1):
        msg = client.messages.create(
            model=config.MODEL,
            max_tokens=config.MAX_TOKENS,
            system=instruction,
            messages=[{"role": "user", "content": user}],
        )
        usage.calls += 1
        usage.input_tokens += msg.usage.input_tokens
        usage.output_tokens += msg.usage.output_tokens
        text = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text").strip()
        try:
            return schema.model_validate_json(_strip_fence(text))
        except (ValidationError, ValueError) as e:
            last_error = e
            usage.failures += 1
            usage.log.append({"attempt": attempt, "error": str(e)[:200]})
    raise RuntimeError(f"No valid output after {config.MAX_RETRIES + 1} attempts: {last_error}")


def _strip_fence(text: str) -> str:
    if text.startswith("```"):
        text = text.split("\n", 1)[1] if "\n" in text else text
        text = text.rsplit("```", 1)[0]
    return text.strip()

What it does, line by line in plain words: asks for JSON that matches your schema, strips a markdown fence if the model added one, validates with Pydantic, retries a limited number of times, counts tokens, and raises a clear error instead of returning garbage.

Step 3: The classifier

# classify_leads.py
"""Read leads.csv, classify each row with Claude, write results.csv. Failed rows go to errors.csv."""
from __future__ import annotations
import csv
import sys
from typing import Literal
from pydantic import BaseModel
import guard
from llm import ask_json, Usage

SYSTEM = (
    "You classify inbound marketing leads for a small software company. "
    "Use only the information given. If something is not stated, choose 'unknown'. "
    "Never invent company facts."
)


class LeadLabel(BaseModel):
    segment: Literal["freelancer", "small_business", "agency", "enterprise", "unknown"]
    intent: Literal["high", "medium", "low"]
    next_action: Literal["send_demo_invite", "send_nurture_email", "no_action"]
    reason: str


def classify_file(client, src: str, dst: str, err: str) -> Usage:
    usage = Usage()
    with open(src, newline="", encoding="utf-8") as f, \
         open(dst, "w", newline="", encoding="utf-8") as out, \
         open(err, "w", newline="", encoding="utf-8") as bad:
        reader = csv.DictReader(f)
        fields = ["id", "segment", "intent", "next_action", "reason"]
        writer = csv.DictWriter(out, fieldnames=fields)
        errors = csv.DictWriter(bad, fieldnames=["id", "error"])
        writer.writeheader()
        errors.writeheader()
        for row in reader:
            note = guard.redact(row.get("message", ""))
            prompt = f"Company size: {row.get('size', '')}\nSource: {row.get('source', '')}\nMessage: {note}"
            try:
                label = ask_json(client, SYSTEM, prompt, LeadLabel, usage)
                writer.writerow({"id": row["id"], **label.model_dump()})
            except RuntimeError as e:
                errors.writerow({"id": row["id"], "error": str(e)[:200]})
    return usage


if __name__ == "__main__":
    import anthropic  # pip install anthropic
    u = classify_file(anthropic.Anthropic(), sys.argv[1], "results.csv", "errors.csv")
    print(f"calls={u.calls} in={u.input_tokens} out={u.output_tokens} failures={u.failures}")

Three details matter for real use. Failed rows are written to errors.csv and never silently dropped. The message is redacted before it is placed in the prompt. And Literal[...] fields mean a typo like "enterprize" is rejected, not accepted.

Step 4: Test it with no API key

A fake client returns scripted replies. This is how you test retries and failure paths without paying for calls.

# tests/fakes.py
from types import SimpleNamespace as NS

def text_msg(text, stop="end_turn", i=10, o=5):
    return NS(content=[NS(type="text", text=text)], stop_reason=stop,
              usage=NS(input_tokens=i, output_tokens=o))

class FakeClient:
    def __init__(self, replies):
        self.replies = list(replies)
        self.calls = []
        self.messages = NS(create=self._create)

    def _create(self, **kw):
        self.calls.append(kw)
        return self.replies.pop(0)

The key tests, in words: the classifier retries bad JSON then succeeds; it raises after the retry budget; it rejects a value outside the allowed list; and, importantly, the text sent to the model contains neither the email nor the phone number from the test lead. Run them with:

python3 -m unittest discover -s tests -v

The complete project used across this course has 23 tests that pass offline, and we confirmed the redaction test fails when the redaction call is removed. Break your own code on purpose once; a test you have never seen fail proves nothing.

Step 5: Run it for real

export ANTHROPIC_API_KEY=...        # your key
python3 classify_leads.py data/leads.csv
# calls=4 in=... out=... failures=0

Open results.csv and errors.csv. Spot-check five rows by hand before trusting the rest.

Common mistakes

• Using json.loads with no validation, then crashing on row 700 of 1,000.
• Retrying forever. Always cap retries.
• Logging the full prompt, which puts personal data in your logs.

Checklist

• Wrapper validates every output against a schema
• Retries are capped and failures are written to a file
• Personal data is redacted before the prompt is built
• Tests pass without an API key, and you have seen one fail on purpose