Claude API Messages Batches: Cutting Production Costs by Up to 50% with Async Processing

import anthropic
import json
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# Sample documents to process
documents = [
    {"id": "doc_001", "content": "A detailed document about AI ethics..."},
    {"id": "doc_002", "content": "A technical paper on machine learning..."},
    {"id": "doc_003", "content": "An analysis of startup funding trends..."},
]
 
# Build batch requests
# Each request needs a unique custom_id to match results back to inputs
batch_requests = []
for doc in documents:
    batch_requests.append({
        "custom_id": doc["id"],  # Required: used to match results with inputs
        "params": {
            "model": "claude-sonnet-4-6",
            "max_tokens": 500,
            "messages": [
                {
                    "role": "user",
                    "content": f"Summarize the following document in 3 sentences:\n\n{doc['content']}"
                }
            ]
        }
    })
 
# Submit the batch
batch = client.beta.messages.batches.create(
    requests=batch_requests
)
 
print(f"Batch ID: {batch.id}")
print(f"Status: {batch.processing_status}")
print(f"Request count: {batch.request_counts.processing}")
# Output:
# Batch ID: msgbatch_01HxxxxxxxxxxxxxxxxxxXXXX
# Status: in_progress
# Request count: 3

import time
import anthropic
 
def wait_for_batch_completion(
    client: anthropic.Anthropic,
    batch_id: str,
    poll_interval_seconds: int = 60,
    max_wait_minutes: int = 120,
) -> anthropic.types.beta.BetaMessageBatch:
    """
    Wait for a batch to finish processing.
 
    Args:
        client: Anthropic client instance
        batch_id: The batch to wait for
        poll_interval_seconds: How often to check status
        max_wait_minutes: Maximum time to wait before raising TimeoutError
 
    Returns:
        Completed batch object
 
    Raises:
        TimeoutError: If max_wait_minutes is exceeded
        RuntimeError: If the batch is canceled
    """
    max_attempts = (max_wait_minutes * 60) // poll_interval_seconds
 
    for attempt in range(max_attempts):
        batch = client.beta.messages.batches.retrieve(batch_id)
        status = batch.processing_status
 
        if status == "ended":
            print(f"Batch completed: {batch_id}")
            print(f"  Succeeded: {batch.request_counts.succeeded}")
            print(f"  Errored: {batch.request_counts.errored}")
            print(f"  Expired: {batch.request_counts.expired}")
            return batch
 
        elif status == "canceling":
            raise RuntimeError(f"Batch is being canceled: {batch_id}")
 
        else:
            # Still in_progress — keep waiting
            remaining = batch.request_counts.processing
            print(f"[{attempt + 1}/{max_attempts}] Processing... ({remaining} remaining)")
            time.sleep(poll_interval_seconds)
 
    raise TimeoutError(
        f"Batch did not complete within {max_wait_minutes} minutes: {batch_id}"
    )

import random
 
def wait_for_batch_with_backoff(
    client: anthropic.Anthropic,
    batch_id: str,
    initial_interval: float = 30.0,
    max_interval: float = 300.0,  # 5-minute maximum
    multiplier: float = 1.5,
    jitter: bool = True,
) -> anthropic.types.beta.BetaMessageBatch:
    """
    Poll for batch completion using exponential backoff.
    Jitter prevents thundering herd issues when running multiple batches.
    """
    interval = initial_interval
 
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
 
        if batch.processing_status == "ended":
            return batch
 
        # Add jitter to spread out requests across multiple batch jobs
        wait_time = interval * (0.5 + random.random()) if jitter else interval
        print(f"Next poll in {wait_time:.1f}s...")
        time.sleep(wait_time)
 
        # Increase interval for next check, capped at max_interval
        interval = min(interval * multiplier, max_interval)

def process_batch_results(
    client: anthropic.Anthropic,
    batch_id: str,
    original_documents: dict[str, dict],
) -> list[dict]:
    """
    Retrieve batch results and match them to source data.
 
    The `.results()` method streams results, making it memory-efficient
    even for very large batches.
    """
    results = []
 
    for result in client.beta.messages.batches.results(batch_id):
        custom_id = result.custom_id
        original_doc = original_documents.get(custom_id, {})
 
        if result.result.type == "succeeded":
            message = result.result.message
            content_text = ""
            for block in message.content:
                if block.type == "text":
                    content_text = block.text
                    break
 
            results.append({
                "id": custom_id,
                "status": "success",
                "original": original_doc,
                "output": content_text,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens,
            })
 
        elif result.result.type == "errored":
            error = result.result.error
            results.append({
                "id": custom_id,
                "status": "error",
                "original": original_doc,
                "error_type": error.type,
                "error_message": str(error),
            })
 
        elif result.result.type == "expired":
            results.append({
                "id": custom_id,
                "status": "expired",
                "original": original_doc,
            })
 
    return results
 
# Usage
documents_map = {doc["id"]: doc for doc in documents}
completed_batch = wait_for_batch_completion(client, batch.id)
results = process_batch_results(client, batch.id, documents_map)
 
# Save results
with open("batch_results.json", "w") as f:
    json.dump(results, f, indent=2)
 
success_count = sum(1 for r in results if r["status"] == "success")
error_count = sum(1 for r in results if r["status"] == "error")
print(f"Done: {success_count} succeeded, {error_count} errored")

import logging
from dataclasses import dataclass, field
from datetime import datetime
import anthropic
 
logger = logging.getLogger(__name__)
 
@dataclass
class BatchJob:
    """Tracks state for a single batch job."""
    job_id: str
    batch_id: str | None = None
    status: str = "pending"
    submitted_at: datetime | None = None
    completed_at: datetime | None = None
    request_count: int = 0
    success_count: int = 0
    error_count: int = 0
    metadata: dict = field(default_factory=dict)
 
class BatchJobManager:
    """
    Production-ready batch job manager.
 
    Handles:
    - Automatic splitting of large request sets into multiple batches
    - Progress tracking and logging
    - Cost estimation from results
    """
 
    def __init__(
        self,
        client: anthropic.Anthropic,
        max_requests_per_batch: int = 5000,
    ):
        self.client = client
        self.max_requests_per_batch = max_requests_per_batch
        self.jobs: dict[str, BatchJob] = {}
 
    def submit_large_workload(
        self,
        all_requests: list[dict],
        job_prefix: str = "job",
    ) -> list[BatchJob]:
        """
        Split a large workload across multiple batches automatically.
        Handles workloads exceeding the 10,000-request limit per batch.
        """
        jobs = []
 
        for i in range(0, len(all_requests), self.max_requests_per_batch):
            chunk = all_requests[i:i + self.max_requests_per_batch]
            job_id = f"{job_prefix}_{i // self.max_requests_per_batch:04d}"
 
            try:
                logger.info(f"Submitting batch: {job_id} ({len(chunk)} requests)")
                batch = self.client.beta.messages.batches.create(requests=chunk)
 
                job = BatchJob(
                    job_id=job_id,
                    batch_id=batch.id,
                    status="submitted",
                    submitted_at=datetime.now(),
                    request_count=len(chunk),
                )
                self.jobs[job_id] = job
                jobs.append(job)
                logger.info(f"Batch created: {batch.id}")
 
            except anthropic.APIError as e:
                logger.error(f"Failed to submit batch: {job_id} — {e}")
                job = BatchJob(
                    job_id=job_id,
                    status="failed",
                    metadata={"error": str(e)},
                )
                self.jobs[job_id] = job
                jobs.append(job)
 
        return jobs
 
    def estimate_cost(
        self,
        results: list[dict],
        model: str = "claude-sonnet-4-6",
    ) -> dict:
        """
        Calculate actual cost from batch results using Batches API pricing.
        """
        # Batches API pricing (50% discount applied)
        pricing = {
            "claude-sonnet-4-6": {"input": 1.50, "output": 7.50},
            "claude-haiku-4-5-20251001": {"input": 0.40, "output": 2.00},
            "claude-opus-4": {"input": 7.50, "output": 37.50},
        }
        model_pricing = pricing.get(model, pricing["claude-sonnet-4-6"])
 
        total_in = sum(r.get("input_tokens", 0) for r in results if r["status"] == "success")
        total_out = sum(r.get("output_tokens", 0) for r in results if r["status"] == "success")
 
        return {
            "total_input_tokens": total_in,
            "total_output_tokens": total_out,
            "input_cost_usd": round((total_in / 1_000_000) * model_pricing["input"], 4),
            "output_cost_usd": round((total_out / 1_000_000) * model_pricing["output"], 4),
            "total_cost_usd": round(
                (total_in / 1_000_000) * model_pricing["input"] +
                (total_out / 1_000_000) * model_pricing["output"],
                4
            ),
        }

def create_product_requests(products: list[dict]) -> list[dict]:
    """Build batch requests for product description generation."""
    requests = []
    for product in products:
        requests.append({
            "custom_id": f"product_{product['id']}",
            "params": {
                "model": "claude-haiku-4-5-20251001",  # Cost-effective for high volume
                "max_tokens": 300,
                "messages": [{
                    "role": "user",
                    "content": f"""Write a compelling 150-word product description for:
 
Product: {product['name']}
Category: {product['category']}
Key specs: {', '.join(product.get('specs', []))}
Price: ${product['price']}
 
Focus on benefits, not just features. End with a subtle call to action."""
                }]
            }
        })
    return requests

def create_sentiment_requests(reviews: list[dict]) -> list[dict]:
    """Build batch requests for structured sentiment analysis."""
    requests = []
    for review in reviews:
        requests.append({
            "custom_id": f"review_{review['id']}",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 150,
                "messages": [{
                    "role": "user",
                    "content": f"""Analyze this customer review and respond with JSON only.
 
Review: {review['text']}
 
Required format:
{{
  "sentiment": "positive|neutral|negative",
  "score": 1-5,
  "categories": ["quality", "price", "shipping", "support"],
  "summary": "Under 20 words",
  "requires_action": true|false
}}"""
                }]
            }
        })
    return requests

TARGET_LANGUAGES = [
    ("en", "English"), ("ja", "Japanese"),
    ("zh", "Simplified Chinese"), ("ko", "Korean"), ("es", "Spanish"),
]
 
def create_translation_requests(articles: list[dict], source_lang: str = "en") -> list[dict]:
    """Generate translation requests for all target languages."""
    requests = []
    for article in articles:
        for lang_code, lang_name in TARGET_LANGUAGES:
            if lang_code == source_lang:
                continue
            requests.append({
                "custom_id": f"article_{article['id']}_{lang_code}",
                "params": {
                    "model": "claude-sonnet-4-6",
                    "max_tokens": len(article["content"]) * 2,
                    "messages": [{
                        "role": "user",
                        "content": f"""Translate to {lang_name}. Maintain tone and formatting.
Return translated text only.
 
---
{article['content']}"""
                    }]
                }
            })
    return requests

def retry_failed_requests(
    client: anthropic.Anthropic,
    original_requests: list[dict],
    results: list[dict],
    max_retries: int = 3,
) -> list[dict]:
    """Automatically resubmit failed and expired requests."""
    failed_ids = {
        r["id"] for r in results
        if r["status"] in ("error", "expired")
    }
 
    if not failed_ids:
        return results
 
    retry_requests = [
        req for req in original_requests
        if req["custom_id"] in failed_ids
    ]
 
    logger.warning(f"Retrying {len(retry_requests)} failed requests (max {max_retries} attempts)")
    successful_results = [r for r in results if r["status"] == "success"]
 
    for attempt in range(max_retries):
        if not retry_requests:
            break
 
        batch = client.beta.messages.batches.create(requests=retry_requests)
        wait_for_batch_completion(client, batch.id)
        retry_results = process_batch_results(client, batch.id, {})
 
        successful_results.extend(r for r in retry_results if r["status"] == "success")
 
        still_failed = {r["id"] for r in retry_results if r["status"] in ("error", "expired")}
        retry_requests = [r for r in retry_requests if r["custom_id"] in still_failed]
        logger.info(f"Retry {attempt + 1}/{max_retries}: {len(retry_requests)} remaining")
 
    # Include permanently failed requests in final output
    for req in retry_requests:
        successful_results.append({
            "id": req["custom_id"],
            "status": "permanent_failure",
        })
 
    return successful_results

request_with_cache = {
    "custom_id": "doc_001",
    "params": {
        "model": "claude-sonnet-4-6",
        "max_tokens": 500,
        "system": [
            {
                "type": "text",
                "text": "You are an expert financial analyst. Apply the following evaluation rubric...(long shared instructions)",
                "cache_control": {"type": "ephemeral"}
            }
        ],
        "messages": [
            {"role": "user", "content": "Analyze this document..."}
        ]
    }
}

def adaptive_polling(client, batch_id, base_interval=60, fast_interval=15):
    """Switch to a tighter poll interval once the batch is 95%+ done."""
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            return batch
 
        rc = batch.request_counts
        total = rc.processing + rc.succeeded + rc.errored + rc.expired
        done = rc.succeeded + rc.errored + rc.expired
        progress = done / total if total else 0
 
        interval = fast_interval if progress >= 0.95 else base_interval
        time.sleep(interval)

[project]
dependencies = [
  "anthropic>=0.40.0,<1.0.0",  # Pin minor while Batches Beta stabilizes
]