Claude API Async Request Handling in Python

Handling Claude API requests asynchronously in Python means using asyncio together with an async-capable HTTP client — either the official SDK's AsyncAnthropic class or a library like httpx — so your program can fire off multiple requests concurrently instead of waiting for each one to finish before starting the next.

This matters because LLM calls are I/O-bound and often slow: a single Claude response can take several seconds, and if you're processing a batch of documents, chat sessions, or background jobs synchronously, you're wasting most of that time just waiting on the network. Async handling lets you run dozens of requests in parallel, respect rate limits with controlled concurrency, and stream tokens back to users without blocking your event loop.

Why Synchronous Calls Don't Scale

A typical synchronous script looks like this:

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

for prompt in prompts:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    print(response.content[0].text)

If each call takes 3 seconds and you have 100 prompts, this takes roughly 5 minutes. With async execution and reasonable concurrency, the same batch can finish in a fraction of that time, bounded mostly by your rate limits rather than total latency.

Using the Async SDK Client

The official Python SDK ships an async client that mirrors the sync one but uses await:

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic(api_key="sk-ant-...")

async def ask(prompt):
    response = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return response.content[0].text

result = asyncio.run(ask("Explain asyncio in one paragraph"))
print(result)

This is the easiest path if you're already committed to the official SDK and just need non-blocking calls inside an async web framework like FastAPI.

Running Requests Concurrently with a Semaphore

The real benefit of async shows up when you batch multiple prompts. Use asyncio.gather to run tasks in parallel, and a Semaphore to cap concurrency so you don't blow through your rate limits:

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic(api_key="sk-ant-...")
sem = asyncio.Semaphore(5)  # max 5 concurrent requests

async def ask(prompt):
    async with sem:
        response = await client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text

async def main(prompts):
    tasks = [ask(p) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

results = asyncio.run(main(["Summarize A", "Translate B", "Classify C"]))

return_exceptions=True is important — without it, one failed request cancels the entire gather call and you lose the rest of your results. Check each item in results for an Exception instance before using it.

Async Streaming for Real-Time Output

If you're building a chat UI or CLI tool, you usually want tokens streamed as they're generated rather than waiting for the full response:

async def stream_answer(prompt):
    async with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(stream_answer("Write a haiku about concurrency"))

This pattern works the same whether you're on the official API or a wrapper — the key is that the streaming context manager yields chunks asynchronously without blocking your event loop.

Doing It with Raw HTTP (httpx)

If you'd rather not depend on the SDK — or you're hitting a Claude-compatible endpoint directly — httpx.AsyncClient works just as well:

import asyncio
import httpx

API_KEY = "sub_live_..."

async def ask(client, prompt):
    resp = await client.post(
        "https://api.subtoapi.app/v1/messages",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "claude-sonnet-4-5",
            "max_tokens": 512,
            "messages": [{"role": "user", "content": prompt}],
        },
    )
    return resp.json()

async def main(prompts):
    async with httpx.AsyncClient(timeout=60) as client:
        return await asyncio.gather(*(ask(client, p) for p in prompts))

results = asyncio.run(main(["Prompt 1", "Prompt 2", "Prompt 3"]))

This same shape works for SubToAPI, which exposes Claude through a standard HTTPS endpoint with sub_live_... application keys — the async code above doesn't change whether you're calling Anthropic directly or through a proxy that adds usage metadata and team-level keys. See the messages reference for the full request schema, streaming docs for server-sent events, and tool use docs if your async requests also need function calling.

Handling Retries and Backoff

Async code still needs retry logic for rate limits (429) and transient server errors (5xx). A simple exponential backoff wrapper:

import asyncio
import random

async def with_retry(coro_fn, *args, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            return await coro_fn(*args)
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            wait = (2 ** attempt) + random.random()
            await asyncio.sleep(wait)

Wrap your ask calls with this instead of calling them bare inside gather, so a single rate-limit hiccup doesn't fail the whole batch.

Practical Tips

• Cap concurrency explicitly. Don't fire 500 requests with bare asyncio.gather — use a Semaphore or a bounded task queue.
• Set timeouts. Both httpx.AsyncClient and the SDK accept a timeout parameter; always set one, since a hung connection in an async batch can stall everything behind it.
• Log per-request errors separately. With return_exceptions=True, inspect and log failures individually rather than letting one bad prompt obscure the rest of the batch.
• Reuse the client. Create one AsyncAnthropic or httpx.AsyncClient instance and reuse it across requests instead of instantiating a new one per call — this keeps connection pooling effective.

Questions

Does async make individual Claude responses faster? No. Async doesn't reduce the latency of a single request — it lets you run many requests concurrently so your total wall-clock time for a batch drops significantly.

What's the difference between asyncio.gather and a task queue? gather is fine for small, fixed batches run once. For long-running or continuous workloads, a bounded queue with worker coroutines gives you better control over backpressure and retry handling.

Can I use async with streaming and tool use together? Yes — the async streaming context manager works the same way whether the response includes plain text or tool-use blocks; you just need to handle both event types inside the async for loop. See /docs/tools for details.