← Back to Blog

If you're manually entering invoice data into a spreadsheet or accounting system, you already know how slow and error-prone that gets. This tutorial walks you through building a multi-agent invoice processing system using the Claude API in Python — something that can read, extract, validate, and structure invoice data automatically, at scale.

What You'll Build

You'll build a production-ready Python system with two specialized Claude agents: one that extracts structured data from raw invoice text, and one that validates that data against business rules. An orchestration loop ties them together, handles retries, and outputs clean JSON you can pipe directly into QuickBooks, a database, or any downstream system. By the end, you'll have something capable of processing 50+ invoices per hour without a human touching them.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (console.anthropic.com)
  • Basic familiarity with Python classes and async concepts
  • pip install anthropic python-dotenv ready to run
  • A handful of sample invoices in plain text or extracted PDF text format
📦 Full Source Code: The complete, working code is built step by step in the sections below. Every snippet is copy-paste ready and connects directly to the next one. By Step 5, you'll have the full system running end to end. No pseudocode, no placeholders.

Step 1: Set Up Your Claude API Environment and Dependencies

First, let's get the environment wired up correctly. Create a project folder and drop a .env file in it with your API key.

.env 
ANTHROPIC_API_KEY=sk-ant-your-key-here

Now install your dependencies and create the base configuration file. This gives every agent in the system a shared client and model reference.

config.py 
import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

# Single shared client — no need to instantiate per agent
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

MODEL = "claude-sonnet-4-5"

# Global processing config
MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 2

Run a quick sanity check before writing any agent logic. This confirms your key works and the SDK is talking to the API correctly.

test_connection.py 
from config import client, MODEL

response = client.messages.create(
    model=MODEL,
    max_tokens=64,
    messages=[{"role": "user", "content": "Reply with: Connected."}]
)

print(response.content[0].text)
# Expected output: Connected.

Step 2: Define the Invoice Extraction and Validation Agents

This is where the real work lives. We're building two agents as Python classes — each with its own system prompt, tool definitions, and a run() method. They don't share state directly; instead, they pass structured data between each other through the orchestrator we build in Step 3.

Invoice extraction agent class with tool definitions

The extraction agent takes raw invoice text and uses a Claude tool call to force structured output. Defining it as a tool (rather than just prompting for JSON) makes the output dramatically more reliable — Claude is explicitly constrained to the schema you define.

agents/extraction_agent.py 
import json
from config import client, MODEL

# Tool definition that forces Claude to return structured invoice data
EXTRACTION_TOOL = {
    "name": "extract_invoice_data",
    "description": "Extract all relevant fields from a raw invoice document.",
    "input_schema": {
        "type": "object",
        "properties": {
            "invoice_number": {
                "type": "string",
                "description": "The invoice ID or number, e.g. INV-2024-0047"
            },
            "vendor_name": {
                "type": "string",
                "description": "Name of the company or person issuing the invoice"
            },
            "vendor_email": {
                "type": "string",
                "description": "Vendor contact email if present, else null"
            },
            "issue_date": {
                "type": "string",
                "description": "Invoice issue date in YYYY-MM-DD format"
            },
            "due_date": {
                "type": "string",
                "description": "Payment due date in YYYY-MM-DD format"
            },
            "line_items": {
                "type": "array",
                "description": "Individual line items on the invoice",
                "items": {
                    "type": "object",
                    "properties": {
                        "description": {"type": "string"},
                        "quantity": {"type": "number"},
                        "unit_price": {"type": "number"},
                        "total": {"type": "number"}
                    },
                    "required": ["description", "quantity", "unit_price", "total"]
                }
            },
            "subtotal": {"type": "number", "description": "Pre-tax total"},
            "tax_amount": {"type": "number", "description": "Tax charged"},
            "total_amount": {"type": "number", "description": "Final amount due"},
            "currency": {"type": "string", "description": "Currency code, e.g. USD"},
            "payment_terms": {
                "type": "string",
                "description": "Payment terms like Net 30, Due on receipt, etc."
            },
            "notes": {
                "type": "string",
                "description": "Any additional notes or special instructions on the invoice"
            }
        },
        "required": [
            "invoice_number", "vendor_name", "issue_date",
            "due_date", "line_items", "subtotal", "tax_amount",
            "total_amount", "currency"
        ]
    }
}

EXTRACTION_SYSTEM_PROMPT = """You are an invoice data extraction specialist.
Your job is to read raw invoice text and extract every piece of structured data
using the extract_invoice_data tool. Always use the tool — never reply in plain text.
If a field is genuinely missing from the document, use null for optional fields.
Normalize all dates to YYYY-MM-DD format. Normalize all amounts to float values."""


class InvoiceExtractionAgent:
    def __init__(self):
        self.client = client
        self.model = MODEL

    def run(self, raw_invoice_text: str) -> dict:
        """
        Sends raw invoice text to Claude and returns extracted structured data.
        Returns a dict with either 'data' (success) or 'error' (failure).
        """
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=EXTRACTION_SYSTEM_PROMPT,
            tools=[EXTRACTION_TOOL],
            # Force tool use so we always get structured output
            tool_choice={"type": "tool", "name": "extract_invoice_data"},
            messages=[
                {
                    "role": "user",
                    "content": f"Extract all invoice data from the following document:\n\n{raw_invoice_text}"
                }
            ]
        )

        # Pull the tool input out of the response
        for block in response.content:
            if block.type == "tool_use" and block.name == "extract_invoice_data":
                return {"status": "success", "data": block.input}

        return {"status": "error", "error": "No tool call found in extraction response"}

Validation agent with rule enforcement

The validation agent takes the extracted JSON and checks it against real business rules — not just schema validation, but things like "does the line item math add up?" and "is the due date after the issue date?" This is where you catch the stuff that would silently break your accounting downstream.

agents/validation_agent.py 
import json
from datetime import datetime
from config import client, MODEL

VALIDATION_TOOL = {
    "name": "validate_invoice_data",
    "description": "Validate extracted invoice data against business rules.",
    "input_schema": {
        "type": "object",
        "properties": {
            "is_valid": {
                "type": "boolean",
                "description": "True if all validation checks pass"
            },
            "confidence_score": {
                "type": "number",
                "description": "0.0 to 1.0 confidence that the extraction was accurate"
            },
            "validation_errors": {
                "type": "array",
                "description": "List of specific validation failures",
                "items": {
                    "type": "object",
                    "properties": {
                        "field": {"type": "string"},
                        "issue": {"type": "string"},
                        "severity": {
                            "type": "string",
                            "enum": ["critical", "warning", "info"]
                        }
                    },
                    "required": ["field", "issue", "severity"]
                }
            },
            "corrected_fields": {
                "type": "object",
                "description": "Any fields Claude was able to auto-correct, keyed by field name"
            }
        },
        "required": ["is_valid", "confidence_score", "validation_errors"]
    }
}

VALIDATION_SYSTEM_PROMPT = """You are an invoice validation specialist. You receive
extracted invoice JSON and must check it against these business rules:

1. MATH: line_item totals must equal quantity * unit_price (allow $0.01 rounding)
2. MATH: subtotal + tax_amount must equal total_amount (allow $0.01 rounding)
3. DATES: due_date must be on or after issue_date
4. REQUIRED: invoice_number, vendor_name, total_amount must be non-null and non-empty
5. AMOUNTS: no amount field should be negative
6. CURRENCY: currency must be a valid 3-letter ISO code

Mark severity as 'critical' for math errors or missing required fields.
Mark severity as 'warning' for date issues or unusual values.
Mark severity as 'info' for minor formatting differences.
Use the validate_invoice_data tool. Never reply in plain text."""


class InvoiceValidationAgent:
    def __init__(self):
        self.client = client
        self.model = MODEL

    def run(self, extracted_data: dict) -> dict:
        """
        Validates extracted invoice data and returns a validation result dict.
        """
        invoice_json = json.dumps(extracted_data, indent=2)

        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=VALIDATION_SYSTEM_PROMPT,
            tools=[VALIDATION_TOOL],
            tool_choice={"type": "tool", "name": "validate_invoice_data"},
            messages=[
                {
                    "role": "user",
                    "content": f"Validate this extracted invoice data:\n\n{invoice_json}"
                }
            ]
        )

        for block in response.content:
            if block.type == "tool_use" and block.name == "validate_invoice_data":
                return {"status": "success", "data": block.input}

        return {"status": "error", "error": "No tool call found in validation response"}

Step 3: Create the Orchestration Loop with State Management

The orchestrator is what connects everything. It loops through your invoices, calls the extraction agent, passes results to the validation agent, tracks state for each invoice, and decides whether to retry or mark something as failed. Think of it as the project manager for your two specialist agents.

orchestrator.py 
import json
import time
import uuid
from datetime import datetime
from typing import Optional
from config import MAX_RETRIES, RETRY_DELAY_SECONDS
from agents.extraction_agent import InvoiceExtractionAgent
from agents.validation_agent import InvoiceValidationAgent


class InvoiceProcessingOrchestrator:
    def __init__(self):
        self.extractor = InvoiceExtractionAgent()
        self.validator = InvoiceValidationAgent()
        # In-memory state store — swap this for a database in production
        self.results = []

    def process_invoice(self, invoice_text: str, invoice_id: Optional[str] = None) -> dict:
        """
        Runs a single invoice through extraction → validation with retry logic.
        Returns a complete processing result record.
        """
        invoice_id = invoice_id or str(uuid.uuid4())[:8]
        processing_record = {
            "invoice_id": invoice_id,
            "processed_at": datetime.utcnow().isoformat(),
            "status": "pending",
            "extracted_data": None,
            "validation_result": None,
            "attempts": 0,
            "errors": []
        }

        # --- Extraction phase ---
        extraction_result = None
        for attempt in range(1, MAX_RETRIES + 1):
            processing_record["attempts"] = attempt
            print(f"  [Extraction] Invoice {invoice_id} — attempt {attempt}/{MAX_RETRIES}")

            extraction_result = self.extractor.run(invoice_text)

            if extraction_result["status"] == "success":
                processing_record["extracted_data"] = extraction_result["data"]
                break
            else:
                error_msg = extraction_result.get("error", "Unknown extraction error")
                processing_record["errors"].append(
                    {"phase": "extraction", "attempt": attempt, "error": error_msg}
                )
                if attempt < MAX_RETRIES:
                    time.sleep(RETRY_DELAY_SECONDS)

        if not processing_record["extracted_data"]:
            processing_record["status"] = "failed"
            print(f"  [FAILED] Invoice {invoice_id} — extraction failed after {MAX_RETRIES} attempts")
            self.results.append(processing_record)
            return processing_record

        # --- Validation phase ---
        validation_result = self.validator.run(processing_record["extracted_data"])

        if validation_result["status"] == "success":
            processing_record["validation_result"] = validation_result["data"]
            val_data = validation_result["data"]

            # Determine final status based on validation outcome
            if val_data["is_valid"]:
                processing_record["status"] = "approved"
            else:
                # Check if any critical errors exist
                critical_errors = [
                    e for e in val_data.get("validation_errors", [])
                    if e.get("severity") == "critical"
                ]
                processing_record["status"] = "failed" if critical_errors else "needs_review"
        else:
            processing_record["status"] = "failed"
            processing_record["errors"].append(
                {"phase": "validation", "error": validation_result.get("error")}
            )

        print(f"  [Done] Invoice {invoice_id} → status: {processing_record['status']}")
        self.results.append(processing_record)
        return processing_record

    def process_batch(self, invoices: list[dict]) -> list[dict]:
        """
        Process a list of invoices. Each item should have 'text' and optional 'id'.
        """
        print(f"\nStarting batch: {len(invoices)} invoice(s)\n{'─' * 40}")
        for i, invoice in enumerate(invoices):
            print(f"\nInvoice {i + 1} of {len(invoices)}")
            self.process_invoice(
                invoice_text=invoice["text"],
                invoice_id=invoice.get("id")
            )

        # Summary stats
        approved = sum(1 for r in self.results if r["status"] == "approved")
        needs_review = sum(1 for r in self.results if r["status"] == "needs_review")
        failed = sum(1 for r in self.results if r["status"] == "failed")

        print(f"\n{'─' * 40}")
        print(f"Batch complete: {approved} approved | {needs_review} need review | {failed} failed")
        return self.results

    def export_results(self, filepath: str = "invoice_results.json"):
        """Write all processing results to a JSON file."""
        with open(filepath, "w") as f:
            json.dump(self.results, f, indent=2)
        print(f"Results exported to {filepath}")

Step 4: Implement Error Handling and Retries

The retry logic in the orchestrator handles transient API errors, but we also need to catch Anthropic SDK exceptions gracefully. Wrap your agent run() methods with this error handler so a single bad invoice doesn't crash your entire batch run.

utils/error_handler.py 
import time
import functools
from anthropic import APIStatusError, APIConnectionError, RateLimitError
from config import RETRY_DELAY_SECONDS


def with_api_retry(max_retries: int = 3):
    """
    Decorator that retries on Anthropic API errors with exponential backoff.
    Use this on any method that calls client.messages.create().
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except RateLimitError as e:
                    wait = RETRY_DELAY_SECONDS * (2 ** attempt)  # exponential backoff
                    print(f"    Rate limited. Waiting {wait}s before retry {attempt}/{max_retries}...")
                    time.sleep(wait)
                    last_error = e
                except APIConnectionError as e:
                    print(f"    Connection error on attempt {attempt}/{max_retries}: {e}")
                    time.sleep(RETRY_DELAY_SECONDS)
                    last_error = e
                except APIStatusError as e:
                    # 5xx errors are retryable; 4xx usually aren't
                    if e.status_code >= 500:
                        print(f"    API server error {e.status_code}, retrying...")
                        time.sleep(RETRY_DELAY_SECONDS)
                        last_error = e
                    else:
                        # 4xx: bad request — don't retry, surface immediately
                        return {"status": "error", "error": f"API error {e.status_code}: {e.message}"}
            return {"status": "error", "error": str(last_error)}
        return wrapper
    return decorator

Apply the decorator to your extraction and validation run() methods by adding @with_api_retry(max_retries=3) directly above each method definition. That's all it takes — the decorator handles the rest transparently.

⚠️ Rate Limit Warning: Claude Sonnet has per-minute token limits. If you're processing a large batch quickly, add a small time.sleep(0.5) between invoices in your orchestration loop. At 50+ invoices per hour, you're well within free-tier limits, but production workloads should implement the exponential backoff shown above.

Step 5: Test with Sample Invoices and Review Output

Let's wire everything together and run it. This main script creates three sample invoices — one clean, one with a math error, and one with missing data — so you can see how each status comes out.

main.py 
from orchestrator import InvoiceProcessingOrchestrator

# Sample invoices — in production these come from PDF text extraction or email parsing
SAMPLE_INVOICES = [
    {
        "id": "INV-001",
        "text": """
        INVOICE
        Invoice Number: INV-2026-0047
        Vendor: Gulf Coast Supply Co.
        Email: [email protected]
        Issue Date: 2026-08-01
        Due Date: 2026-08-31
        Payment Terms: Net 30

        Line Items:
        - Office Chairs (qty: 10, unit price: $149.99, total: $1,499.90)
        - Standing Desks (qty: 5, unit price: $399.00, total: $1,995.00)
        - Cable Management Kits (qty: 15, unit price: $24.99, total: $374.85)

        Subtotal: $3,869.75
        Tax (7%): $270.88
        Total Due: $4,140.63
        Currency: USD

        Notes: Please remit payment via ACH. Thank you for your business!
        """
    },
    {
        "id": "INV-002",
        "text": """
        Invoice #: 88291
        From: Naples Printing Services
        Date Issued: August 5, 2026
        Payment Due: August 20, 2026

        Services:
        - Brochure Printing 500 units @ $0.85 each = $425.00
        - Logo Design Revisions (1 hour) @ $120.00 = $120.00
        - Rush Delivery Fee = $75.00

        Subtotal: $620.00
        Tax: $43.40
        TOTAL: $999.99

        Currency: USD
        Terms: Due on receipt
        """
        # Note: $620 + $43.40 = $663.40, not $999.99 — validation should flag this
    },
    {
        "id": "INV-003",
        "text": """
        INVOICE

        Vendor: Southwest Florida IT Solutions
        Services rendered: Network setup and security audit
        Amount: $2,500.00
        Currency: USD