Run Parallel Requests

Why Run Requests in Parallel

Many real tasks involve many independent requests: summarizing 200 support tickets, translating 50 lesson pages, classifying 1,000 reviews, or generating descriptions for a product catalog. Sent one after another, these take a long time. Sent in parallel with the async client, they finish many times faster (Lesson 71).

Parallelism has limits, though. Your organization has rate limits, every request costs money, and some requests fail. Running parallel requests well means managing all three.

Understanding Rate Limits

Your organization's limits depend on its usage tier (Lesson 2) and are shown in the Console under Rate limits. They limit:

• requests per minute, and
• tokens per minute, for input and for output.

If parallel requests exceed a limit, the API returns 429 rate_limit_error. The SDK retries 429 responses automatically (twice by default, Lesson 10), but a sustained overload keeps failing.

The concurrency you can use depends on how long each request takes and how many tokens it uses. As a rough guide:

requests per minute ≈ concurrency × (60 / average seconds per request)

With 10 concurrent requests averaging 6 seconds each, you send about 100 requests per minute. Compare this with your limits, including the token limits, and start lower than the maximum.

Limiting Concurrency with a Semaphore

semaphore = asyncio.Semaphore(5)


async def limited(coroutine_function, *args):
    async with semaphore:
        return await coroutine_function(*args)

All tasks are created at once, but only five at a time run inside async with semaphore:. The rest wait for a free slot (Lesson 74). Changing one number adjusts the pressure you put on the API.

Handling Rate Limits Gracefully

When you do hit a rate limit despite the semaphore, wait and try again. The retry-after response header, when present, says how many seconds to wait:

import random

import anthropic


async def call_with_retry(make_call, attempts: int = 4):
    for attempt in range(1, attempts + 1):
        try:
            return await make_call()
        except anthropic.RateLimitError as error:
            if attempt == attempts:
                raise
            header = error.response.headers.get("retry-after")
            try:
                delay = float(header)
            except (TypeError, ValueError):
                delay = 2 ** attempt
            await asyncio.sleep(delay + random.uniform(0, 1))

• make_call is a function that creates a new coroutine for each attempt. A coroutine can only be awaited once, so retries need a fresh one.
• The retry-after header is used when available; otherwise the delay grows exponentially: 2, 4, 8 seconds.
• random.uniform(0, 1) adds jitter, a small random extra delay, so many waiting tasks do not all retry at exactly the same moment and trigger the limit again.

This adds to the SDK's built-in retries, which handle brief spikes. Lesson 146 covers backoff strategies in depth.

Collecting Results and Errors

For independent items, one failure should not lose the other results. asyncio.gather(..., return_exceptions=True) returns exceptions as results, or each task can catch its own errors and return a result object that records success or failure, as the practical example does.

Parallel Requests vs. Message Batches

For thousands of items that are not needed right away, the Batches API (Lessons 131–140) is usually cheaper and simpler.

Practical Example: A Parallel Ticket Summarizer

This program summarizes a list of support tickets in parallel. It limits concurrency, retries rate-limited requests, shows progress, records failures per ticket, and reports totals at the end.

"""parallel_summarizer.py: summarize many texts in parallel, safely."""

import asyncio
import random
import time
from dataclasses import dataclass

import anthropic
from anthropic import AsyncAnthropic

CONCURRENCY = 5
MODEL = "claude-haiku-5-5"

TICKETS = [
    "I paid for the Django course but it still shows as locked on my dashboard.",
    "The video in lesson 4 stops playing after two minutes on Firefox.",
    "Can I get a certificate if I finish the free internship program?",
    "Quiz 3 marks my correct answer about list slicing as wrong.",
    "How do I change the email address on my account?",
    "The mobile app logs me out every time I switch to another app.",
    "Is there a discount for students in India?",
    "My project submission upload fails with a 'file too large' error at 12 MB.",
]


@dataclass
class Outcome:
    index: int
    summary: str | None
    error: str | None
    output_tokens: int


async def call_with_retry(make_call, attempts: int = 4):
    for attempt in range(1, attempts + 1):
        try:
            return await make_call()
        except anthropic.RateLimitError as error:
            if attempt == attempts:
                raise
            header = error.response.headers.get("retry-after")
            try:
                delay = float(header)
            except (TypeError, ValueError):
                delay = 2 ** attempt
            await asyncio.sleep(delay + random.uniform(0, 1))


async def summarize(client: AsyncAnthropic, semaphore: asyncio.Semaphore,
                    index: int, ticket: str) -> Outcome:
    def make_call():
        return client.messages.create(
            model=MODEL,
            max_tokens=1024,
            messages=[{"role": "user", "content":
                       f"<ticket>\n{ticket}\n</ticket>\n\n"
                       "Summarize this support ticket in at most 10 words, starting with its category "
                       "(billing, technical, account, or question) followed by a colon."}],
        )

    async with semaphore:
        try:
            message = await call_with_retry(make_call)
        except anthropic.APIError as error:
            return Outcome(index, None, type(error).__name__, 0)

    text = "".join(b.text for b in message.content if b.type == "text").strip()
    return Outcome(index, text, None, message.usage.output_tokens)


async def main() -> list[Outcome]:
    semaphore = asyncio.Semaphore(CONCURRENCY)
    outcomes: list[Outcome] = []

    async with AsyncAnthropic() as client:
        tasks = [summarize(client, semaphore, i, t) for i, t in enumerate(TICKETS, start=1)]
        for finished in asyncio.as_completed(tasks):
            outcome = await finished
            outcomes.append(outcome)
            print(f"\rProgress: {len(outcomes)}/{len(TICKETS)}", end="", flush=True)
    print()
    return sorted(outcomes, key=lambda outcome: outcome.index)


if __name__ == "__main__":
    start = time.perf_counter()
    results = asyncio.run(main())
    elapsed = time.perf_counter() - start

    for outcome in results:
        print(f"{outcome.index:>2}. {outcome.summary or 'FAILED: ' + str(outcome.error)}")

    failures = sum(outcome.error is not None for outcome in results)
    tokens = sum(outcome.output_tokens for outcome in results)
    print(f"\n{len(results) - failures} succeeded, {failures} failed, "
          f"{tokens} output tokens, {elapsed:.1f}s with concurrency {CONCURRENCY}")

Example output:

Progress: 8/8
 1. billing: Paid Django course still shows as locked
 2. technical: Lesson 4 video stops after two minutes on Firefox
 3. question: Certificate eligibility for the free internship program
 4. technical: Quiz 3 wrongly marks correct list slicing answer
 5. account: How to change the account email address
 6. technical: Mobile app logs out when switching apps
 7. question: Student discount availability in India
 8. technical: Project upload fails with 12 MB file error

8 succeeded, 0 failed, 167 output tokens, 3.1s with concurrency 5

How It Works

1. MODEL = "claude-haiku-5-5": short, high-volume tasks like ticket summaries are a good fit for a fast, inexpensive model (Lesson 12).
2. summarize() acquires the semaphore before calling the API, so at most CONCURRENCY requests are in flight. The semaphore is released when the async with block ends, even if an error occurs.
3. make_call() returns a new coroutine every time it is called. call_with_retry() calls it again for each attempt, because an awaited coroutine cannot be reused.
4. Errors are caught per ticket, and become an Outcome with an error value, so one failure never stops the others.
5. asyncio.as_completed() delivers outcomes as they finish, so the progress line updates live. \r rewrites the same line.
6. Results are sorted by index at the end, restoring the original order for the report.
7. The summary line reports successes, failures, total output tokens (for cost estimation, Lesson 30), and the elapsed time.

Increase CONCURRENCY gradually while watching for rate-limit retries. When retries become frequent, you have reached the practical limit for your tier.

Common Mistakes

Launching Everything at Once

What happens: hundreds of simultaneous requests cause rate-limit errors and failed items.

Fix: limit concurrency with a semaphore or worker pool.

Reusing a Coroutine in a Retry Loop

What happens: RuntimeError: cannot reuse already awaited coroutine.

Fix: pass a function that creates a new coroutine for each attempt.

Retrying Without Delay or Jitter

What happens: all tasks retry at the same moment and hit the limit again.

Fix: respect retry-after, use exponential backoff, and add jitter.

Letting One Exception Cancel the Whole Job

What happens: gather() raises on the first error, and the other results are lost.

Fix: catch errors per item, or use return_exceptions=True.

Using Parallel Requests for Huge, Non-Urgent Jobs

What happens: higher cost and constant pressure on rate limits.

Fix: use the Message Batches API.

Best Practices

• Choose concurrency from your rate limits, and start low.
• Limit concurrency with a semaphore.
• Retry 429 responses with backoff and jitter, respecting retry-after.
• Handle errors per item, and report them at the end.
• Show progress for long jobs.
• Use a fast model for simple, high-volume tasks.
• Track tokens for cost.
• Prefer the Batches API for large jobs that can wait.

Exercise

1. Run parallel_summarizer.py with CONCURRENCY set to 1, 3, and 8. Record the total time for each.
2. Add a retries field to Outcome that records how many rate-limit retries each ticket needed.
3. Add a per-request time limit of 30 seconds with asyncio.wait_for(), and record timeouts as errors.
4. Read the tickets from a text file (one ticket per line), and write the results to a CSV file with the columns index, summary, and error.

Step 3: wrap the call inside summarize():

async with semaphore:
    try:
        message = await asyncio.wait_for(call_with_retry(make_call), timeout=30)
    except asyncio.TimeoutError:
        return Outcome(index, None, "timeout", 0)
    except anthropic.APIError as error:
        return Outcome(index, None, type(error).__name__, 0)

asyncio.TimeoutError is the name that works on every supported Python version. On Python 3.11+, it is the same class as the built-in TimeoutError.

Step 4:

import csv
from pathlib import Path

TICKETS = [line.strip() for line in Path("tickets.txt").read_text(encoding="utf-8").splitlines()
           if line.strip()]

# After the results are collected:
with open("summaries.csv", "w", newline="", encoding="utf-8") as file:
    writer = csv.writer(file)
    writer.writerow(["index", "summary", "error"])
    for outcome in results:
        writer.writerow([outcome.index, outcome.summary or "", outcome.error or ""])

newline="" is required by the csv module to avoid blank lines on Windows.