LLM Batch Processing with Anthropic Message Batches API

pip install anthropic

import anthropic
import json
import time
from pathlib import Path
 
 
client = anthropic.Anthropic()
 
 
def create_log_analysis_batch(log_entries: list[dict]) -> str:
    """Create a batch request to analyze multiple log entries."""
 
    requests = []
    for i, entry in enumerate(log_entries):
        requests.append({
            "custom_id": f"log-{i}-{entry.get('timestamp', 'unknown')}",
            "params": {
                "model": "claude-haiku-4-5",   # Cheaper model for batch
                "max_tokens": 500,
                "messages": [{
                    "role": "user",
                    "content": f"""Classify this log entry:
 
Log: {entry.get('message', '')}
Level: {entry.get('level', 'unknown')}
Service: {entry.get('service', 'unknown')}
Timestamp: {entry.get('timestamp', 'unknown')}
 
Return JSON: {{"category": "error|warning|info|debug", "root_cause": "brief description", "action_required": true|false, "priority": "critical|high|medium|low"}}"""
                }]
            }
        })
 
    batch = client.beta.messages.batches.create(requests=requests)
    print(f"Batch created: {batch.id}")
    print(f"Requests: {batch.request_counts.processing} processing")
    return batch.id

def wait_for_batch(batch_id: str, poll_interval: int = 60) -> object:
    """Poll until batch completes. Typical wait: 1-24 hours."""
 
    print(f"Waiting for batch {batch_id}...")
 
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
 
        counts = batch.request_counts
        total = counts.processing + counts.succeeded + counts.errored + counts.canceled + counts.expired
        done = counts.succeeded + counts.errored + counts.canceled + counts.expired
 
        print(f"Progress: {done}/{total} | Succeeded: {counts.succeeded} | Errors: {counts.errored}")
 
        if batch.processing_status == "ended":
            print(f"Batch completed!")
            return batch
 
        time.sleep(poll_interval)
 
 
def get_batch_results(batch_id: str) -> dict[str, dict]:
    """Retrieve all results from a completed batch."""
    results = {}
 
    for result in client.beta.messages.batches.results(batch_id):
        custom_id = result.custom_id
 
        if result.result.type == "succeeded":
            response_text = result.result.message.content[0].text
            try:
                # Parse JSON response
                parsed = json.loads(response_text.strip())
                results[custom_id] = {"status": "success", "data": parsed}
            except json.JSONDecodeError:
                results[custom_id] = {"status": "success", "data": {"raw": response_text}}
 
        elif result.result.type == "errored":
            error = result.result.error
            results[custom_id] = {
                "status": "error",
                "error_type": error.type,
                "error": str(error)
            }
 
        elif result.result.type == "expired":
            results[custom_id] = {"status": "expired"}
 
    return results

import boto3
from datetime import datetime
 
 
def audit_security_groups_batch():
    """Analyze all AWS Security Groups overnight for policy violations."""
    ec2 = boto3.client("ec2", region_name="ap-south-1")
 
    # Get all security groups
    sgs = ec2.describe_security_groups()["SecurityGroups"]
    print(f"Auditing {len(sgs)} security groups...")
 
    # Create batch request — each SG analyzed by Claude
    requests = []
    for sg in sgs:
        sg_summary = json.dumps({
            "GroupId": sg["GroupId"],
            "GroupName": sg["GroupName"],
            "Description": sg["Description"],
            "InboundRules": sg.get("IpPermissions", [])[:10],
            "OutboundRules": sg.get("IpPermissionsEgress", [])[:5]
        }, indent=2)
 
        requests.append({
            "custom_id": sg["GroupId"],
            "params": {
                "model": "claude-haiku-4-5",
                "max_tokens": 800,
                "messages": [{
                    "role": "user",
                    "content": f"""You are a cloud security auditor. Analyze this AWS Security Group:
 
{sg_summary}
 
Identify security issues and return JSON:
{{
  "risk_level": "critical|high|medium|low|info",
  "issues": [
    {{"rule": "description", "risk": "what this enables", "fix": "how to fix"}}
  ],
  "compliant": true|false,
  "summary": "one sentence"
}}
 
Flag: open ports to 0.0.0.0/0 (especially SSH 22, RDP 3389, all traffic), overly permissive rules, and unnecessary exposure."""
                }]
            }
        })
 
    # Submit batch
    batch = client.beta.messages.batches.create(requests=requests)
    batch_id = batch.id
 
    # Save batch ID for later retrieval (process async)
    Path("audit_batch_id.txt").write_text(batch_id)
    print(f"Batch submitted: {batch_id}")
    print("Results available in ~1-4 hours. Run retrieve_audit_results.py tomorrow.")
 
    return batch_id
 
 
def retrieve_audit_results(batch_id: str):
    """Run this the next morning to get results."""
    batch = client.beta.messages.batches.retrieve(batch_id)
 
    if batch.processing_status != "ended":
        print(f"Still processing: {batch.request_counts.processing} remaining")
        return
 
    results = get_batch_results(batch_id)
 
    # Generate report
    critical = []
    high = []
    medium = []
 
    for sg_id, result in results.items():
        if result["status"] != "success":
            continue
        data = result["data"]
        risk = data.get("risk_level", "unknown")
        if risk == "critical":
            critical.append({"sg": sg_id, **data})
        elif risk == "high":
            high.append({"sg": sg_id, **data})
        elif risk == "medium":
            medium.append({"sg": sg_id, **data})
 
    # Print report
    print(f"\n{'='*60}")
    print(f"SECURITY GROUP AUDIT REPORT — {datetime.utcnow().strftime('%Y-%m-%d')}")
    print(f"{'='*60}")
    print(f"Analyzed: {len(results)} security groups")
    print(f"Critical: {len(critical)} | High: {len(high)} | Medium: {len(medium)}")
 
    if critical:
        print(f"\n🔴 CRITICAL ISSUES ({len(critical)} security groups):")
        for item in critical:
            print(f"\n  {item['sg']}: {item.get('summary', '')}")
            for issue in item.get("issues", [])[:2]:
                print(f"    - {issue.get('rule', '')}: {issue.get('fix', '')}")
 
    # Save full report
    with open("security_audit_report.json", "w") as f:
        json.dump({"critical": critical, "high": high, "medium": medium}, f, indent=2)
 
    print(f"\nFull report saved to security_audit_report.json")
 
 
# Run overnight
if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "retrieve":
        batch_id = Path("audit_batch_id.txt").read_text().strip()
        retrieve_audit_results(batch_id)
    else:
        audit_security_groups_batch()

def estimate_batch_cost(num_requests: int, avg_input_tokens: int = 500, avg_output_tokens: int = 300) -> dict:
    """Estimate batch vs real-time cost."""
 
    # claude-haiku-4-5 pricing (batch is 50% off real-time)
    realtime_input_cost_per_mtok = 1.0    # $1/M tokens
    realtime_output_cost_per_mtok = 5.0   # $5/M tokens
    batch_discount = 0.5
 
    total_input_tokens = num_requests * avg_input_tokens
    total_output_tokens = num_requests * avg_output_tokens
 
    realtime_cost = (
        (total_input_tokens / 1_000_000 * realtime_input_cost_per_mtok) +
        (total_output_tokens / 1_000_000 * realtime_output_cost_per_mtok)
    )
 
    batch_cost = realtime_cost * batch_discount
 
    return {
        "requests": num_requests,
        "realtime_cost_usd": round(realtime_cost, 4),
        "batch_cost_usd": round(batch_cost, 4),
        "savings_usd": round(realtime_cost - batch_cost, 4),
        "savings_pct": "50%"
    }
 
 
# Example
print(estimate_batch_cost(10000))
# {'requests': 10000, 'realtime_cost_usd': 2.0, 'batch_cost_usd': 1.0, 'savings_usd': 1.0, 'savings_pct': '50%'}

# Run audit submission every night at 11 PM
0 23 * * * cd /opt/infra-tools && python audit.py >> /var/log/audit.log 2>&1
 
# Run result retrieval every morning at 7 AM
0 7 * * * cd /opt/infra-tools && python audit.py retrieve >> /var/log/audit.log 2>&1