How to Use the Asynchronous Anthropic Client with aiohttp

Install the anthropic[aiohttp] extra and pass DefaultAioHttpClient() to the http_client parameter of AsyncAnthropic to replace the default httpx transport with aiohttp.

The anthropics/anthropic-sdk-python repository ships with two asynchronous HTTP backends for the AsyncAnthropic client. While the SDK defaults to httpx, you can configure the asynchronous client with aiohttp by installing an optional extra and passing a custom transport class, enabling better concurrency characteristics for specific workloads without modifying your application logic.

Installing the aiohttp Backend

The aiohttp support is provided through the httpx-aiohttp extra dependency. Install it alongside the main SDK package:

pip install anthropic[aiohttp]

This command pulls in httpx-aiohttp, which itself depends on aiohttp, making the DefaultAioHttpClient class available for import.

Configuring the Asynchronous Transport

Internally, AsyncAnthropic initializes its HTTP layer through the http_client argument. According to the source code in src/anthropic/_base_client.py, if no client is supplied, the SDK falls back to DefaultAsyncHttpxClient (lines 1453–1504). When the aiohttp extra is present, the module defines DefaultAioHttpClient (lines 1506–1515). The high-level client stores the chosen transport in _client (as seen in src/anthropic/_client.py, lines 56–70) and uses it for all API requests.

To switch transports, instantiate DefaultAioHttpClient and pass it to the constructor:

from anthropic import AsyncAnthropic, DefaultAioHttpClient

client = AsyncAnthropic(
    http_client=DefaultAioHttpClient()
)

All subsequent API calls (messages.create, completions.create, models.list) use this underlying aiohttp session automatically.

Basic Async Implementation

Use the client as an asynchronous context manager to ensure proper connection cleanup:

import os
import asyncio
from anthropic import AsyncAnthropic, DefaultAioHttpClient

async def main() -> None:
    async with AsyncAnthropic(
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        response = await client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[
                {"role": "user", "content": "Tell me a short joke."}
            ],
        )
        print(response.content[0].text)

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

Streaming with aiohttp

Streaming responses work identically regardless of transport. Set stream=True and iterate over the async generator:

import os
import asyncio
from anthropic import AsyncAnthropic, DefaultAioHttpClient

async def main() -> None:
    async with AsyncAnthropic(
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        http_client=DefaultAioHttpClient(),
    ) as client:
        async with client.completions.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=256,
            prompt="Write a haiku about the sunrise.",
            stream=True,
        ) as stream:
            async for token in stream:
                print(token.text, end="", flush=True)

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

Advanced aiohttp Configuration

For production workloads requiring custom connection limits, timeouts, or proxy settings, subclass the underlying transport directly.

Custom Client Sessions

Subclass HttpxAiohttpClient (the base class behind DefaultAioHttpClient) to inject a custom aiohttp.ClientSession configuration:

import aiohttp
from httpx_aiohttp import HttpxAiohttpClient
from anthropic import AsyncAnthropic

class MyAioHttpClient(HttpxAiohttpClient):
    def __init__(self):
        timeout = aiohttp.ClientTimeout(total=120)
        connector = aiohttp.TCPConnector(limit=200)
        super().__init__(timeout=timeout, connector=connector)

async def main():
    async with AsyncAnthropic(http_client=MyAioHttpClient()) as client:
        response = await client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello"}]
        )
        print(response.content[0].text)

This pattern exposes the full aiohttp configuration surface while maintaining compatibility with the Anthropic SDK's abstract interface.

Summary

• Install the extra: Use pip install anthropic[aiohttp] to enable the DefaultAioHttpClient class.
• Pass the transport: Instantiate DefaultAioHttpClient() and provide it to the http_client parameter of AsyncAnthropic.
• Source locations: The transport selection logic resides in src/anthropic/_base_client.py (lines 1453–1515), while the high-level client interface is defined in src/anthropic/_client.py (lines 56–70).
• API compatibility: All methods including streaming work unchanged when switching from httpx to aiohttp.
• Customization: Subclass HttpxAiohttpClient for advanced aiohttp session configuration.

Frequently Asked Questions

What is the difference between DefaultAsyncHttpxClient and DefaultAioHttpClient?

DefaultAsyncHttpxClient is the default transport based on httpx.AsyncClient with tuned defaults for timeouts and connection limits. DefaultAioHttpClient is an optional transport provided by the httpx-aiohttp extra that replaces the httpx transport with an aiohttp implementation, potentially offering better concurrency for specific async workloads.

Do I need to change my API calls when switching to aiohttp?

No. The AsyncAnthropic class abstracts the HTTP layer behind a common interface. Once you pass http_client=DefaultAioHttpClient(), all methods including messages.create, completions.create, and streaming responses function identically to the default httpx implementation.

How do I handle connection pooling with aiohttp?

By default, DefaultAioHttpClient manages its own connection pool. For custom pooling behavior, subclass HttpxAiohttpClient and pass a configured aiohttp.TCPConnector to the superclass constructor, then provide your subclass instance to the http_client parameter.

Is aiohttp faster than httpx for Anthropic API calls?

Performance depends on your specific workload and Python version. The aiohttp transport may offer advantages for high-concurrency scenarios with many simultaneous connections, while httpx provides broader HTTP/2 support. The SDK allows you to benchmark both by simply swapping the http_client argument without code changes.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}