How to stream OpenAI-compatible chat completions

Streaming chat completions transforms the user experience from waiting for a full response to seeing tokens appear in real time. The OpenAI-compatible streaming API uses Server-Sent Events (SSE) over HTTP, delivering each token as a separate JSON chunk. This guide walks through the complete implementation: constructing the request, parsing the event stream, handling edge cases, and verifying the integration works end to end.

Step 1: Understand the streaming response format

When you set "stream": true in a chat completions request, the server responds with Content-Type: text/event-stream. Each chunk arrives as an SSE event with a data: field containing JSON. The stream terminates with a data: [DONE] sentinel.

A typical chunk looks like this:

data: {"id":"chatcmpl-abc123","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}

Key fields to handle:

choices[0].delta.content — the incremental token (may be null or absent)
choices[0].delta.role — appears only in the first chunk, typically "assistant"
choices[0].finish_reason — null while streaming, "stop" or "length" at the end
usage — appears only in the final chunk when stream_options.include_usage is true

Step 2: Build the streaming request

Use any HTTP client that supports streaming responses. The request body mirrors the non-streaming format with one addition:

import httpx
import json

url = "https://api.openai.com/v1/chat/completions"
headers = {
    "Authorization": "Bearer YOUR_API_KEY",
    "Content-Type": "application/json",
    "Accept": "text/event-stream",  # important for some proxies
}
payload = {
    "model": "gpt-4o-mini",
    "messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain streaming in three sentences."}
    ],
    "stream": True,
    "stream_options": {"include_usage": True},  # optional, gets token counts at end
    "max_tokens": 200,
    "temperature": 0.7,
}

The Accept: text/event-stream header isn’t strictly required by the spec but prevents some CDNs and proxies from buffering the response.

Step 3: Parse the SSE stream

SSE is a line-oriented protocol. Each event consists of field: value lines terminated by a blank line. The data: field may span multiple lines for large payloads. A minimal parser:

def parse_sse_lines(response_iter):
    """Yield parsed JSON objects from an SSE response iterator."""
    buffer = ""
    for line in response_iter:
        line = line.decode("utf-8") if isinstance(line, bytes) else line
        if line.startswith("data: "):
            buffer += line[6:]  # strip "data: "
        elif line == "data: [DONE]\n":
            break
        elif line == "\n" and buffer:
            yield json.loads(buffer)
            buffer = ""
        # ignore comment lines (": ...") and other fields

With httpx, you iterate over response.iter_lines():

with httpx.stream("POST", url, headers=headers, json=payload, timeout=60.0) as response:
    response.raise_for_status()
    for chunk in parse_sse_lines(response.iter_lines()):
        delta = chunk["choices"][0]["delta"]
        if content := delta.get("content"):
            print(content, end="", flush=True)
        if chunk["choices"][0]["finish_reason"]:
            print("\n[stream complete]")
            if "usage" in chunk:
                print(f"Tokens: {chunk['usage']}")

Step 4: Handle the first chunk and role delta

The first chunk often contains only the role:

data: {"id":"chatcmpl-abc123","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}

Your parser should not crash when delta.content is absent. A robust handler:

def handle_chunk(chunk):
    choice = chunk["choices"][0]
    delta = choice["delta"]
    
    # First chunk may have role only
    if "role" in delta:
        # Initialize message buffer, set role if tracking conversation
        pass
    
    # Content chunks
    if content := delta.get("content"):
        yield content
    
    # Terminal chunk
    if choice["finish_reason"]:
        return {"finish_reason": choice["finish_reason"], "usage": chunk.get("usage")}

Step 5: Implement retry and error handling

Streaming connections fail in ways non-streaming requests don’t: mid-stream network interruptions, provider timeouts, or rate limits after the first few tokens. Implement retry with exponential backoff, but only for recoverable errors — don’t retry a 400 or 401.

import time
from httpx import HTTPStatusError, ReadTimeout, ConnectError

RETRY_STATUS = {429, 500, 502, 503, 504}
MAX_RETRIES = 3
BASE_DELAY = 1.0

def stream_with_retry(url, headers, payload):
    attempt = 0
    while True:
        try:
            with httpx.stream("POST", url, headers=headers, json=payload, timeout=60.0) as response:
                if response.status_code in RETRY_STATUS:
                    raise HTTPStatusError(
                        f"Status {response.status_code}", request=response.request, response=response
                    )
                response.raise_for_status()
                yield from parse_sse_lines(response.iter_lines())
                return  # success
        except (HTTPStatusError, ReadTimeout, ConnectError) as e:
            attempt += 1
            if attempt > MAX_RETRIES:
                raise
            delay = BASE_DELAY * (2 ** (attempt - 1))
            time.sleep(delay)

Important: You cannot resume a stream from the middle. A retry restarts the generation from token 0. For user-facing apps, consider showing what arrived before the failure, then restarting silently or with a toast notification.

Step 6: Add request-level timeout and cancellation

Long generations can hang. Set two timeouts: a connect/read timeout on the HTTP client, and an application-level deadline.

import asyncio
import httpx

async def stream_with_deadline(url, headers, payload, deadline_seconds=30):
    """Stream with a hard deadline; cancels cleanly on timeout."""
    timeout = httpx.Timeout(connect=10.0, read=deadline_seconds)
    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream("POST", url, headers=headers, json=payload) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                # parse SSE as before
                pass

For synchronous code, run the stream in a thread with concurrent.futures and cancel the future on deadline. Always close the response to release the connection back to the pool.

Step 7: Verify the integration end to end

Run these checks before shipping:

Basic stream — Send a short prompt ("Count to five.") and confirm tokens arrive incrementally, not all at once.
Finish reason — Verify the final chunk has finish_reason: "stop" (or "length" if you hit max_tokens).
Usage accounting — With stream_options.include_usage: true, confirm the final chunk includes usage.prompt_tokens and usage.completion_tokens.
Empty delta handling — Some providers send keep-alive chunks with empty delta. Your parser should ignore them without error.
Error mid-stream — Simulate a network kill (e.g., tc qdisc add dev lo root netem loss 50%) and confirm your retry logic triggers and surfaces a sensible error to the UI.
Cancellation — Abort a long request client-side and verify no resource leaks (open sockets, threads).

A quick verification script:

def verify_stream():
    test_payload = {
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "Count to three."}],
        "stream": True,
        "stream_options": {"include_usage": True},
        "max_tokens": 20,
    }
    tokens = []
    usage = None
    finish_reason = None
    
    for chunk in stream_with_retry(url, headers, test_payload):
        if content := chunk["choices"][0]["delta"].get("content"):
            tokens.append(content)
        if chunk["choices"][0]["finish_reason"]:
            finish_reason = chunk["choices"][0]["finish_reason"]
            usage = chunk.get("usage")
    
    assert tokens, "No tokens received"
    assert finish_reason in ("stop", "length"), f"Unexpected finish_reason: {finish_reason}"
    assert usage and usage["completion_tokens"] > 0, "Usage missing or zero"
    print(f"OK: {''.join(tokens)} | usage={usage} | finish={finish_reason}")

verify_stream()

Step 8: Common pitfalls and fixes

Symptom	Cause	Fix
All tokens arrive at once	Proxy/CDN buffering response	Set `Accept: text/event-stream`; disable buffering in nginx (`proxy_buffering off;`) or Cloudflare
`json.JSONDecodeError` on chunk	Concatenated `data:` lines not joined	Accumulate lines until blank line, then parse
`KeyError: 'content'`	First chunk has only `role`	Guard with `.get("content")`
Connection hangs indefinitely	No read timeout	Set `read` timeout on client; add application deadline
Retry duplicates output	Restarted stream re-emits tokens	Track emitted tokens; on retry, suppress duplicates or restart UI
`finish_reason: null` on last chunk	Provider bug or stream cut	Treat missing `finish_reason` + closed connection as `"stop"`

Step 9: Production hardening

Beyond the basics, production systems need:

Structured logging: Log request_id (from response headers or chunk id), model, token counts, latency to first token (TTFT), and total latency.
Metrics: Emit histograms for TTFT, tokens/second, error rates by type.
Circuit breaker: Stop hammering a provider that returns 5xx or 429 repeatedly.
Fallback routing: If your gateway supports multiple providers, fail over to a healthy one on repeated errors. n4n.ai handles this automatically by routing around degraded providers while preserving the same OpenAI-compatible contract.
Client-side buffering: For UX, batch tokens into ~50ms flushes rather than rendering each character, reducing layout thrashing.

Step 10: Testing with a mock server

Unit tests shouldn’t hit real APIs. Spin up a local SSE mock:

# test_mock_server.py
from http.server import HTTPServer, BaseHTTPRequestHandler
import json, time, threading

CHUNKS = [
    {"id": "test", "object": "chat.completion.chunk", "created": 1, "model": "test", "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}]},
    {"id": "test", "object": "chat.completion.chunk", "created": 1, "model": "test", "choices": [{"index": 0, "delta": {"content": "Hello"}, "finish_reason": None}]},
    {"id": "test", "object": "chat.completion.chunk", "created": 1, "model": "test", "choices": [{"index": 0, "delta": {"content": " world"}, "finish_reason": "stop"}]},
]

class MockHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.end_headers()
        for chunk in CHUNKS:
            self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
            time.sleep(0.05)
        self.wfile.write(b"data: [DONE]\n\n")

def run_mock(port=8999):
    server = HTTPServer(("localhost", port), MockHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server

Point your client at http://localhost:8999/v1/chat/completions and assert the parser emits "Hello world" with correct finish reason.

Streaming is the default for any interactive LLM feature. The SSE format is simple but unforgiving: a missed newline, a buffered proxy, or an unhandled empty delta breaks the experience. Follow the steps above, verify each failure mode, and you’ll ship a streaming integration that feels instant and stays reliable under load.