Claude AI Agents for Document Processing & Workflow Automation

import base64
import pypdf
from docx import Document
from anthropic import Anthropic

client = Anthropic()

def load_document(file_path: str) -> dict:
    """
    Load a document and prepare it for Claude vision API.
    Returns a dict with document type, content, and metadata.
    """
    if file_path.endswith('.pdf'):
        # Extract text and get first page as image
        reader = pypdf.PdfReader(file_path)
        text = "\n".join([page.extract_text() for page in reader.pages])
        page_count = len(reader.pages)

        # Convert first page to image for vision
        import pdf2image
        images = pdf2image.convert_from_path(file_path, first_page=1, last_page=1)
        image_bytes = io.BytesIO()
        images[0].save(image_bytes, format='PNG')
        image_base64 = base64.standard_b64encode(image_bytes.getvalue()).decode()

        return {
            'type': 'pdf',
            'text': text,
            'image_base64': image_base64,
            'page_count': page_count,
            'file_name': file_path.split('/')[-1]
        }

    elif file_path.endswith('.docx'):
        # Extract text from Word document
        doc = Document(file_path)
        text = "\n".join([para.text for para in doc.paragraphs])

        return {
            'type': 'docx',
            'text': text,
            'page_count': 1,
            'file_name': file_path.split('/')[-1]
        }

    elif file_path.endswith(('.jpg', '.jpeg', '.png')):
        # Load image directly
        with open(file_path, 'rb') as f:
            image_base64 = base64.standard_b64encode(f.read()).decode()

        return {
            'type': 'image',
            'image_base64': image_base64,
            'page_count': 1,
            'file_name': file_path.split('/')[-1]
        }

    else:
        raise ValueError(f"Unsupported file type: {file_path}")


def classify_document(doc: dict) -> dict:
    """
    Classify a document type using Claude vision.
    Returns document type and confidence score.
    """
    prompt = """Classify this document into one of these categories:
- INVOICE: Payment request for goods/services
- CONTRACT: Legal agreement between parties
- PO: Purchase order
- COMPLIANCE: Regulatory/policy document
- EXPENSE_REPORT: Employee expense submission
- OTHER: Doesn't fit above categories

Respond with ONLY JSON:
{
  "document_type": "CATEGORY",
  "confidence": 0.95,
  "reasoning": "brief reason"
}"""

    image_block = {
        "type": "image",
        "source": {
            "type": "base64",
            "media_type": "image/png",
            "data": doc['image_base64']
        }
    } if 'image_base64' in doc else None

    messages = []
    if image_block:
        messages.append({
            "role": "user",
            "content": [image_block, {"type": "text", "text": prompt}]
        })
    else:
        messages.append({
            "role": "user",
            "content": f"{prompt}\n\nDocument text:\n{doc.get('text', 'No text available')[:2000]}"
        })

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        messages=messages
    )

    return json.loads(response.content[0].text)

from anthropic.lib.agents import Agent, tool
from typing import Any

@tool
def extract_invoice_data(invoice_text: str, invoice_image_base64: str) -> dict:
    """Extract structured data from an invoice."""
    prompt = """Extract invoice data. Return ONLY valid JSON:
{
  "vendor_name": "...",
  "invoice_number": "...",
  "invoice_date": "YYYY-MM-DD",
  "due_date": "YYYY-MM-DD",
  "total_amount": 1000.00,
  "currency": "USD",
  "line_items": [
    {"description": "...", "quantity": 1, "unit_price": 100.00, "amount": 100.00}
  ],
  "extraction_confidence": {
    "vendor_name": 0.95,
    "total_amount": 0.98,
    "overall": 0.96
  }
}"""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": "image/png", "data": invoice_image_base64}
                },
                {"type": "text", "text": prompt}
            ]
        }]
    )

    return json.loads(response.content[0].text)

@tool
def extract_contract_data(contract_text: str) -> dict:
    """Extract key terms from a contract."""
    prompt = """Extract contract data. Return ONLY valid JSON:
{
  "contract_type": "SERVICE_AGREEMENT|NDA|PURCHASE|OTHER",
  "parties": ["Party A", "Party B"],
  "effective_date": "YYYY-MM-DD",
  "term_months": 12,
  "auto_renewal": true,
  "termination_notice_days": 30,
  "key_obligations": ["obligation 1", "obligation 2"],
  "payment_terms": "NET_30",
  "liability_cap": 100000,
  "extraction_confidence": {
    "parties": 0.98,
    "effective_date": 0.85,
    "overall": 0.90
  }
}"""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"{prompt}\n\nContract text:\n{contract_text[:5000]}"
        }]
    )

    return json.loads(response.content[0].text)

@tool
def validate_extraction(
    extracted_data: dict,
    document_type: str,
    business_rules: dict
) -> dict:
    """Validate extracted data against business rules."""
    validation_results = {
        "passed": True,
        "errors": [],
        "warnings": [],
        "requires_review": False
    }

    # Check confidence scores
    overall_conf = extracted_data.get('extraction_confidence', {}).get('overall', 0)
    if overall_conf < 0.85:
        validation_results['requires_review'] = True
        validation_results['warnings'].append(f"Low confidence: {overall_conf:.0%}")

    # Document-specific validation
    if document_type == 'invoice':
        amount = extracted_data.get('total_amount', 0)
        if amount > business_rules.get('max_invoice_amount', 100000):
            validation_results['errors'].append(f"Invoice exceeds max amount: ${amount}")
            validation_results['passed'] = False

        if not extracted_data.get('vendor_name'):
            validation_results['errors'].append("Missing vendor name")
            validation_results['passed'] = False

    elif document_type == 'contract':
        term = extracted_data.get('term_months', 0)
        if term > business_rules.get('max_contract_term_months', 60):
            validation_results['warnings'].append(f"Long term: {term} months")
            validation_results['requires_review'] = True

    return validation_results

# Build the document processing agent
doc_processor_agent = Agent(
    client=client,
    model="claude-sonnet-4-6",
    system="""You are a document processing agent. For each document:
1. Classify the document type using vision
2. Extract relevant structured data
3. Validate against business rules
4. Flag low-confidence fields for human review

Always be conservative — if you're unsure, flag for review.""",
    tools=[extract_invoice_data, extract_contract_data, validate_extraction],
    max_tokens=2048,
    max_iterations=8
)

# Process a document
def process_document(file_path: str) -> dict:
    doc = load_document(file_path)
    classification = classify_document(doc)

    result = doc_processor_agent.run(
        f"Process this {classification['document_type']} document: {doc['file_name']}"
    )

    return {
        "file": doc['file_name'],
        "classification": classification,
        "extraction_result": result.output,
        "token_usage": result.usage.total_tokens
    }