Streaming Chat Completions Correctly: SSE, Usage Chunks and Retries

When you call an LLM API with stream: true, you're opting into a real-time conversation over Server-Sent Events (SSE). The response arrives as a sequence of data: lines, each containing a JSON chunk. Done right, this gives the user a smooth typewriter effect. Done wrong, you get truncated responses, missing token counts, or silent failures.

This article covers how to consume streaming chat completions correctly using OpenAI-compatible endpoints, including handling usage chunks, building robust retry logic, and debugging common SSE pitfalls. Code examples work with any provider that mirrors the OpenAI streaming format, such as TokenShop.

How SSE Streaming Works in Practice

A streaming request to /v1/chat/completions with "stream": true returns an HTTP response with Content-Type: text/event-stream. Each event is a line starting with data: followed by a JSON object. The stream ends with data: [DONE].

A typical chunk looks like this:

data: {"id":"chatcmpl-abc123","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}

The delta field is the key: it contains only the new content since the last chunk. Your client must accumulate these deltas to reconstruct the full message.

Common mistake: treating each chunk as a complete response. Never parse a single delta.content as the final answer — always concatenate.

Handling the Usage Chunk

One detail many implementations miss: when streaming completes, the final chunk (before [DONE]) may include a usage field. This chunk has choices: [] (empty array) and an explicit usage object:

data: {"id":"chatcmpl-abc123","object":"chat.completion.chunk","created":1712345678,"model":"alibaba/qwen-3-32b","choices":[],"usage":{"prompt_tokens":45,"completion_tokens":120,"total_tokens":165}}

Not all providers emit this chunk. As of recent reports, OpenAI does not include usage in streaming mode by default — you must set "stream_options": {"include_usage": true}. Some open-model APIs like those available via TokenShop may include usage automatically, but the safest approach is to always request it explicitly.

Python example with usage tracking:

import json
from openai import OpenAI

client = OpenAI(base_url="https://tokshop.xyz/v1", api_key="your-key")

response = client.chat.completions.create(
    model="alibaba/qwen-3-32b",
    messages=[{"role": "user", "content": "Write a haiku about streaming APIs."}],
    stream=True,
    stream_options={"include_usage": True}
)

full_content = ""
usage = None

for chunk in response:
    if chunk.choices and chunk.choices[0].delta.content:
        full_content += chunk.choices[0].delta.content
    if hasattr(chunk, 'usage') and chunk.usage:
        usage = chunk.usage

print("Full response:", full_content)
if usage:
    print(f"Tokens: {usage.total_tokens} (prompt: {usage.prompt_tokens}, completion: {usage.completion_tokens})")

This pattern lets you show tokens to the user immediately after the last character appears, without a separate API call.

Building a Robust Streaming Client

A production streaming client needs three things: buffering, error detection, and timeout handling.

Buffering and Concatenation

Never assume chunks arrive in order or at predictable sizes. Use a simple accumulator:

def consume_stream(response_iter):
    buffer = ""
    for chunk in response_iter:
        if chunk.choices:
            for choice in chunk.choices:
                if choice.delta.content:
                    buffer += choice.delta.content
                if choice.finish_reason == "stop":
                    yield buffer, True
        # handle usage as shown above
    yield buffer, True

Timeout Handling

Network blips happen. Set both a connection timeout and a per-chunk read timeout. With the OpenAI Python SDK:

client = OpenAI(
    base_url="https://tokshop.xyz/v1",
    api_key="your-key",
    timeout=30.0,        # total timeout
    max_retries=2        # built-in retry
)

For lower-level control with httpx or requests, set stream=True and use a read_timeout on each chunk iteration.

Retry Logic That Respects Streams

Retrying a streaming request is different from retrying a non-streaming one. You cannot simply replay the same HTTP connection. Instead, implement a strategy:

1. Idempotency keys: Send an X-Idempotency-Key header with the request. If the connection drops mid-stream, the server can resume from the last processed chunk (if supported). Not all providers support this, but it costs nothing to include.

2. Exponential backoff with jitter: On network errors or HTTP 5xx, wait before retrying. A simple implementation:

import time
import random

def stream_with_retry(client, model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True,
                stream_options={"include_usage": True}
            )
            for chunk in response:
                yield chunk
            return  # success
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)

3. Avoid retrying on 4xx errors: A 400 (bad request) or 402 (insufficient balance) will not succeed on retry. For HTTP 402, check your account balance. TokenShop, for example, returns 402 when the prepaid ledger drops below $0.001 — a clear signal to top up.

Common Pitfalls and How to Avoid Them

Pitfall 1: Ignoring the [DONE] Signal

Some clients stop reading after the last data: line that has content. But the server may still send the usage chunk after that. Always read until you see data: [DONE] or the connection closes cleanly.

Pitfall 2: Not Handling Empty choices

When a usage-only chunk arrives, choices is an empty list. Code that assumes chunk.choices[0] will crash:

# WRONG
for chunk in response:
    content = chunk.choices[0].delta.content  # IndexError on usage chunk

# RIGHT
for chunk in response:
    if chunk.choices:
        content = chunk.choices[0].delta.content

Pitfall 3: Mixing Streaming and Non-Streaming Code Paths

If your app sometimes streams and sometimes doesn't, keep the response handling separate. A non-streaming response returns a single object with choices[0].message.content, while streaming returns an iterator. Trying to unify them with isinstance checks is fragile — use two distinct functions.

Pitfall 4: Forgetting to Set stream_options

Without "include_usage": true, many providers omit the usage chunk entirely. You'll get the content stream, then [DONE], and never see the token count. Always set this option if you need usage data.

When to Stream vs. When to Wait

Streaming adds complexity. Use it when:

• You're building a chat UI where latency perception matters
• You want to show partial reasoning (e.g., chain-of-thought)
• You need to process tokens as they arrive (e.g., real-time translation)

Avoid streaming when:

• You only need the final answer and can wait 1-3 seconds
• You're doing batch processing (streaming adds overhead)
• You're on a very constrained network (SSE connections can be flaky)

For non-streaming use, a simple client.chat.completions.create(stream=False) works fine and is easier to debug.

Conclusion

Streaming chat completions via SSE is a powerful pattern, but the devil is in the details: accumulate deltas, handle the usage chunk, set timeouts, and implement retry logic that respects stream semantics. Most OpenAI-compatible providers, including those listed on TokenShop's pricing page (like Qwen3 32B at $0.30/M input tokens or DeepSeek V3.2 at $0.40/M), support the same streaming format, so the code above works with minimal changes.

Start with the Python SDK's built-in streaming, add usage tracking, and layer on retry logic only when you see connection issues. Your users will thank you for the snappy typewriter effect — and your logs will thank you for the robust error handling.