SSE streaming client

Every major LLM API (OpenAI, Anthropic, vLLM, TGI) uses Server-Sent Events for streaming responses. When you set stream: true, the server does not wait until generation is complete. It sends each token (or small chunk of tokens) as an SSE event the moment it is decoded. This dramatically improves perceived latency: the user sees the first token in milliseconds instead of waiting seconds for the full response.

Building a robust SSE client is not as trivial as it looks. The SSE protocol has its own format, edge cases around reconnection, and you need to handle timing carefully to measure TTFT (time to first token) and ITL (inter-token latency) accurately.

The SSE protocol

Server-Sent Events is an HTTP standard (defined in the HTML spec, not a separate RFC). The response has Content-Type: text/event-stream and consists of UTF-8 text lines:

data: {"id":"cmpl-abc","object":"text_completion","choices":[{"text":" Hello"}]}

data: {"id":"cmpl-abc","object":"text_completion","choices":[{"text":" world"}]}

data: [DONE]

Each event is a block of lines starting with data:, separated by a blank line. The final event is typically data: [DONE]. Some servers also send id: and event: fields, which are useful for reconnection but often omitted by LLM APIs.

A minimal client with requests

The simplest streaming client in Python uses requests with stream=True:

import requests
import json
import time

def stream_completion(url, prompt, model, max_tokens=128):
    """Stream tokens from an OpenAI-compatible endpoint."""
    payload = {
        "model": model,
        "prompt": prompt,
        "max_tokens": max_tokens,
        "stream": True,
    }

    t_start = time.perf_counter()
    t_first_token = None
    tokens = []
    token_times = []

    response = requests.post(
        f"{url}/v1/completions",
        json=payload,
        stream=True,
        headers={"Accept": "text/event-stream"},
    )
    response.raise_for_status()

    for line in response.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data:"):
            continue

        data = line[len("data:"):].strip()
        if data == "[DONE]":
            break

        chunk = json.loads(data)
        token_text = chunk["choices"][0]["text"]

        t_now = time.perf_counter()
        if t_first_token is None:
            t_first_token = t_now

        tokens.append(token_text)
        token_times.append(t_now)

    t_end = time.perf_counter()

    ttft = t_first_token - t_start if t_first_token else None
    itl_values = [
        token_times[i] - token_times[i - 1]
        for i in range(1, len(token_times))
    ]
    avg_itl = sum(itl_values) / len(itl_values) if itl_values else None

    return {
        "text": "".join(tokens),
        "num_tokens": len(tokens),
        "ttft_ms": ttft * 1000 if ttft else None,
        "avg_itl_ms": avg_itl * 1000 if avg_itl else None,
        "total_time_ms": (t_end - t_start) * 1000,
    }

This works, but it has limitations. The requests library blocks on iter_lines(), so you can only handle one stream at a time per thread.

A better client with httpx

For production use, I prefer httpx because it supports both sync and async modes and handles SSE edge cases better:

import httpx
import json
import time

def stream_with_httpx(url, prompt, model, max_tokens=128):
    """Streaming client using httpx for better connection handling."""
    payload = {
        "model": model,
        "prompt": prompt,
        "max_tokens": max_tokens,
        "stream": True,
    }

    tokens = []
    t_start = time.perf_counter()
    t_first = None

    with httpx.Client(timeout=60.0) as client:
        with client.stream(
            "POST",
            f"{url}/v1/completions",
            json=payload,
        ) as response:
            response.raise_for_status()
            buffer = ""

            for chunk in response.iter_text():
                buffer += chunk
                while "\n\n" in buffer:
                    event, buffer = buffer.split("\n\n", 1)
                    for line in event.split("\n"):
                        if line.startswith("data:"):
                            data = line[5:].strip()
                            if data == "[DONE]":
                                break
                            parsed = json.loads(data)
                            token = parsed["choices"][0]["text"]
                            if t_first is None:
                                t_first = time.perf_counter()
                            tokens.append(token)

    t_end = time.perf_counter()
    return {
        "text": "".join(tokens),
        "ttft_ms": (t_first - t_start) * 1000 if t_first else None,
        "total_ms": (t_end - t_start) * 1000,
        "tokens": len(tokens),
    }

The key difference is the explicit buffer management. SSE events are delimited by double newlines (\n\n), and network chunks do not always align with event boundaries. The buffer accumulates raw text and splits on \n\n to extract complete events.

Handling edge cases

Production SSE clients need to handle several things that the basic examples above skip:

• Reconnection: if the connection drops mid-stream, you lose the rest of the response. The SSE spec defines a Last-Event-ID header for reconnection, but most LLM APIs do not support it. In practice, you retry the full request.
• Timeouts: set both a connection timeout and a read timeout. Long generations can take minutes, so the read timeout should be generous. But if no data arrives for 30 seconds, something is wrong.
• Backpressure: if your client processes tokens slower than the server generates them, the TCP buffer fills up and eventually the server blocks. This is rare in practice because token generation is much slower than network throughput, but it matters for logging-heavy clients.
• Empty tokens: some servers send events with empty token text (whitespace, empty string). Handle them gracefully.

# Robust timeout configuration
client = httpx.Client(
    timeout=httpx.Timeout(
        connect=5.0,       # 5s to establish connection
        read=30.0,         # 30s max between chunks
        write=5.0,         # 5s to send request
        pool=10.0,         # 10s to get a connection from pool
    )
)

Measuring streaming metrics

The two metrics that matter for streaming are TTFT and ITL. TTFT is the time from sending the request to receiving the first token. It reflects the prefill phase (processing the input prompt). ITL is the time between consecutive tokens during decode.

For benchmarking, I collect every token timestamp and compute percentiles:

import statistics

def analyze_stream_metrics(token_times):
    """Compute streaming metrics from token timestamps."""
    if len(token_times) < 2:
        return {}

    itl_values = [
        (token_times[i] - token_times[i - 1]) * 1000
        for i in range(1, len(token_times))
    ]

    return {
        "itl_p50_ms": statistics.median(itl_values),
        "itl_p99_ms": sorted(itl_values)[int(len(itl_values) * 0.99)],
        "itl_mean_ms": statistics.mean(itl_values),
        "itl_stdev_ms": statistics.stdev(itl_values) if len(itl_values) > 1 else 0,
    }

ITL variance is an underappreciated metric. High variance means some tokens arrive in bursts (vLLM's continuous batching can cause this when new requests join the batch, briefly increasing the per-step computation). Consistent ITL means smoother user experience.

Streaming does not make generation faster. It makes the wait feel shorter. That distinction matters for UX but also for architecture: a streaming client consumes a connection for the full generation duration, which affects your load balancer and connection pool sizing.

Tomorrow I will extend this to an async batch client with asyncio that can handle hundreds of concurrent streams.