All articles

If you've ever watched someone manually key invoice data into a spreadsheet for hours, you already know the problem. Invoice processing is slow, error-prone, and a total waste of skilled labor. In this tutorial, I'll show you how to build a production-ready multi-agent AI system using the Claude API that extracts, validates, and routes invoice documents automatically, with Python and the Anthropic SDK doing the heavy lifting.

What You'll Build

By the end of this tutorial, you'll have a working Python system that uses three coordinated Claude agents to process invoice PDFs and images end to end. The orchestrator agent delegates tasks to a document analyzer and a validation agent, each with their own tools, and produces clean structured JSON output you can pipe directly into your accounting or ERP system. This is the exact kind of intelligent document processing automation we build for clients at Naples AI, and now you can build it yourself.

Prerequisites

  • Python 3.11 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • Basic familiarity with Python classes and dictionaries
  • The anthropic and python-dotenv packages (we'll install these in Step 1)
  • A sample invoice image or PDF to test with (a JPG screenshot of any invoice works fine)
Full Source Code Note: All the code snippets in the steps below are complete and working. Each step builds on the previous one, and by Step 6 you'll have the full system assembled. I recommend following the steps in order rather than jumping to the end, because the agent architecture makes more sense once you see it built piece by piece.

Step 1: Set Up Your Python Environment and Anthropic SDK

Start by creating a clean project directory and a virtual environment. This keeps your dependencies isolated and avoids the classic "it works on my machine" problem.

terminal 
mkdir invoice-processor
cd invoice-processor
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install anthropic python-dotenv

Next, create a .env file in your project root and add your API key. Never hardcode secrets directly in your Python files.

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

Now create the main project file. This is where everything will live.

invoice_processor.py 
import os
import json
import base64
import anthropic
from dotenv import load_dotenv

load_dotenv()

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

That's the foundation. The client object is what every agent in the system will use to call Claude.

Step 2: Define Invoice Processing Tools for Extraction, Validation, and Routing

Claude's tool use feature lets the model call functions you define, and return structured data instead of free-form text. This is what makes a multi-agent system predictable enough to run in production. We'll define three tools: one for extracting invoice fields, one for validating the data, and one for routing the invoice to the right workflow.

invoice_processor.py (continued) 
# Tool definitions tell Claude what functions are available and what shape the inputs take.
# Claude will decide when and how to call these based on the task.

EXTRACTION_TOOLS = [
    {
        "name": "extract_invoice_data",
        "description": (
            "Extract structured data fields from an invoice document. "
            "Use this tool to pull vendor info, line items, totals, and dates."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "vendor_name": {
                    "type": "string",
                    "description": "The name of the vendor or supplier on the invoice"
                },
                "vendor_address": {
                    "type": "string",
                    "description": "Full mailing address of the vendor"
                },
                "invoice_number": {
                    "type": "string",
                    "description": "The invoice ID or reference number"
                },
                "invoice_date": {
                    "type": "string",
                    "description": "Date the invoice was issued, in YYYY-MM-DD format"
                },
                "due_date": {
                    "type": "string",
                    "description": "Payment due date, in YYYY-MM-DD format"
                },
                "line_items": {
                    "type": "array",
                    "description": "List of 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": "Sum of all line items before tax"
                },
                "tax_amount": {
                    "type": "number",
                    "description": "Total tax amount applied"
                },
                "total_amount": {
                    "type": "number",
                    "description": "Final total including tax"
                },
                "currency": {
                    "type": "string",
                    "description": "Currency code, e.g. USD, EUR"
                },
                "payment_terms": {
                    "type": "string",
                    "description": "Payment terms such as Net 30, Due on Receipt, etc."
                }
            },
            "required": [
                "vendor_name", "invoice_number", "invoice_date",
                "line_items", "subtotal", "total_amount", "currency"
            ]
        }
    }
]

VALIDATION_TOOLS = [
    {
        "name": "validate_invoice_data",
        "description": (
            "Validate extracted invoice data for mathematical accuracy and completeness. "
            "Check that line item totals add up, required fields are present, and dates are logical."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "is_valid": {
                    "type": "boolean",
                    "description": "True if the invoice passes all validation checks"
                },
                "errors": {
                    "type": "array",
                    "description": "List of validation errors found, empty if invoice is valid",
                    "items": {"type": "string"}
                },
                "warnings": {
                    "type": "array",
                    "description": "Non-blocking issues that may need human review",
                    "items": {"type": "string"}
                },
                "calculated_subtotal": {
                    "type": "number",
                    "description": "Subtotal recalculated from line items for cross-checking"
                },
                "discrepancy_amount": {
                    "type": "number",
                    "description": "Difference between stated and calculated subtotal, 0 if none"
                }
            },
            "required": ["is_valid", "errors", "warnings", "calculated_subtotal", "discrepancy_amount"]
        }
    }
]

ROUTING_TOOLS = [
    {
        "name": "route_invoice",
        "description": (
            "Determine the correct processing workflow for this invoice based on its "
            "total amount, vendor, and validation status."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "routing_decision": {
                    "type": "string",
                    "enum": ["auto_approve", "manager_review", "reject", "accounting_queue"],
                    "description": "Where to send this invoice for next processing step"
                },
                "reason": {
                    "type": "string",
                    "description": "Plain-English explanation of the routing decision"
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "urgent"],
                    "description": "Processing priority level"
                }
            },
            "required": ["routing_decision", "reason", "priority"]
        }
    }
]

Step 3: Create the Document Analyzer Agent

The document analyzer agent takes a raw invoice image and runs the extraction tool to pull structured data out of it. I'm keeping each agent as its own class so you can test them independently or swap one out without breaking the others. That separation is what makes this a real multi-agent system rather than just one big prompt.

invoice_processor.py (continued) 
class DocumentAnalyzerAgent:
    """
    Handles the first stage: reading raw invoice documents and extracting structured fields.
    Accepts a file path to an invoice image (JPG, PNG, or PDF rendered as image).
    """

    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.model = "claude-sonnet-4-6"

    def _encode_image(self, image_path: str) -> tuple[str, str]:
        """Read an image file and return base64-encoded content plus its media type."""
        extension = image_path.lower().split(".")[-1]
        media_type_map = {
            "jpg": "image/jpeg",
            "jpeg": "image/jpeg",
            "png": "image/png",
            "gif": "image/gif",
            "webp": "image/webp"
        }
        media_type = media_type_map.get(extension, "image/jpeg")

        with open(image_path, "rb") as f:
            encoded = base64.standard_b64encode(f.read()).decode("utf-8")

        return encoded, media_type

    def analyze(self, image_path: str) -> dict:
        """
        Send the invoice image to Claude with the extraction tool.
        Returns the extracted invoice data as a dictionary.
        """
        print(f"[Analyzer] Processing invoice: {image_path}")

        image_data, media_type = self._encode_image(image_path)

        messages = [
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data
                        }
                    },
                    {
                        "type": "text",
                        "text": (
                            "Please extract all invoice data from this document using the "
                            "extract_invoice_data tool. Be precise with numbers and dates. "
                            "If a field is not present on the invoice, omit it from the output."
                        )
                    }
                ]
            }
        ]

        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            tools=EXTRACTION_TOOLS,
            messages=messages
        )

        # Pull the tool use block out of the response content
        for block in response.content:
            if block.type == "tool_use" and block.name == "extract_invoice_data":
                print("[Analyzer] Extraction complete.")
                return block.input

        # If Claude didn't call the tool, something went wrong
        raise ValueError(
            f"Analyzer agent did not return tool output. Stop reason: {response.stop_reason}"
        )

Step 4: Create the Data Validator Agent

The validation agent takes the structured data from the analyzer and checks it for errors. It recalculates line item totals, looks for missing required fields, and flags anything suspicious. I added explicit error handling here because bad invoice data is the rule, not the exception, especially if you're processing documents from dozens of different vendors.

invoice_processor.py (continued) 
class DataValidatorAgent:
    """
    Validates extracted invoice data for mathematical accuracy, completeness,
    and logical consistency. Returns a structured validation report.
    """

    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.model = "claude-sonnet-4-6"

    def validate(self, invoice_data: dict) -> dict:
        """
        Run validation checks on extracted invoice data.
        Returns a validation result dict with is_valid, errors, warnings, and discrepancy info.
        """
        print("[Validator] Running validation checks...")

        # Serialize the extracted data so Claude can reason over it
        invoice_json = json.dumps(invoice_data, indent=2)

        messages = [
            {
                "role": "user",
                "content": (
                    f"Validate the following extracted invoice data using the validate_invoice_data tool.\n\n"
                    f"Check for:\n"
                    f"1. Mathematical accuracy: do line item quantities * unit_prices equal their totals?\n"
                    f"2. Does the sum of line item totals match the stated subtotal?\n"
                    f"3. Are all required fields present and non-empty?\n"
                    f"4. Is the invoice date before or on the due date?\n"
                    f"5. Are all numeric values positive?\n\n"
                    f"Invoice data:\n{invoice_json}"
                )
            }
        ]

        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=1024,
                tools=VALIDATION_TOOLS,
                messages=messages
            )
        except anthropic.APIError as e:
            # Surface API errors clearly rather than letting them fail silently
            raise RuntimeError(f"Validation agent API call failed: {e}") from e

        for block in response.content:
            if block.type == "tool_use" and block.name == "validate_invoice_data":
                result = block.input
                status = "PASSED" if result["is_valid"] else "FAILED"
                print(f"[Validator] Validation {status}. Errors: {len(result['errors'])}")
                return result

        raise ValueError(
            f"Validator agent did not return tool output. Stop reason: {response.stop_reason}"
        )

Step 5: Implement the Orchestrator Agent

The orchestrator is the brain of the system. It receives the validated invoice data and the validation report, then decides where the invoice should go next. It uses the routing tool to make that decision in a structured, auditable way. Think of the orchestrator as a senior AP clerk who reviews what the junior staff found and makes the final call.

invoice_processor.py (continued) 
class OrchestratorAgent:
    """
    Coordinates the full invoice processing pipeline.
    Takes extraction and validation results, then applies business routing rules
    to determine the correct downstream workflow for each invoice.
    """

    def __init__(self, client: anthropic.Anthropic):
        self.client = client
        self.model = "claude-sonnet-4-6"
        # Invoices above this threshold require manager review regardless of validity
        self.high_value_threshold = 5000.00

    def route(self, invoice_data: dict, validation_result: dict) -> dict:
        """
        Apply routing logic based on invoice data and validation outcome.
        Returns a routing decision with reason and priority.
        """
        print("[Orchestrator] Determining routing decision...")

        invoice_summary = json.dumps(invoice_data, indent=2)
        validation_summary = json.dumps(validation_result, indent=2)
        total = invoice_data.get("total_amount", 0)

        routing_context = (
            f"Business routing rules:\n"
            f"- Invoices with validation errors must be REJECTED or sent to manager_review\n"
            f"- Invoices over ${self.high_value_threshold:,.2f} require manager_review\n"
            f"- Valid invoices under ${self.high_value_threshold:,.2f} can be auto_approved\n"
            f"- Invoices with warnings (but no errors) go to accounting_queue for soft review\n"
            f"- Set priority to 'urgent' if due_date is within 3 days\n\n"
            f"Invoice total: ${total:,.2f}\n\n"
            f"Extracted invoice data:\n{invoice_summary}\n\n"
            f"Validation result:\n{validation_summary}"
        )

        messages = [
            {
                "role": "user",
                "content": (
                    f"Based on the invoice data and validation results below, use the "
                    f"route_invoice tool to determine the correct processing workflow.\n\n"
                    f"{routing_context}"
                )
            }
        ]

        response = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            tools=ROUTING_TOOLS,
            messages=messages
        )

        for block in response.content:
            if block.type == "tool_use" and block.name == "route_invoice":
                decision = block.input
                print(f"[Orchestrator] Routing to: {decision['routing_decision']} ({decision['priority']} priority)")
                return decision

        raise ValueError(
            f"Orchestrator agent did not return tool output. Stop reason: {response.stop_reason}"
        )

Step 6: Build the Main Run Loop with Tool Execution

Now we wire everything together into a single process_invoice function that runs the full pipeline from raw image to final JSON output. I'm also adding an InvoiceProcessingSystem wrapper class so you can instantiate the whole system in one line. At the bottom, I've included an example runner with a mock invoice for cases where you don't have a real image file handy.

invoice_processor.py (continued) 
class InvoiceProcessingSystem:
    """
    Top-level system that coordinates all three agents into a single processing pipeline.
    Instantiate once and call process_invoice() for each document.
    """

    def __init__(self):
        self.analyzer = DocumentAnalyzerAgent(client)
        self.validator = DataValidatorAgent(client)
        self.orchestrator = OrchestratorAgent(client)

    def process_invoice(self, image_path: str) -> dict:
        """
        Run the full multi-agent pipeline on a single invoice image.
        Returns a complete processing result including extracted data, validation, and routing.
        """
        print(f"\n{'='*50}")
        print(f"Starting invoice processing pipeline")
        print(f"{'='*50}\n")

        # Stage 1: Extract structured data from the raw document
        try:
            extracted_data = self.analyzer.analyze(image_path)
        except (ValueError, FileNotFoundError) as e:
            return self._build_error_result("extraction", str(e))

        # Stage 2: Validate the extracted data for accuracy and completeness
        try:
            validation_result = self.validator.validate(extracted_data)
        except (ValueError, RuntimeError) as e:
            return self._build_error_result("validation", str(e))

        # Stage 3: Orchestrator applies business rules and decides routing
        try:
            routing_decision = self.orchestrator.route(extracted_data, validation_result)
        except ValueError as e:
            return self._build_error_result("routing", str(e))

        # Assemble the final result object
        final_result = {
            "status": "processed",
            "pipeline_stages_completed": ["extraction", "validation", "routing"],
            "invoice_data": extracted_data,
            "validation": validation_result,
            "routing": routing_decision
        }

        print(f"\n{'='*50}")
        print("Pipeline complete.")
        print(f"{'='*50}\n")

        return final_result

    def _build_error_result(self, failed_stage: str, error_message: str) -> dict:
        """Return a consistent error structure when a pipeline stage fails."""
        print(f"[System] Pipeline failed at stage: {failed_stage}")
        return {
            "status": "error",
            "failed_stage": failed_stage,
            "error": error_message,
            "invoice_data": None,
            "validation": None,
            "routing": None
        }


def create_mock_invoice_image() -> str:
    """
    Creates a simple text file that simulates invoice content for testing.
    In production, replace this with a real image path.
    Replace this with a real invoice image path when testing against actual documents.
    """
    import tempfile
    from PIL import Image, ImageDraw, ImageFont
    import textwrap

    invoice_text = textwrap.dedent("""
        INVOICE

        Vendor: Gulf Coast Office Supplies LLC
        Address: 3400 Tamiami Trail N, Naples, FL 34103
        Invoice #: INV-2026-00847
        Invoice Date: 2026-10-15
        Due Date: 2026-11-14
        Payment Terms: Net 30

        Line Items:
        1. Printer Paper (Case)  Qty: 10  Unit: $45.00  Total: $450.00
        2. Toner Cartridges      Qty: 4   Unit: $89.99  Total: $359.96
        3. Desk Organizers       Qty: 6   Unit: $22.50  Total: $135.00

        Subtotal: $944.96
        Tax (7%): $66.15
        TOTAL DUE: $1,011.11
        Currency: USD
    """).strip()

    # Create a simple white image with the invoice text rendered on it
    img = Image.new("RGB", (800, 600), color=(255, 255, 255))
    draw = ImageDraw.Draw(img)
    draw.multiline_text((40, 40), invoice_text, fill=(0, 0, 0), spacing=8)

    tmp = tempfile.NamedTemporaryFile(suffix=".png",