Streaming chat completions over SSE, cost in the last chunk

Streaming makes a chat UI feel instant, but it also changes how billing works: usage isn't known until the model stops generating, so it has to travel in the stream itself rather than in a response header. This post covers parsing genmux's SSE stream by hand (useful if you're not on the OpenAI SDK, or want to understand what the SDK is doing for you) and what happens if you cut the connection early.

The wire format

POST /v1/chat/completions with "stream": true returns text/event-stream: a sequence of data: {...}\n\n lines, each a JSON chat.completion.chunk, ending with a literal data: [DONE]\n\n. Every chunk except the last carries a delta with the next slice of content:

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"The"},"finish_reason":null}]}

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":" Volga"},"finish_reason":null}]}

data: {"id":"gnx_9f2c...","object":"chat.completion.chunk","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":12,"completion_tokens":47,"total_tokens":59,"cost":0.071,"cost_source":"estimated"}}

data: [DONE]

The last chunk before [DONE] has an empty choices[].delta and a populated top-level usage — that's the one line in the whole stream that tells you what the call cost. cost_source is always "estimated" for streamed calls: upstream vendors don't report exact usage for streamed output, so genmux computes it from the token counts it can see.

Parsing it in Python

import json
import requests

def stream_chat(model: str, messages: list[dict], **kwargs):
    with requests.post(
        "https://api.genmux.tech/v1/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": model, "messages": messages, "stream": True, **kwargs},
        stream=True,
        timeout=60,
    ) as resp:
        resp.raise_for_status()
        usage = None
        for line in resp.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data: "):
                continue
            payload = line[len("data: "):]
            if payload == "[DONE]":
                break
            chunk = json.loads(payload)
            delta = chunk["choices"][0]["delta"] if chunk["choices"] else {}
            if "content" in delta:
                print(delta["content"], end="", flush=True)
            if chunk.get("usage"):
                usage = chunk["usage"]
        print()
        return usage

usage = stream_chat("deepseek-v4-flash", [{"role": "user", "content": "Describe the Volga in one sentence."}])
print(usage)  # {'prompt_tokens': 12, 'completion_tokens': 47, 'total_tokens': 59, 'cost': 0.071, 'cost_source': 'estimated'}

Two details worth keeping: check chunk["choices"] before indexing into it (the final chunk's array can be empty), and check chunk.get("usage") on every chunk rather than assuming it's only on the last one — that's the correct way to write code that keeps working if a future chunk format changes.

Parsing it in Node

async function streamChat(model, messages, extra = {}) {
  const resp = await fetch("https://api.genmux.tech/v1/chat/completions", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.GENMUX_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ model, messages, stream: true, ...extra }),
  });

  const reader = resp.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let usage = null;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n\n");
    buffer = lines.pop(); // keep the incomplete tail for the next read

    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const payload = line.slice(6);
      if (payload === "[DONE]") return usage;
      const chunk = JSON.parse(payload);
      const delta = chunk.choices[0]?.delta ?? {};
      if (delta.content) process.stdout.write(delta.content);
      if (chunk.usage) usage = chunk.usage;
    }
  }
  return usage;
}

If you're already on the openai Node or Python SDK, none of the above parsing is your problem — pass stream: true (Python: stream=True) and read chunk.usage on whatever chunk has it; the SDK handles line splitting and JSON parsing for you.

Aborting mid-stream

If your client closes the connection or the user navigates away before [DONE], genmux bills for the tokens generated up to that point — never more, and never a flat "cancelled" charge. There's an important asymmetry worth designing around:

Practically, this means an abort button in a chat UI is safe to wire up without any special-casing on the billing side: cut the connection whenever the user wants, and the invoice reflects exactly what they saw on screen.

Reconciling after the fact

If your process crashes mid-stream and you lose the usage chunk, the call isn't lost — every chat call gets a genmux job id (the id field on every chunk), and GET /v1/generations/{id} returns the full record including usage and cost once the job settles. For a periodic reconciliation job, GET /v1/generations?capability=text.chat lists every chat call on the account, which is the more robust source of truth than anything your client happened to capture live.

Getting started

Streaming works identically across every text model on the platform — DeepSeek, Qwen3, Doubao, and more as they come online — no per-model quirks in the wire format. Full endpoint reference is in the docs; for reasoning models, pair this with how thinking is billed since reasoning tokens stream through the same delta.content field before the final answer.