Client code: streaming, async, protocols

Most tutorials about LLM inference focus on the server. Fair enough: that's where the GPU lives and where the hard optimization happens. But the client code matters more than people think. A bad client can block on a streaming response and waste the low-latency advantage of token-by-token delivery. A synchronous client sending requests one at a time will never saturate your server. And a client that doesn't handle mid-stream errors will leave users staring at a broken response.

Today I want to walk through the three layers of building a good inference client: the streaming protocol, the async transport, and the patterns that make production clients robust.

Server-Sent Events (SSE)

The dominant protocol for streaming LLM responses is Server-Sent Events over HTTP. The OpenAI API uses it, vLLM uses it, and most serving frameworks follow the same format. SSE is simple: the server holds the HTTP connection open and sends lines prefixed with data:, each containing a JSON chunk with one or more tokens.

# What an SSE stream looks like on the wire
data: {"choices":[{"delta":{"content":"Hello"},"index":0}]}

data: {"choices":[{"delta":{"content":" world"},"index":0}]}

data: {"choices":[{"delta":{"content":"!"},"index":0}]}

data: [DONE]

Each data: line is followed by a blank line (the SSE event separator). The client reads line by line, parses each JSON payload, extracts the token delta, and appends it to the growing response. The stream ends with data: [DONE].

The simplest Python client using requests:

import requests, json

def stream_chat(prompt, base_url="http://localhost:8000"):
    resp = requests.post(
        f"{base_url}/v1/chat/completions",
        json={
            "model": "llama-3.1-70b",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True,
        },
        stream=True,
    )
    resp.raise_for_status()

    for line in resp.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data: "):
            continue
        payload = line[len("data: "):]
        if payload == "[DONE]":
            break
        chunk = json.loads(payload)
        token = chunk["choices"][0]["delta"].get("content", "")
        print(token, end="", flush=True)

This works for demos but has a problem: requests is synchronous. While you're reading one stream, the entire thread is blocked. You can't send a second request concurrently.

Going async with httpx or aiohttp

For production clients, you need async I/O. Python's asyncio lets you run many concurrent requests on a single thread, each consuming its own SSE stream independently.

import httpx, asyncio, json

async def stream_chat_async(prompt, client, base_url="http://localhost:8000"):
    tokens = []
    async with client.stream(
        "POST",
        f"{base_url}/v1/chat/completions",
        json={
            "model": "llama-3.1-70b",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True,
        },
    ) as resp:
        resp.raise_for_status()
        async for line in resp.aiter_lines():
            if not line or not line.startswith("data: "):
                continue
            payload = line[len("data: "):]
            if payload == "[DONE]":
                break
            chunk = json.loads(payload)
            token = chunk["choices"][0]["delta"].get("content", "")
            tokens.append(token)
    return "".join(tokens)

async def main():
    async with httpx.AsyncClient(timeout=120) as client:
        prompts = ["Explain transformers", "What is KV cache?", "How does batching work?"]
        tasks = [stream_chat_async(p, client) for p in prompts]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r[:100], "...")

asyncio.run(main())

With httpx.AsyncClient, all three requests run concurrently. Each one reads its SSE stream as tokens arrive, without blocking the others. The connection pool inside httpx handles TCP reuse and connection limits.

WebSockets: the alternative

Some inference servers (notably TensorRT-LLM's Triton backend) support WebSocket streaming instead of SSE. WebSockets are bidirectional and have lower per-message overhead than SSE (no HTTP headers per chunk). The tradeoff: WebSockets require more complex client code, don't work through all proxies, and aren't cacheable.

import websockets, asyncio, json

async def stream_ws(prompt, ws_url="ws://localhost:8000/v1/stream"):
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps({
            "prompt": prompt,
            "max_tokens": 256,
        }))
        async for msg in ws:
            data = json.loads(msg)
            if data.get("done"):
                break
            print(data["token"], end="", flush=True)

For most deployments, SSE over HTTP is the right choice. It's simpler, better supported by load balancers and CDNs, and the per-token overhead of SSE framing is negligible compared to the actual token generation time.

gRPC: structured and typed

For internal service-to-service calls (not user-facing), gRPC with server-side streaming offers strong typing via protobuf, connection multiplexing via HTTP/2, and built-in deadline propagation. NVIDIA Triton Inference Server exposes a gRPC API, and some teams use gRPC between their API gateway and the inference backend.

The downside: gRPC streaming is harder to debug (binary protocol), requires protobuf schema management, and doesn't play well with browser clients. Use it for backend-to-backend, SSE for client-facing.

Production patterns

Beyond the basic streaming loop, production clients need several things:

• Timeout handling. Set both a connection timeout (how long to wait for the server to start responding) and a read timeout (how long between chunks). If a server hangs mid-generation, you don't want the client to wait forever. A 120-second total timeout with a 30-second inter-chunk timeout is a reasonable starting point.
• Retry with backoff. If the server returns 429 (rate limited) or 503 (overloaded), retry with exponential backoff. Don't retry on 400 (bad request) or mid-stream errors (the partial response is already delivered).
• Partial response recovery. If the stream breaks after 200 tokens, you have a partial response. Decide whether to return it to the user (with a note that generation was interrupted) or retry from scratch. For chat applications, returning the partial response is usually better.
• Concurrency limits. Even with async, don't fire 1000 concurrent requests at a server with 32 slots. Use a semaphore to cap concurrency at a level the server can handle.

import asyncio

async def bounded_inference(prompts, client, max_concurrent=32):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited(prompt):
        async with semaphore:
            return await stream_chat_async(prompt, client)

    return await asyncio.gather(*[limited(p) for p in prompts])

The OpenAI Python SDK as a reference

The openai Python package is worth studying as a reference client implementation. It handles SSE parsing, async/sync modes, retry logic, streaming type safety, and timeout configuration. If you're building a client for a custom inference endpoint, start with httpx (which the OpenAI SDK uses internally) and add the patterns above.

The server does the hard work of generating tokens. The client's job is to not waste that work: consume tokens as they arrive, handle failures gracefully, and keep the pipeline full with concurrent requests. Get these right and your inference stack feels fast end to end.

Up next in the Deep Implementation phase: building the autoregressive decoder loop in PyTorch, where we go from calling APIs to implementing the generation loop ourselves.