Batch Processing API: Scaling for High Volume

Overview

Batch APIs process multiple requests asynchronously at 50% discount.

Perfect for agents analyzing reports, generating content, or processing data.

---

When to Use Batch

Batch vs Real-Time Tradeoff

Factor	Real-Time API	Batch API
Latency	1-10 seconds	Hours to days
Cost	Full price	50% discount
Volume	Single requests	Bulk processing
Use Case	Chat, live	Reports, analysis

---

Claude Batch API

Submitting Batch Requests

from anthropic import Anthropic
import json

class BatchProcessing:
    def __init__(self):
        self.client = Anthropic()

    def create_batch_request(self, requests: list[dict]):
        """Submit multiple requests as batch"""

        # Format requests for batch
        batch_requests = []

        for i, request in enumerate(requests):
            batch_requests.append({
                "custom_id": f"request_{i}",
                "params": {
                    "model": "claude-3-5-sonnet-20241022",
                    "max_tokens": 1024,
                    "messages": [
                        {"role": "user", "content": request["prompt"]}
                    ]
                }
            })

        # Submit batch
        batch = self.client.messages.batches.create(
            requests=batch_requests
        )

        return batch.id

    def check_batch_status(self, batch_id: str):
        """Check if batch completed"""

        batch = self.client.messages.batches.retrieve(batch_id)

        return {
            "status": batch.processing_status,
            "succeeded": batch.request_counts.succeeded,
            "failed": batch.request_counts.errored,
            "pending": batch.request_counts.processing
        }

    def get_batch_results(self, batch_id: str):
        """Retrieve results after completion"""

        batch = self.client.messages.batches.retrieve(batch_id)

        if batch.processing_status != "completed":
            return None  # Not done yet

        results = {}

        for result in batch.results:
            results[result.custom_id] = {
                "output": result.result.message.content[0].text,
                "usage": result.result.usage
            }

        return results

---

Use Cases for Batch

Case 1: Report Generation

class BatchReportGeneration:
    """Generate reports for 1000 products"""

    def generate_reports_batch(self, products: list[dict]):
        """Batch generate product reports"""

        requests = []

        for product in products:
            requests.append({
                "prompt": f"""Generate marketing report for:
                Product: {product['name']}
                Category: {product['category']}
                Price: ${product['price']}

                Include: Overview, target market, positioning"""
            })

        # Submit batch
        batch_id = self.create_batch_request(requests)

        # Wait for completion (async)
        # Check status in background

        return batch_id

Case 2: Content Moderation

class BatchModeration:
    """Moderate 10k user comments"""

    def moderate_comments_batch(self, comments: list[str]):
        """Batch moderate comments"""

        requests = []

        for i, comment in enumerate(comments):
            requests.append({
                "prompt": f"""Moderate this comment:
                "{comment}"

                Response: [APPROVED|FLAGGED|REMOVED]
                Reason: [brief explanation]"""
            })

        return self.create_batch_request(requests)

---

Cost Math

Savings Calculation

10,000 requests @ 500 tokens each

REAL-TIME API:
  Cost per request: 500 × $3/1M = $0.0015
  Total: 10,000 × $0.0015 = $15.00

BATCH API:
  Cost per request: 500 × $1.50/1M = $0.00075
  Total: 10,000 × $0.00075 = $7.50

SAVINGS: 50% ($7.50)

Monthly at 1M requests:
  Real-time: $3,000
  Batch: $1,500
  Annual savings: $18,000!

---

Polling for Results

Checking Batch Status

import time

class BatchPolling:
    def __init__(self):
        self.client = Anthropic()

    def wait_for_batch_completion(self, batch_id: str, check_interval=30):
        """Poll until batch completes"""

        while True:
            batch = self.client.messages.batches.retrieve(batch_id)

            status = batch.processing_status

            print(f"Status: {status}")
            print(f"  Succeeded: {batch.request_counts.succeeded}")
            print(f"  Failed: {batch.request_counts.errored}")
            print(f"  Processing: {batch.request_counts.processing}")

            if status == "completed":
                return batch

            # Wait before next check
            time.sleep(check_interval)

---

3 Warnings ⚠️

Warning 1: Latency Mismatch

# ❌ WRONG
# Use batch for real-time needs
batch_id = submit_batch(requests)
result = get_batch_results(batch_id)  # Wait hours!

# User waiting for response!

# ✅ RIGHT
# Use batch only for async work
batch_id = submit_batch(reports)
# Process in background
notify_user_when_done(batch_id)

Warning 2: No Retries

# ❌ WRONG
# Assume all requests succeed
batch = submit_batch(requests)
results = get_results(batch)
# Some failed!

# ✅ RIGHT
# Track failures and retry
batch = submit_batch(requests)
results = get_results(batch)

failed = [r for r in results if r["failed"]]
if failed:
    resubmit_batch(failed)

Warning 3: Cost of Mistakes

# ❌ WRONG
# Submit 1M+ requests carelessly
# Realize mistakes after submission

# ✅ RIGHT
# Validate before batch submit
validate_requests(sample)  # Test first
if valid:
    submit_batch(requests)

---

Last Updated: August 9, 2026