Many Prompts at Once
Fan out prompts with asyncio, bound concurrency to the rate limit, keep results in order, and let one item fail alone.
results = await asyncio.gather(*(call(p) for p in prompts), return_exceptions=True)A thousand prompts run serially is a thousand round trips end to end. These calls are almost entirely waiting, so concurrency is the fix, and asyncio with the async client is the cheapest form of it: one thread, many in-flight requests.
Bounding the fan-out
Unbounded fan-out does not go faster, it just converts throughput into 429s. An asyncio.Semaphore is the limiter: acquire inside each task, so at most N calls are open at once. Pick N from the published requests-per-minute limit divided by the average call duration, then lower it until the error rate is near zero.
Order and partial failure
asyncio.gather returns results positionally, in the order the awaitables were passed, not the order they completed, so zipping back against the input list is safe. With return_exceptions=True a failure becomes a value in that list instead of an exception that abandons the batch, and you decide per item whether to retry, skip or log it.
Shorthand and longhand
asyncio.gather(*tasks) is the one-liner. The longhand is asyncio.TaskGroup (Python 3.11+), which creates tasks in a with block and cancels the siblings if one raises. Gather with return_exceptions=True is what you want for a batch where items are independent; a TaskGroup is what you want when any failure makes the rest pointless.
What trips people up
If nothing is latency-sensitive, do not fan out at all: the Batches endpoint, client.messages.batches.create(requests=[...]), runs the same work asynchronously at half the price. Its results also come back in arbitrary order, keyed by the custom_id you supplied, so never match them up by position.
Example
When to use it
- Classifying a backlog of ten thousand support tickets
- Re-summarising every document after a prompt change
- Enriching a CSV export without exceeding the per-minute limit
- Keeping one malformed row from aborting an overnight job
More examples
The real async client, bounded
The production shape: AsyncAnthropic, one semaphore shared by every task, and gather collecting results in input order. Note the semaphore is released before the text is assembled, which keeps the slot held only for the duration of the network call.
import asyncio, anthropic
client = anthropic.AsyncAnthropic(max_retries=4)
async def summarise(doc: str, limit: asyncio.Semaphore) -> str:
async with limit:
reply = await client.messages.create(
model="claude-opus-5-5", max_tokens=512,
messages=[{"role": "user", "content": doc}],
)
return "".join(b.text for b in reply.content if b.type == "text")
async def run(docs: list[str]) -> list[str | Exception]:
limit = asyncio.Semaphore(8) # 8 concurrent calls, tuned by observation
return await asyncio.gather(
*(summarise(d, limit) for d in docs), return_exceptions=True
)
summaries = asyncio.run(run(load_docs()))Streaming results as they finish
asyncio.as_completed yields in completion order, which is the opposite trade from gather: you get early results but lose the input ordering, so you have to carry the mapping yourself. Reach for it when the batch is long enough that waiting for the slowest item is unacceptable.
async def as_they_finish(docs, limit):
async def tagged(doc):
try:
return doc, await summarise(doc, limit)
except Exception as exc: # one item failing is not the batch failing
return doc, exc
# Iterated as a plain iterator, as_completed yields fresh coroutines, not
# the tasks you passed in, so the input cannot be recovered by looking the
# yielded object up in a dict. Carry it in the task's own result instead.
tasks = [asyncio.create_task(tagged(d)) for d in docs]
for coro in asyncio.as_completed(tasks):
yield await coro # (doc, summary or exception), as it lands
# Use this when a long batch should write results to disk or to a queue
# incrementally, rather than holding a thousand summaries in memory.Half price, if latency does not matter
The Batches endpoint for overnight work: half the per-token price, no rate-limit choreography, and no concurrency code at all. The cost is latency measured in hours and a polling loop. Results arrive in arbitrary order, so custom_id is how you match a result to its input; matching by position is a bug that will eventually bite.
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
batch = client.messages.batches.create(requests=[
Request(
custom_id=f"doc-{doc.id}", # your key; results are keyed by it
params=MessageCreateParamsNonStreaming(
model="claude-opus-5-5", max_tokens=512,
messages=[{"role": "user", "content": doc.body}],
),
)
for doc in docs
])
while client.messages.batches.retrieve(batch.id).processing_status != "ended":
time.sleep(30)
for entry in client.messages.batches.results(batch.id):
if entry.result.type == "succeeded": # also: errored, canceled, expired
store(entry.custom_id, entry.result.message)
Discussion