Python: parallel requests to Claude via AsyncOpenAI

Classifying a thousand support tickets or extracting data from a hundred documents one request at a time is slow: each answer takes seconds, and the script spends almost all of that time waiting. AsyncOpenAI with asyncio sends parallel requests, and a semaphore keeps you from sending too many at once. Below are working examples for Zukko: the basic pattern, retries on errors, stopping when the balance runs out, cost tracking, and CSV processing.

Basic example: parallel requests with a semaphore

There's one dependency: pip install openai. The key goes in the ZUKKO_API_KEY environment variable.

import asyncio
import os

from openai import AsyncOpenAI

MODEL = "claude-sonnet-5"

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
)


async def classify(text: str, semaphore: asyncio.Semaphore) -> str:

asyncio.gather starts all coroutines at once, but only five get inside async with semaphore at the same time. The rest wait for a free slot. Results come back in the same order as the texts, even if the responses arrive out of order.

Retries on 429 and 503

The official SDK already retries on its own: by default up to two times with growing backoff on 408, 409, 429, and 5xx, and on network failures. For batch processing, raise the number of attempts and set a timeout:

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
    timeout=60.0,
)

You usually don't need your own retry loop on top of that. But errors that are pointless to retry need explicit handling:

• 401 — invalid key;
• 402 — insufficient balance, so every following request will fail too;
• 403 — the model isn't in the key's allowed list;
• 413 — the text is too large and needs to be shortened or split.

Stopping the whole batch on 402

If the balance has run out, there's no point sending another thousand requests and collecting a thousand identical errors. asyncio.TaskGroup (Python 3.11+) cancels all remaining tasks as soon as one of them fails:

import asyncio
import os

from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
)


class StopBatch(Exception):
    pass

Errors on individual texts (for example, 413 on an oversized document) don't stop the batch — they land in the results as an ERROR ... string, and you can rerun them later.

Cost tracking from usage

Every response has response.usage with the number of input (prompt_tokens) and output (completion_tokens) tokens. Zukko prices per 1M tokens: claude-sonnet-5 — $0.3 / $0.3, gpt-5.6-luna — $0.1 / $0.1, glm-5.3-flash — $0.04 / $0.04.

PRICES = {  # $ per 1M tokens: (input, output)
    "claude-sonnet-5": (0.3, 0.3),
    "gpt-5.6-luna": (0.1, 0.1),
    "glm-5.3-flash": (0.04, 0.04),
}


def cost_usd(model: str, usage) -> float:
    price_in, price_out = PRICES[model]
    return (usage.prompt_tokens * price_in + usage.completion_tokens * price_out) / 1_000_000

An upfront estimate: 10,000 texts with 200 input and 5 output tokens each is 2,000,000 input and 50,000 output tokens.

Prices can change, so check them against the "Models" page in the dashboard. Which model gives acceptable accuracy on your texts is something to test yourself: label 50–100 examples by hand and compare the answers of two or three models against your labels.

The gateway returns the exact cost of each request in the X-Zukko-Cost-Micros header (in microdollars). Headers are available through with_raw_response:

raw = await client.chat.completions.with_raw_response.create(
    model="claude-sonnet-5",
    messages=[{"role": "user", "content": "Topic in one word: shares went up"}],
    max_tokens=10,
)
completion = raw.parse()
micros = int(raw.headers.get("X-Zukko-Cost-Micros", "0"))
print(completion.choices[0].message.content, f"${micros / 1_000_000:.6f}")

More on headers and key limits is in the key limits and cache↗ section.

Processing a CSV and saving results

A typical task: a file with a text column, and you need to add a category column.

import asyncio
import csv
import os

from openai import AsyncOpenAI, APIStatusError, APIConnectionError

client = AsyncOpenAI(
    base_url="https://api.zukko.pro/v1",
    api_key=os.environ["ZUKKO_API_KEY"],
    max_retries=5,
)


async def process_row(row: dict, semaphore: asyncio.Semaphore) -> dict:

Each row is processed independently: a network failure or an error on one text doesn't bring down the whole file. Rows with an ERROR status are easy to filter out and rerun.

How to avoid 429

1. Cap concurrency with a semaphore. Start at 5 and raise it until 429 responses appear.
2. Keep the SDK retries (max_retries): it will ride out short bursts of 429 on its own.
3. Check the key limits in the dashboard↗: if the daily spend limit is used up, the key also answers 429, and retries won't help until the next day.

FAQ

What semaphore size should I pick?

Start with 5. If there are no 429 responses and processing is slow, try 10. If 429s appear, go back to a lower value. The optimum depends on request length and load, so find it empirically.

How do I stop processing with Ctrl+C?

On Ctrl+C, asyncio.run cancels the main task, and all pending requests with it. To keep finished results, save them in batches (see the next question).

How do I save progress in case processing is interrupted?

Split the list into batches and save results after each one:

async def run_in_batches(texts: list[str], batch_size: int = 100):
    semaphore = asyncio.Semaphore(5)
    for start in range(0, len(texts), batch_size):
        batch = texts[start:start + batch_size]
        results = await asyncio.gather(*(classify(t, semaphore) for t in batch))
        save_results(start, results)  # your function: a file or a database
        print(f"Done {start + len(batch)} of {len(texts)}")

On restart, begin from the first unsaved batch.

Can I do the same synchronously?

Yes, with OpenAI and concurrent.futures.ThreadPoolExecutor, but for hundreds of parallel requests the async client is simpler and lighter on resources. Both client variants are covered in the Python docs↗.