Send Async Messages

The Basic Pattern

Sending a message with the async client is the same as with the sync client, plus await:

import asyncio

from anthropic import AsyncAnthropic


async def main() -> None:
    async with AsyncAnthropic() as client:
        message = await client.messages.create(
            model="claude-opus-5-5",
            max_tokens=1024,
            system="You are a concise programming tutor.",
            messages=[{"role": "user", "content": "What is a Python decorator? Two sentences."}],
        )

    print("".join(b.text for b in message.content if b.type == "text"))
    print(f"[stop_reason={message.stop_reason}, output_tokens={message.usage.output_tokens}]")


asyncio.run(main())

await client.messages.create(...) returns the same Message object as the sync client (Lesson 14), so everything you know about reading responses applies: content blocks, stop reasons, usage, and IDs.

What await Does Here

coroutine = client.messages.create(...)   # 1. Creates a coroutine; nothing is sent yet
message = await coroutine                  # 2. Runs it: sends the request, waits, returns the Message

1. Calling an async method returns a coroutine: a description of work that has not started.
2. await runs the coroutine. While it waits for the network, the event loop can run other coroutines. When the response arrives, execution continues on the next line with the result.

This two-step nature is what makes concurrency possible: you can create several coroutines first, then run them together with asyncio.gather() (Lesson 71) or as tasks (Lesson 74).

Note: writing await directly in front of each call runs the calls one after another, just like sync code. Concurrency only happens when you run several coroutines together.

a = await client.messages.create(...)   # Waits for A
b = await client.messages.create(...)   # Starts B only after A finished: sequential

a, b = await asyncio.gather(             # A and B run at the same time: concurrent
    client.messages.create(...),
    client.messages.create(...),
)

Every Resource Has an Async Version

All SDK resources work the same way on the async client:

Structured Output

from pydantic import BaseModel


class Sentiment(BaseModel):
    label: str
    confidence: float


result = await client.messages.parse(
    model="claude-opus-5-5",
    max_tokens=1024,
    output_format=Sentiment,
    messages=[{"role": "user", "content": "Classify: 'The course was excellent!'"}],
)
print(result.parsed_output.label)   # positive

Paginated Lists

List methods return async iterators that fetch further pages automatically. Iterate them with async for:

async for model in client.models.list():
    print(model.id)

Raw Responses

raw = await client.messages.with_raw_response.create(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)
print(raw.headers.get("request-id"))
message = await raw.parse()

With the async client, parse() on a raw response is also a coroutine, so it must be awaited.

Errors Are the Same Exceptions

The async client raises the same exception classes as the sync client (Lesson 9). Catch them with ordinary try/except around the await:

import anthropic

try:
    message = await client.messages.create(...)
except anthropic.RateLimitError:
    ...
except anthropic.APIStatusError as error:
    print(error.status_code)

The exception is raised at the await, where the coroutine actually runs. Lesson 77 covers errors in concurrent code.

Practical Example: An Async Helper Module

Applications usually wrap common operations in small, focused helper coroutines. This module provides three: plain text answers, structured answers, and token counting. A demo script uses them together.

claude_async.py

"""claude_async.py: small async helpers built on AsyncAnthropic."""

from typing import TypeVar

from anthropic import AsyncAnthropic
from pydantic import BaseModel

MODEL = "claude-opus-5-5"
T = TypeVar("T", bound=BaseModel)


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


async def ask(client: AsyncAnthropic, question: str, system: str = "", max_tokens: int = 1024) -> str:
    request = {"model": MODEL, "max_tokens": max_tokens,
               "messages": [{"role": "user", "content": question}]}
    if system:
        request["system"] = system
    message = await client.messages.create(**request)
    if message.stop_reason == "refusal":
        raise ValueError("Claude declined this request.")
    return text_of(message)


async def ask_structured(client: AsyncAnthropic, prompt: str, model_type: type[T]) -> T:
    result = await client.messages.parse(
        model=MODEL,
        max_tokens=1024,
        output_config={"effort": "low"},
        output_format=model_type,
        messages=[{"role": "user", "content": prompt}],
    )
    if result.stop_reason != "end_turn" or result.parsed_output is None:
        raise ValueError(f"No structured result (stop_reason={result.stop_reason}).")
    return result.parsed_output


async def count_tokens(client: AsyncAnthropic, text: str) -> int:
    count = await client.messages.count_tokens(
        model=MODEL, messages=[{"role": "user", "content": text}]
    )
    return count.input_tokens

review_report.py

"""review_report.py: analyze several reviews concurrently with the async helpers."""

import asyncio

from anthropic import AsyncAnthropic
from pydantic import BaseModel

from claude_async import ask, ask_structured, count_tokens

REVIEWS = [
    "The Django course was clear and the projects were practical. Loved it!",
    "Videos are good, but the quizzes have several wrong answers. Please fix them.",
    "Too fast for beginners. I got lost in week two.",
]


class ReviewAnalysis(BaseModel):
    sentiment: str        # positive, negative, or mixed
    main_point: str


async def analyze(client: AsyncAnthropic, review: str) -> tuple[ReviewAnalysis, int]:
    prompt = (f"<review>\n{review}\n</review>\n\nClassify the sentiment as positive, negative, "
              "or mixed, and state the main point in under 12 words.")
    analysis, tokens = await asyncio.gather(
        ask_structured(client, prompt, ReviewAnalysis),
        count_tokens(client, review),
    )
    return analysis, tokens


async def main() -> None:
    async with AsyncAnthropic() as client:
        results = await asyncio.gather(*(analyze(client, review) for review in REVIEWS))
        summary = await ask(
            client,
            "Write one sentence for the course team summarizing these points:\n"
            + "\n".join(f"- {analysis.main_point}" for analysis, _ in results),
        )

    for (analysis, tokens), review in zip(results, REVIEWS):
        print(f"[{analysis.sentiment:<8}] {analysis.main_point}  ({tokens} tokens) <- {review[:35]}...")
    print(f"\nTeam summary: {summary}")


if __name__ == "__main__":
    asyncio.run(main())

Example output:

[positive] Clear lessons and practical projects  (19 tokens) <- The Django course was clear and the...
[mixed   ] Good videos, but quiz answers contain errors  (21 tokens) <- Videos are good, but the quizzes ha...
[negative] Pace is too fast for beginners  (15 tokens) <- Too fast for beginners. I got lost...

Team summary: Students value the clear lessons and practical projects, but the quiz errors and the fast pace for beginners need attention.

How It Works

1. The helpers take the client as a parameter (Lesson 72), so one client created in main() is shared by everything.
2. ask() builds the request, adds the system prompt only when given (Lesson 22), awaits the response, and turns a refusal into an exception, so callers receive either text or an error.
3. ask_structured() is generic: TypeVar("T", bound=BaseModel) and type[T] tell type checkers that passing ReviewAnalysis returns a ReviewAnalysis. It awaits messages.parse() and checks the stop reason before returning the parsed model (Lesson 36).
4. count_tokens() awaits the token counting endpoint (Lesson 46).
5. analyze() runs two independent requests for the same review, the analysis and the token count, concurrently with asyncio.gather().
6. main() analyzes all reviews concurrently, so this is concurrency at two levels: three reviews at once, each with two requests at once. It then makes one final request that depends on all the results, so it runs after the gather().
7. Results stay in order, because gather() preserves input order, so zip(results, REVIEWS) pairs each analysis with its review.

Common Mistakes

Awaiting Calls One by One When They Could Run Together

What happens: async code that runs no faster than sync code.

Fix: collect independent coroutines and run them with asyncio.gather().

Forgetting await on parse() for Raw Responses

What happens: you get a coroutine instead of a Message.

Fix: message = await raw.parse() with the async client.

Using for Instead of async for on Paginated Lists

What happens: TypeError, because the async paginator is not a regular iterator.

Fix: use async for.

Running Dependent Requests Concurrently

What happens: a request that needs another request's result runs before that result exists.

Fix: run dependent steps after awaiting the steps they depend on, as main() does with the summary.

Best Practices

• await every async SDK call, including parse() on raw responses.
• Run independent requests concurrently, and dependent ones in sequence.
• Wrap common operations in small helper coroutines that receive the client.
• Check stop reasons in helpers, and convert unusable results into clear exceptions.
• Use async for with paginated lists.

Exercise

1. Add a translate(client, text, language) helper to claude_async.py, and translate one review into Tamil and Hindi concurrently.
2. Write an async function that lists all available models with async for and prints their IDs and maximum output tokens.
3. Change analyze() so that it awaits the two requests sequentially. Measure the difference in total time.
4. Explain why the final ask() call in main() cannot run concurrently with the reviews' analyses.

Step 2:

async def show_models(client: AsyncAnthropic) -> None:
    async for model in client.models.list():
        print(f"{model.id:<24} max output: {model.max_tokens}")

async for lets the paginator fetch more pages on demand, awaiting each page request.

Step 4: the summary prompt is built from the analyses' results (analysis.main_point). Those results only exist after the analyses have completed. A step that depends on other steps' outputs must run after them. Only independent operations can run concurrently.