Async SDK Intro

What "Async" Means

Most Claude requests spend nearly all of their time waiting: for the network, and for the model to generate a response. With the synchronous client (Anthropic), your program does nothing else during that wait. The thread is blocked.

Asynchronous programming lets a program start a task, and while that task waits, work on other tasks. Python supports this with the asyncio library and the keywords async and await. The Anthropic SDK provides an asynchronous client, AsyncAnthropic, with the same methods and parameters as the synchronous one.

import asyncio

from anthropic import AsyncAnthropic

client = AsyncAnthropic()


async def main() -> None:
    message = await client.messages.create(
        model="claude-opus-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Explain async programming in one sentence."}],
    )
    print("".join(b.text for b in message.content if b.type == "text"))


asyncio.run(main())

Compared with the synchronous version, three things change: the class is AsyncAnthropic, the call is preceded by await, and the code runs inside an async def function started by asyncio.run().

The Core Concepts

How Waiting Becomes Concurrency

Synchronous: three requests, one after another
  request A ███████████
  request B            ███████████
  request C                       ███████████        total ≈ 3 × one request

Asynchronous: three requests, concurrently
  request A ███████████
  request B ███████████
  request C ███████████                                total ≈ 1 × one request

While request A waits for Claude, the event loop starts request B, then C. All three wait at the same time. When each response arrives, the event loop resumes the coroutine that was waiting for it.

Important: async does not make a single request faster. It lets many waiting operations overlap. The work is still I/O-bound waiting on Claude; async simply stops your program from wasting that time.

Sync vs. Async Client

Everything learned so far, including messages, system prompts, tokens, stop reasons, structured outputs, and errors, applies unchanged. Only the calling style differs.

When Async Is Useful

When async is not needed: simple scripts that make one request at a time, synchronous frameworks such as Flask or traditional Django views, and notebooks used for single experiments. Lesson 80 compares the options in detail.

Practical Example: Sequential vs. Concurrent Requests

This script asks five questions twice: once one after another with the synchronous client, and once concurrently with the asynchronous client. It measures the total time of each approach.

"""sync_vs_async.py: compare sequential and concurrent requests."""

import asyncio
import time

from anthropic import Anthropic, AsyncAnthropic

QUESTIONS = [
    "What is a Python list? One sentence.",
    "What is a Python tuple? One sentence.",
    "What is a Python set? One sentence.",
    "What is a Python dictionary? One sentence.",
    "What is a Python generator? One sentence.",
]
REQUEST = {"model": "claude-opus-5-5", "max_tokens": 1024}


def text_of(message) -> str:
    return "".join(b.text for b in message.content if b.type == "text").strip()


def run_sequential() -> list[str]:
    client = Anthropic()
    return [
        text_of(client.messages.create(**REQUEST, messages=[{"role": "user", "content": q}]))
        for q in QUESTIONS
    ]


async def run_concurrent() -> list[str]:
    async with AsyncAnthropic() as client:
        coroutines = [
            client.messages.create(**REQUEST, messages=[{"role": "user", "content": q}])
            for q in QUESTIONS
        ]
        messages = await asyncio.gather(*coroutines)
    return [text_of(m) for m in messages]


if __name__ == "__main__":
    start = time.perf_counter()
    run_sequential()
    sequential = time.perf_counter() - start

    start = time.perf_counter()
    answers = asyncio.run(run_concurrent())
    concurrent = time.perf_counter() - start

    for question, answer in zip(QUESTIONS, answers):
        print(f"- {question.split('?')[0]}? {answer[:70]}")
    print(f"\nSequential: {sequential:.1f}s | Concurrent: {concurrent:.1f}s "
          f"| Speed-up: {sequential / concurrent:.1f}x")

Example output (timings vary):

- What is a Python list? A list is an ordered, mutable collection of items written in square
- What is a Python tuple? A tuple is an ordered, immutable collection written in parentheses.
- What is a Python set? A set is an unordered collection of unique items.
- What is a Python dictionary? A dictionary stores key-value pairs with fast lookup by key.
- What is a Python generator? A generator produces values lazily, one at a time, using yield.

Sequential: 14.2s | Concurrent: 3.4s | Speed-up: 4.2x

How It Works

1. run_sequential() sends each request and waits for it before sending the next one, so the total time is roughly the sum of all five.
2. run_concurrent() creates five coroutines without awaiting them individually. Calling client.messages.create(...) on the async client returns a coroutine object, a request ready to run, but not yet running.
3. asyncio.gather(*coroutines) runs all five concurrently and waits until every one has finished. It returns the results in the same order as the coroutines were given, regardless of which finished first, so zip(QUESTIONS, answers) matches questions to answers correctly.
4. async with AsyncAnthropic() as client creates the client and closes its connections when the block ends (Lesson 79).
5. asyncio.run(run_concurrent()) starts the event loop, runs the coroutine to completion, and closes the loop.

The concurrent version is limited by the slowest single request plus a little overhead, not by the sum. With more requests, the gap grows, until rate limits become the constraint (Lesson 75).

Common Mistakes

Forgetting await

message = client.messages.create(...)   # With AsyncAnthropic: this is a coroutine, not a Message
print(message.content)                  # AttributeError: 'coroutine' object has no attribute 'content'

Python also warns "coroutine ... was never awaited".

Fix: message = await client.messages.create(...).

Using await Outside an async def Function

What happens: SyntaxError: 'await' outside async function in normal scripts.

Fix: put async code in an async def function and start it with asyncio.run(). In Jupyter notebooks, top-level await is allowed, because the notebook already runs an event loop.

Calling asyncio.run() Inside a Running Event Loop

What happens: RuntimeError: asyncio.run() cannot be called from a running event loop, for example inside Jupyter or inside an async web framework.

Fix: in those environments, use await directly instead of asyncio.run().

Mixing the Sync Client into Async Code

What happens: Anthropic().messages.create(...) inside an async def function blocks the whole event loop, so nothing else can run until it finishes.

Fix: use AsyncAnthropic in async code.

Expecting Async to Speed Up One Request

What happens: no improvement for single requests.

Fix: use async when you have many concurrent operations; use streaming to improve the perceived speed of one request.

Best Practices

• Use AsyncAnthropic in async programs, and the sync client elsewhere.
• Always await SDK calls on the async client.
• Start async programs with asyncio.run(), once, at the top level.
• Run independent requests concurrently with asyncio.gather(), with concurrency limits for large numbers (Lesson 75).
• Close async clients with async with or await client.close().
• Keep blocking code out of the event loop.

Exercise

1. Run sync_vs_async.py with 3 questions and with 10 questions. How does the speed-up change?
2. Remove await from a single async create() call and observe the error and the warning.
3. Write an async function ask(question: str) -> str and use it with asyncio.gather() to answer three questions.
4. Explain why asyncio.gather() returns results in the original order, even though requests finish at different times.

Step 3:

import asyncio

from anthropic import AsyncAnthropic

client = AsyncAnthropic()


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


async def main() -> None:
    answers = await asyncio.gather(
        ask("What is recursion? One sentence."),
        ask("What is a closure? One sentence."),
        ask("What is a decorator? One sentence."),
    )
    for answer in answers:
        print("-", answer)


asyncio.run(main())

Step 4: gather() keeps track of each coroutine's position in its argument list. When a coroutine finishes, its result is stored at that position. gather() returns the list only when every coroutine has finished, so the order always matches the input, whichever request completed first.