How We Cut Our Nightly Document Pipeline Costs in Half with the Claude Batch API

If you've ever paused at your monthly Claude API bill, this post might help. I started out processing thousands of contracts with synchronous API calls, only to watch the charges blow past my estimates. The root cause was architectural: I was assuming every request needed an immediate response, when in reality most document processing didn't need an answer right away.

The Claude Message Batches API is an asynchronous batch processing interface that handles up to 10,000 requests at once at 50% lower cost compared to the standard API. The tradeoff is simple: results come back within 24 hours in exchange for half the price. Combine that with prompt caching and you get additional discounts on the input token side.

This API isn't the right fit for every situation. But for workloads like contract extraction, invoice parsing, or overnight content classification — where results just need to be ready by end of day — the story changes entirely. Let's walk through the prerequisites and the actual architectural design.

How the Batch API Works

Core Flow

The Batch API is straightforward. Submit a bundle of requests to POST /v1/messages/batches and a batch_id comes back immediately. Actual processing happens asynchronously on Anthropic's side, and there are two ways to check for completion.

One thing worth noting: webhooks only deliver a completion notification. The actual results still need to be downloaded separately via GET /v1/messages/batches/{id}/results. It's easy to assume the webhook payload contains the results — knowing this upfront saves wasted effort.

Each request is identified by a custom_id. When results come in, this ID is used to map responses back to their originating requests. In a multi-tenant environment, prefixing IDs like tenant_a::doc_001 makes separating results much easier.

How the Cost Levers Stack

Prompt caching and batch discounts apply independently. If your pipeline repeatedly uses the same system prompt or document template, both discounts apply simultaneously.

There's a common misconception here. You'll often see people multiply the 50% batch discount by the cache hit discount and claim "5–10% of standard cost," but the cache discount applies only to input tokens. For workloads like contract parsing — where the system prompt is thousands of tokens and the output is a short JSON of a few hundred tokens — the effective savings are substantial. For workloads with long outputs, the effect is less dramatic. I'd recommend calculating your own input/output ratio first.

Deciding Whether a Workload Is a Good Fit

Each request within a batch is fully isolated. Sequential agent workflows where the output of one request feeds into the next cannot be handled by the Batch API. Think of this less as a limitation and more as a design principle.

FastAPI + PostgreSQL + Batch API Pipeline Example

The code below uses the anthropic Python SDK and should be treated as a conceptual reference. SDK internal type paths change frequently across versions, so for production use, run pip show anthropic on your installed version and adjust import paths accordingly. Redis client code is written against redis.asyncio (redis-py 4.2+).

Accepting Requests and Decoupling the Job Queue

When a user uploads a document, return a job_id immediately and hand off the actual processing to the background.

from fastapi import FastAPI
from uuid import uuid4
import json
 
app = FastAPI()
 
@app.post("/documents/process")
async def submit_document(document: DocumentRequest):
    job_id = str(uuid4())
 
    await db.execute(
        "INSERT INTO processing_jobs (job_id, status, document_content) "
        "VALUES ($1, 'pending', $2)",
        job_id, document.content
    )
 
    await redis.rpush("batch_queue", json.dumps({
        "job_id": job_id,
        "content": document.content,
        "custom_id": f"job::{job_id}"
    }))
 
    return {"job_id": job_id, "status": "queued"}

The key idea is accumulation. Instead of submitting immediately per request, you wait until a certain number of items have queued up or a certain amount of time has passed, then submit them all as a single batch.

Batch Submission Worker

In an async environment, you must use the AsyncAnthropic client. Calling the synchronous client inside async def blocks the event loop, and async for won't work when iterating results either.

import anthropic
import asyncio
import json
 
client = anthropic.AsyncAnthropic()
 
BATCH_SIZE = 500
POLL_INTERVAL_SEC = 300
 
async def drain_queue_atomically(count: int) -> list[str]:
    # LRANGE + LTRIM are not atomic — race conditions can cause duplicates or dropped items
    # In production, use the Lua script below or BLMOVE for atomic handling
    lua = """
    local items = redis.call('LRANGE', KEYS[1], 0, tonumber(ARGV[1]) - 1)
    if #items > 0 then
        redis.call('LTRIM', KEYS[1], #items, -1)
    end
    return items
    """
    return await redis.eval(lua, 1, "batch_queue", BATCH_SIZE)
 
async def submit_batch_worker():
    while True:
        pending = await drain_queue_atomically(BATCH_SIZE)
 
        if not pending:
            await asyncio.sleep(60)
            continue
 
        requests = []
        job_ids = []
 
        for raw in pending:
            data = json.loads(raw)
            job_ids.append(data["job_id"])
 
            requests.append({
                "custom_id": data["custom_id"],
                "params": {
                    "model": "claude-sonnet-4-6",
                    "max_tokens": 4096,
                    "system": [{
                        "type": "text",
                        "text": CONTRACT_EXTRACTION_PROMPT,
                        "cache_control": {"type": "ephemeral"}
                    }],
                    "messages": [{
                        "role": "user",
                        "content": data["content"]
                    }]
                }
            })
 
        batch = await client.beta.messages.batches.create(requests=requests)
 
        await db.executemany(
            "UPDATE processing_jobs SET batch_id = $1, status = 'submitted' "
            "WHERE job_id = $2",
            [(batch.id, jid) for jid in job_ids]
        )
 
        print(f"Batch submitted: {batch.id}, request count: {len(requests)}")
        await asyncio.sleep(POLL_INTERVAL_SEC)

The atomic queue draining via Lua is important. If you call LRANGE and then LTRIM separately, new items can arrive between the two calls or another worker can read the same range, causing duplicates or dropped items. Use EVAL to handle it atomically, or consider using streams (XADD/XREADGROUP) instead of a list.

Also worth noting: the cache_control attached to the system prompt. When the same system prompt is reused across hundreds of requests in a batch, cache hits occur and input token costs drop significantly.

Polling for Results and Writing to the DB

async def poll_batch_results():
    active_batches = await db.fetch(
        "SELECT DISTINCT batch_id FROM processing_jobs WHERE status = 'submitted'"
    )
 
    for row in active_batches:
        batch_id = row["batch_id"]
        batch = await client.beta.messages.batches.retrieve(batch_id)
 
        if batch.processing_status != "ended":
            continue
 
        async for result in await client.beta.messages.batches.results(batch_id):
            job_id = result.custom_id.replace("job::", "")
 
            if result.result.type == "succeeded":
                extracted = result.result.message.content[0].text
 
                await db.execute(
                    "UPDATE processing_jobs SET status = 'completed', result = $1 "
                    "WHERE job_id = $2",
                    extracted, job_id
                )
 
                await db.execute(
                    "SELECT pg_notify('job_completed', $1)",
                    json.dumps({"job_id": job_id})
                )
 
            elif result.result.type == "errored":
                await db.execute(
                    "UPDATE processing_jobs SET status = 'failed' WHERE job_id = $1",
                    job_id
                )

The result stream is consumed via iteration, so there's no need to load tens of thousands of results into memory at once.

When You Need Large Outputs

As of 2026, beta features that increase the output token limit for batch requests are rolling out gradually. However, the exact beta header names and limits change over time, so before using them in production, check the official release notes and use the headers exactly as documented. Conceptually, you specify beta headers like this:

# Conceptual example — verify actual beta header name and supported limits in official docs
client = anthropic.AsyncAnthropic(
    default_headers={"anthropic-beta": "<latest-long-output-beta-header>"}
)

This is useful for pipelines that need 128k or more tokens of output, such as generating long-form reports or rewriting full contract documents. Since this is a beta feature, always validate actual behavior in staging before moving to production.

Tradeoffs

At a Glance

Common Mistakes in Practice

1. Setting batch size too small Submitting batches of 100 requests frequently diminishes the benefit of the Batch API. Accumulate hundreds to thousands of requests before submitting when possible.

2. Delaying result storage Batch results are automatically deleted after 29 days. You must include logic to immediately move results to your own storage (S3, GCS, DB) as soon as a batch reaches the ended state.

3. Retrying the entire batch when some requests fail Batch results come back as a mix of succeeded and errored items. Only the failed items should be retried; resubmitting the entire batch charges you twice for the ones that already succeeded.

4. Assuming results are immediately available after a webhook arrives A webhook is just a completion notification — result data must be fetched via a separate API call. A common failure pattern is trying to parse results directly in the webhook handler.

5. Choosing the wrong first workload Trying to convert a user-facing pipeline to batch processing from the start will break UX because of the 24-hour latency. Start by migrating workloads that are already acceptable to run overnight — metadata extraction from documents accumulated the previous day, scheduled content classification, report generation.

Building Durable Workflows with Temporal

A simple polling script loses state if the server restarts. In production, you can use Temporal to achieve automatic crash recovery and idempotency.

Each Activity is automatically retried on failure, and batch_id is durably persisted in workflow state — so even if the worker dies, it can pick up right where it left off after restarting.

Adoption Checklist

If you've decided to actually adopt the Batch API, working through the following in order will help reduce trial and error.

• Have you measured your current workload's input-to-output token ratio? The higher the input share, the greater the caching savings.
• Does the target workload satisfy independence between requests? If there are sequential dependencies, you need workflow orchestration rather than batch processing.
• Is your UX acceptable if users receive results within 24 hours rather than immediately?
• Will your batch submission trigger prioritize a size threshold or a time threshold?
• Is the process of pulling items from the queue into a batch atomic (Lua/streams for Redis, transactions for a relational DB)?
• Is there logic to immediately transfer results to your own storage when a batch reaches ended (to handle the 29-day deletion policy)?
• Is there a separate path to selectively reprocess only errored items?
• If using webhooks, does your handler download results separately?
• Is batch_id durably persisted to survive worker crashes (Temporal, a DB status column, etc.)?

This formula is useful for getting a quick sense of monthly costs:

Estimated monthly cost ≈
  (monthly input tokens × input rate × (1 - cache hit rate × cache discount rate) × 0.5)
+ (monthly output tokens × output rate × 0.5)

The × 0.5 is the batch discount; caching multiplies only the input term. Plug in your estimated cache hit rate from your logs and you'll get a concrete sense of how much this pipeline actually saves.

References

• Introducing the Message Batches API | Anthropic Blog
• Batch processing | Claude Platform Docs
• Batch processing with Message Batches API | Claude Cookbook
• Claude Batch API in Practice | claudeapi.com
• Optimizing costs with Anthropic's API batching and caching | ai.moda
• Using Anthropic's Message Batches API with Temporal | Steve Kinney
• Anthropic Batch API in Production | Dotzlaw Consulting
• Anthropic API Pricing in 2026 | finout.io
• Claude Cost Optimization 2026 | pecollective.com
• Anthropic Batch API for Asynchronous Multi-Tenant AI Processing | DEV Community