n4nAI
Open console →
API integrationHow-toPython Raw REST Calls (requests/httpx)

Async LLM API calls in Python with httpx.AsyncClient

Learn to build concurrent, resilient LLM API calls in Python using httpx.AsyncClient with retries, streaming, routing, and usage metering.

n4n Team3 min read654 words

Audio narration

Coming soon — every post will get a voice note here.

Most LLM integrations bottleneck on synchronous I/O the moment you fan out to multiple prompts. Using a python httpx asyncclient llm api call pattern lets you issue dozens of concurrent requests against an OpenAI-compatible endpoint without threading headaches. This guide walks through a production-shaped setup: connection pooling, retries, streaming, and structured error handling.

Step 1: Install dependencies and configure the client

Install httpx with pip. Use a recent version (0.27+) that supports the async transport improvements and proper connection recycling.

pip install "httpx>=0.27"

Create exactly one AsyncClient per process and reuse it. The client manages a connection pool, TCP keep-alive, and DNS caching. Creating a new client per request will exhaust file descriptors under load.

import httpx

client = httpx.AsyncClient(
    base_url="https://api.openai.com/v1",
    headers={"Authorization": "Bearer $API_KEY"},
    timeout=httpx.Timeout(30.0, connect=5.0),
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
)

If you control the gateway, enable HTTP/2 by passing http2=True. Most LLM gateways accept HTTP/1.1 fine; HTTP/2 helps when you push past 100 concurrent streams. Always close the client on shutdown:

async def shutdown():
    await client.aclose()

Step 2: Send a single chat completion

Define an async function that posts to /chat/completions. The python httpx asyncclient llm api request shape mirrors the OpenAI REST contract exactly, so any OpenAI-compatible endpoint works without SDK lock-in.

import asyncio

async def complete(prompt: str, model: str = "gpt-4o-mini") -> dict:
    resp = await client.post(
        "/chat/completions",
        json={
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
        },
    )
    resp.raise_for_status()
    return resp.json()

async def main():
    data = await complete("Summarize: async I/O matters.")
    print(data["choices"][0]["message"]["content"])

asyncio.run(main())

Verify success

Run the script. A generated string prints to stdout. Internally, check resp.status_code == 200 and that data["choices"] is a non-empty list. A 401 means the bearer token is missing or malformed. A 429 means you hit a rate limit and should back off (see Step 4).

Step 3: Fan out concurrent requests

The real value of async is cheap concurrency. Use asyncio.gather to batch prompts. The python httpx asyncclient llm api client pools connections, so 50 parallel calls add little overhead beyond socket memory.

async def batch(prompts: list[str]) -> list[object]:
    tasks = [complete(p) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

async def main():
    prompts = [f"Translate to French: {i}" for i in range(20)]
    results = await batch(prompts)
    ok = [r for r in results if isinstance(r, dict)]
    print(f"{len(ok)}/{len(prompts)} succeeded")
    for r in results:
        if isinstance(r, Exception):
            print("failed:", type(r).__name__)

asyncio.run(main())

return_exceptions=True prevents one bad response from killing the whole batch. In production, route exceptions to a dead-letter queue or retry bucket instead of printing.

Step 4: Retries, timeouts, and fallback

Networks flap. Providers return 503. Wrap the call in a bounded retry loop with exponential backoff. httpx raises httpx.TimeoutException on slow connections and httpx.HTTPStatusError on 4xx/5xx when you call raise_for_status().

async def complete_retry(prompt: str, attempts: int = 3) -> dict:
    for i in range(attempts):
        try:
            return await complete(prompt)
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            if i == attempts - 1:
                raise
            await asyncio.sleep(2 ** i)

A gateway such as n4n.ai provides automatic fallback when a provider is rate-limited or degraded, but your client still owns timeout budgets and retry cadence. Do not assume the server will mask a dead TCP connection—local timeouts are mandatory.

Step 5: Stream tokens

For chat UIs, block on a full response is unacceptable. Use client.stream with stream: True in the payload. The server returns Server-Sent Events. Parse each data: line as JSON.

import json

async def stream_complete(prompt: str):
    async with client.stream(
        "POST",
        "/chat/completions",
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True,
        },
    ) as resp:
        async for line in resp.aiter_lines():
            if not line.startswith("data: "):
                continue
            payload = line[6:]
            if payload == "[DONE]":
                break
            yield json.loads(payload)["choices"][0]["delta"].get("content", "")

async def main():
    async for tok in stream_complete("Tell me a joke"):
        print(tok, end="", flush=True)

asyncio.run(main())

You get incremental strings instead of one multi-second wait. Always handle the [DONE] sentinel and JSON decode errors defensively.

Step 6: Pass routing and cache hints

Multi-provider gateways accept headers to pin a backend or enable prompt caching. n4n.ai honors client routing directives and forwards provider cache-control hints, so set them explicitly when you need determinism or cost control.

headers = {
    "X-Route-To": "provider:anthropic",
    "Cache-Control": "max-age=3600",
}
resp = await client.post(
    "/chat/completions",
    json={
        "model": "claude-3-5-sonnet",
        "messages": [{"role": "user", "content": "Long static context..."}],
    },
    headers=headers,
)

Against a raw provider these headers are ignored but harmless. In a routed setup they cut latency and token cost by reusing cached prefix computations.

Step 7: Measure usage and verify metering

OpenAI-compatible responses embed a usage object. Read it per call to track token spend.

data = await complete("Explain asyncio")
usage = data["usage"]
print("prompt", usage["prompt_tokens"], "completion", usage["completion_tokens"])

When you route through a gateway with per-token usage metering, the same usage block reflects the actual billed counts. Log it alongside a request ID:

import logging
logging.info("req=%s prompt=%d completion=%d",
             data["id"], usage["prompt_tokens"], usage["completion_tokens"])

End-to-end verification

Run the Step 3 batch with five prompts. Confirm:

  • Status codes are 200 or exceptions caught and logged.
  • Streaming (Step 5) yields multiple non-empty chunks.
  • Usage prints non-zero token counts.

That proves the python httpx asyncclient llm api pipeline works from single call to concurrent fan-out.

Step 8: Bound concurrency with a worker pool

asyncio.gather fires all tasks at once. If your provider quota is strict, wrap calls in a semaphore or queue.

sem = asyncio.Semaphore(10)

async def bounded_complete(p: str) -> dict:
    async with sem:
        return await complete_retry(p)

async def bounded_batch(prompts: list[str]) -> list[object]:
    return await asyncio.gather(
        *(bounded_complete(p) for p in prompts),
        return_exceptions=True,
    )

This caps simultaneous connections at 10 regardless of list size. Combine with the retry logic from Step 4 and the streaming generator from Step 5 for a complete client module.

Keep the AsyncClient at module scope, never per request. Use async with for streams to guarantee socket release. The pattern above holds up under sustained load and keeps your LLM calls cheap, observable, and free of thread pools.

Tagspythonhttpxasyncllm-api

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
API integrationBuilding a minimal Python LLM client with httpxBuild a python minimal llm client httpx from scratch with auth, streaming, retries, and OpenAI-compatible calls in this hands-on tutorial for engineers.API integrationHandling server-sent events from LLM APIs in PythonLearn how to handle python server-sent events llm api responses in Python with httpx and requests, including SSE parsing, retries, and streaming pitfalls.API integrationHow to parse streaming JSON chunks from an LLM API in PythonLearn how to python parse streaming json llm responses in Python using httpx: buffer SSE chunks, split lines, and decode JSON deltas safely.API integrationPython httpx vs requests for calling LLM APIsA pragmatic head-to-head comparison of python httpx vs requests llm API calls across capabilities, latency, ergonomics, and verdicts for engineers.API integrationStreaming LLM responses in Python with httpx and SSEStream LLM responses in Python with httpx and SSE. Build a minimal OpenAI-compatible client: sync/async code, error handling, retries, and production checklist.API integrationBuilding a retry-with-backoff wrapper for LLM REST callsBuild a robust python retry backoff llm rest api client with httpx. Hands-on tutorial covering exponential backoff, jitter, and flaky mock testing.

All python raw rest calls (requests/httpx) posts →