Inspecting Streaming Usage Tokens in Your Terminal
Parsing streaming usage without losing the last chunk
A streaming response can finish printing text before it finishes reporting usage. If your client stops at a finish_reason, your terminal looks correct while your token ledger stays empty. The fix is to read the stream through its termination marker and inspect usage independently of content.
I use Fluxo as the relay here: one key and one base URL, https://api.fluxorelay.com, across Claude, GPT, Gemini, DeepSeek, Kimi, GLM, Qwen, Grok and more. It works with Claude Code, Codex, Gemini CLI and any OpenAI SDK. For this walkthrough, curl exposes the actual events, and Python handles their framing.
Set the key and resolve the model ID
Run these commands in Bash with curl and Python 3 installed. Read the key without putting its value into shell history:
read -rsp 'Fluxo API key: ' FLUXO_API_KEY; printf '\n'; export FLUXO_API_KEY
Fetch the model catalog. The commands use the OpenAI-compatible /v1 routes beneath the base URL:
curl --fail-with-body -sS https://api.fluxorelay.com/v1/models -H "Authorization: Bearer $FLUXO_API_KEY" | python3 -m json.tool
Find GPT-6 Astra and copy its returned id. Display names and request identifiers can differ, so replace the placeholder below with that exact ID: — export FLUXO_MODEL='REPLACE_WITH_RETURNED_ID'
Build an SSE parser that preserves usage objects
Server-sent events are separated by blank lines. Network reads are not event boundaries, and an event can contain multiple data lines. This parser collects those lines, decodes complete events, and remembers the latest non-null usage object. Paste this command to create usage.py:
printf '%s\n' 'import json, sys' 'parts = []' 'usage = None' 'done = False' 'def consume(parts):' ' global usage, done' ' if not parts:' ' return' ' payload = "\n".join(parts)' ' if payload == "[DONE]":' ' done = True' ' return' ' event = json.loads(payload)' ' if "error" in event:' ' raise RuntimeError(event["error"])' ' if event.get("usage") is not None:' ' usage = event["usage"]' 'for raw in sys.stdin:' ' line = raw.rstrip("\r\n")' ' if not line:' ' consume(parts)' ' parts = []' ' if done:' ' break' ' elif line.startswith("data:"):' ' value = line[5:]' ' parts.append(value[1:] if value.startswith(" ") else value)' 'if parts and not done:' ' consume(parts)' 'if not done:' ' raise SystemExit("Stream ended without [DONE]; usage may be incomplete")' 'if usage is None:' ' raise SystemExit("Stream completed without a usage object")' 'print(json.dumps(usage, indent=2))' > usage.py
The parser never indexes choices[0]. A usage-only event can have an empty choices array. It also ignores SSE comments and other fields because only data lines contribute to the JSON payload.
Request usage and save the stream for replay
Generate the request body with Python so the model ID is safely JSON-escaped:
python3 -c 'import json, os; print(json.dumps({"model": os.environ["FLUXO_MODEL"], "messages": [{"role": "user", "content": "Reply with one short sentence about streaming APIs."}], "stream": True, "stream_options": {"include_usage": True}}))' > request.json
Enable pipeline failure reporting, then capture and parse the response:
set -o pipefail
curl --fail-with-body -sS -N https://api.fluxorelay.com/v1/chat/completions -H "Authorization: Bearer $FLUXO_API_KEY" -H 'Content-Type: application/json' --data-binary @request.json | tee stream.sse | python3 usage.py
The include_usage option requests accounting in the stream; actual fields depend on the model and relay mapping. If the request rejects that option, inspect the error before changing it. Silently dropping the option can leave a working text stream without the accounting you wanted.
Check cache details before calculating token costs
The output preserves the complete usage object, including any nested token details. Fluxo bills per real token, including cache reads and writes, with no monthly fee. Keep those categories intact rather than flattening everything into one total.
- Inspect prompt_tokens, completion_tokens, and total_tokens when those fields are present.
- Preserve cache read and write fields exactly as returned.
- Treat missing cache fields as unknown unless documented otherwise.
- Avoid adding cached tokens twice when they overlap prompt totals.
Replay missing usage without another API request
Run python3 usage.py < stream.sse to reproduce parsing locally. A missing termination marker suggests an interrupted stream; a completed stream without usage needs a closer look at the response contract. Neither case should become a zero-token entry.
Create a Fluxo account if you would like to try this with your own terminal workflow. Users also earn a referral share when people they invite pay. If you need help interpreting a captured response, leave a note on the contact page after signing in.