Use asyncio

asyncio: The Engine Behind the Async Client

AsyncAnthropic provides awaitable SDK calls. asyncio, Python's built-in library for asynchronous programming, provides the tools to organize those calls: running them concurrently, collecting results, handling them as they finish, limiting how many run at once, applying time limits, and cancelling work that is no longer needed.

This lesson covers the asyncio tools you will use most with the Claude SDK:

Tool	Purpose
asyncio.run()	Start the event loop and run your main coroutine
asyncio.gather()	Run coroutines concurrently and collect all results in order
asyncio.create_task()	Start a coroutine in the background
asyncio.as_completed()	Handle results in the order they finish
asyncio.TaskGroup	Run related tasks together, with automatic cleanup on failure (Python 3.11+)
asyncio.Semaphore	Limit how many operations run at once
asyncio.Queue	Pass work between producers and workers
asyncio.timeout() / asyncio.wait_for()	Apply time limits
asyncio.to_thread()	Run blocking code without freezing the event loop

---

asyncio.run(): The Entry Point

asyncio.run(main())

asyncio.run() creates an event loop, runs main() until it completes, then closes the loop. Call it once, at the top level of a program (Lesson 72). In Jupyter notebooks, a loop is already running, so use await main() instead.

---

gather(): All Results, in Order

answers = await asyncio.gather(ask("Q1"), ask("Q2"), ask("Q3"))

• Runs all coroutines concurrently.
• Returns a list of results in the same order as the arguments.
• By default, if one coroutine raises an exception, gather() raises it immediately. The other coroutines keep running, but their results are lost.

return_exceptions=True

results = await asyncio.gather(*coroutines, return_exceptions=True)
for result in results:
    if isinstance(result, Exception):
        print("failed:", type(result).__name__)
    else:
        print("ok:", result)

With return_exceptions=True, exceptions are returned as results instead of raised, so one failure does not hide the other results. This is usually what you want for independent requests, such as processing many documents (Lesson 77).

---

create_task(): Start Now, Await Later

create_task() schedules a coroutine to start immediately in the background and returns a Task that you can await later:

summary_task = asyncio.create_task(ask("Summarize this article: ..."))
await do_other_work()                 # The summary request runs meanwhile
summary = await summary_task          # Get the result when needed

Keep a reference to every task you create, and await it, or cancel it, eventually. Tasks without references can be garbage-collected, and exceptions in tasks that are never awaited are easy to miss.

---

as_completed(): Results as They Finish

gather() waits for everything. as_completed() yields results as each one finishes, which is useful for showing progress or acting on fast results immediately:

coroutines = [ask(q) for q in questions]
for finished in asyncio.as_completed(coroutines):
    answer = await finished
    print("done:", answer[:60])

Results arrive in completion order, not input order. To know which input a result belongs to, return the input together with the result, for example as a tuple (index, answer).

---

TaskGroup: Structured Concurrency (Python 3.11+)

A TaskGroup runs related tasks together and guarantees cleanup: when the async with block ends, all tasks have finished, and if one task fails, the others are cancelled and the error is raised:

async with asyncio.TaskGroup() as group:
    task_a = group.create_task(ask("Q1"))
    task_b = group.create_task(ask("Q2"))
print(task_a.result(), task_b.result())

Use a TaskGroup when tasks belong together and a failure in one makes the others pointless. Use gather(..., return_exceptions=True) when tasks are independent and you want every result, including failures. On Python 3.10, which the SDK supports, TaskGroup is not available, so use gather().

---

Semaphore: Limit Concurrency

Starting 500 requests at once will hit your rate limits (Lesson 75). A semaphore allows only a fixed number of operations inside a block at the same time:

limit = asyncio.Semaphore(5)


async def limited_ask(question: str) -> str:
    async with limit:                     # At most 5 coroutines inside this block at once
        return await ask(question)


answers = await asyncio.gather(*(limited_ask(q) for q in questions))

All coroutines start, but only five at a time pass async with limit:. The others wait until a slot frees up.

---

Queue: Producers and Workers

For continuous or large workloads, a queue decouples producers, which add work, from a fixed pool of workers, which process it. Workers take items with await queue.get() and call queue.task_done() when finished. await queue.join() waits until every item has been processed. The practical example below uses this pattern.

---

Time Limits

# Python 3.11+
async with asyncio.timeout(30):
    answer = await ask("Explain the GIL.")

# Any supported Python version
answer = await asyncio.wait_for(ask("Explain the GIL."), timeout=30)

Both cancel the operation if it does not finish in time and raise TimeoutError (asyncio.TimeoutError on Python 3.10). The SDK also has its own per-request timeouts. Lesson 78 explains how the two kinds of time limits work together.

---

to_thread(): Blocking Code Inside Async Programs

Some libraries are synchronous, for example sqlite3, many file operations, or image processing. Calling them directly inside a coroutine blocks the whole event loop. asyncio.to_thread() runs them in a separate thread and lets you await the result:

rows = await asyncio.to_thread(load_history_from_sqlite, session_id)

---

Cancellation

Tasks can be cancelled with task.cancel(). The task receives asyncio.CancelledError at its current await, so cleanup in finally blocks and async with blocks still runs, which closes any open Claude streams (Lesson 60). Let CancelledError propagate: catching it and continuing prevents tasks from shutting down correctly.

---

Practical Example: A Worker Pool with a Queue

This program processes a list of questions with a fixed number of workers. Each worker repeatedly takes a question from a queue, asks Claude, records the result, and moves on. The number of workers limits concurrency, and failures are recorded instead of stopping the program.

"""question_workers.py: process questions with a fixed pool of async workers."""

import asyncio
import time
from dataclasses import dataclass

import anthropic
from anthropic import AsyncAnthropic

WORKER_COUNT = 3
QUESTIONS = [
    "What is a Python list comprehension? One sentence.",
    "What does the 'with' statement do? One sentence.",
    "What is a virtual environment? One sentence.",
    "What is PEP 8? One sentence.",
    "What does 'pip freeze' do? One sentence.",
    "What is a docstring? One sentence.",
]


@dataclass
class Result:
    index: int
    question: str
    answer: str | None
    error: str | None
    seconds: float


async def worker(name: str, client: AsyncAnthropic, queue: asyncio.Queue, results: list[Result]) -> None:
    while True:
        index, question = await queue.get()
        start = time.perf_counter()
        try:
            message = await client.messages.create(
                model="claude-opus-5-5",
                max_tokens=1024,
                messages=[{"role": "user", "content": question}],
            )
            answer = "".join(b.text for b in message.content if b.type == "text").strip()
            results.append(Result(index, question, answer, None, time.perf_counter() - start))
            print(f"[{name}] finished question {index}")
        except anthropic.APIError as error:
            results.append(Result(index, question, None, type(error).__name__, time.perf_counter() - start))
            print(f"[{name}] question {index} failed: {type(error).__name__}")
        finally:
            queue.task_done()


async def main() -> list[Result]:
    queue: asyncio.Queue = asyncio.Queue()
    for item in enumerate(QUESTIONS, start=1):
        queue.put_nowait(item)

    results: list[Result] = []
    async with AsyncAnthropic() as client:
        workers = [
            asyncio.create_task(worker(f"worker-{n}", client, queue, results))
            for n in range(1, WORKER_COUNT + 1)
        ]
        await queue.join()                 # Wait until every question has been processed
        for task in workers:
            task.cancel()                  # Workers loop forever; stop them now
        await asyncio.gather(*workers, return_exceptions=True)

    return sorted(results, key=lambda result: result.index)


if __name__ == "__main__":
    start = time.perf_counter()
    for result in asyncio.run(main()):
        status = result.answer[:60] if result.answer else f"ERROR: {result.error}"
        print(f"{result.index}. ({result.seconds:.1f}s) {status}")
    print(f"Total: {time.perf_counter() - start:.1f}s with {WORKER_COUNT} workers")

Example output:

[worker-2] finished question 2
[worker-1] finished question 1
[worker-3] finished question 3
[worker-2] finished question 4
[worker-3] finished question 6
[worker-1] finished question 5
1. (2.6s) A list comprehension builds a new list from an iterable in a
2. (2.2s) The 'with' statement manages resources, such as files, and a
3. (2.9s) A virtual environment is an isolated Python environment with
4. (2.4s) PEP 8 is Python's official style guide for writing readable
5. (2.7s) 'pip freeze' lists installed packages with their exact versi
6. (2.1s) A docstring is a string literal that documents a module, fun
Total: 5.4s with 3 workers

How It Works

1. queue.put_nowait(item) fills the queue with (index, question) pairs. enumerate(..., start=1) numbers the questions, so results can be sorted back into order later.
2. Each worker runs an endless loop: await queue.get() waits for the next item, the worker processes it, and queue.task_done() in finally marks it as finished, whether it succeeded or failed.
3. Three workers means at most three requests run at the same time. The worker count works like a semaphore and keeps you within rate limits.
4. await queue.join() waits until task_done() has been called once for every item put into the queue.
5. Stopping workers: the workers would otherwise wait forever for new items. task.cancel() raises CancelledError at their await queue.get(), and gather(..., return_exceptions=True) waits for all of them to finish cancelling, without raising the CancelledErrors.
6. Failures are recorded as results, so one failed question does not stop the others.
7. Results are sorted by index at the end, because workers finish in an unpredictable order.

With six questions and three workers, the total time is about two "rounds" of requests. Producers can also keep adding items while workers run, which makes this pattern suitable for continuous workloads such as processing incoming support tickets.

---

Common Mistakes

Calling asyncio.run() More Than Once with Shared Resources

What happens: clients bound to a closed event loop fail (Lesson 72).

Fix: one asyncio.run(main()) per program.

Unbounded Concurrency

What happens: hundreds of simultaneous requests trigger rate-limit errors.

Fix: limit concurrency with a semaphore or a worker pool.

Losing Task References and Exceptions

What happens: background tasks fail silently, or are garbage-collected before finishing.

Fix: keep references, and always await or cancel tasks.

Blocking the Event Loop

What happens: synchronous calls inside coroutines freeze all other tasks.

Fix: use async libraries, or asyncio.to_thread().

Swallowing CancelledError

What happens: tasks cannot be cancelled cleanly, and programs hang on shutdown.

Fix: let cancellation propagate, and use finally for cleanup.

---

Best Practices

• Start async programs with one asyncio.run(main()).
• Use gather() for a fixed set of concurrent operations, with return_exceptions=True when they are independent.
• Use as_completed() to show progress or handle fast results first.
• Use TaskGroup for related tasks that should fail together (Python 3.11+).
• Limit concurrency with semaphores or worker pools.
• Use queues for large or continuous workloads.
• Apply time limits, and keep blocking code off the event loop.

---

Exercise

1. Rewrite the worker pool using asyncio.gather() with an asyncio.Semaphore(3) instead of a queue. Compare the code and the total time.
2. Use asyncio.as_completed() to print each answer as soon as it arrives, with its question number.
3. On Python 3.11 or later, run two related requests in a TaskGroup, and observe what happens when one of them uses an invalid model name.
4. Add a producer to the worker example that adds three more questions to the queue after a 2-second delay, while the workers are running.

Show answers for steps 2 and 4

Step 2:

async def numbered(client, index, question):
    message = await client.messages.create(
        model="claude-opus-5-5", max_tokens=1024,
        messages=[{"role": "user", "content": question}],
    )
    return index, "".join(b.text for b in message.content if b.type == "text").strip()


async def main():
    async with AsyncAnthropic() as client:
        coroutines = [numbered(client, i, q) for i, q in enumerate(QUESTIONS, start=1)]
        for finished in asyncio.as_completed(coroutines):
            index, answer = await finished
            print(f"{index}: {answer[:60]}")

Returning the index together with the answer identifies each result, because completion order differs from input order.

Step 4:

async def producer(queue: asyncio.Queue, start_index: int) -> None:
    await asyncio.sleep(2)
    for offset, question in enumerate(["What is a lambda?", "What is a set?", "What is a module?"]):
        await queue.put((start_index + offset, question + " One sentence."))

In main(), start it with producer_task = asyncio.create_task(producer(queue, len(QUESTIONS) + 1)), and await producer_task before await queue.join(). Otherwise join() could return before the producer has added its items, when the queue is briefly empty.