Consuming a streamed reply: server-sent events, line by line

What streaming actually is

Without streaming, a model API generates the whole reply, then sends it. A 500-token answer at 50 tokens per second means ten seconds of nothing, then everything. With streaming, the server starts sending tokens as it produces them, over a single HTTP response that stays open until the generation ends.

The wire format is server-sent events, and it is text:

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"The"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" answer"}}

Each event is a few lines; a blank line ends it. The data: line carries JSON. That is the whole protocol. A course on how a model decides those tokens lives elsewhere; this lesson is about reading them.

Reading it with requests

import json, requests

r = session.post(
    url,
    headers=headers,
    json={**payload, "stream": True},
    stream=True,                       # do not read the body up front
    timeout=(5, 120),
)
r.raise_for_status()

parts = []
for line in r.iter_lines(decode_unicode=True):
    if not line or not line.startswith("data:"):
        continue
    data = line[len("data:"):].strip()
    if data == "[DONE]":               # OpenAI's terminator; Anthropic sends a message_stop event
        break
    event = json.loads(data)
    text = extract_text(event)         # provider-specific; see below
    if text:
        parts.append(text)
        print(text, end="", flush=True)

print()
full = "".join(parts)

Three details carry the weight.

stream=True on the requests call tells it not to download the body before returning. Without it, iter_lines still works but only after the entire reply has arrived, which defeats the purpose.

iter_lines yields each line as it is received. decode_unicode=True gives you str rather than bytes.

flush=True on print. When stdout is a terminal Python flushes on each newline, and you are printing none. When stdout is a pipe or a file it is block-buffered and flushes only when the buffer fills. Either way, without flush=True the tokens arrive but do not appear, and you conclude the server is not streaming.

The extract step differs per provider

OpenAI's chunks look like:

def extract_text(event):
    choices = event.get("choices") or []
    if not choices:
        return ""
    return choices[0].get("delta", {}).get("content") or ""

Anthropic's:

def extract_text(event):
    if event.get("type") == "content_block_delta":
        return event["delta"].get("text", "")
    return ""

Use .get() everywhere here. The stream carries events that are not text — a start event, usage figures at the end, a stop reason — and indexing them as if they were text raises KeyError mid-stream, at which point the connection is dropped and the partial answer is lost.

With the SDK

with client.messages.stream(model=m, max_tokens=500, messages=msgs) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()

The SDK parses the events and hands you text. get_final_message() returns the assembled message with the usage figures, which a later lesson uses for cost. OpenAI's is for chunk in client.chat.completions.create(..., stream=True), with the text in chunk.choices[0].delta.content, sometimes None.

Accumulate, because nobody else will

A stream gives you fragments. If you want the full answer afterwards — to save it, to parse JSON out of it, to log it — you must collect it yourself. The parts list above is not optional. Appending to a list and joining once at the end is the right pattern; full += text in the loop copies the growing string every time, which is quadratic and noticeable past a few thousand fragments.

A stream can end early

The server can close the connection at any point: a network fault, a timeout, a server-side error surfaced as an error event. Your loop ends and you have half an answer. Check the terminal event before trusting the text. OpenAI sets finish_reason on the last chunk ("stop" is good, "length" means the token limit cut it off); Anthropic sends a message_delta with stop_reason. If neither arrived, the stream broke and the partial text should be marked as partial, not saved as an answer.

Also apply the read timeout. It counts time between bytes, and with streaming the gaps are short, so a 30-second read timeout catches a genuinely stuck stream without failing a long healthy one. A stream is the one place where a short read timeout is fine.

Turning it into a generator

Wrap the loop in a function with yield and the caller can consume it like any sequence, print it, or feed it to a web interface:

def stream_text(session, url, headers, payload):
    with session.post(url, headers=headers, json={**payload, "stream": True},
                      stream=True, timeout=(5, 30)) as r:
        r.raise_for_status()
        for line in r.iter_lines(decode_unicode=True):
            if line.startswith("data:") and line[5:].strip() != "[DONE]":
                text = extract_text(json.loads(line[5:]))
                if text:
                    yield text

The with ensures the connection is released when the caller stops early. Module 9 plugs this exact function into a web interface.

Try this now

Run the raw requests version against Ollama's OpenAI-compatible endpoint, which streams for free. Remove flush=True and run it with output piped through cat to watch the whole answer appear at once. Then put it back.