Claude API Async Python Requests Example

If you're sending more than one Claude API request per script run, synchronous requests calls will bottleneck you fast — each call blocks until the model finishes generating, and generation can take several seconds. The fix is to run your Claude API requests asynchronously with httpx.AsyncClient and asyncio, so multiple prompts are in flight at the same time instead of queued one after another.

Below is a working pattern for calling Claude's Messages API asynchronously in Python, covering single async calls, batching many calls concurrently, limiting concurrency so you don't blow through rate limits, and handling retries on transient errors.

Why requests isn't enough

The standard requests library is synchronous — it opens a connection, waits for the full response, and only then lets your code continue. For a single Claude call that's fine. For 50 calls in a loop, you're paying the full round-trip latency 50 times in sequence, even though the API could easily handle several requests in parallel.

asyncio with an async HTTP client (httpx is the standard choice since aiohttp doesn't play perfectly with all JSON streaming patterns) lets you fire off many requests and await them together.

A single async request

import os
import httpx
import asyncio

API_URL = "https://api.anthropic.com/v1/messages"
API_KEY = os.environ["CLAUDE_API_KEY"]

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

async def main():
    result = await ask_claude("Summarize async/await in two sentences.")
    print(result)

asyncio.run(main())

This is functionally identical to a requests.post call, but because it's defined as a coroutine, you can run many of these concurrently instead of one at a time.

Running multiple requests concurrently

The real payoff comes from asyncio.gather:

async def main():
    prompts = [
        "Explain the GIL in one paragraph.",
        "What is backpressure in streaming systems?",
        "Give a one-line definition of idempotency.",
    ]

    results = await asyncio.gather(*(ask_claude(p) for p in prompts))

    for prompt, result in zip(prompts, results):
        print(f"Q: {prompt}\nA: {result}\n")

asyncio.run(main())

All three requests go out nearly simultaneously. Total wall-clock time is close to the slowest single request, not the sum of all three — this is the difference that matters when you're processing batches of documents, user messages, or evaluation prompts.

Limiting concurrency with a semaphore

Firing off hundreds of requests at once will trip rate limits or exhaust connection pools. Use an asyncio.Semaphore to cap how many requests run at the same time:

sem = asyncio.Semaphore(5)  # max 5 concurrent requests

async def ask_claude_limited(prompt: str) -> str:
    async with sem:
        return await ask_claude(prompt)

async def main():
    prompts = [f"Give a fun fact about number {i}" for i in range(50)]
    results = await asyncio.gather(*(ask_claude_limited(p) for p in prompts))
    print(len(results), "responses received")

asyncio.run(main())

This keeps throughput high without overwhelming the API. Start with a concurrency of 5–10 and tune based on how many rate-limit errors you see.

Handling retries on transient errors

Network blips and 429/5xx responses happen. A minimal retry wrapper with exponential backoff:

async def ask_claude_with_retry(prompt: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            return await ask_claude(prompt)
        except httpx.HTTPStatusError as e:
            if e.response.status_code in (429, 500, 502, 503) and attempt < retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            raise

Wrap this (not the bare ask_claude) inside your semaphore-limited function for production use.

Async streaming

For streaming responses, httpx supports client.stream with Server-Sent Events, which you parse line by line as chunks arrive instead of waiting for the full generation:

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

You can run several of these streaming coroutines concurrently with asyncio.gather the same way as the non-streaming version.

If you'd rather not manage API keys and rate limits yourself

The patterns above work directly against Anthropic's API, but if your team shares one Claude subscription and needs per-application keys, usage tracking, and a dashboard instead of raw credentials scattered across scripts, that's exactly what SubToAPI is for. It turns your existing Claude access into an HTTPS API with sub_live_... application keys, streaming, and tool use support — the async code above works the same way, you just point it at https://api.subtoapi.app/v1/messages with Authorization: Bearer $SUBTOAPI_KEY instead of the x-api-key header. Full request/response shapes are in the docs and streaming setup is covered in /docs/streaming. Plans start at €9/month with a free trial — see /pricing.

Key takeaways

• Use httpx.AsyncClient instead of requests for any workload with more than one Claude call per run.
• asyncio.gather parallelizes independent prompts; wall-clock time drops to roughly the slowest single call.
• Always cap concurrency with a Semaphore — unbounded concurrency will hit rate limits.
• Wrap calls in retry logic with backoff for 429/5xx responses.
• Async streaming uses client.stream with SSE line parsing instead of waiting on the full response body.

FAQ

Does Claude's API have a native async Python SDK? Anthropic's official Python SDK supports both sync and async clients out of the box (AsyncAnthropic). The httpx-based pattern shown here is useful when you want direct control over connection pooling, retries, or when working through a proxy service rather than the SDK.

How many concurrent Claude requests can I safely run? It depends on your account's rate limits. Start with a semaphore value of 5–10 concurrent requests and increase gradually while watching for 429 responses; back off with exponential retry when you hit them.

Is aiohttp or httpx better for Claude API calls? httpx is generally preferred for Claude API work because its streaming API (client.stream) handles Server-Sent Events cleanly and its sync/async interfaces share the same request-building code, which simplifies testing.