Batch Processing — Message Batches API for High-Volume Work

⚡ Exam Tested 7 min +30 XP

💡

THE ANALOGY

Overnight cloud photo processing vs same-day in-store printing. Overnight is 50% cheaper, handles thousands at once — but results come the next morning. The Batch API is the overnight option: cheaper and scalable, but never for anything time-sensitive.

⚠️ EXAM TRAP — The Wrong Answer People Choose

Using Batch API for anything with a latency requirement. It has NO guaranteed SLA — up to 24 hours. The exam specifically tests that you know not to use it for CI/CD pipeline checks, user-blocking features, or real-time workflows.

KEY POINTS

1 Batch API: 50% cost reduction. Processing time up to 24 hours. No latency SLA.

2 Use for: nightly jobs, non-urgent large volumes, cost-sensitive backfills.

3 Never use for: CI/CD checks, user-blocking features, any latency requirement.

4 custom_id field: YOUR correlation key — essential for partial failure recovery.

5 Partial failure: resubmit only failed requests by custom_id — not the entire batch.

The Batch API Decision Rule

Has a latency requirement?        → Real-time API
User is waiting?                  → Real-time API
CI/CD pipeline step?              → Real-time API (batch can take 24h)

Non-urgent overnight processing?  → Batch API
Volume > 1,000 + cost sensitive?  → Consider Batch API
Backfill operation?               → Batch API

Complete Batch Workflow

import anthropic, time

client = anthropic.Anthropic()

# 1. Build with custom_ids for correlation
requests = [
    {
        "custom_id": f"inv-{inv_id}",         # YOUR key — persists in response
        "params": {
            "model":      "claude-sonnet-4-6",
            "max_tokens": 2048,
            "tools":       [invoice_extraction_tool],
            "tool_choice": {"type": "tool", "name": "extract_invoice"},
            "messages": [{"role": "user", "content": f"Extract:\n{inv_text}"}]
        }
    }
    for inv_id, inv_text in invoices.items()
]

# 2. Submit
batch = client.beta.messages.batches.create(requests=requests)
batch_id = batch.id

# 3. Poll (batch is asynchronous — must check status)
while True:
    status = client.beta.messages.batches.retrieve(batch_id)
    if status.processing_status == "ended":
        break
    time.sleep(60)  # check every minute

# 4. Separate successes from failures
succeeded, failed = [], []
for result in client.beta.messages.batches.results(batch_id):
    inv_id = result.custom_id.replace("inv-", "")
    if result.result.type == "succeeded":
        succeeded.append({"id": inv_id, "data": parse_tool_result(result.result.message)})
    else:
        failed.append({"id": inv_id, "custom_id": result.custom_id,
                        "error": result.result.error.type})

print(f"Succeeded: {len(succeeded)}, Failed: {len(failed)}")

# 5. Resubmit ONLY failed — not the whole batch
if failed:
    failed_ids = {f["custom_id"] for f in failed}
    retry_requests = [r for r in requests if r["custom_id"] in failed_ids]
    retry_batch = client.beta.messages.batches.create(requests=retry_requests)
    print(f"Resubmitting {len(retry_requests)} failed as batch {retry_batch.id}")

Key Takeaways

50% cheaper, up to 24h — non-urgent work only
Never for latency-sensitive — CI/CD, user-blocking, real-time
custom_id is essential — enables partial failure recovery
Resubmit only failures — don’t resubmit successful requests
Poll for completion — asynchronous, not synchronous