Claude API Request Queue Management in Python

Claude API Request Queue Management in Python

When you call the Claude API directly, you're bound by Anthropic's rate limits — requests per minute, tokens per minute, and concurrent connection caps that vary by model and account tier. The moment your application does more than trickle traffic (batch jobs, multi-user apps, background workers), you need a request queue that controls concurrency, retries failed calls intelligently, and doesn't silently drop work when limits are hit.

A request queue for the Claude API in Python typically means: a bounded worker pool (using asyncio.Semaphore or a thread pool), a asyncio.Queue or queue.Queue holding pending prompts, exponential backoff on 429 and 529 responses, and some visibility into queue depth so you know when you're falling behind. Below is a practical pattern you can drop into a backend service, plus the tradeoffs of rolling this yourself versus offloading it.

Why You Can't Just Fire Requests in a Loop

The naive approach — looping over a list of prompts and calling the API sequentially or with asyncio.gather on everything at once — breaks in two predictable ways:

• Sequential calls are slow. Each request waits for the previous one to finish, even though Claude can handle multiple concurrent requests per account.
• Unbounded concurrency hits rate limits. Firing 200 requests at once gets you a wall of 429 Too Many Requests responses, and if you don't handle them, those requests are just lost.

A queue with a concurrency cap solves both: it keeps enough requests in flight to use your available throughput, without exceeding it.

Building a Bounded Worker Pool

The core building block is a semaphore-limited async worker pool. Here's a minimal, production-shaped example:

import asyncio
import random
import httpx

CONCURRENCY = 5
MAX_RETRIES = 5
BASE_DELAY = 1.0

class ClaudeQueue:
    def __init__(self, api_key: str, concurrency: int = CONCURRENCY):
        self.api_key = api_key
        self.semaphore = asyncio.Semaphore(concurrency)
        self.client = httpx.AsyncClient(timeout=60.0)

    async def send(self, prompt: str) -> dict:
        async with self.semaphore:
            return await self._request_with_retry(prompt)

    async def _request_with_retry(self, prompt: str) -> dict:
        for attempt in range(MAX_RETRIES):
            resp = await self.client.post(
                "https://api.anthropic.com/v1/messages",
                headers={
                    "x-api-key": self.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}],
                },
            )
            if resp.status_code == 200:
                return resp.json()
            if resp.status_code in (429, 529):
                delay = BASE_DELAY * (2 ** attempt) + random.uniform(0, 0.5)
                retry_after = resp.headers.get("retry-after")
                if retry_after:
                    delay = max(delay, float(retry_after))
                await asyncio.sleep(delay)
                continue
            resp.raise_for_status()
        raise RuntimeError(f"Exceeded retries for prompt: {prompt[:40]}")

async def main():
    queue = ClaudeQueue(api_key="sk-ant-...")
    prompts = [f"Summarize topic {i}" for i in range(50)]
    results = await asyncio.gather(*(queue.send(p) for p in prompts))
    print(f"Completed {len(results)} requests")

asyncio.run(main())

This gives you a hard cap on concurrent requests, jittered exponential backoff on rate-limit responses, and respect for retry-after headers when Anthropic sends one. The jitter matters — without it, retries from multiple workers synchronize and hit the API in bursts, which just triggers more 429s.

Adding Priority and Backpressure

If some requests matter more than others (interactive user queries vs. batch summarization jobs), a plain FIFO queue isn't enough. Python's asyncio.PriorityQueue or queue.PriorityQueue lets you tag jobs:

import asyncio

work_queue = asyncio.PriorityQueue()

async def enqueue(priority: int, prompt: str):
    await work_queue.put((priority, prompt))

async def worker(queue_obj: ClaudeQueue):
    while True:
        priority, prompt = await work_queue.get()
        try:
            await queue_obj.send(prompt)
        finally:
            work_queue.task_done()

Lower numbers run first, so interactive requests (priority 0) jump ahead of batch jobs (priority 10). Pair this with backpressure: if work_queue.qsize() exceeds a threshold, reject new low-priority jobs with a 503 rather than letting the queue grow unbounded in memory.

Monitoring Queue Health

A queue you can't observe is a queue that fails silently. Track at minimum:

• Queue depth — items waiting, sampled every few seconds.
• Retry rate — how often you hit 429/529, as a leading indicator you need more capacity or fewer concurrent workers.
• p95 latency per job, not just average — queued jobs can look fine on average while individual users wait far too long.

Exposing these as Prometheus counters or just logging them at intervals is usually enough to catch a queue that's quietly backing up before it becomes an outage.

When to Stop Building This Yourself

The pattern above is solid for a single service hitting one Anthropic account. It gets more complex once you need per-team quotas, multiple API keys for different environments, streaming responses alongside queued batch work, or usage accounting across a product with paying customers. At that point you're maintaining rate-limit logic, retry logic, and auth/billing logic that has nothing to do with your actual product.

This is the gap SubToAPI fills: it sits in front of your Claude access and exposes a standard HTTPS API with its own sub_live_... application keys, built-in handling for streaming and tool use, and per-key usage metadata — so your queue workers call a stable endpoint instead of juggling Anthropic's rate limits and retry semantics directly. You still queue and prioritize work on your side, but the backend plumbing for auth, streaming, and usage tracking is handled. Check the quickstart or the messages and streaming docs to see how request shapes compare to calling Anthropic directly, and pricing if you're evaluating cost against running your own key rotation and retry stack.

questions

Do I need a queue if my app only makes a few Claude requests per minute? No. If your traffic is well under Anthropic's rate limits, a simple sequential or lightly concurrent call pattern is fine. Queues earn their complexity once you have bursty traffic, batch jobs, or multiple users hitting the API at once.

What's the difference between asyncio.Queue and multiprocessing.Queue for this use case? asyncio.Queue is for I/O-bound concurrency within a single event loop — ideal for API calls, since the bottleneck is network wait time, not CPU. multiprocessing.Queue is for CPU-bound work split across processes, which rarely applies to calling an HTTP API.

How many concurrent Claude requests should my queue allow? Start conservatively (3–5) and increase while watching your 429 rate. The right number depends on your account's rate limit tier and the size of your prompts, so tune it empirically rather than hardcoding a guess.