Claude API Python Async Client Example

If you're building a Python service that calls Claude at any real volume — a web backend, a batch job, a worker queue — a synchronous client will bottleneck you. Every blocking HTTP call ties up a thread while it waits for a response that can take several seconds, especially with longer completions. An async client lets you fire off dozens of requests concurrently and let the event loop handle the waiting.

This article walks through a complete, working async Python client for calling Claude, built on httpx and asyncio. It covers single requests, concurrent batches with a semaphore to cap parallelism, retry logic for rate limits, and async streaming. The same pattern applies whether you're calling Anthropic directly or routing through a gateway like SubToAPI.

Why async matters for Claude API calls

A typical Claude request takes 2–15 seconds depending on output length. If you're processing 100 documents synchronously at 5 seconds each, that's over 8 minutes of wall-clock time with your CPU doing nothing but waiting. With asyncio and a connection pool, you can run those same 100 requests with, say, 10 concurrent connections, and finish in under a minute — the network wait time overlaps instead of stacking up.

The tradeoff is you need to handle concurrency limits (both your own and the API's rate limits) and write slightly more careful error handling, since failures in an async batch don't stop the whole program the way an unhandled exception in a sync script might.

Basic async client with httpx

httpx is the standard choice here — it has a first-class async API that mirrors requests, supports HTTP/2, and handles connection pooling automatically.

import httpx
import asyncio
import os

API_KEY = os.environ["ANTHROPIC_API_KEY"]
BASE_URL = "https://api.anthropic.com/v1"

async def call_claude(prompt: str, client: httpx.AsyncClient) -> str:
    response = await client.post(
        f"{BASE_URL}/messages",
        headers={
            "x-api-key": API_KEY,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        },
        json={
            "model": "claude-sonnet-4-5",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": prompt}],
        },
        timeout=60.0,
    )
    response.raise_for_status()
    data = response.json()
    return data["content"][0]["text"]

async def main():
    async with httpx.AsyncClient() as client:
        result = await call_claude("Explain async/await in one paragraph.", client)
        print(result)

asyncio.run(main())

Note the shared httpx.AsyncClient() instance — create it once and reuse it across requests rather than opening a new client per call. It maintains a connection pool, which matters a lot when you're making many concurrent requests.

Running requests concurrently with a semaphore

The naive approach — just wrapping everything in asyncio.gather — will fire all requests at once, which can trip rate limits or exhaust your connection pool. Use a Semaphore to cap concurrency to a sane number:

import asyncio
import httpx

CONCURRENCY_LIMIT = 8

async def bounded_call(prompt: str, client: httpx.AsyncClient, sem: asyncio.Semaphore) -> str:
    async with sem:
        return await call_claude(prompt, client)

async def process_batch(prompts: list[str]) -> list[str]:
    sem = asyncio.Semaphore(CONCURRENCY_LIMIT)
    async with httpx.AsyncClient() as client:
        tasks = [bounded_call(p, client, sem) for p in prompts]
        return await asyncio.gather(*tasks, return_exceptions=True)

prompts = [f"Summarize document {i} in two sentences." for i in range(50)]
results = asyncio.run(process_batch(prompts))

for i, r in enumerate(results):
    if isinstance(r, Exception):
        print(f"Request {i} failed: {r}")
    else:
        print(f"Request {i}: {r[:80]}...")

return_exceptions=True is important — without it, one failed request cancels the whole gather call and you lose results from everything else that was still in flight.

Retrying on rate limits (429s)

Even with a concurrency cap, you'll hit 429s under load. A simple exponential backoff with jitter handles this without extra dependencies:

import random

async def call_with_retry(prompt: str, client: httpx.AsyncClient, max_retries: int = 5) -> str:
    for attempt in range(max_retries):
        try:
            return await call_claude(prompt, client)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429 and attempt < max_retries - 1:
                wait = (2 ** attempt) + random.uniform(0, 1)
                await asyncio.sleep(wait)
                continue
            raise
    raise RuntimeError("Max retries exceeded")

Wrap your batch calls with this instead of calling call_claude directly, and transient rate limiting stops being a problem that kills your job.

Async streaming responses

For chat interfaces or long-form generation where you want tokens as they arrive, httpx supports async streaming via Server-Sent Events:

async def stream_claude(prompt: str, client: httpx.AsyncClient):
    async with client.stream(
        "POST",
        f"{BASE_URL}/messages",
        headers={
            "x-api-key": API_KEY,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        },
        json={
            "model": "claude-sonnet-4-5",
            "max_tokens": 1024,
            "stream": True,
            "messages": [{"role": "user", "content": prompt}],
        },
    ) as response:
        async for line in response.aiter_lines():
            if line.startswith("data: "):
                yield line[6:]

async def main():
    async with httpx.AsyncClient() as client:
        async for chunk in stream_claude("Write a short poem about rivers."):
            print(chunk)

asyncio.run(main())

This same shape works whether you're calling Anthropic's endpoint directly or an API-compatible layer — streaming is covered in more depth in the streaming docs if you're routing requests through SubToAPI.

Using this pattern with SubToAPI

If you're issuing calls from a team of developers or multiple services and want per-key usage tracking without managing your own key rotation and rate-limit logic, SubToAPI gives you an HTTPS endpoint (https://api.subtoapi.app/v1/messages) that accepts the same async client code above — just swap the base URL and use a sub_live_... key with a standard Authorization: Bearer header instead of x-api-key. Usage metadata, streaming, and tool calls all work the same way. Check the quickstart and Messages API reference for the exact request shape, and tool use docs if your async batch jobs need function calling. Plans start at €9/month with a free trial at signup, and pricing details are on the pricing page.

The core async pattern — shared client, semaphore-bounded concurrency, retry with backoff — doesn't change based on which endpoint you're hitting. Get that right once and it scales from a 10-request script to a production batch pipeline.

Questions

Does the official Anthropic Python SDK support async natively? Yes — the anthropic package ships an AsyncAnthropic client that wraps the same patterns shown here (httpx under the hood). The manual httpx approach is useful when you need custom retry logic, connection pooling tuned to your workload, or want to avoid the extra dependency.

How many concurrent requests can I safely run? It depends on your rate limits and the complexity of your prompts. Start with a semaphore limit of 5–10 and increase it while watching for 429 responses; back off if you see them increase.

Can I use this async pattern with streaming and tool use together? Yes. Set "stream": True in the request body and handle tool_use content blocks as they arrive in the SSE stream, accumulating partial JSON until the block closes before executing the tool call.