Building a Data Extraction Pipeline with Claude Code: Complete Guide

Introduction to Data Extraction Pipelines with Claude Code

Modern applications sit on top of oceans of unstructured data — PDFs, emails, web pages, scanned receipts, and legacy database dumps. Building a reliable pipeline that converts this chaos into clean, structured records has traditionally required brittle regex patterns, fragile OCR integrations, and endless manual tuning. Claude Code changes the equation by giving you an AI agent that can read, reason about, and transform data directly from your terminal, orchestrating extraction logic with the same tooling you already use for software development.

In this guide, you'll learn what a Claude Code-powered extraction pipeline is, why it matters, and how to build one end-to-end with practical, runnable examples.

What Is a Data Extraction Pipeline with Claude Code?

A data extraction pipeline is a sequence of steps that ingests raw, heterogeneous documents and emits structured, validated data ready for storage or downstream analysis. When powered by Claude Code, the pipeline delegates the "understanding" step — parsing ambiguous layouts, inferring schemas, normalizing values — to Claude's language and vision capabilities, while keeping orchestration, validation, and persistence in deterministic code.

Claude Code is Anthropic's CLI-based agentic coding assistant. It can read files, run shell commands, edit source, and call the Claude API. This makes it uniquely suited for extraction work because it can:

• Inspect raw input files and decide on an extraction strategy.
• Generate and refine parser code iteratively.
• Invoke the Claude API with structured output schemas.
• Validate results against a schema and retry on failure.
• Write cleaned records to a database, CSV, or JSONL file.

Why It Matters

Traditional extraction pipelines fail in predictable ways. Regex breaks when a vendor changes an invoice layout. OCR libraries misread tables. Hand-coded parsers require constant maintenance. By placing Claude in the loop, you gain three concrete advantages:

• Resilience: Claude handles layout variation, typos, and mixed formats without retraining.
• Speed of iteration: Describe the desired output in natural language and let Claude generate the glue code.
• Structured output: The Claude API supports JSON schema enforcement, so downstream systems receive predictable records.

The result is a pipeline that is cheaper to build, easier to maintain, and dramatically more accurate on messy real-world inputs.

Prerequisites and Project Setup

Before building, ensure you have the following installed and configured:

• Node.js 18+ and Python 3.10+
• The Claude Code CLI: npm install -g @anthropic-ai/claude-code
• An Anthropic API key exported as ANTHROPIC_API_KEY
• A working directory with sample input documents

Create the project skeleton:

mkdir extraction-pipeline && cd extraction-pipeline
mkdir -p input output src schemas
echo '{"version":"0.1.0"}' > package.json
pip install anthropic pydantic python-dotenv

Verify Claude Code is available:

claude --version

Designing the Pipeline Architecture

A robust pipeline separates concerns into discrete stages. The architecture we'll build has five stages:

• Ingest: Discover and load raw files from the input/ directory.
• Classify: Determine document type and route to the appropriate schema.
• Extract: Call Claude with a strict JSON schema and the document content.
• Validate: Use Pydantic to enforce types and required fields.
• Persist: Write validated records to output/records.jsonl.

This separation lets you retry individual stages, swap out implementations, and monitor each step independently.

Defining Schemas with Pydantic

Schemas are the contract between Claude's flexible output and your deterministic downstream code. Pydantic models give you runtime validation and can be serialized to JSON Schema for the Claude API.

Create src/schemas.py:

from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum


class DocumentType(str, Enum):
    INVOICE = "invoice"
    RECEIPT = "receipt"
    CONTRACT = "contract"
    UNKNOWN = "unknown"


class LineItem(BaseModel):
    description: str = Field(..., description="Item or service description")
    quantity: float = Field(1, ge=0)
    unit_price: float = Field(..., ge=0)
    total: float = Field(..., ge=0)


class InvoiceRecord(BaseModel):
    document_type: DocumentType = DocumentType.INVOICE
    invoice_number: str
    vendor_name: str
    vendor_address: Optional[str] = None
    issue_date: str = Field(..., description="ISO 8601 date YYYY-MM-DD")
    due_date: Optional[str] = None
    currency: str = Field("USD", max_length=3)
    line_items: List[LineItem]
    subtotal: float
    tax: float = 0.0
    total: float
    notes: Optional[str] = None

This model will be converted into a JSON schema that Claude must satisfy. The field descriptions guide Claude toward correct extraction.

Building the Extraction Core

The extraction core calls the Claude API with the document text and the JSON schema. We use the tool_use mechanism to enforce structured output, which is more reliable than prompting for raw JSON.

Create src/extractor.py:

import json
from anthropic import Anthropic
from schemas import InvoiceRecord

client = Anthropic()

SCHEMA = InvoiceRecord.model_json_schema()

EXTRACTION_TOOL = {
    "name": "save_invoice",
    "description": "Save a structured invoice record extracted from the document.",
    "input_schema": SCHEMA,
}

SYSTEM_PROMPT = (
    "You are a precise data extraction engine. Read the provided document "
    "and extract invoice fields exactly as defined by the save_invoice tool. "
    "If a field is not present, omit it rather than guessing. Always use "
    "ISO 8601 dates and numeric amounts without currency symbols."
)


def extract_invoice(document_text: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        tools=[EXTRACTION_TOOL],
        tool_choice={"type": "tool", "name": "save_invoice"},
        messages=[
            {
                "role": "user",
                "content": f"Extract the invoice from this document:\n\n{document_text}",
            }
        ],
    )

    for block in response.content:
        if block.type == "tool_use" and block.name == "save_invoice":
            return block.input

    raise RuntimeError("Claude did not return a tool_use block")

Forcing tool_choice guarantees Claude returns a structured object conforming to the schema. This eliminates most JSON parsing failures.

Handling Different File Formats

Real pipelines receive PDFs, images, and HTML, not just plain text. Add a loader module that normalizes inputs into text or base64-encoded images for Claude's vision support.

Create src/loaders.py:

import base64
from pathlib import Path


def load_file(path: Path) -> dict:
    suffix = path.suffix.lower()
    if suffix in {".txt", ".md", ".csv", ".json", ".eml"}:
        return {"type": "text", "content": path.read_text(encoding="utf-8", errors="replace")}
    if suffix in {".png", ".jpg", ".jpeg", ".webp"}:
        media = "image/png" if suffix == ".png" else "image/jpeg"
        data = base64.standard_b64encode(path.read_bytes()).decode("ascii")
        return {"type": "image", "media_type": media, "data": data}
    if suffix == ".pdf":
        # Use pdftotext or pdfplumber as a fallback text layer
        import subprocess
        text = subprocess.check_output(["pdftotext", "-layout", str(path), "-"]).decode("utf-8", "replace")
        return {"type": "text", "content": text}
    raise ValueError(f"Unsupported file type: {suffix}")

For image-based inputs, modify the extractor to send a content block with type: "image" alongside the prompt. Claude's vision capabilities then handle OCR implicitly.

Orchestrating the Full Pipeline

The orchestrator ties ingestion, loading, extraction, validation, and persistence together. It also handles retries and logs failures for later review.

Create src/pipeline.py:

import json
import logging
from pathlib import Path
from datetime import datetime
from pydantic import ValidationError
from loaders import load_file
from extractor import extract_invoice
from schemas import InvoiceRecord

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

INPUT_DIR = Path("input")
OUTPUT_FILE = Path("output/records.jsonl")
ERROR_FILE = Path("output/errors.jsonl")


def run():
    OUTPUT_FILE.parent.mkdir(exist_ok=True)
    records = []
    errors = []

    for path in sorted(INPUT_DIR.iterdir()):
        if not path.is_file():
            continue
        logger.info(f"Processing {path.name}")
        try:
            loaded = load_file(path)
            raw = extract_invoice(loaded["content"]) if loaded["type"] == "text" else extract_invoice_image(loaded)
            record = InvoiceRecord(**raw)
            records.append(record.model_dump(mode="json"))
            logger.info(f"  OK: invoice {record.invoice_number} -> ${record.total}")
        except ValidationError as e:
            logger.error(f"  Validation failed for {path.name}: {e}")
            errors.append({"file": path.name, "error": e.json(), "ts": datetime.utcnow().isoformat()})
        except Exception as e:
            logger.error(f"  Extraction failed for {path.name}: {e}")
            errors.append({"file": path.name, "error": str(e), "ts": datetime.utcnow().isoformat()})

    with OUTPUT_FILE.open("w") as f:
        for r in records:
            f.write(json.dumps(r) + "\n")
    with ERROR_FILE.open("w") as f:
        for e in errors:
            f.write(json.dumps(e) + "\n")

    logger.info(f"Done. {len(records)} records, {len(errors)} errors.")


def extract_invoice_image(loaded: dict) -> dict:
    from anthropic import Anthropic
    from schemas import InvoiceRecord
    client = Anthropic()
    tool = {"name": "save_invoice", "description": "Save invoice.", "input_schema": InvoiceRecord.model_json_schema()}
    resp = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        max_tokens=4096,
        tools=[tool],
        tool_choice={"type": "tool", "name": "save_invoice"},
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": loaded["media_type"], "data": loaded["data"]}},
                {"type": "text", "text": "Extract the invoice fields using the save_invoice tool."},
            ],
        }],
    )
    for b in resp.content:
        if b.type == "tool_use":
            return b.input
    raise RuntimeError("No tool_use returned")


if __name__ == "__main__":
    run()

Run the pipeline from the project root:

python -m src.pipeline

Using Claude Code to Build and Iterate

Instead of writing all this by hand, you can ask Claude Code to scaffold and refine the pipeline. From the project directory, launch the agent:

claude
> Read the files in input/ and design a Pydantic schema that captures the fields present across all invoices. Write it to src/schemas.py.
> Now write src/extractor.py that calls the Claude API with tool_use to enforce that schema.
> Run the pipeline against input/ and fix any validation errors you find.

Claude Code will inspect the actual documents, propose a schema grounded in real data, generate the extractor, and self-correct by reading error output. This iterative loop is where the productivity gains compound — the agent closes the gap between intent and working code in minutes rather than hours.

Best Practices

• Enforce structure with tools, not prompts. tool_choice guarantees valid JSON; free-form JSON prompts do not.
• Validate twice. The API enforces schema shape, but Pydantic adds domain rules like total >= subtotal + tax.
• Log everything. Write failed extractions to a separate file so you can review and improve prompts over time.
• Use the cheapest capable model. Start with Claude Sonnet for extraction; drop to Haiku for classification once prompts stabilize.
• Cache where possible. Hash input files and store extracted records keyed by hash to avoid reprocessing unchanged documents.
• Handle PII. Redact or encrypt sensitive fields before logging, and avoid sending regulated data to the API without a compliance review.
• Version your schemas. Bump the schema version when fields change so downstream consumers can migrate safely.
• Test with golden examples. Keep a folder of known-good documents and assert extraction output in CI.

Extending the Pipeline

Once the core works, common extensions include adding a classification stage that routes documents to different schemas, integrating a queue (Redis, SQS) for asynchronous processing at scale, writing records directly to a warehouse like Snowflake or BigQuery, and adding a human-in-the-loop review step for low-confidence extractions. Each of these is a small addition because the pipeline's modular design keeps stages independent.

A classification stage might look like this:

def classify(document_text: str) -> str:
    resp = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=20,
        messages=[{"role": "user", "content": f"Classify this document as invoice, receipt, or contract. Reply with one word.\n\n{document_text[:2000]}"}],
    )
    return resp.content[0].text.strip().lower()

Route the result to the matching schema and extractor, and your pipeline now handles heterogeneous document streams.

Conclusion

Building a data extraction pipeline with Claude Code combines the flexibility of large language models with the rigor of typed, validated code. By letting Claude handle the messy work of understanding unstructured documents and keeping orchestration deterministic, you get a system that is accurate on day one and adaptable as inputs evolve. Start with a clear schema, enforce structure through tool use, validate with Pydantic, and let Claude Code iterate on the glue code — you'll have a production-grade extraction pipeline running in an afternoon rather than a sprint.