) (1).png)
Naples AI
What You'll Build
If you've ever watched someone manually key invoice data into a spreadsheet for hours, you already know why this matters. In this tutorial, you'll build a production-ready multi-agent invoice processor using the Claude API that automatically extracts structured data from invoice text, validates it, and routes documents based on business rules. By the end, you'll have a working Python system that handles the entire pipeline from raw invoice input to validated, routed output without a human touching it.
Prerequisites
- Python 3.9 or higher installed
- An Anthropic API key (get one at console.anthropic.com)
- The Anthropic Python SDK installed (
pip install anthropic) - Basic familiarity with Python classes and functions
- A terminal or command prompt you're comfortable using
Step 1: Set Up Your Claude API Client and Environment
The first thing you need is a clean environment setup. You want your API key loaded from an environment variable, not hardcoded into the file, especially if you're ever pushing this to GitHub.
Create a file called invoice_processor.py and start with this foundation block. This sets up the Anthropic client, imports everything you'll need, and defines the base configuration for the model you'll use throughout the project.
import os
import json
import anthropic
from typing import Any
# Load your Anthropic API key from the environment
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
# We're using claude-sonnet-4-6 throughout this project
MODEL_NAME = "claude-sonnet-4-6"
# Approval threshold: invoices above this dollar amount get flagged for human review
APPROVAL_THRESHOLD = 5000.00
Set your API key in your terminal before running anything. On Mac or Linux, run export ANTHROPIC_API_KEY=your_key_here. On Windows, use set ANTHROPIC_API_KEY=your_key_here. This keeps your credentials out of your source code.
Step 2: Define Tool Functions for Invoice Extraction
This is where the Claude API integration gets interesting. Claude supports tool use, which means you can define structured functions and Claude will decide when to call them and what arguments to pass. For invoice processing, this is the cleanest approach because you get structured JSON output every time instead of parsing free-form text.
You're going to define three tools: one for extracting invoice fields, one for validating the data, and one for routing the document. Claude acts as the orchestrator and decides which tool to call based on the context you give it.
invoice_processor.py (continued)
# Tool definitions passed to Claude so it knows what functions are available
INVOICE_TOOLS = [
{
"name": "extract_invoice_data",
"description": (
"Extract structured data fields from raw invoice text. "
"Call this first when processing any invoice document."
),
"input_schema": {
"type": "object",
"properties": {
"invoice_number": {
"type": "string",
"description": "The unique invoice identifier or number"
},
"vendor_name": {
"type": "string",
"description": "The name of the vendor or supplier"
},
"invoice_date": {
"type": "string",
"description": "The date on the invoice in YYYY-MM-DD format"
},
"due_date": {
"type": "string",
"description": "The 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"}
}
}
},
"subtotal": {
"type": "number",
"description": "Invoice subtotal before tax"
},
"tax_amount": {
"type": "number",
"description": "Total tax amount applied"
},
"total_amount": {
"type": "number",
"description": "Final total amount due"
},
"currency": {
"type": "string",
"description": "Currency code, e.g. USD"
}
},
"required": [
"invoice_number", "vendor_name", "invoice_date",
"total_amount", "currency"
]
}
},
{
"name": "validate_invoice",
"description": (
"Validate extracted invoice data for completeness and consistency. "
"Check that totals add up, required fields are present, and dates are valid."
),
"input_schema": {
"type": "object",
"properties": {
"is_valid": {
"type": "boolean",
"description": "Whether the invoice passed all validation checks"
},
"validation_errors": {
"type": "array",
"description": "List of validation error messages if any",
"items": {"type": "string"}
},
"confidence_score": {
"type": "number",
"description": "Confidence score from 0.0 to 1.0 for the extraction quality"
}
},
"required": ["is_valid", "validation_errors", "confidence_score"]
}
},
{
"name": "route_invoice",
"description": (
"Determine where the invoice should be sent based on its content and validation status. "
"Route to the correct department or workflow queue."
),
"input_schema": {
"type": "object",
"properties": {
"route": {
"type": "string",
"description": "Routing destination",
"enum": [
"auto_approve",
"manager_review",
"finance_team",
"error_queue"
]
},
"reason": {
"type": "string",
"description": "Plain-English explanation of why this routing decision was made"
},
"priority": {
"type": "string",
"description": "Processing priority level",
"enum": ["low", "normal", "high", "urgent"]
}
},
"required": ["route", "reason", "priority"]
}
}
]
The schema design here matters. The more precise your field descriptions are, the more reliably Claude will populate them correctly. I learned this the hard way by leaving vague descriptions on currency and date fields and getting inconsistent formats back.
Step 3: Build the Multi-Agent Orchestrator
Now comes the main class that ties everything together. The orchestrator manages the conversation loop with Claude, processes tool calls as they come back, and assembles the final result from each agent's output. Think of it as the coordinator that hands off work between your extraction, validation, and routing agents.
invoice_processor.py (continued)
class InvoiceProcessingOrchestrator:
"""
Multi-agent orchestrator that coordinates Claude tool calls
to extract, validate, and route invoice documents.
"""
def __init__(self):
self.client = client
self.model = MODEL_NAME
self.tools = INVOICE_TOOLS
def process_invoice(self, invoice_text: str) -> dict[str, Any]:
"""
Main entry point. Takes raw invoice text and returns a
fully processed result with extracted data, validation,
and routing decision.
"""
print(f"\n{'='*50}")
print("Starting invoice processing pipeline...")
print(f"{'='*50}\n")
messages = [
{
"role": "user",
"content": (
f"Process this invoice document completely. "
f"First extract all data fields, then validate the extraction, "
f"then route it appropriately.\n\nInvoice text:\n{invoice_text}"
)
}
]
# Storage for results collected across multiple tool calls
extracted_data = {}
validation_result = {}
routing_decision = {}
# Agentic loop: keep calling Claude until it stops using tools
while True:
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
tools=self.tools,
messages=messages
)
print(f"Agent response -- stop reason: {response.stop_reason}")
# If Claude is done with tools, break out of the loop
if response.stop_reason == "end_turn":
print("Pipeline complete.")
break
# Process each content block in the response
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_use_id = block.id
print(f"\nAgent calling tool: {tool_name}")
# Dispatch to the appropriate handler
result = self._handle_tool_call(
tool_name, tool_input,
extracted_data, validation_result
)
# Store results by tool name for final assembly
if tool_name == "extract_invoice_data":
extracted_data = tool_input
elif tool_name == "validate_invoice":
validation_result = tool_input
elif tool_name == "route_invoice":
routing_decision = tool_input
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": json.dumps(result)
})
# Append Claude's response and tool results back into the conversation
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
return {
"extracted_data": extracted_data,
"validation": validation_result,
"routing": routing_decision,
"status": "processed"
}
def _handle_tool_call(
self,
tool_name: str,
tool_input: dict,
extracted_data: dict,
validation_result: dict
) -> dict:
"""
Routes tool calls to the correct handler and returns a result
that gets fed back into the conversation as a tool_result message.
"""
if tool_name == "extract_invoice_data":
return self._run_extraction(tool_input)
elif tool_name == "validate_invoice":
return self._run_validation(tool_input, extracted_data)
elif tool_name == "route_invoice":
return self._run_routing(tool_input)
else:
return {"error": f"Unknown tool: {tool_name}"}
Step 4: Implement Validation and Error Handling
The individual handler methods are where your business logic lives. The extraction handler acknowledges the data and echoes it back so Claude knows the tool call succeeded. The validation handler runs your actual checks, like verifying that line item totals add up to the subtotal. The routing handler applies your approval threshold rules.
invoice_processor.py (continued)
def _run_extraction(self, tool_input: dict) -> dict:
"""
Handles the extract_invoice_data tool call.
In production you might persist this to a database here.
"""
print(f" Extracted invoice number: {tool_input.get('invoice_number', 'N/A')}")
print(f" Vendor: {tool_input.get('vendor_name', 'N/A')}")
print(f" Total: {tool_input.get('currency', 'USD')} {tool_input.get('total_amount', 0):.2f}")
return {
"success": True,
"message": "Invoice data extracted successfully",
"fields_extracted": list(tool_input.keys())
}
def _run_validation(self, tool_input: dict, extracted_data: dict) -> dict:
"""
Handles the validate_invoice tool call.
Runs programmatic checks on top of Claude's self-reported validation.
"""
errors = list(tool_input.get("validation_errors", []))
# Programmatic check: verify line item totals match the subtotal
line_items = extracted_data.get("line_items", [])
if line_items:
calculated_subtotal = sum(
item.get("total", 0) for item in line_items
)
reported_subtotal = extracted_data.get("subtotal", 0)
# Allow a small floating point tolerance
if abs(calculated_subtotal - reported_subtotal) > 0.02:
errors.append(
f"Line item sum ({calculated_subtotal:.2f}) does not match "
f"reported subtotal ({reported_subtotal:.2f})"
)
# Verify total = subtotal + tax
reported_total = extracted_data.get("total_amount", 0)
reported_subtotal = extracted_data.get("subtotal", 0)
tax_amount = extracted_data.get("tax_amount", 0)
expected_total = reported_subtotal + tax_amount
if extracted_data and abs(reported_total - expected_total) > 0.02:
errors.append(
f"Total ({reported_total:.2f}) does not match "
f"subtotal + tax ({expected_total:.2f})"
)
is_valid = len(errors) == 0
confidence = tool_input.get("confidence_score", 0.0)
print(f" Validation passed: {is_valid}")
if errors:
for err in errors:
print(f" Validation error: {err}")
print(f" Confidence score: {confidence:.2f}")
return {
"success": True,
"is_valid": is_valid,
"errors_found": errors,
"confidence_score": confidence
}
def _run_routing(self, tool_input: dict) -> dict:
"""
Handles the route_invoice tool call.
Applies the approval threshold rule before confirming the route.
"""
route = tool_input.get("route", "error_queue")
reason = tool_input.get("reason", "")
priority = tool_input.get("priority", "normal")
print(f" Routing to: {route}")
print(f" Priority: {priority}")
print(f" Reason: {reason}")
return {
"success": True,
"route_confirmed": route,
"priority_confirmed": priority,
"message": f"Invoice routed to {route} with {priority} priority"
}
def run_example():
"""
Runs a sample invoice through the full processing pipeline
so you can see what the output looks like.
"""
sample_invoice = """
INVOICE
Invoice Number: INV-2026-00847
Date: 2026-10-15
Due Date: 2026-11-14
From:
Gulf Coast Tech Supplies LLC
3400 Tamiami Trail N, Naples, FL 34103
To:
Coastal Property Management Group
800 Fifth Ave S, Naples, FL 34102
Line Items:
1. Network Switch (24-port) Qty: 2 Unit Price: $349.00 Total: $698.00
2. Cat6 Cable (500ft spool) Qty: 5 Unit Price: $89.00 Total: $445.00
3. Rack Mount Kit Qty: 2 Unit Price: $75.00 Total: $150.00
4. On-site Installation Qty: 8 Unit Price: $125.00 Total: $1,000.00
Subtotal: $2,293.00
Tax (7%): $160.51
TOTAL: $2,453.51
Currency: USD
Payment Terms: Net 30
"""
orchestrator = InvoiceProcessingOrchestrator()
result = orchestrator.process_invoice(sample_invoice)
print("\n" + "="*50)
print("FINAL PROCESSING RESULT")
print("="*50)
print(json.dumps(result, indent=2))
if __name__ == "__main__":
run_example()
Example Invoice Processing Output
Here's what you'll actually see in your terminal when you run this against the sample invoice. The pipeline logs each step so you can watch the agents work in sequence.
Terminal output
==================================================
Starting invoice processing pipeline...
==================================================
Agent response -- stop reason: tool_use
Agent calling tool: extract_invoice_data
Extracted invoice number: INV-2026-00847
Vendor: Gulf Coast Tech Supplies LLC
Total: USD 2453.51
Agent response -- stop reason: tool_use
Agent calling tool: validate_invoice
Validation passed: True
Confidence score: 0.97
Agent response -- stop reason: tool_use
Agent calling tool: route_invoice
Routing to: auto_approve
Priority: normal
Reason: Invoice is valid with high confidence and total is below the $5,000 approval threshold.
Agent response -- stop reason: end_turn
Pipeline complete.
==================================================
FINAL PROCESSING RESULT
==================================================
{
"extracted_data": {
"invoice_number": "INV-2026-00847",
"vendor_name": "Gulf Coast Tech Supplies LLC",
"invoice_date": "2026-10-15",
"due_date": "2026-11-14",
"line_items": [
{"description": "Network Switch (24-port)", "quantity": 2, "unit_price": 349.00, "total": 698.00},
{"description": "Cat6 Cable (500ft spool)", "quantity": 5, "unit_price": 89.00, "total": 445.00},
{"description": "Rack Mount Kit", "quantity": 2, "unit_price": 75.00, "total": 150.00},
{"description": "On-site Installation", "quantity": 8, "unit_price": 125.00, "total": 1000.00}
],
"subtotal": 2293.00,
"tax_amount": 160.51,
"total_amount": 2453.51,
"currency": "USD"
},
"validation": {
"is_valid": true,
"validation_errors": [],
"confidence_score": 0.97
},
"routing": {
"route": "auto_approve",
"reason": "Invoice is valid with high confidence and total is below the $5,000 approval threshold.",
"priority": "normal"
},
"status": "processed"
}
How It Works
The system runs a conversation loop with Claude where each turn either returns a tool call or signals that the job is done. Claude reads the invoice text, decides to call extract_invoice_data first, and populates the schema with what it finds. Your code handles that tool call, stores the extracted data, and feeds the result back into the conversation.
Claude then calls validate_invoice based on what it extracted. Your handler runs the programmatic checks (like verifying that line item totals actually add up) on top of whatever Claude flags. The combined result goes back into the conversation, and Claude makes the final routing call based on everything it knows, including the total amount relative to your threshold.
The key design here is that Claude is the decision-maker for what to do next, but your Python code owns the business logic for how each tool executes. That separation is what makes this pattern work for real production systems.
Common Errors and Fixes
Error 1: AuthenticationError on startup
Exact error: anthropic.AuthenticationError: Error code: 401 -- invalid_api_key
This means your API key isn't being read from the environment. Double-check that you ran export ANTHROPIC_API_KEY=your_key_here in the same terminal session where you're running the script. A common mistake is setting it in one terminal window and running the script in another. You can verify it's set by running echo $ANTHROPIC_API_KEY before executing your script.
Error 2: Infinite loop in the agentic loop
Exact error: The script runs forever and never prints "Pipeline complete."
This usually happens when tool results aren't being appended back into the messages list correctly. Claude keeps seeing a conversation that never resolved, so it keeps asking to call tools. Make sure you're appending both