Building a Reliable Async Python Client for AI Applications

An AI application may look correct while its model-provider client is quietly accumulating failure modes: invalid payloads reach the network, every request opens a new connection, a slow stream occupies a worker forever, cancellation is swallowed, and retries multiply cost.

The fix is not “use async everywhere.” The fix is to make the system’s contracts, ownership, and failure policy explicit.

We will derive that design from a deliberately naive client.

1. The naive synchronous client

import requests

def complete(prompt):
    response = requests.post(
        "https://provider.example/v1/chat",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": "large-model", "prompt": prompt},
    )
    return response.json()["text"]

This works in a demo because the happy path hides every unanswered question:

Is prompt valid, and what shape does the response have?
How long may connection, upload, and response reading take?
Who owns the HTTP connection pool?
Which failures are safe to retry?
What happens if the caller disconnects?
How many calls may run concurrently?
How do tests replace the real provider?

The first invariant is already unclear: the function should either return a valid domain result or raise a known domain error. A raw dictionary and arbitrary library exceptions cannot protect it.

2. The first production failures

Suppose this function runs inside an asynchronous FastAPI endpoint. requests.post() blocks the thread executing it. If it runs on the event-loop thread, unrelated requests cannot make progress until it returns.

That reveals an important distinction:

Concurrency means multiple operations can make progress during overlapping time periods.
Parallelism means multiple operations execute at the same instant, usually on multiple cores.

Python's asyncio provides cooperative concurrency. A task runs until it reaches an operation that actually suspends, normally an await. The event loop can then run another ready task. Calling an async def function only creates a coroutine object; it does not run or schedule it. Creating a task schedules the coroutine, and awaiting it lets the caller observe its result or failure. This model is intended for I/O-bound work; CPU-heavy tokenization or local inference still blocks the loop and belongs in a thread/process boundary or a separate worker. Python's asyncio documentation describes it as a library for concurrent code using async/await.

Replacing def with async def while keeping blocking calls changes nothing useful:

async def complete(prompt):
    return requests.post(...)  # Still blocks the event-loop thread.

The rule is narrower: use asynchronous functions when their work must await asynchronous I/O. Keep validation, policy calculations, and ordinary transformations synchronous. Async is a resource-coordination mechanism, not a default function style.

3. Deriving typed boundaries

Python type hints help humans, editors, and static checkers reason about code, but they do not validate untrusted JSON at runtime. Provider input and output cross a trust boundary, so they need runtime validation.

from typing import Literal
from pydantic import BaseModel, ConfigDict, Field

class Message(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    role: Literal["system", "user", "assistant"]
    content: str = Field(min_length=1)

class CompletionRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    model: str = Field(min_length=1)
    messages: list[Message] = Field(min_length=1)
    temperature: float = Field(ge=0, le=2)

class Usage(BaseModel):
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)

class Completion(BaseModel):
    request_id: str
    text: str
    usage: Usage

Pydantic guarantees the shape of the resulting model after successful validation, not the truthfulness of the provider's claims. It may coerce values by default, so strict mode is appropriate when silent conversion would weaken the boundary. It also ignores extra fields by default, which is why extra="forbid" is a deliberate compatibility choice here. These behaviours are documented in Pydantic models.

This gives us two separate validation responsibilities:

Validate caller input before spending network capacity or money.
Validate provider output before allowing it into trusted application code.

Do not bury business rules inside a provider model. “Messages must fit this schema” is boundary validation. “This tenant may spend only $10 today” belongs to an application policy service.

For internal value objects that need no parsing—retry settings, limits, or computed decisions—a frozen dataclass is often simpler:

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class RetryPolicy:
    max_attempts: int = 3
    base_delay_seconds: float = 0.25
    max_delay_seconds: float = 4.0

Frozen objects reduce accidental mutation, but “frozen” is shallow: a mutable object stored inside can still change. Avoid shared mutable defaults; use factories such as Field(default_factory=list) when a collection really is required.

4. Deriving asynchronous execution

An asynchronous HTTP client lets the task yield while waiting for sockets. It does not make the provider faster; it prevents idle network time from monopolizing the event-loop thread.

import httpx
from pydantic import ValidationError

class ProviderClient:
    def __init__(self, http: httpx.AsyncClient, base_url: str) -> None:
        self._http = http
        self._base_url = base_url

    async def complete(self, request: CompletionRequest) -> Completion:
        response = await self._http.post(
            f"{self._base_url}/v1/chat",
            json=request.model_dump(mode="json"),
        )
        response.raise_for_status()
        try:
            return Completion.model_validate(response.json())
        except (ValueError, ValidationError) as exc:
            raise InvalidProviderResponse("provider returned an invalid body") from exc

The client method is async because it awaits network I/O. Model construction remains synchronous. The injected AsyncClient can be shared between tasks and provides connection pooling. HTTPX's async documentation specifically warns against repeatedly creating clients in a hot loop because that defeats pooling.

If several independent calls form one operation, use structured concurrency:

import asyncio

async with asyncio.TaskGroup() as group:
    summary = group.create_task(client.complete(summary_request))
    keywords = group.create_task(client.complete(keyword_request))

result = summary.result(), keywords.result()

A TaskGroup owns its child tasks, waits for them on exit, and cancels siblings when one fails. That gives failures a lexical lifetime instead of leaving “fire-and-forget” tasks running after their caller has failed. Python documents the resulting multi-failure behaviour as an ExceptionGroup in coroutines and tasks.

5. Resource ownership

An HTTP client owns sockets and a connection pool. Creating one per model call wastes handshakes; keeping one forever without closing it leaks resources.

The clean invariant is: the application composition root owns long-lived resources; request handlers borrow them.

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    timeout = httpx.Timeout(connect=3, read=30, write=10, pool=2)
    limits = httpx.Limits(max_connections=100, max_keepalive_connections=20)

    async with httpx.AsyncClient(timeout=timeout, limits=limits) as http:
        app.state.provider = ProviderClient(http, "https://provider.example")
        yield

app = FastAPI(lifespan=lifespan)

The asynchronous context manager makes ownership visible: acquire before yield, release after it, including during exceptional shutdown. FastAPI dependency injection can expose the already-created ProviderClient to endpoints; it should not construct a new HTTP pool per request. FastAPI's dependency documentation frames dependencies as declared application requirements, while the lifespan remains the owner of this shared resource.

For background jobs, the worker process—not an HTTP request—owns the client for the worker's lifetime. Durable model work should be represented as a queue job with persistent state; an untracked in-process task can disappear when the API process restarts.

6. Timeouts and cancellation

“Timeout = 30 seconds” is underspecified. At least five waits matter:

acquiring a connection from the pool;
connecting to the provider;
writing the request;
waiting between response bytes;
completing the whole logical operation, including retries and backoff.

HTTPX supports separate pool, connect, write, and read timeouts, and enforces timeouts by default. An outer asyncio.timeout() can enforce the total deadline:

import asyncio

async def complete_with_deadline(
    client: ProviderClient,
    request: CompletionRequest,
    deadline_seconds: float,
) -> Completion:
    try:
        async with asyncio.timeout(deadline_seconds):
            return await client.complete(request)
    except TimeoutError as exc:
        raise ProviderTimeout("completion deadline exceeded") from exc

Cancellation is different from an operational timeout. It means the caller no longer wants the work—for example, the request disconnected or a sibling task failed. Python delivers cancellation by raising asyncio.CancelledError at a suspension point. Cleanup belongs in finally, and cancellation should normally be re-raised:

try:
    return await operation()
except asyncio.CancelledError:
    logger.info("provider_call_cancelled")
    raise
finally:
    release_local_resource()

Do not translate cancellation into “provider unavailable,” and do not catch BaseException around business logic. TaskGroup and asyncio.timeout() use cancellation internally; swallowing it can break structured-concurrency guarantees. This is an explicit warning in the Python task-cancellation documentation.

Streaming adds another ownership boundary. The response must remain open while the async iterator is consumed:

from collections.abc import AsyncIterator

async def stream_text(self, request: CompletionRequest) -> AsyncIterator[str]:
    async with self._http.stream(
        "POST",
        f"{self._base_url}/v1/chat/stream",
        json=request.model_dump(mode="json"),
    ) as response:
        response.raise_for_status()
        async for line in response.aiter_lines():
            if line:
                yield line

An async generator naturally couples incremental production, backpressure, and cleanup: the producer advances only when the consumer asks for the next item, and exiting the context closes the response. Each stream event still needs validation before becoming a trusted domain event.

7. Error modelling

Library exceptions describe transport mechanics; application exceptions should describe decisions callers can make.

class ProviderError(Exception):
    """Base error exposed by the package."""

class InvalidRequest(ProviderError):
    pass  # Non-recoverable without changing input.

class AuthenticationFailed(ProviderError):
    pass  # Non-recoverable without configuration change.

class ProviderTimeout(ProviderError):
    pass  # Potentially recoverable.

class RateLimited(ProviderError):
    def __init__(self, retry_after: float | None = None):
        self.retry_after = retry_after

class ProviderUnavailable(ProviderError):
    pass  # Potentially recoverable.

class InvalidProviderResponse(ProviderError):
    pass  # Provider violated its response contract.

class StreamInterrupted(ProviderError):
    pass  # Partial output may already be visible.

Translate exceptions once, near the provider boundary, and preserve the original cause with raise DomainError(...) from exc. HTTPX distinguishes request failures from HTTP status failures in its exception hierarchy; Pydantic raises ValidationError when output cannot satisfy the response model.

Do not make retryable: bool the only meaning of an error. Retryability depends on operation semantics, attempt count, deadline, provider instructions, and whether output has already escaped to the caller.

8. Retries and concurrency limits

A retry is a new request, not a continuation of the old one. The provider may have completed and billed the first request even if the response was lost.

Therefore retry only when all of these are true:

The failure is plausibly transient: selected network errors, 429, or selected 5xx responses.
The deadline permits another attempt.
The operation is safe to duplicate, or the provider supports an idempotency key.
No streamed output has been exposed to the caller.

Use bounded exponential backoff with jitter and respect Retry-After when valid. Release concurrency capacity during backoff; otherwise sleeping retries occupy permits while useful work waits.

import asyncio
import random

def backoff(attempt: int, policy: RetryPolicy) -> float:
    ceiling = min(policy.max_delay_seconds, policy.base_delay_seconds * 2**attempt)
    return random.uniform(0, ceiling)

async def call_one_attempt(self, request: CompletionRequest) -> Completion:
    async with self._semaphore:
        return await self._send_and_validate(request)

The semaphore protects provider capacity and local memory from an unbounded number of in-flight calls. HTTP connection limits protect sockets. They solve related but different problems: a semaphore is an application admission rule; the pool is a transport resource limit. Neither replaces provider rate limiting, which may also require a token bucket or queue based on requests and tokens per minute.

Retry policy is a function passed into the client, not scattered sleep() calls. Functions are values in Python, so policies, clocks, and jitter sources can be injected for deterministic tests. Decorators can package cross-cutting policy, but a generic retry decorator is dangerous if it catches cancellation, retries validation errors, or hides whether a streamed request emitted data.

9. Testing strategy

Reliable async tests assert failure behaviour, not only happy-path JSON.

Unit tests

Keep pure decisions synchronous and cheap to test:

which statuses are retryable;
backoff bounds;
exception translation;
request and response validation;
log redaction.

Property-based thinking asks for invariants over classes of inputs: backoff never exceeds its cap, negative token counts never validate, and arbitrary unknown fields never enter a strict boundary. A library such as Hypothesis can generate those cases, but the important part is specifying the invariant first.

Client tests without real network calls

Inject an httpx.AsyncClient configured with MockTransport and assert the outgoing method, headers, payload, and mapped result. Use pytest.mark.asyncio for coroutine tests and fixtures for resource setup/teardown.

import httpx
import pytest

@pytest.mark.asyncio
async def test_rejects_invalid_provider_response():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"text": 42})

    transport = httpx.MockTransport(handler)
    async with httpx.AsyncClient(transport=transport) as http:
        client = ProviderClient(http, "https://mock.provider")
        with pytest.raises(InvalidProviderResponse):
            await client.complete(valid_request())

Integration tests

Run a small mock provider that can deliberately:

return malformed JSON;
delay connection or chunks;
emit 429, 401, and 503;
disconnect halfway through a stream;
record idempotency keys and request counts.

Then test the complete boundary through HTTP. HTTPX's ASGI transport can call an async test application in process.

The essential cancellation test starts a request, confirms it reached a suspension point, cancels its task, and asserts:

CancelledError reaches the caller;
the response and permits are released;
no retry occurs;
no completion log falsely reports success.

Avoid mocking every method on ProviderClient; that can prove only that the mock behaves as configured. Mock the external boundary and let serialization, error mapping, streaming, and cleanup execute for real.

10. Final architecture

The package can remain small because each module owns one kind of decision:

ai_provider/
├── models.py       # Pydantic boundary models and stream events
├── errors.py       # Stable domain exception hierarchy
├── policies.py     # Retry, deadline, and concurrency decisions
├── client.py       # Request orchestration and exception translation
├── streaming.py    # Framing, event validation, partial-stream rules
├── config.py       # Frozen settings/value objects
└── logging.py      # Structured, redacted event helpers

tests/
├── unit/           # Pure validation and policy tests
├── client/         # MockTransport success/failure tests
└── integration/    # Stateful mock-provider tests

The dependency direction is deliberate:

models and errors know nothing about HTTP;
policies operate on domain facts, not FastAPI objects;
the client depends on those modules and on an injected HTTP client;
FastAPI or a queue worker acts as the composition root and owns resources;
tests replace only the external boundary.

Configuration should be loaded and validated once at startup, then passed explicitly. Secrets should never appear in model representations or logs. Structured logs should include a correlation ID, provider, model, attempt number, latency, status, and token usage—but not prompts, credentials, or raw provider payloads by default.

11. Key invariants

A production design is easier to reconstruct from invariants than from framework syntax:

Trusted code receives only validated models. Both caller input and provider output are boundary data.
The event loop never performs blocking I/O or unbounded CPU work. Async provides cooperative I/O concurrency, not CPU parallelism.
Every resource has one visible owner. The application or worker owns the HTTP pool; individual calls borrow it.
Every wait is bounded. Transport timeouts and an overall deadline protect different waits.
Cancellation propagates after cleanup. It is control flow, not an ordinary provider failure.
Retries are bounded and semantically safe. Transience alone does not make a request safe to duplicate.
Concurrency is admitted deliberately. Semaphores, connection limits, and provider quotas protect different capacities.
Streams validate incrementally and close reliably. Once partial output escapes, transparent retry is normally unsafe.
Callers see stable domain errors. Transport and validation details remain available as chained causes.
Tests control time and the external boundary. They exercise timeouts, cancellation, malformed responses, retry exhaustion, and cleanup—not just successful completions.

That is the real role of production Python in an AI system: not clever syntax, but executable boundaries around uncertainty. The model may remain probabilistic; resource ownership, validation, failure classification, and recovery policy should not be.