OpenAI SDK: Async client & concurrency

Each API call spends almost all its time waiting for the model. Classifying 1,000 support tickets one at a time at 1.5 s each takes 25 minutes, and your CPU is idle for all of it. Because the work is waiting, not computing, AsyncOpenAI with asyncio can run many requests at once (Advanced Python part 17). The discipline is doing it inside your rate limits, with a semaphore.

1. AsyncOpenAI: same API, with await

real API

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    r = await client.responses.create(model="gpt-5.6-luna", input="Say hi")
    print(r.output_text)

asyncio.run(main())

2. Bounded concurrency with a semaphore

import asyncio
import time
from fake_openai import fake_async_client

tickets = [f"Ticket {i}: my order is late" for i in range(40)]

async def classify_all(client, texts, limit):
    sem = asyncio.Semaphore(limit)                      # at most `limit` requests in flight

    async def classify(text):
        async with sem:
            r = await client.responses.create(model="gpt-5.6-luna",
                                              instructions="Label: delivery, refund or other.",
                                              input=text)
            return r.output_text

    return await asyncio.gather(*(classify(t) for t in tickets))

async def main():
    await fake_async_client().responses.create(model="warm-up", input="x")   # exclude one-time import cost

    for limit in (1, 10):
        client = fake_async_client(latency=0.1)          # every fake request takes 0.1 s
        start = time.perf_counter()
        labels = await classify_all(client, tickets, limit)
        print(f"limit={limit:2}: {len(labels)} results in {time.perf_counter() - start:.2f} s, "
              f"peak in flight = {client.fake.peak}")

asyncio.run(main())

Output

limit= 1: 40 results in 4.60 s, peak in flight = 1
limit=10: 40 results in 0.47 s, peak in flight = 10

Forty requests at 0.1 s each: about 4 s one at a time, about 0.4 s with ten in flight. The semaphore is the difference between “fast” and “429 Too Many Requests”. An unbounded gather over 10,000 items opens 10,000 requests at once and trips every rate limit you have.

3. Rate limits: RPM and TPM

4. Errors inside gather

import asyncio
import openai
from fake_openai import fake_async_client

async def main():
    # the fake server rejects "item 2" with 400 Bad Request; retries are off so we see it at once
    def responder(body):
        if body["input"] == "item 2":
            return {"status": 400, "message": "input too long"}
        return f"label for {body['input']}"
    client = fake_async_client(responder=responder, max_retries=0)

    async def one(i):
        r = await client.responses.create(model="gpt-5.6-luna", input=f"item {i}")
        return r.output_text

    results = await asyncio.gather(*(one(i) for i in range(4)), return_exceptions=True)
    for i, res in enumerate(results):
        if isinstance(res, openai.APIStatusError):
            print(i, "failed:", type(res).__name__, res.status_code)
        else:
            print(i, res)

asyncio.run(main())

Output

0 label for item 0
1 label for item 1
2 failed: BadRequestError 400
3 label for item 3

With return_exceptions=True, one bad item does not cancel the others. You get a result or an exception per item, and can retry or log the failures. In a real pipeline, also write each result to disk as it completes, so a crash at item 900 does not lose items 1–899.

Recap