How to Build a Streaming Inference API with Python and asyncio

When users submit a prompt to a language model, they're often waiting 5–30 seconds for a complete response. If you buffer everything server-side and return it in one shot, users see a blank screen and either refresh or give up. Streaming fixes this: the first token arrives in under a second, output flows visibly, and you avoid proxy timeouts on long responses.

Here's how to wire this correctly with Python and asyncio — including the non-obvious parts most tutorials skip.

Why asyncio fits this workload

Streaming LLM responses are IO-bound. Each token arrives asynchronously from an upstream model API; your server is mostly waiting, not computing. asyncio lets a single process handle hundreds of concurrent streaming sessions without spawning threads per connection.

The pattern is straightforward: an async generator yields chunks as they arrive from the upstream model, and FastAPI wraps that generator in a StreamingResponse. What makes it production-worthy is the error handling around it.

The async generator

import json
import httpx
from typing import AsyncGenerator

API_KEY = "your-api-key-here"

async def stream_inference(
    prompt: str,
    model: str = "gpt-4o",
) -> AsyncGenerator[str, None]:
    """Stream tokens from an upstream LLM API."""
    async with httpx.AsyncClient(timeout=60.0) as client:
        async with client.stream(
            "POST",
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": True,
            },
        ) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                if not line.startswith("data: "):
                    continue
                chunk = line[6:]
                if chunk == "[DONE]":
                    break
                try:
                    data = json.loads(chunk)
                    content = data["choices"][0]["delta"].get("content", "")
                    if content:
                        yield content
                except (json.JSONDecodeError, KeyError):
                    continue

Use httpx.AsyncClient with .stream(), not requests with stream=True. The requests library is synchronous under the hood — it blocks the event loop, killing your concurrency.

aiter_lines() reads lines from the SSE stream without buffering the full body. Each line starting with data: is a JSON chunk from the model.

The FastAPI endpoint

import asyncio
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class InferenceRequest(BaseModel):
    prompt: str
    model: str = "gpt-4o"

async def sse_format(generator: AsyncGenerator[str, None]) -> AsyncGenerator[bytes, None]:
    """Wrap token chunks in Server-Sent Events format."""
    try:
        async for chunk in generator:
            payload = json.dumps({"token": chunk})
            yield f"data: {payload}\n\n".encode()
    except asyncio.CancelledError:
        return  # Client disconnected — stop generating
    finally:
        yield b"data: [DONE]\n\n"

@app.post("/inference/stream")
async def inference_stream(request: InferenceRequest):
    if not request.prompt.strip():
        raise HTTPException(status_code=400, detail="Empty prompt")

    generator = stream_inference(request.prompt, request.model)
    return StreamingResponse(
        sse_format(generator),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

Two headers matter here. Cache-Control: no-cache tells proxies not to buffer the response. X-Accel-Buffering: no is specifically for nginx — without it, nginx buffers SSE output until its proxy buffer fills, which silently turns your streaming response back into a delayed batch response. This header is easy to miss and causes hours of debugging.

Handling disconnects and timeouts

The most common failure mode: a client disconnects mid-stream, but the server keeps generating and spending tokens for nobody. FastAPI/Starlette propagates disconnects as asyncio.CancelledError through the generator. The try/except block in sse_format above handles this cleanly.

You also need timeouts. A single request timeout isn't enough — you need a per-token timeout to catch stalled connections where the model stops emitting:

async def stream_with_timeout(
    prompt: str,
    model: str,
    total_timeout: float = 120.0,
    token_timeout: float = 10.0,
) -> AsyncGenerator[str, None]:
    gen = stream_inference(prompt, model)
    loop = asyncio.get_event_loop()
    deadline = loop.time() + total_timeout

    while True:
        remaining = deadline - loop.time()
        if remaining <= 0:
            raise TimeoutError("Total inference timeout exceeded")

        try:
            chunk = await asyncio.wait_for(
                gen.__anext__(),
                timeout=min(token_timeout, remaining),
            )
            yield chunk
        except StopAsyncIteration:
            break
        except asyncio.TimeoutError:
            raise TimeoutError(f"No token in {token_timeout}s — connection stalled")

total_timeout caps the full request. token_timeout catches gaps in token delivery — rare in lab conditions, common at the tail of production traffic.

Backpressure for slow clients

If a client consumes chunks slowly (slow JS renderer, poor connection), the server can accumulate unbounded memory buffering tokens the client hasn't read. A bounded asyncio.Queue fixes this by blocking the producer when the consumer falls behind:

async def buffered_stream(
    prompt: str,
    model: str,
    buffer_size: int = 20,
) -> AsyncGenerator[str, None]:
    queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=buffer_size)

    async def producer() -> None:
        try:
            async for chunk in stream_inference(prompt, model):
                await queue.put(chunk)
        finally:
            await queue.put(None)  # sentinel

    task = asyncio.create_task(producer())
    try:
        while True:
            chunk = await queue.get()
            if chunk is None:
                break
            yield chunk
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

The maxsize=20 means the producer blocks after 20 unread tokens. This prevents a 10 000-token response from eating your server's memory when the client is on a slow connection.

Testing it end-to-end

A quick sanity check with curl:

curl -N -X POST http://localhost:8000/inference/stream \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain async/await in three sentences"}'

The -N flag disables curl's own buffering. You should see tokens appear one by one. If they arrive all at once after a delay, check your nginx X-Accel-Buffering header and your media_type="text/event-stream" on the StreamingResponse.

For load testing concurrent streams, a simple Python script using httpx.AsyncClient works well and is easier to instrument than generic HTTP benchmarking tools.

The takeaway

Building a streaming inference API in Python is not difficult, but several details determine whether it works under load:

• Use httpx.AsyncClient with .stream(), never the synchronous requests library
• Add X-Accel-Buffering: no for nginx deployments — this one trips up most people
• Handle asyncio.CancelledError in your generator to stop token generation on disconnect
• Apply both total-request and per-token timeouts
• Use a bounded queue if clients can be slow consumers

Once you have the streaming layer right, make sure your API enforces proper authentication and rate limiting on top of it — security hardening checklists for API layers are a useful reference when hardening this in production.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.