AsyncAnthropic: Async Client for the Claude API

# Synchronous (blocks the thread)
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(...)

# Asynchronous (non-blocking)
import anthropic
async_client = anthropic.AsyncAnthropic()

async def create_message():
    return await async_client.messages.create(...)

import asyncio
import anthropic
from environs import Env

env = Env()
env.read_env()

client = anthropic.AsyncAnthropic(api_key=env.str("ANTHROPIC_API_KEY"))

async def chat(message: str) -> str:
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}]
    )
    return response.content[0].text

async def main():
    answer = await chat("Explain async/await in three sentences")
    print(answer)

asyncio.run(main())

async def stream_chat(message: str) -> str:
    chunks = []
    async with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
            chunks.append(text)
        final = await stream.get_final_message()
    print()
    return "".join(chunks)

async def analyze_batch(texts: list[str]) -> list[str]:
    """Sends all requests in parallel."""
    async def analyze_one(text: str) -> str:
        response = await client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=256,
            messages=[{"role": "user", "content": f"Sentiment: {text}. One word."}]
        )
        return response.content[0].text.strip()

    return await asyncio.gather(*[analyze_one(t) for t in texts])

async def main():
    texts = ["Excellent product!", "Terrible service", "It is fine, nothing special"]
    results = await analyze_batch(texts)
    for text, sentiment in zip(texts, results):
        print(f"{sentiment}: {text}")

asyncio.run(main())

async def safe_chat(message: str) -> str | None:
    for attempt in range(3):
        try:
            response = await client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=512,
                messages=[{"role": "user", "content": message}]
            )
            return response.content[0].text
        except anthropic.RateLimitError:
            if attempt == 2:
                raise
            await asyncio.sleep(2 ** attempt)
        except anthropic.APIError as exc:
            print(f"API error: {exc}")
            return None

    return None