Anthropic Message Batches API: High-Volume Agent Task Processing at Lower Cost

TL;DR:

• Anthropic’s Message Batches API processes up to 10,000 Claude requests per batch at up to 50% lower cost than real-time API calls
• Batches run asynchronously: you submit, poll for completion, and retrieve results — no need for concurrent request management
• It’s ideal for agent tasks that don’t require immediate responses: data extraction, document classification, bulk content generation, evaluation runs

If your agent pipeline processes hundreds or thousands of similar tasks, you’re probably overpaying and overcomplicating your concurrency logic. The Message Batches API handles both problems.

What the Batches API Does

The standard Anthropic Messages API is synchronous: you send a request, you wait for a response. For production agent pipelines processing large volumes of work — batch data extraction, document classification, content generation, running evaluation suites — this means managing a pool of concurrent workers, rate limit retries, and error handling across potentially thousands of individual connections.

The Message Batches API changes the shape of the problem. You submit up to 10,000 requests in a single API call, each with its own custom_id for tracking. Anthropic processes the batch within 24 hours (typically much faster for smaller batches). You poll for completion and retrieve results as a JSONL stream — one result per line, keyed to your original custom_id.

Pricing is up to 50% lower per token than the synchronous API. For Claude Sonnet, that’s a meaningful saving on workloads that process millions of tokens per day.

When to Use It

The Batches API is the right choice when:

• Results aren’t needed in real time. Document processing, overnight data extraction runs, batch evaluation, content enrichment.
• You’re processing many structurally similar tasks. The API handles the concurrency; you just compose your requests.
• Cost matters more than latency. Up to 50% token cost reduction for the same model and same results.

It’s the wrong choice when:

• Users are waiting for a response. Interactive chat, streaming output, real-time reasoning — use the synchronous API.
• Tasks are chained and sequential. If step 2 depends on step 1’s output, batching doesn’t help; you need synchronous calls in sequence.

Basic Implementation

import anthropic
import json
import time

client = anthropic.Anthropic()

# Prepare batch requests
documents = [
    {"id": "doc-001", "text": "Invoice #1234 from Acme Corp, £4,500 due 30 days..."},
    {"id": "doc-002", "text": "Contract renewal for SaaS subscription, £12,000/year..."},
    # ... up to 10,000 more
]

requests = [
    anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
        custom_id=doc["id"],
        params={
            "model": "claude-sonnet-4-5",
            "max_tokens": 256,
            "messages": [{
                "role": "user",
                "content": f"""Extract from this document:
- document_type (invoice/contract/other)
- amount (number or null)
- currency (string or null)  
- counterparty (string)
- due_date (ISO date or null)

Return valid JSON only.

Document: {doc["text"]}"""
            }]
        }
    )
    for doc in documents
]

# Submit the batch
batch = client.messages.batches.create(requests=requests)
print(f"Batch submitted: {batch.id}")
print(f"Status: {batch.processing_status}")

Polling for Completion

def wait_for_batch(client, batch_id, poll_interval=30):
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        status = batch.processing_status
        
        print(f"Status: {status} | "
              f"Succeeded: {batch.request_counts.succeeded} | "
              f"Errored: {batch.request_counts.errored} | "
              f"Processing: {batch.request_counts.processing}")
        
        if status == "ended":
            return batch
        
        time.sleep(poll_interval)

batch = wait_for_batch(client, batch.id)

Retrieving Results

Results stream as JSONL — memory-efficient for large batches:

results = {}
errors = []

for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        # Parse the structured output
        content = result.result.message.content[0].text
        try:
            parsed = json.loads(content)
            results[result.custom_id] = parsed
        except json.JSONDecodeError:
            errors.append({
                "id": result.custom_id,
                "error": "JSON parse failed",
                "raw": content
            })
    elif result.result.type == "errored":
        errors.append({
            "id": result.custom_id,
            "error": result.result.error.type
        })

print(f"Extracted {len(results)} documents, {len(errors)} errors")

Agent Pipeline Integration Pattern

For agent workflows that generate tasks dynamically:

class BatchAgentPipeline:
    def __init__(self, client, model="claude-sonnet-4-5"):
        self.client = client
        self.model = model
        self.pending_requests = []
    
    def add_task(self, task_id: str, prompt: str, max_tokens: int = 512):
        self.pending_requests.append(
            anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
                custom_id=task_id,
                params={
                    "model": self.model,
                    "max_tokens": max_tokens,
                    "messages": [{"role": "user", "content": prompt}]
                }
            )
        )
    
    def flush(self) -> dict:
        """Submit all pending tasks as a batch and return results."""
        if not self.pending_requests:
            return {}
        
        batch = self.client.messages.batches.create(
            requests=self.pending_requests
        )
        self.pending_requests = []
        
        # Wait for completion
        while True:
            batch = self.client.messages.batches.retrieve(batch.id)
            if batch.processing_status == "ended":
                break
            time.sleep(15)
        
        # Collect results
        return {
            result.custom_id: result.result.message.content[0].text
            for result in self.client.messages.batches.results(batch.id)
            if result.result.type == "succeeded"
        }

# Usage
pipeline = BatchAgentPipeline(client)

for article_id, article_text in articles.items():
    pipeline.add_task(
        f"summary-{article_id}",
        f"Summarise this article in 2 sentences:\n\n{article_text}"
    )
    pipeline.add_task(
        f"tags-{article_id}",
        f"List 5 relevant tags for this article as a JSON array:\n\n{article_text}"
    )

results = pipeline.flush()

Handling Errors and Partial Failures

Batches can partially succeed. Always check request_counts and handle errored results:

batch = client.messages.batches.retrieve(batch_id)
counts = batch.request_counts

if counts.errored > 0:
    # Collect failed IDs for retry
    failed_ids = []
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "errored":
            error_type = result.result.error.type
            if error_type in ("overloaded_error", "api_error"):
                # Retryable — add back to a new batch
                failed_ids.append(result.custom_id)
            elif error_type == "invalid_request_error":
                # Not retryable — fix the request
                print(f"Invalid request for {result.custom_id}")

Cost Comparison

For a pipeline processing 1,000 documents with Claude Sonnet at ~1,000 input tokens and ~256 output tokens per document:

At scale — 100,000 documents per day — the saving becomes significant. The Batches API also removes the complexity of managing a worker pool, retry logic, and rate limit backoff, which simplifies your pipeline considerably.

For workloads that fit the async constraint, it’s the obvious choice.