Using the Python OpenAI Async Client

The OpenAI Python library ships both a synchronous client and an async client. The python openai async client, exposed as AsyncOpenAI, provides the same request-building interface as the sync client but executes calls through async/await on the event loop. The synchronous client blocks the calling thread while a request is in flight. In a script that sends one request at a time, that is rarely a problem. In a web server handling many users, or a batch job that needs dozens of completions, the blocking behavior serializes work that could otherwise overlap.

What the AsyncOpenAI Client Changes

The synchronous client and the async client share the same request-building interface. The difference is in how the request is executed. With the sync client, client.chat.completions.create(...) blocks until the response is ready. With AsyncOpenAI, the same call is a coroutine and must be awaited:

from openai import AsyncOpenAI

client = AsyncOpenAI()

async def get_summary(text: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": f"Summarize this in one sentence: {text}"}
        ],
    )
    return response.choices[0].message.content

The method names, parameters, and response objects match the synchronous client. If you already know how to build a request with OpenAI, you know how to build one with AsyncOpenAI. The async client is built on httpx.AsyncClient rather than the blocking httpx.Client, which is what allows the request to yield control back to the event loop while the network call is pending.

Running Concurrent Requests with asyncio.gather

The practical benefit of the async client appears when you issue several independent requests. Awaiting them one after another is no faster than calling the sync client sequentially. The point is to start them together and let them complete in parallel:

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def classify(text: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": f"Classify this as positive or negative: {text}"}
        ],
        max_tokens=10,
    )
    return response.choices[0].message.content

async def main() -> None:
    texts = [
        "The service was fast and friendly.",
        "The package arrived damaged.",
        "The refund took three weeks.",
    ]
    results = await asyncio.gather(*(classify(t) for t in texts))
    for text, label in zip(texts, results):
        print(f"{text!r} -> {label}")

asyncio.run(main())

asyncio.gather schedules all three coroutines on the event loop. Each await client.chat.completions.create(...) suspends that coroutine until its response arrives, so the three requests are in flight simultaneously. The total wall time is close to the slowest single request rather than the sum of all three.

One caveat: asyncio.gather returns results in the order the coroutines were passed, not in completion order. If you need to process results as they arrive, use asyncio.as_completed instead.

Streaming Responses Without Blocking

When a completion is long, waiting for the full response before doing anything adds latency. The async client supports the same streaming interface as the sync client. Set stream=True and iterate with async for:

from openai import AsyncOpenAI

client = AsyncOpenAI()

async def stream_answer(question: str) -> None:
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)

Each iteration yields a chunk as soon as it arrives from the API. The event loop is free to run other coroutines between chunks, which matters when the same process is also serving HTTP requests or handling other I/O.

Error Handling and Retry Behavior

The async client raises the same exception types as the sync client. openai.APIError is the base class; openai.RateLimitError, openai.APIConnectionError, and openai.AuthenticationError are common subclasses. You handle them the same way you would with the sync client:

from openai import AsyncOpenAI, RateLimitError, APIConnectionError

client = AsyncOpenAI()

async def safe_completion(prompt: str) -> str | None:
    try:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
        )
        return response.choices[0].message.content
    except RateLimitError:
        # The library retries with backoff by default, so this
        # usually means the retries were exhausted.
        return None
    except APIConnectionError:
        return None

The library applies automatic retries with exponential backoff for transient failures such as connection errors and 429 rate limits. You can control that behavior with the max_retries argument when constructing the client. If you need custom backoff, set max_retries=0 and implement your own retry loop, but be careful to respect the Retry-After header returned by the API.

Timeouts and Cancellation

Network calls can hang. The async client accepts a timeout argument, either as a single float for all phases or as a Timeout object with separate read, connect, and write values:

from openai import AsyncOpenAI, Timeout

client = AsyncOpenAI(
    timeout=Timeout(connect=5.0, read=60.0, write=10.0),
)

Because the request is a coroutine, it also participates in normal asyncio cancellation. If the surrounding task is cancelled, the pending request is aborted and the coroutine raises asyncio.CancelledError. This is useful in a web server where a client disconnects and you want to stop the upstream API call rather than let it run to completion.

When the Async Client Does Not Help

The async client only helps when there is other work to overlap with the network I/O. A simple script that sends one request and prints the result gains nothing from async; the event loop has nothing to do while the request is pending. The same applies if you await requests strictly one after another.

Async also does not make CPU-bound work faster. If your code spends significant time processing tokens or building prompts locally, that work still occupies the event loop and blocks other coroutines. For heavy local processing, offload it to a thread pool with asyncio.to_thread or run it in a separate process.

There is a practical constraint on how many concurrent requests are useful. The OpenAI API applies rate limits per account and per model. Issuing hundreds of concurrent requests will produce 429 responses, and the retry logic will start backing off. The right concurrency level depends on your rate limit, so start with a modest number of concurrent requests and increase it only after observing the rate-limit responses.

Managing the Client Lifecycle

Creating an AsyncOpenAI client allocates an underlying httpx.AsyncClient, which maintains a connection pool. Creating a new client for every request discards that pool and forces a new TLS handshake and connection setup each time. In a long-running process, create the client once and reuse it.

If you are using FastAPI or another async framework, you can create the client at startup and close it at shutdown:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from openai import AsyncOpenAI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.openai_client = AsyncOpenAI()
    yield
    await app.state.openai_client.close()

app = FastAPI(lifespan=lifespan)

Calling close() releases the underlying connection pool. If you skip it, the connections are eventually cleaned up by garbage collection, but explicit closing avoids warnings and makes resource usage predictable. In a short-lived script, asyncio.run tears down the event loop and the client's resources are released when the process exits.