C.W.K.
Stream
Lesson 06 of 06 · published

Files API and Batch API: Async-First Workflows

~16 min · files-api, batch-api, async-workflows

Level 0Observer
0 XP0/64 lessons0/13 achievements
0/150 XP to next level150 XP to go0% complete

Files API: upload once, reference many

The Files API lets you upload PDFs, images, and other documents once, get back a file_id, and reference it from messages.create() instead of re-uploading bytes every call. Useful for long-lived analysis sessions over the same document set.

Batch API: 50% off, 24-hour SLA

The Batch API runs a list of independent requests asynchronously. The trade: 24-hour completion SLA (often much faster) in exchange for ~50% off per-token pricing. Perfect for evaluations, bulk classification, content backfills, and anything that does not need a real-time response.

When neither fits

If you need real-time streaming for an interactive user, neither helps. If you are running 100 evals where the answer can land tomorrow, batch wins on cost. If the same 50MB PDF feeds a hundred chat turns this week, files cuts upload latency to zero on every turn after the first.

Principle: Real-time has a cost. Async workflows are where you reclaim it.

Code

Upload a PDF once and reference it across turns·python
from anthropic import Anthropic

client = Anthropic()

# 1. Upload (once)
uploaded = client.beta.files.upload(
    file=("contract.pdf", open("/path/to/contract.pdf", "rb"), "application/pdf"),
)
print("file id:", uploaded.id)

# 2. Reference by file_id in subsequent calls (no re-upload)
resp = client.beta.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "document", "source": {"type": "file", "file_id": uploaded.id}},
                {"type": "text", "text": "Summarize section 5."},
            ],
        }
    ],
    betas=["files-api-2025-04-14"],
)
Batch 100 classifications for half price·python
# Submit
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"row-{i}",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 32,
                "system": "Reply with one word: positive, negative, neutral.",
                "messages": [{"role": "user", "content": text}],
            },
        }
        for i, text in enumerate(corpus)
    ]
)
print("batch id:", batch.id, "status:", batch.processing_status)

# Poll (in production, use webhooks or scheduled checks)
import time
while True:
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(30)

# Stream results — one JSON line per request
for item in client.messages.batches.results(batch.id):
    print(item.custom_id, item.result.message.content[0].text)

External links

Exercise

Move one bulk job (eval run, daily classification, dataset annotation) to the Batch API. Measure the cost difference and the actual completion time vs the 24-hour SLA.
Hint
Most batches finish well inside an hour — but plan for 24h so you do not block business on the optimistic case.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.