Batch Processing API

Overview

Batch APIs process multiple requests asynchronously at 50% discount.

Perfect for agents analyzing reports, generating content, or processing data.

---

When to Use Batch

Batch vs Real-Time Tradeoff

Factor	Real-Time API	Batch API
Latency	1-10 seconds	Hours to days
Cost	Full price	50% discount
Volume	Single requests	Bulk processing
Use Case	Chat, live	Reports, analysis

---

Claude Batch API

Submitting Batch Requests

from anthropic import Anthropic
import json

class BatchProcessing:
 def __init__(self):
 self.client = Anthropic()

 def create_batch_request(self, requests: list[dict]):
 """Submit multiple requests as batch"""

 # Format requests for batch
 batch_requests = []

 for i, request in enumerate(requests):
 batch_requests.append({
 "custom_id": f"request_{i}",
 "params": {
 "model": "claude-3-5-sonnet-20241022",
 "max_tokens": 1024,
 "messages": [
 {"role": "user", "content": request["prompt"]}
]
 }
 })

 # Submit batch
 batch = self.client.messages.batches.create(
 requests=batch_requests
)

 return batch.id

 def check_batch_status(self, batch_id: str):
 """Check if batch completed"""

 batch = self.client.messages.batches.retrieve(batch_id)

 return {
 "status": batch.processing_status,
 "succeeded": batch.request_counts.succeeded,
 "failed": batch.request_counts.errored,
 "pending": batch.request_counts.processing
 }

 def get_batch_results(self, batch_id: str):
 """Retrieve results after completion"""

 batch = self.client.messages.batches.retrieve(batch_id)

 if batch.processing_status != "completed":
 return None # Not done yet

 results = {}

 for result in batch.results:
 results[result.custom_id] = {
 "output": result.result.message.content[0].text,
 "usage": result.result.usage
 }

 return results

---

Use Cases for Batch

Case 1: Report Generation

class BatchReportGeneration:
 """Generate reports for 1000 products"""

 def generate_reports_batch(self, products: list[dict]):
 """Batch generate product reports"""

 requests = []

 for product in products:
 requests.append({
 "prompt": f"""Generate marketing report for:
 Product: {product['name']}
 Category: {product['category']}
 Price: ${product['price']}

 Include: Overview, target market, positioning"""
 })

 # Submit batch
 batch_id = self.create_batch_request(requests)

 # Wait for completion (async)
 # Check status in background

 return batch_id

Case 2: Content Moderation

class BatchModeration:
 """Moderate 10k user comments"""

 def moderate_comments_batch(self, comments: list[str]):
 """Batch moderate comments"""

 requests = []

 for i, comment in enumerate(comments):
 requests.append({
 "prompt": f"""Moderate this comment:
 "{comment}"

 Response: [APPROVED|FLAGGED|REMOVED]
 Reason: [brief explanation]"""
 })

 return self.create_batch_request(requests)

-

Cost Math

Savings Calculation

10,000 requests @ 500 tokens each

REAL-TIME API:
 Cost per request: 500 × $3/1M = $0.0015
 Total: 10,000 × $0.0015 = $15.00

BATCH API:
 Cost per request: 500 × $1.50/1M = $0.00075
 Total: 10,000 × $0.00075 = $7.50

SAVINGS: 50% ($7.50)

Monthly at 1M requests:
 Real-time: $3,000
 Batch: $1,500
 Annual savings: $18,000!

---

Polling for Results

Checking Batch Status

import time

class BatchPolling:
 def __init__(self):
 self.client = Anthropic()

 def wait_for_batch_completion(self, batch_id: str, check_interval=30):
 """Poll until batch completes"""

 while True:
 batch = self.client.messages.batches.retrieve(batch_id)

 status = batch.processing_status

 print(f"Status: {status}")
 print(f" Succeeded: {batch.request_counts.succeeded}")
 print(f" Failed: {batch.request_counts.errored}")
 print(f" Processing: {batch.request_counts.processing}")

 if status == "completed":
 return batch

 # Wait before next check
 time.sleep(check_interval)

---

3 Warnings

Warning 1: Latency Mismatch

# WRONG
# Use batch for real-time needs
batch_id = submit_batch(requests)
result = get_batch_results(batch_id) # Wait hours!

# User waiting for response!

# RIGHT
# Use batch only for async work
batch_id = submit_batch(reports)
# Process in background
notify_user_when_done(batch_id)

Warning 2: No Retries

# WRONG
# Assume all requests succeed
batch = submit_batch(requests)
results = get_results(batch)
# Some failed!

# RIGHT
# Track failures and retry
batch = submit_batch(requests)
results = get_results(batch)

failed = [r for r in results if r["failed"]]
if failed:
 resubmit_batch(failed)

Warning 3: Cost of Mistakes

# WRONG
# Submit 1M+ requests carelessly
# Realize mistakes after submission

# RIGHT
# Validate before batch submit
validate_requests(sample) # Test first
if valid:
 submit_batch(requests)

-

Last Updated: August 9, 2026