How to parse a streaming chat completion in Python

Streaming chat completions arrive as Server-Sent Events (SSE) — a text-based protocol where each line starts with data: and payloads are JSON fragments. Most tutorials stop at “print each chunk.” This one builds a production-ready parser that reassembles deltas, handles tool calls, and survives real-world provider quirks. You’ll walk away with a single StreamParser class you can drop into any project.

Prerequisites

• Python 3.10+
• httpx (or requests if you prefer synchronous code)
• An OpenAI-compatible endpoint — any provider that emits SSE works

pip install httpx

The examples target an OpenAI-compatible /v1/chat/completions endpoint with stream=true. If you’re using n4n.ai, the same endpoint works — just swap the base URL and key.

The raw stream: what actually hits the wire

Before writing a parser, look at the raw bytes. Fire this once to see the format:

import httpx

url = "https://api.openai.com/v1/chat/completions"
headers = {"Authorization": "Bearer $OPENAI_API_KEY"}
payload = {
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Count to 3"}],
    "stream": True,
}

with httpx.stream("POST", url, headers=headers, json=payload, timeout=30) as resp:
    for line in resp.iter_lines():
        print(repr(line))

Expected output (truncated):

'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"One"},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":","},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":" two"},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":","},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":" three"},"finish_reason":null}]}'
'data: {"id":"chatcmpl-...","object":"chat.completion.chunk","created":1700000000,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}'
'data: [DONE]'

Key observations:

• Each event is a line prefixed with data:
• Payloads are JSON objects with a choices[0].delta field
• The first chunk often carries only role: "assistant"
• Content arrives in arbitrary-sized fragments
• The final chunk has finish_reason: "stop" (or "tool_calls", "length", etc.)
• A literal data: [DONE] line terminates the stream

Minimal working parser

Start with a function that yields parsed deltas. This handles the happy path and nothing else:

import json
from typing import Iterator

def parse_sse_stream(lines: Iterator[str]) -> Iterator[dict]:
    """Yield parsed delta objects from an SSE line iterator."""
    for line in lines:
        line = line.strip()
        if not line or not line.startswith("data: "):
            continue
        payload = line[6:]  # strip "data: "
        if payload == "[DONE]":
            break
        try:
            chunk = json.loads(payload)
        except json.JSONDecodeError:
            continue  # malformed line, skip
        delta = chunk.get("choices", [{}])[0].get("delta", {})
        if delta:
            yield delta

Usage:

with httpx.stream("POST", url, headers=headers, json=payload, timeout=30) as resp:
    for delta in parse_sse_stream(resp.iter_lines()):
        if "content" in delta:
            print(delta["content"], end="", flush=True)
print()

Expected output:

One, two, three

This works for simple text. It breaks immediately when tool calls enter the picture.

Tool calls: why the naive parser fails

When the model emits a function call, the delta structure changes. A tool-call chunk looks like:

{
  "choices": [{
    "index": 0,
    "delta": {
      "tool_calls": [{
        "index": 0,
        "id": "call_abc123",
        "type": "function",
        "function": {"name": "get_weather", "arguments": "{\"location\":"}
      }]
    },
    "finish_reason": null
  }]
}

Notice:

• tool_calls is a list, not a single object
• Each call has an index — multiple calls can interleave
• arguments arrives as a string fragment, not parsed JSON
• The id and name may appear in the first fragment; subsequent fragments only carry arguments

A parser must accumulate fragments per tool_calls[index] and only emit a complete call when finish_reason == "tool_calls".

Building a production StreamParser

Here is a complete, typed class that handles text, tool calls, reasoning tokens (for models that emit them), and the [DONE] sentinel. It exposes a simple callback interface so you can plug in logging, UI updates, or token counting.

import json
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum


class DeltaType(Enum):
    TEXT = "text"
    TOOL_CALL = "tool_call"
    REASONING = "reasoning"  # e.g., o1-style reasoning tokens


@dataclass
class ToolCallDelta:
    index: int
    id: Optional[str] = None
    name: Optional[str] = None
    arguments: str = ""  # accumulated JSON string fragments

    def to_dict(self) -> dict:
        return {
            "index": self.index,
            "id": self.id,
            "type": "function",
            "function": {"name": self.name, "arguments": self.arguments},
        }


@dataclass
class StreamParser:
    """
    Incrementally parse an OpenAI-compatible SSE stream.

    Call `feed(line)` for each raw SSE line. The parser invokes callbacks
    as deltas arrive and completes when it sees [DONE].
    """
    on_text: Callable[[str], None] = lambda _: None
    on_tool_call: Callable[[ToolCallDelta], None] = lambda _: None
    on_reasoning: Callable[[str], None] = lambda _: None
    on_finish: Callable[[str], None] = lambda _: None  # finish_reason

    _tool_calls: dict[int, ToolCallDelta] = field(default_factory=dict, init=False)
    _finished: bool = field(default=False, init=False)

    def feed(self, line: str) -> None:
        if self._finished:
            return
        line = line.strip()
        if not line or not line.startswith("data: "):
            return
        payload = line[6:]
        if payload == "[DONE]":
            self._finished = True
            return
        try:
            chunk = json.loads(payload)
        except json.JSONDecodeError:
            return

        choice = chunk.get("choices", [{}])[0]
        delta = choice.get("delta", {})
        finish_reason = choice.get("finish_reason")

        # Text content
        if content := delta.get("content"):
            self.on_text(content)

        # Reasoning tokens (some providers use "reasoning" or "reasoning_content")
        for key in ("reasoning", "reasoning_content"):
            if reasoning := delta.get(key):
                self.on_reasoning(reasoning)

        # Tool calls — accumulate by index
        for tc in delta.get("tool_calls", []):
            idx = tc["index"]
            if idx not in self._tool_calls:
                self._tool_calls[idx] = ToolCallDelta(index=idx)
            acc = self._tool_calls[idx]
            if tc.get("id"):
                acc.id = tc["id"]
            if tc.get("function", {}).get("name"):
                acc.name = tc["function"]["name"]
            if tc.get("function", {}).get("arguments"):
                acc.arguments += tc["function"]["arguments"]
            self.on_tool_call(acc)

        # Finish
        if finish_reason:
            # Emit any completed tool calls
            for acc in self._tool_calls.values():
                self.on_tool_call(acc)
            self.on_finish(finish_reason)
            self._finished = True

    def reset(self) -> None:
        self._tool_calls.clear()
        self._finished = False

Wiring it up

def main():
    url = "https://api.openai.com/v1/chat/completions"
    headers = {"Authorization": "Bearer $OPENAI_API_KEY"}
    payload = {
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "What's the weather in Tokyo?"}],
        "tools": [{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get current weather",
                "parameters": {
                    "type": "object",
                    "properties": {"location": {"type": "string"}},
                    "required": ["location"],
                },
            },
        }],
        "stream": True,
    }

    parser = StreamParser(
        on_text=lambda t: print(f"[text] {t}", end="", flush=True),
        on_tool_call=lambda tc: print(f"\n[tool_call] {tc.to_dict()}"),
        on_reasoning=lambda r: print(f"[reasoning] {r}", end="", flush=True),
        on_finish=lambda fr: print(f"\n[finish] {fr}"),
    )

    with httpx.stream("POST", url, headers=headers, json=payload, timeout=60) as resp:
        for line in resp.iter_lines():
            parser.feed(line)

if __name__ == "__main__":
    main()

Expected output (model decides to call the tool):

[tool_call] {'index': 0, 'id': 'call_abc123', 'type': 'function', 'function': {'name': 'get_weather', 'arguments': '{"location":"Tokyo"}'}}
[finish] tool_calls

If the model answers directly without tools:

[text] The weather in Tokyo is currently...
[finish] stop

Handling provider quirks

Real endpoints deviate from the spec. Here are the ones you’ll hit and how the parser above deals with them — or where you need to extend it.

1. Missing choices array

Some providers return a bare delta object. Guard with .get("choices", [{}])[0] as shown.

2. role in the first chunk only

The first delta often contains only {"role": "assistant"}. The parser ignores it because content, reasoning, and tool_calls are absent. If you need the role, add:

if role := delta.get("role"):
    self.on_role(role)  # add a callback

3. Reasoning tokens under different keys

OpenAI uses reasoning (for o1) or reasoning_content (some compat layers). The parser checks both. Add more keys to the tuple if your provider invents a new one.

4. Tool call id and name split across chunks

The parser accumulates into ToolCallDelta and re-emits on every fragment. Your on_tool_call callback receives the current accumulated state. If you only want the final object, check finish_reason == "tool_calls" in on_finish and read parser._tool_calls (or expose a get_tool_calls() method).

5. Empty chunks and keep-alive newlines

Some proxies send blank lines or data: with no payload. The if not line or not line.startswith("data: "): continue guard handles both.

6. Chunked transfer encoding splitting a single SSE event across iter_lines() calls

httpx.iter_lines() handles HTTP chunking correctly — it yields complete lines. If you read raw bytes with iter_raw() or aiter_bytes(), you must buffer until \n\n (SSE event boundary). Stick with iter_lines().

Async version for FastAPI / Starlette

If you’re building a proxy or gateway, you need async. The parser logic is identical; only the I/O changes.

import httpx
from typing import AsyncIterator

class AsyncStreamParser(StreamParser):
    """Same callbacks, but feed() is async-ready (no blocking)."""
    pass  # feed() is already non-blocking

async def stream_completion(
    url: str,
    headers: dict,
    payload: dict,
    parser: AsyncStreamParser,
) -> None:
    async with httpx.AsyncClient(timeout=60) as client:
        async with client.stream("POST", url, headers=headers, json=payload) as resp:
            async for line in resp.aiter_lines():
                parser.feed(line)

Token counting without buffering the full response

If you need per-request token usage for billing or logging, you have two options:

1. Provider returns usage in the final chunk — some do, many don’t.
2. Count locally — use tiktoken on accumulated text and tool-call arguments.

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o-mini")

class CountingParser(StreamParser):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.prompt_tokens = 0
        self.completion_tokens = 0

    def feed(self, line: str) -> None:
        # ... same as parent, but track tokens
        super().feed(line)
        # Note: this only counts streamed deltas, not the prompt.
        # For prompt tokens, encode the request messages before sending.

Testing the parser without an API key

Unit tests should not hit the network. Feed the parser canned lines:

def test_tool_call_accumulation():
    parser = StreamParser(
        on_tool_call=lambda tc: None,
        on_finish=lambda fr: None,
    )
    lines = [
        'data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":"{\\"location\\":"}}]}}]}',
        'data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\\"Tokyo\\""}}]}}]}',
        'data: {"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}',
        'data: [DONE]',
    ]
    for line in lines:
        parser.feed(line)
    assert parser._tool_calls[0].arguments == '{"location":"Tokyo"}'
    assert parser._tool_calls[0].name == "get_weather"

Run with pytest -q. No mocks, no network, deterministic.

Checklist before shipping

• Handle finish_reason values: stop, length, tool_calls, content_filter, function_call (legacy)
• Set a read timeout on the HTTP client (30–60s typical)
• Propagate provider usage chunk if present — parse chunk.get("usage") on the final event
• Log raw lines at debug level for post-mortem debugging
• Add a max_tokens guard on accumulated text if you stream to a UI with limited buffer

TL;DR

• Streaming chat completions are SSE: data: <json>\n\n
• Parse line by line, strip data: , skip [DONE]
• Accumulate delta.content for text, delta.tool_calls[index].function.arguments for tools
• Tool calls arrive fragmented — reassemble by index
• The parser above is ~80 lines, typed, tested, and handles the quirks you’ll actually see

Drop StreamParser into your codebase, wire the callbacks, and you’re done.