Create an Async Client

The AsyncAnthropic Class

The asynchronous client is created like the synchronous one:

from anthropic import AsyncAnthropic

client = AsyncAnthropic()

It reads the same environment variables (ANTHROPIC_API_KEY first, Lesson 7), accepts the same constructor options, and has the same resources: client.messages, client.models, client.beta, and so on. The difference is that its methods return coroutines that you await.

Option	Example	Same as sync (Lesson 10)?
api_key	AsyncAnthropic(api_key=load_secret(...))	Yes
timeout	AsyncAnthropic(timeout=60.0)	Yes
max_retries	AsyncAnthropic(max_retries=4)	Yes
base_url	AsyncAnthropic(base_url="https://gateway.example.com")	Yes
default_headers	AsyncAnthropic(default_headers={"anthropic-workspace-id": "wrkspc_..."})	Yes
http_client	AsyncAnthropic(http_client=DefaultAioHttpClient())	Async-specific HTTP clients

Like the sync client, creating an AsyncAnthropic object does not contact the API. Network activity starts with the first request.

---

One Client, Many Requests

An async client manages a connection pool: open HTTPS connections reused across requests. Concurrent requests share the pool, so one client per application is the right default:

client = AsyncAnthropic()

answers = await asyncio.gather(
    client.messages.create(...),   # All three share the same client and its connections
    client.messages.create(...),
    client.messages.create(...),
)

Creating a new client for each request throws away the pool and adds connection setup time to every call.

---

The Event-Loop Rule

Async connections belong to the event loop that was running when they were opened. That leads to one important rule:

Use an async client within a single event loop. Do not reuse the same client across separate asyncio.run() calls.

client = AsyncAnthropic()

asyncio.run(task_one(client))   # Opens connections on event loop #1, then closes loop #1
asyncio.run(task_two(client))   # Event loop #2 tries to reuse connections from the closed loop #1

The second call can fail with errors such as RuntimeError: Event loop is closed, because pooled connections from the first loop are still referenced.

Fixes:

• Call asyncio.run() once, with one top-level main() coroutine that does all the work, or
• create the client inside the coroutine that runs on the loop, for example with async with AsyncAnthropic() as client: inside main().

Web frameworks such as FastAPI run a single event loop for the whole application, so a client created at startup (in the lifespan, Lesson 68) is safe for every request.

---

Client Lifecycle Patterns

Pattern 1: Scripts, with async with

async def main() -> None:
    async with AsyncAnthropic() as client:
        ...   # All requests here

asyncio.run(main())

async with creates the client inside the running loop and closes its connections when the block ends (Lesson 79).

Pattern 2: Applications, Created at Startup

@asynccontextmanager
async def lifespan(app):
    app.state.claude = AsyncAnthropic()
    yield
    await app.state.claude.close()

The client lives for the whole application, on the application's event loop, and is closed cleanly at shutdown.

Pattern 3: Libraries, Client Passed In

class Summarizer:
    def __init__(self, client: AsyncAnthropic):
        self.client = client

Reusable code accepts a client instead of creating one. The application controls creation, configuration, and closing, and tests can pass a fake client (Lessons 181–182).

---

The aiohttp Backend

By default, the async client uses httpx2 for HTTP. The SDK also supports aiohttp as an alternative backend, which the SDK documentation recommends for improved concurrency performance:

python -m pip install "anthropic[aiohttp]"

from anthropic import AsyncAnthropic, DefaultAioHttpClient


async def main() -> None:
    async with AsyncAnthropic(http_client=DefaultAioHttpClient()) as client:
        message = await client.messages.create(
            model="claude-opus-5-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello!"}],
        )

• DefaultAioHttpClient() is an HTTP client configured with the SDK's defaults, using aiohttp.
• Your code does not change otherwise: requests, streaming, and errors work the same way.
• Consider it for applications with many concurrent requests or streams. For occasional requests, the default backend is fine.

Custom HTTP Settings

To customize the default httpx2-based backend, for example to add a proxy, while keeping the SDK's defaults, use DefaultAsyncHttpxClient:

from anthropic import AsyncAnthropic, DefaultAsyncHttpxClient

client = AsyncAnthropic(http_client=DefaultAsyncHttpxClient(proxy="http://proxy.internal:8080"))

---

Per-Request Options

with_options() works exactly as on the sync client (Lesson 10). It returns a copy of the client with changed settings, sharing the same connection pool:

fast = client.with_options(timeout=15.0, max_retries=1)
message = await fast.messages.create(...)

---

Practical Example: A Shared Async Client Module

This module provides one correctly managed async client for a whole program: configured from environment variables, created inside the running event loop, and closed once at the end. Other modules receive the client instead of creating their own.

claude_runtime.py

"""claude_runtime.py: create and manage the application's async Claude client."""

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from anthropic import AsyncAnthropic, DefaultAioHttpClient


@asynccontextmanager
async def claude_client() -> AsyncIterator[AsyncAnthropic]:
    """Create the shared async client inside the running event loop, and close it at the end."""
    options = {
        "timeout": float(os.environ.get("CLAUDE_TIMEOUT", "60")),
        "max_retries": int(os.environ.get("CLAUDE_MAX_RETRIES", "2")),
    }
    if os.environ.get("CLAUDE_HTTP_BACKEND") == "aiohttp":
        options["http_client"] = DefaultAioHttpClient()

    async with AsyncAnthropic(**options) as client:
        yield client

summarizer.py

"""summarizer.py: reusable logic that receives a client instead of creating one."""

from anthropic import AsyncAnthropic


class Summarizer:
    def __init__(self, client: AsyncAnthropic, model: str = "claude-opus-5-5"):
        self.client = client
        self.model = model

    async def summarize(self, text: str, words: int = 30) -> str:
        message = await self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content":
                       f"<text>\n{text}\n</text>\n\nSummarize the text above in at most {words} words."}],
        )
        return "".join(b.text for b in message.content if b.type == "text").strip()

main.py

"""main.py: one event loop, one client, shared by all components."""

import asyncio

from claude_runtime import claude_client
from summarizer import Summarizer

ARTICLES = [
    "FastAPI is a modern Python web framework that uses type hints for validation and documentation...",
    "SQLite is a self-contained, serverless database engine stored in a single file...",
]


async def main() -> None:
    async with claude_client() as client:
        summarizer = Summarizer(client)
        summaries = await asyncio.gather(*(summarizer.summarize(text) for text in ARTICLES))
    for summary in summaries:
        print("-", summary)


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

Example output:

- FastAPI is a modern Python web framework that uses type hints to validate data and generate API documentation automatically.
- SQLite is a lightweight, serverless database that stores an entire database in a single file.

How It Works

1. claude_client() is an async context manager, written with @asynccontextmanager. Code before yield runs when the async with block starts, and the cleanup runs when it ends. Here, the cleanup is handled by the inner async with AsyncAnthropic(...), which closes the client.
2. Configuration comes from environment variables (Lesson 8): timeout, retries, and optionally the aiohttp backend through CLAUDE_HTTP_BACKEND=aiohttp. The defaults apply when they are not set.
3. The client is created inside main(), which runs on the event loop started by asyncio.run(). That follows the event-loop rule.
4. Summarizer receives the client in its constructor. It never creates or closes clients, so it can be reused anywhere, and tested with a fake client.
5. asyncio.gather(*(...)) runs both summaries concurrently with the shared client. The generator expression creates one coroutine per article.
6. asyncio.run(main()) is called once, so there is exactly one event loop for the program's lifetime.

---

Common Mistakes

Reusing a Client Across asyncio.run() Calls

What happens: Event loop is closed errors, or hangs, on later calls.

Fix: one asyncio.run(main()) per program, with the client created inside it.

Creating a Client per Request

What happens: slower requests and wasted resources.

Fix: share one client, and use with_options() for per-request variations.

Never Closing the Client

What happens: warnings about unclosed connections, and leaked resources in long-running services.

Fix: use async with, or await client.close() at shutdown.

Creating Clients Deep Inside Library Code

What happens: components cannot share connections, and tests cannot replace the client.

Fix: pass the client in from the application.

Using the aiohttp Backend Without the Extra

What happens: an import error, because aiohttp is not installed.

Fix: install anthropic[aiohttp].

---

Best Practices

• Create one async client per application (per credential), on the application's event loop.
• Manage its lifetime with async with, or startup and shutdown hooks.
• Pass the client into components instead of creating clients inside them.
• Configure it from the environment: timeout, retries, and backend.
• Consider the aiohttp backend for high concurrency.
• Use with_options() for per-request settings.

---

Exercise

1. Write a script that calls asyncio.run() twice with the same client, and observe the behavior. Then fix it by using a single main().
2. Run main.py with CLAUDE_HTTP_BACKEND=aiohttp (after installing the extra) and confirm that it works the same way.
3. Add a translate(text, language) method to Summarizer, and run a summary and a translation concurrently.
4. Explain why Summarizer should not call client.close().

Show answers for steps 1 and 4

Step 1: the fixed version moves both tasks into one coroutine:

async def main() -> None:
    async with AsyncAnthropic() as client:
        await task_one(client)
        await task_two(client)

asyncio.run(main())

Both tasks run on the same event loop, and the client is created and closed on that loop.

Step 4: Summarizer does not own the client. It received it from the application, which may share the same client with other components. If Summarizer closed it, every other component using that client would fail. The code that creates a resource is responsible for closing it.