Most LLM SDKs pull in more dependencies and abstraction than a direct integration needs. This tutorial builds a python minimal llm client httpx that speaks the OpenAI-compatible REST contract, handles streaming, and retries on transient failures in well under 200 lines. You will end up with a client you can point at any compliant endpoint, including your own proxy or a multi-provider gateway.
Prerequisites
- Python 3.10 or newer (uses
list[dict]typing) httpxinstalled:pip install httpx- An API key for an OpenAI-compatible service. OpenAI works; so does any gateway that exposes
/v1/chat/completions.
python -m venv .venv && source .venv/bin/activate
pip install httpxhttpx is the modern choice: it supports HTTP/2, async, and connection pooling out of the box, unlike requests, which is sync-only and has stalled on major feature work. For a minimal client, that matters.
Core client skeleton
We start with a thin wrapper around httpx.Client. The only persistent state is the base URL, auth header, and a reusable connection pool.
import httpx
class MinimalLLMClient:
def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
self.base_url = base_url.rstrip("/")
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
self.client = httpx.Client(timeout=timeout, headers=self.headers)That is the entire setup. httpx.Client handles connection reuse, so repeated calls to the same host are cheap.
Non-streaming chat completion
The simplest useful method posts a JSON body to /chat/completions and returns the parsed response. We call raise_for_status() immediately so HTTP errors surface as exceptions.
def chat(self, model: str, messages: list[dict], temperature: float = 0.7) -> dict:
resp = self.client.post(
f"{self.base_url}/chat/completions",
json={"model": model, "messages": messages, "temperature": temperature},
)
resp.raise_for_status()
return resp.json()Test it from the REPL:
client = MinimalLLMClient("https://api.openai.com/v1", "sk-your-key")
out = client.chat("gpt-4o-mini", [{"role": "user", "content": "Say hi in three words."}])
print(out["choices"][0]["message"]["content"])Expected output (varies by model):
Hello! How can I help?The returned dict mirrors the OpenAI schema. Access out["usage"] for token counts if the server populates it.
Streaming tokens
For interactive apps you want tokens as they generate. The OpenAI streaming format is newline-delimited data: {json} frames ending with data: [DONE]. httpx.stream gives us an iterator over lines without buffering the full response.
import json
def stream_chat(self, model: str, messages: list[dict], temperature: float = 0.7):
with self.client.stream(
"POST",
f"{self.base_url}/chat/completions",
json={"model": model, "messages": messages, "temperature": temperature, "stream": True},
) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if not line.startswith("data: "):
continue
payload = line[len("data: "):].strip()
if payload == "[DONE]":
break
yield json.loads(payload)Consume it like this:
for chunk in client.stream_chat("gpt-4o-mini", [{"role": "user", "content": "Count to 3."}]):
delta = chunk["choices"][0].get("delta", {})
if "content" in delta:
print(delta["content"], end="", flush=True)
print()Expected output prints 1 2 3 (or similar) incrementally, not all at once. The generator yields one JSON object per server-sent event.
Retries on transient errors
Providers rate-limit and occasionally 503. A minimal retry loop with exponential backoff keeps the client usable in production without a heavy framework.
from httpx import HTTPStatusError, RequestError
import time
def chat_with_retry(self, model: str, messages: list[dict], retries: int = 3, backoff: float = 0.5) -> dict:
for attempt in range(retries):
try:
return self.chat(model, messages)
except HTTPStatusError as e:
if e.response.status_code in (429, 500, 502, 503):
time.sleep(backoff * (2 ** attempt))
continue
raise
except RequestError:
time.sleep(backoff * (2 ** attempt))
continue
raise RuntimeError("Exhausted retries")This catches 429 and common 5xx, sleeps backoff * 2**attempt, and rethrows permanent errors. Wire it into your call sites instead of chat when resilience matters.
Pointing at a multi-provider gateway
The client above is agnostic to the backend. If you point it at n4n.ai, its OpenAI-compatible endpoint addresses 240+ models and automatically falls back when a provider is rate-limited or degraded, so the same chat() call works without branching logic. You only change the base URL and key.
client = MinimalLLMClient("https://api.n4n.ai/v1", "your-gateway-key")
out = client.chat("anthropic/claude-3.5-sonnet", [{"role": "user", "content": "Ping"}])The request shape is identical; the gateway forwards cache-control hints and per-token metering comes back in the usage block.
Reading usage and billing signals
OpenAI-compatible responses include a usage object. Capture it for logging or cost tracking.
out = client.chat("gpt-4o-mini", [{"role": "user", "content": "Hi"}])
print(out.get("usage"))Example output:
{"prompt_tokens": 5, "completion_tokens": 8, "total_tokens": 13}If your endpoint supports client routing directives (e.g., x-routing-key header), add them to self.headers or pass per-request headers via client.post(headers=...).
Complete client listing
For reference, here is the full class with streaming and retries combined.
import httpx
import json
import time
from httpx import HTTPStatusError, RequestError
class MinimalLLMClient:
def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
self.base_url = base_url.rstrip("/")
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
self.client = httpx.Client(timeout=timeout, headers=self.headers)
def chat(self, model: str, messages: list[dict], temperature: float = 0.7) -> dict:
resp = self.client.post(
f"{self.base_url}/chat/completions",
json={"model": model, "messages": messages, "temperature": temperature},
)
resp.raise_for_status()
return resp.json()
def chat_with_retry(self, model: str, messages: list[dict], retries: int = 3, backoff: float = 0.5) -> dict:
for attempt in range(retries):
try:
return self.chat(model, messages)
except HTTPStatusError as e:
if e.response.status_code in (429, 500, 502, 503):
time.sleep(backoff * (2 ** attempt))
continue
raise
except RequestError:
time.sleep(backoff * (2 ** attempt))
continue
raise RuntimeError("Exhausted retries")
def stream_chat(self, model: str, messages: list[dict], temperature: float = 0.7):
with self.client.stream(
"POST",
f"{self.base_url}/chat/completions",
json={"model": model, "messages": messages, "temperature": temperature, "stream": True},
) as resp:
resp.raise_for_status()
for line in resp.iter_lines():
if not line.startswith("data: "):
continue
payload = line[len("data: "):].strip()
if payload == "[DONE]":
break
yield json.loads(payload)Async variant
If your app is built on asyncio, swap httpx.Client for httpx.AsyncClient. The method shapes stay the same.
import httpx
class AsyncMinimalLLMClient:
def __init__(self, base_url: str, api_key: str, timeout: float = 30.0):
self.base_url = base_url.rstrip("/")
self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
self.client = httpx.AsyncClient(timeout=timeout, headers=self.headers)
async def chat(self, model: str, messages: list[dict]) -> dict:
resp = await self.client.post(
f"{self.base_url}/chat/completions",
json={"model": model, "messages": messages},
)
resp.raise_for_status()
return resp.json()
async def close(self):
await self.client.aclose()Usage:
async def main():
c = AsyncMinimalLLMClient("https://api.openai.com/v1", "sk-...")
out = await c.chat("gpt-4o-mini", [{"role": "user", "content": "Hi"}])
print(out["choices"][0]["message"]["content"])
await c.close()
import asyncio
asyncio.run(main())The async path gives you concurrency across many in-flight requests without threading.
Extending the python minimal llm client httpx
The base is deliberately small. From here you can:
- Add a
list_modelsmethod hitting/models. - Pass
response_format={"type": "json_object"}for structured output. - Inject
extra_headersper call to honor provider-specific cache hints. - Wrap
stream_chatin an async generator for WebSocket frontends.
Keep the surface area tight. A python minimal llm client httpx built on raw REST avoids SDK lock-in and makes the wire protocol visible, which is exactly what you want when debugging production latency or token accounting. The OpenAI-compatible contract is stable enough that you rarely need SDK version bumps—your own code is the only moving part.