Anthropic Batch API for Asynchronous Multi-Tenant AI Processing: Cutting Claude Costs by 50% Without Sacrificing Feature Responsiveness

Anthropic's Batch API can cut Claude AI costs by 50% for asynchronous workloads, according to a developer at CitizenApp. The team reduced their monthly Claude bill from approximately $8,000 to $4,000 by offloading 65% of their document summarization, compliance classification, and policy analysis requests to batch processing. The architecture separates synchronous user-facing features from background jobs, enabling the cost savings without sacrificing responsiveness for requests that need immediate responses.

I'm going to be blunt: if you're running a SaaS that uses Claude, and you haven't implemented the Batch API yet, you're hemorrhaging money. Not metaphorically—literally 50% off the per-token cost for any request that doesn't need a response in the next 60 seconds. At CitizenApp, we process thousands of document summaries, compliance classifications, and policy analyses daily. Our first instinct was synchronous: user uploads document → Claude responds in real-time → dashboard updates. It felt responsive. It also felt like burning cash. Then we realized: 80% of those requests don't actually need synchronous responses. A user uploads a document for summarization? They're fine waiting 5–30 minutes. A tenant wants to bulk-classify 500 policies? That's explicitly an async job. The Batch API is built for this, and it's absurdly underutilized. Here's how I wired it into CitizenApp—and why you should too. The 50% cost reduction is the headline, but the real win is architectural. Batching forces you to separate concerns: This separation is healthy . You stop cramming everything through Claude's sync API and actually think about what needs to be urgent. In CitizenApp's case, we reduced our monthly Claude bill from ~$8k to ~$4k while actually increasing feature coverage. The key was offloading 65% of our workload to batches. Here's the flow: User Request ↓ FastAPI Endpoint Validate, Enqueue ↓ PostgreSQL Queue Table ↓ Batch Processor reads queue, submits to Anthropic ↓ Anthropic Batch Job runs in background ↓ Webhook/Polling Handler gets results ↓ PostgreSQL LISTEN/NOTIFY ↓ WebSocket → React 19 Dashboard real-time update Let's build it. python models.py from sqlalchemy import Column, String, Integer, Text, DateTime, Enum, ForeignKey from sqlalchemy.orm import declarative base from datetime import datetime import enum Base = declarative base class AIJobStatus str, enum.Enum : QUEUED = "queued" SUBMITTED = "submitted" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" class AIJob Base : tablename = "ai jobs" id = Column String, primary key=True tenant id = Column String, ForeignKey "tenants.id" , nullable=False, index=True user id = Column String, nullable=False job type = Column String, nullable=False "summarize", "classify", etc. status = Column Enum AIJobStatus , default=AIJobStatus.QUEUED, index=True input data = Column Text, nullable=False JSON stringified result = Column Text, nullable=True Result from Claude batch id = Column String, nullable=True, index=True Anthropic batch ID request id = Column String, nullable=True Within batch created at = Column DateTime, default=datetime.utcnow, index=True completed at = Column DateTime, nullable=True error message = Column Text, nullable=True python api/ai.py from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel import uuid import json from sqlalchemy.orm import Session from database import get db from models import AIJob, AIJobStatus router = APIRouter class SummarizeRequest BaseModel : document text: str max length: int = 500 @router.post "/api/ai/summarize" async def queue summarize request: SummarizeRequest, db: Session = Depends get db , tenant id: str = Depends get tenant id , user id: str = Depends get user id , : """ Queue a document for summarization async . Returns immediately with job ID. User receives result via WebSocket when batch completes. """ job id = str uuid.uuid4 job = AIJob id=job id, tenant id=tenant id, user id=user id, job type="summarize", input data=json.dumps { "document text": request.document text, "max length": request.max length, } , status=AIJobStatus.QUEUED, db.add job db.commit return { "job id": job id, "status": "queued", "message": "Your request is queued. You'll receive results in 5-30 minutes.", } I prefer a separate worker process via Celery or APScheduler that runs periodically. Here's APScheduler for simplicity: python python workers/batch processor.py from anthropic import Anthropic from sqlalchemy.orm import Session from database import SessionLocal from models import AIJob, AIJobStatus import json from datetime import datetime client = Anthropic def process batch jobs : """ Run every 5 minutes via APScheduler . Collects queued jobs and submits to Anthropic Batch API. """ db = SessionLocal Get all queued jobs batch size: 10k requests per batch max queued jobs = db.query AIJob .filter AIJob.status == AIJobStatus.QUEUED .limit 100 .all if not queued jobs: db.close return Build batch request requests = job map = {} for job in queued jobs: input data = json.loads job.input data if job.job type == "summarize": message = f"Summarize the following document in {input data 'max length' } words:\n\n{input data 'document text' }" elif job.job type == "classify": message = f"Classify this text into one of: {input data 'categories' }\n\nText: {input data 'text' }" else: continue request id = job.id job map request id = job.id requests.append { "custom id": request id, "params": { "model": "claude-3-5-sonnet-20241022", "max tokens": 1024, "messages": {"role": "user", "content": message} , }, } Submit batch to Anthropic batch = client.beta.messages.batches.create requests=requests, betas= "batch-2024-09-24" , Mark all jobs as submitted for job in queued jobs: job.status = AIJobStatus.SUBMITTED job.batch id = batch.id db.commit print f"Submitted batch {batch.id} with {len requests } requests" db.close def poll batch results : """ Run every 30 seconds. Checks submitted batches for completion, stores results. """ db = SessionLocal Get all submitted jobs submitted jobs = db.query AIJob .filter AIJob.status == AIJobStatus.SUBMITTED .all batch ids = set job.batch id for job in submitted jobs for batch id in batch ids: batch = client.beta.messages.batches.retrieve batch id, betas= "batch-2024-09-24" if batch.processing status == "in progress": continue if batch.processing status == "expired": Mark jobs as failed for job in submitted jobs: if job.batch id == batch id: job.status = AIJobStatus.FAILED job.error message = "Batch expired" db.commit continue Batch complete — fetch results results = client.beta.messages.batches.results batch id, betas= "batch-2024-09-24" for result in results: request id = result.custom id job = db.query AIJob .filter AIJob.id == request id .first if not job: continue if result.result.type == "succeeded": job.status = AIJobStatus.COMPLETED job.result = result.result.message.content 0 .text elif result.result.type == "errored": job.status = AIJobStatus.FAILED job.error message = result.result.error.message job.completed at = datetime.utcnow db.commit Notify connected clients next step notify clients batch complete batch id db.close