Batch API Workflows in Production

What batch is really for

The Batch API is the right tool for: nightly classification, periodic backfills, large evaluation runs, dataset annotation, content audits. Anything where 'the answer can land within 24 hours' is true. The trade is ~50% off per-token pricing for that flexibility.

Custom IDs are how you re-key results

Each batch request takes a custom_id. The results stream back keyed by that id, in arbitrary order. Map results back to your domain rows by custom_id, not by position. Treat the batch input list and output stream as two independent collections joined on the id.

Polling, webhooks, and idempotent retry

For light workloads, polling every 30-60 seconds is fine. For production, prefer webhooks (when offered) or scheduled checks. Always make the result-processing step idempotent — re-pulling a finished batch should not double-write your downstream store.

Principle: Batch is not 'slow API'. It is 'cheaper API for work that can wait'. Use the difference, do not pretend it is the same as Messages.

Code

Submit, poll, write idempotently·python

from anthropic import Anthropic
import time, json

client = Anthropic()

batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": row["id"],
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 64,
                "system": "Reply with one word: positive, negative, or neutral.",
                "messages": [{"role": "user", "content": row["text"]}],
            },
        }
        for row in rows
    ]
)

while True:
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(60)

# Idempotent write: upsert by custom_id
for item in client.messages.batches.results(batch.id):
    label = item.result.message.content[0].text.strip()
    db.upsert("sentiment", {"id": item.custom_id, "label": label})

Per-row error handling·python

for item in client.messages.batches.results(batch.id):
    if item.result.type == "succeeded":
        save(item.custom_id, item.result.message)
    elif item.result.type == "errored":
        log.warning("batch row failed", id=item.custom_id, error=item.result.error)
        retry_queue.add(item.custom_id)

Batch API Workflows in Production

What batch is really for

Custom IDs are how you re-key results

Polling, webhooks, and idempotent retry

Code

External links

Exercise

Progress

Comments 0