Async HTTP with Python

The nightly digest job pushes 200 chunks through a summarization endpoint. Each call takes about 12 seconds — generation is slow — and the job ran sequentially: request, wait, request, wait. Two hundred times twelve is forty minutes of a worker idling between responses, every night. The fix is not a faster model; it is overlapping the waits.

The sequential baseline

import httpx, os, time

URL = "https://api.openai.com/v1/chat/completions"
HEADERS = {"Authorization": "Bearer " + os.environ["OPENAI_API_KEY"]}
prompts = [f"Summarize chunk {i}" for i in range(200)]

def payload(p: str) -> dict:
    return {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": p}]}

def summarize_all() -> list[str]:
    out = []
    with httpx.Client(timeout=60.0) as client:
        for p in prompts:
            r = client.post(URL, headers=HEADERS, json=payload(p))
            out.append(r.json()["choices"][0]["message"]["content"])
    return out

start = time.perf_counter()
summarize_all()
print(f"{time.perf_counter() - start:.1f}s")

Output: 2398.6s — forty minutes, nearly all of it waiting on a socket.

Overlap the waits

import asyncio, httpx, os, time

async def summarize_all() -> list[str]:
    async with httpx.AsyncClient(timeout=60.0) as client:
        tasks = [
            client.post(URL, headers=HEADERS, json=payload(p))
            for p in prompts
        ]
        responses = await asyncio.gather(*tasks)
    return [r.json()["choices"][0]["message"]["content"] for r in responses]

start = time.perf_counter()
results = asyncio.run(summarize_all())
print(f"{time.perf_counter() - start:.1f}s", len(results))

Output: 14.2s 200

Four lines do the real work:

• AsyncClient(...) — pooling lives on the client. A client per request would renegotiate TLS two hundred times and erase most of the gain.
• tasks = [... for p in prompts] — building the list sends nothing yet; each client.post(...) produces an awaitable that is still cold.
• await asyncio.gather(*tasks) — starts all of them on one event loop. While one response streams in, the loop yields to others; the idle waits are what actually overlap.
• asyncio.run(...) — owns the event loop for the whole script. Calling it inside a running loop (Jupyter, an async framework) raises RuntimeError; await is the form to use there.

Wall time fell from 2398.6s to 14.2s because two hundred network waits now overlap instead of queueing. Speed introduced two problems of its own — nothing bounds the burst, and a single hung connection can stall the batch — and the httpx version this runs on is pinned in Environments and Dependency Hygiene. The payload shapes being indexed come from Python Data Structures for API Work.

The 14.2-second run above hides two failures that show up the first time a provider has a bad minute. One hung TCP connection parks its task indefinitely, and gather propagates the first exception while the other 199 results — summaries you paid for — are never collected.

Give each phase of the request its own budget

httpx ships a flat 5-second default applied to connect, read, write, and pool alike. Five seconds is fine for a handshake and fatal for generation, which routinely runs longer. Specify them separately:

import httpx

timeout = httpx.Timeout(
    connect=5.0,    # TCP + TLS handshake: fail fast, the address is wrong or down
    read=60.0,      # generation: this is the budget the model actually needs
    write=10.0,     # uploading the prompt
    pool=5.0,       # waiting for a pooled connection slot
)

async with httpx.AsyncClient(timeout=timeout) as client:
    ...

A read that exceeds its budget raises httpx.ReadTimeout. That is one failed request, not a reason to abort the batch.

Keep the batch alive when one request dies

responses = await asyncio.gather(*tasks, return_exceptions=True)

ok, failed = [], []
for prompt, r in zip(prompts, responses):
    if isinstance(r, Exception):
        failed.append((prompt, type(r).__name__))
    else:
        ok.append(r.json()["choices"][0]["message"]["content"])

print(f"ok={len(ok)} failed={len(failed)}")
print(failed[:2])

Output:

ok=197 failed=3
[('Summarize chunk 88', 'httpx.ReadTimeout'), ('Summarize chunk 151', 'httpx.ConnectTimeout')]

• return_exceptions=True — exceptions become values in the results list, at the position of their task, so zip still pairs each prompt with its outcome. Without it, one failure discards a batch that already succeeded.
• isinstance(r, Exception) — the list is heterogeneous now: payloads and exceptions side by side. Type-check before indexing into a response.

Handling follows the classification, not the habit:

A timeout without a budget is an outage discovered late. The connect/read split decides whether you learn about a dead endpoint in five seconds or discover a stalled generation sixty seconds in.

Firing all 200 requests at once worked in the demo and returned eight 429s in production. Providers meter requests per minute and tokens per minute; a burst over either limit gets throttled, and every rejected request is work your pipeline must schedule again. Concurrency needs a ceiling.

Cap what is in flight

import asyncio, httpx

TIMEOUT = httpx.Timeout(connect=5.0, read=60.0, write=10.0, pool=5.0)

async def summarize_all(limit: int = 8) -> tuple[list[str], list]:
    sem = asyncio.Semaphore(limit)
    ok, errors = [], []

    async def one(client: httpx.AsyncClient, p: str) -> None:
        async with sem:                     # only `limit` tasks hold this at once
            try:
                r = await client.post(URL, headers=HEADERS, json=payload(p))
                r.raise_for_status()
                ok.append(r.json()["choices"][0]["message"]["content"])
            except httpx.HTTPError as exc:
                errors.append((p, type(exc).__name__))

    async with httpx.AsyncClient(timeout=TIMEOUT) as client:
        await asyncio.gather(*(one(client, p) for p in prompts))

    return ok, errors

ok, errors = asyncio.run(summarize_all())
print(len(ok), len(errors))

Output: 200 0

Two details carry the design:

• async with sem wraps only the request. The slot is released the moment the response lands — before append — so the queue keeps moving instead of waiting on bookkeeping.
• The semaphore is created outside the per-task coroutine: one instance for the whole batch. Declaring it inside one() would give every task its own private permit and bound nothing.

What the ceiling costs and buys

Measured on the same 200-chunk batch:

The flow:

200 prompts
     │
     ▼
┌────────────────┐  acquire   ┌─────────────────┐
│ asyncio.gather │───────────▶│  Semaphore(8)   │
│ task factory   │◀───────────│  8 permits      │
└────────────────┘  release   └────────┬────────┘
                                       │ 8 requests in flight
                                       ▼
                             provider API (rate quota)
                                       │
                                       ▼
                                ok[] / errors[]

Semaphore(8) is the throughput knob: raise it until the 429 column stops being zero, then stop. Throughput that ignores the quota line trades a thirty-second saving for a retry storm — and the retry policy that handles those responses when they do appear is Rate Limiting and Retries.