Build a JSON Extractor

The Goal

A JSON extractor reads unstructured text, such as customer emails, forms, or notes, and returns records that a program can store and check. It is the most common production use of Claude after chat: a stream of messy text goes in, and rows for a database come out. The project's value lies in what happens between the model and the database. Every record must match a schema, every value must pass the rules that matter to the business, and every failure must be visible, so that nothing invalid is stored silently.

The project combines structured output (Lessons 101 to 105) with the validation and recovery ideas from earlier lessons. It keeps the model's output and the application's acceptance of it as two separate steps.

What You Will Build

An extract_order function that:

1. sends one email to the model with a schema that describes the record,
2. checks that the answer is complete,
3. validates it with a Pydantic model that enforces the business rules,
4. returns either a valid record or the reason it was rejected,
5. processes a batch and reports how many records succeeded.

Project Structure

json_extractor/
├── extractor.py    # the record model, the extraction call, and the batch runner
└── demo.py         # a stand-in client that returns three answers, one of them invalid

Step 1: The Record

The record model states what a valid order looks like. Each field has a type, and the constraints express the rules the business cares about. The model is the single source of truth: its schema is sent to Claude, and its validation is applied to the answer:

"""extractor.py: extract order records from text, validated before they are accepted."""

from anthropic import transform_schema
from pydantic import BaseModel, Field, ValidationError


class Order(BaseModel):
    customer: str = Field(min_length=1, description="The sender's full name")
    item: str = Field(min_length=1, description="The product name, as written")
    quantity: int = Field(ge=1, le=100, description="Whole number of units, from 1 to 100")
    delivery_date: str = Field(description="Requested delivery date in YYYY-MM-DD format")

The schema sent to the API is produced by transform_schema, which moves the numeric limits into the field descriptions (Lesson 102). The model's own validation still enforces the limits, which is why the record is checked again after the answer arrives.

Step 2: One Extraction

The extraction sends the email in tags, asks for the schema-shaped answer, checks that the response is complete, and validates the text. The function returns either the record or a reason, never an exception that would stop the batch:

MODEL = "claude-sonnet-5-5"
SCHEMA = transform_schema(Order)


def extract_order(client, email: str) -> tuple[Order | None, str | None]:
    response = client.messages.create(
        model=MODEL,
        max_tokens=300,
        output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
        messages=[{"role": "user", "content": f"Extract the order from this email.\n<email>\n{email}\n</email>"}],
    )
    if response.stop_reason != "end_turn":
        return None, f"stopped: {response.stop_reason}"
    text = "".join(block.text for block in response.content if block.type == "text")
    try:
        return Order.model_validate_json(text), None
    except ValidationError as error:
        return None, error.errors()[0]["msg"]

The order of checks matters. A truncated answer is caught by its stop reason before parsing is attempted, because truncated JSON would fail for a less useful reason. The validation then catches a complete answer that breaks a rule.

Step 3: A Batch

The batch runner processes every email, keeps the accepted records and the rejections, and reports the totals. Rejections are kept with their reasons, so a person can review them:

def extract_batch(client, emails: dict[str, str]) -> dict:
    accepted: dict[str, Order] = {}
    rejected: dict[str, str] = {}
    for email_id, email in emails.items():
        order, reason = extract_order(client, email)
        if order is None:
            rejected[email_id] = reason
        else:
            accepted[email_id] = order
    return {"accepted": accepted, "rejected": rejected}

The rejected dictionary is as important as the accepted one. Without it, the batch would report success for the records it stored and silently lose the others.

Step 4: A Demonstration

The stand-in returns a valid record for the first email, a record with a quantity of zero for the second, and a record missing its item for the third. The batch shows how each outcome is handled:

"""demo.py: a batch of three emails, one of which fails validation."""

import json
from types import SimpleNamespace

from extractor import extract_batch


ANSWERS = [
    {"customer": "Ravi Kumar", "item": "Python Basics", "quantity": 2, "delivery_date": "2026-11-20"},
    {"customer": "Meena", "item": "Django Kit", "quantity": 0, "delivery_date": "2026-11-21"},
    {"customer": "Arjun", "quantity": 1, "delivery_date": "2026-11-22"},
]


class ScriptedClient:
    def __init__(self):
        self.messages = self
        self.turn = 0

    def create(self, **request):
        answer = ANSWERS[self.turn]
        self.turn += 1
        return SimpleNamespace(stop_reason="end_turn",
                               content=[SimpleNamespace(type="text", text=json.dumps(answer))])


if __name__ == "__main__":
    emails = {"e1": "Ravi here...", "e2": "Meena here...", "e3": "Arjun here..."}
    result = extract_batch(ScriptedClient(), emails)
    for email_id, order in result["accepted"].items():
        print(email_id, "ok", order)
    for email_id, reason in result["rejected"].items():
        print(email_id, "rejected:", reason)
    print(f"{len(result['accepted'])} of {len(emails)} records accepted")

Output:

e1 ok customer='Ravi Kumar' item='Python Basics' quantity=2 delivery_date='2026-11-20'
e2 rejected: Input should be greater than or equal to 1
e3 rejected: Field required
2 of 3 records accepted

How It Works

1. The first record passes every rule, so it is accepted, and its printed form shows the validated values.
2. The second record has a quantity of zero. The model's own limit does not stop it, because the stand-in is not bound by the schema, and the Pydantic validation rejects it with the rule's message.
3. The third record is missing its item. The validation reports a required field, which names the problem exactly.
4. The summary line counts accepted records against the total, which is the figure a team tracks for extraction quality.
5. Both rejections are kept with their reasons, so the batch can be reviewed or retried with a corrected prompt, following the recovery pattern from Lesson 105.

Going Live

Replace the scripted client with Anthropic(), and the same batch runs against the API. In production the schema-constrained output makes a malformed answer rare, but the validation remains, because a well-formed record can still break a business rule, and the model's answer is never trusted without the check.

Where to Store the Records

The accepted records go to a database table whose columns match the model's fields, and the rejected ones go to a review table with the email identifier and the reason. Storing both is what makes the extraction auditable, and it is the pattern from the data-pipeline lessons in Section 2 (Lesson 109).