Claude API Real-time Multimodal Agent Architecture: Design Patterns & Implementation

#!/usr/bin/env python3
# invoice_processing_agent.py
 
import anthropic
import base64
import json
import os
from pathlib import Path
from datetime import datetime
from typing import Optional
import subprocess
 
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
# Define tools the agent can use
TOOLS = [
    {
        "name": "record_invoice",
        "description": "Record extracted invoice data to Google Sheets",
        "input_schema": {
            "type": "object",
            "properties": {
                "invoice_number": {
                    "type": "string",
                    "description": "Invoice ID (e.g., INV-2026-001)"
                },
                "amount": {
                    "type": "number",
                    "description": "Invoice amount in USD"
                },
                "invoice_date": {
                    "type": "string",
                    "description": "Invoice date (YYYY-MM-DD format)"
                },
                "vendor_name": {
                    "type": "string",
                    "description": "Vendor/company name"
                },
                "status": {
                    "type": "string",
                    "enum": ["pending", "approved", "flagged"],
                    "description": "Initial approval status"
                }
            },
            "required": ["invoice_number", "amount", "invoice_date", "vendor_name"]
        }
    },
    {
        "name": "flag_for_review",
        "description": "Flag invoice for manual human review",
        "input_schema": {
            "type": "object",
            "properties": {
                "invoice_number": {"type": "string"},
                "reason": {
                    "type": "string",
                    "enum": [
                        "amount_unusual",
                        "duplicate_suspected",
                        "missing_fields",
                        "date_inconsistency",
                        "vendor_not_recognized"
                    ]
                },
                "severity": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "Review urgency"
                }
            },
            "required": ["invoice_number", "reason"]
        }
    },
    {
        "name": "notify_slack",
        "description": "Send Slack notification to accounting team",
        "input_schema": {
            "type": "object",
            "properties": {
                "channel": {"type": "string", "description": "#accounting or #alerts"},
                "message": {"type": "string"},
                "attachment_data": {
                    "type": "object",
                    "description": "Optional structured data to display"
                }
            },
            "required": ["channel", "message"]
        }
    }
]
 
def encode_image(image_path: str) -> tuple[str, str]:
    """Encode image to base64 and determine MIME type"""
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")
 
    extension = Path(image_path).suffix.lower()
    mime_types = {
        ".jpg": "image/jpeg",
        ".jpeg": "image/jpeg",
        ".png": "image/png",
        ".gif": "image/gif",
        ".webp": "image/webp"
    }
    mime_type = mime_types.get(extension, "image/jpeg")
 
    return image_data, mime_type
 
async def process_invoice_stream(image_path: str) -> dict:
    """
    Process invoice image with streaming.
    Outputs are generated and tool calls are collected as they arrive.
    """
 
    image_data, mime_type = encode_image(image_path)
 
    print(f"\n📄 Processing: {Path(image_path).name}")
    print("=" * 70)
 
    accumulated_text = ""
    tool_calls = []
 
    # Stream the response
    with client.messages.stream(
        model="claude-opus-4-6",
        max_tokens=1024,
        tools=TOOLS,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": mime_type,
                            "data": image_data
                        }
                    },
                    {
                        "type": "text",
                        "text": """Analyze this invoice image and:
 
1. Extract: invoice number, amount (USD), invoice date, vendor name
2. Use record_invoice to save to our system
3. If amount exceeds $10,000 or vendor is unknown, use flag_for_review
4. Send a summary to #accounting via Slack
 
Be thorough but efficient."""
                    }
                ]
            }
        ]
    ) as stream:
        for event in stream:
            # Collect text output
            if hasattr(event, "delta") and hasattr(event.delta, "text"):
                text = event.delta.text
                accumulated_text += text
                print(text, end="", flush=True)
 
            # Collect tool uses
            if hasattr(event, "content"):
                for block in event.content:
                    if hasattr(block, "type") and block.type == "tool_use":
                        tool_calls.append(block)
 
    print("\n")
 
    # Execute collected tool calls
    tool_results = execute_tools(tool_calls)
 
    return {
        "image": Path(image_path).name,
        "reasoning": accumulated_text,
        "tools_executed": tool_results,
        "timestamp": datetime.now().isoformat()
    }
 
def execute_tools(tool_calls: list) -> list:
    """Execute the agent's tool calls"""
 
    results = []
 
    for tool in tool_calls:
        print(f"🔧 Executing: {tool.name}")
        print(f"   Input: {json.dumps(tool.input, indent=2)}")
 
        # Simulate tool execution
        if tool.name == "record_invoice":
            result = f"✅ Recorded invoice {tool.input['invoice_number']} " \
                    f"for ${tool.input['amount']:,.2f}"
 
        elif tool.name == "flag_for_review":
            result = f"🚩 Flagged {tool.input['invoice_number']} " \
                    f"({tool.input['reason']})"
 
        elif tool.name == "notify_slack":
            result = f"📢 Slack notification sent to {tool.input['channel']}"
 
        else:
            result = f"Unknown tool: {tool.name}"
 
        results.append({"tool": tool.name, "result": result})
        print(f"   {result}\n")
 
    return results
 
# Usage
if __name__ == "__main__":
    import asyncio
 
    test_image = "sample_invoice.jpg"
 
    if Path(test_image).exists():
        result = asyncio.run(process_invoice_stream(test_image))
        print("\nFinal Result:")
        print(json.dumps(result, indent=2))
    else:
        print(f"⚠️  Test image '{test_image}' not found")

#!/usr/bin/env python3
# chart_anomaly_detector.py
 
import anthropic
import base64
import json
import os
from pathlib import Path
from datetime import datetime
import subprocess
 
client = anthropic.Anthropic()
 
async def detect_anomalies(image_path: str, historical_data: dict) -> dict:
    """
    Analyze chart image for anomalies using Vision.
 
    Args:
        image_path: Path to dashboard screenshot or chart
        historical_data: Reference baseline (e.g., {"avg": 5000, "stddev": 250})
    """
 
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")
 
    print(f"🔍 Analyzing chart: {Path(image_path).name}")
 
    # Call Claude with Vision
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data
                        }
                    },
                    {
                        "type": "text",
                        "text": f"""Analyze this chart and return JSON:
 
{{
    "metric_name": "what metric is shown?",
    "current_value": numeric value,
    "previous_value": previous data point,
    "change_percent": (current - previous) / previous * 100,
    "trend": "up" | "down" | "stable",
    "data_quality": "complete" | "incomplete" | "missing",
    "anomaly_detected": true | false,
    "anomaly_type": "spike | drop | reversal | missing_data",
    "confidence": 0.0 to 1.0,
    "recommendation": "action to take"
}}
 
Reference baseline:
{json.dumps(historical_data)}
 
Flag as anomaly if:
- Change exceeds ±25%
- Trend reverses (was up, now down)
- Data is incomplete
- Confidence < 0.8 (may indicate chart issue)"""
                    }
                ]
            }
        ]
    )
 
    # Parse JSON response
    response_text = response.content[0].text
 
    # Extract JSON (handle markdown code blocks)
    import re
    json_match = re.search(r'```json\n(.*?)\n```', response_text, re.DOTALL)
    if json_match:
        analysis = json.loads(json_match.group(1))
    else:
        analysis = json.loads(response_text)
 
    print(f"✅ Analysis complete: {analysis['metric_name']}")
 
    # Act on anomalies
    if analysis["anomaly_detected"]:
        severity = "critical" if abs(analysis["change_percent"]) > 50 else "warning"
        await notify_slack_anomaly(analysis, severity)
 
    return analysis
 
async def notify_slack_anomaly(analysis: dict, severity: str):
    """Send formatted Slack notification"""
 
    emoji = {"critical": "🔴", "warning": "🟡", "info": "🟢"}[severity]
    color = {"critical": "danger", "warning": "warning", "info": "good"}[severity]
 
    message = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{emoji} Anomaly Detected"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Metric*\n{analysis['metric_name']}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Change*\n{analysis['change_percent']:+.1f}%"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Type*\n{analysis['anomaly_type']}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Confidence*\n{analysis['confidence']:.0%}"
                    }
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Recommended Action*\n{analysis['recommendation']}"
                }
            }
        ]
    }
 
    webhook = os.environ.get("SLACK_WEBHOOK_URL")
    if webhook:
        subprocess.run([
            "curl", "-X", "POST", webhook,
            "-H", "Content-Type: application/json",
            "-d", json.dumps(message)
        ], capture_output=True)
 
# Example usage
if __name__ == "__main__":
    import asyncio
 
    baseline = {
        "metric": "Daily Revenue",
        "avg": 15000,
        "stddev": 2000,
        "normal_range": [13000, 17000]
    }
 
    result = asyncio.run(detect_anomalies("dashboard.png", baseline))
    print(json.dumps(result, indent=2))

from PIL import Image
import os
 
def optimize_for_vision(image_path: str, target_size: int = 768) -> str:
    """
    Compress and resize image for Vision API while preserving information.
 
    Reduces costs by:
    - Lowering token count (smaller images = fewer tokens)
    - Faster processing
    - Smaller storage footprint
    """
 
    img = Image.open(image_path)
 
    # Resize maintaining aspect ratio
    img.thumbnail((target_size, target_size))
 
    # Save as optimized JPEG (quality 80% is usually imperceptible)
    output = image_path.replace(".", "_opt.")
    img.save(output, "JPEG", quality=80, optimize=True)
 
    original_kb = os.path.getsize(image_path) / 1024
    optimized_kb = os.path.getsize(output) / 1024
    savings = (1 - optimized_kb / original_kb) * 100
 
    print(f"✅ Optimized: {original_kb:.1f}KB → {optimized_kb:.1f}KB ({savings:.0f}% savings)")
 
    return output

def batch_process_documents(image_paths: list, batch_size: int = 3) -> list:
    """
    Process multiple documents efficiently.
 
    Trade-off: Processing 3 images together costs less than 3 separate calls,
    but provides slightly less focused analysis per image.
    Sweet spot: 3-5 images per batch.
    """
 
    batches = [
        image_paths[i:i + batch_size]
        for i in range(0, len(image_paths), batch_size)
    ]
 
    results = []
    for batch in batches:
        # Single API call for multiple images
        result = process_batch_of_images(batch)
        results.extend(result)
 
    return results
 
def process_batch_of_images(images: list) -> list:
    """Process multiple images in one API call"""
 
    content = []
 
    # Add images
    for img_path in images:
        with open(img_path, "rb") as f:
            img_data = base64.b64encode(f.read()).decode()
 
        content.append({
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/png",
                "data": img_data
            }
        })
 
    # Add instruction
    content.append({
        "type": "text",
        "text": "Analyze each image and extract key information. Return JSON array."
    })
 
    # Single API call processes all images
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": content}]
    )
 
    return json.loads(response.content[0].text)

import hashlib
 
CACHE_DIR = Path("/tmp/vision_cache")
CACHE_EXPIRY = 86400  # 24 hours
 
def get_cached_analysis(image_path: str) -> Optional[dict]:
    """
    Return cached Vision analysis if available and fresh.
 
    This prevents re-analyzing identical images, which saves significantly
    when processing repeated screenshots (dashboards, etc.).
    """
 
    # Generate cache key from image content
    with open(image_path, "rb") as f:
        image_hash = hashlib.sha256(f.read()).hexdigest()
 
    cache_file = CACHE_DIR / f"{image_hash}.json"
 
    # Check if cache is fresh
    if cache_file.exists():
        age = time.time() - cache_file.stat().st_mtime
 
        if age < CACHE_EXPIRY:
            with open(cache_file) as f:
                print(f"📦 Using cached analysis (age: {age/3600:.1f}h)")
                return json.load(f)
 
    return None
 
def save_analysis_cache(image_path: str, analysis: dict):
    """Save Vision analysis results to cache"""
 
    with open(image_path, "rb") as f:
        image_hash = hashlib.sha256(f.read()).hexdigest()
 
    cache_file = CACHE_DIR / f"{image_hash}.json"
    cache_file.parent.mkdir(exist_ok=True)
 
    with open(cache_file, "w") as f:
        json.dump(analysis, f)

import time
from functools import wraps
from typing import Callable, TypeVar
 
T = TypeVar('T')
 
def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
    """
    Retry decorator with exponential backoff.
 
    Handles:
    - RateLimitError (429): Use exponential backoff
    - APIError (5xx): Retry with increasing delay
    - APIConnectionError: Network issues
    - ValidationError (400): Don't retry (user input error)
    """
 
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            delay = base_delay
 
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
 
                except anthropic.RateLimitError as e:
                    if attempt == max_retries - 1:
                        raise
 
                    print(f"⏳ Rate limited. Waiting {delay:.1f}s...")
                    time.sleep(delay)
                    delay *= 2
 
                except anthropic.APIError as e:
                    if attempt == max_retries - 1:
                        raise
 
                    if e.status_code >= 500:
                        print(f"❌ Server error ({e.status_code}). Retrying in {delay:.1f}s...")
                        time.sleep(delay)
                        delay *= 2
                    else:
                        raise  # Don't retry client errors
 
                except anthropic.APIConnectionError as e:
                    if attempt == max_retries - 1:
                        raise
 
                    print(f"🌐 Connection error. Retrying in {delay:.1f}s...")
                    time.sleep(delay)
                    delay *= 1.5
 
            raise RuntimeError(f"Failed after {max_retries} attempts")
 
        return wrapper
 
    return decorator
 
# Usage
@retry_with_backoff(max_retries=3, base_delay=1.0)
def call_vision_api(image_path: str) -> dict:
    """Resilient Vision API call"""
 
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
 
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": [{
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": image_data
                }
            }]
        }]
    )
 
    return json.loads(response.content[0].text)

# ❌ Too broad (dangerous)
TOOLS = [
    "execute_any_code",
    "access_any_database",
    "send_email_to_anyone"
]
 
# ✅ Scoped properly (safe)
TOOLS = [
    "add_invoice_to_accounting_sheet",
    "flag_duplicate_invoices_only",
    "notify_accounting_channel_on_slack"
]

def validate_tool_result(tool_name: str, result: dict) -> bool:
    """Ensure tool results are well-formed"""
 
    if tool_name == "record_invoice":
        required_fields = ["invoice_id", "amount", "timestamp"]
        return all(field in result for field in required_fields)
 
    if tool_name == "notify_slack":
        return result.get("status") == "sent"
 
    return True

def log_agent_execution(
    agent_name: str,
    input_image: str,
    analysis: dict,
    tools_called: list,
    api_tokens: dict
):
    """Log all agent activity for auditing and cost tracking"""
 
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "agent": agent_name,
        "input": {
            "image_size_kb": os.path.getsize(input_image) / 1024,
            "image_hash": hashlib.sha256(
                open(input_image, "rb").read()
            ).hexdigest()
        },
        "analysis_summary": {
            "fields_extracted": len(analysis),
            "confidence_avg": sum(
                v.get("confidence", 1.0) for v in analysis.values()
            ) / max(len(analysis), 1)
        },
        "tools_executed": [
            {"name": t.name, "status": "success"} for t in tools_called
        ],
        "api_usage": {
            "input_tokens": api_tokens.get("input_tokens"),
            "output_tokens": api_tokens.get("output_tokens"),
            "cost_usd": api_tokens.get("cost_usd")
        }
    }
 
    # Save to database or log file
    with open(f"/var/log/agents/{agent_name}.jsonl", "a") as f:
        f.write(json.dumps(log_entry) + "\n")