AI on Demand: Token streaming

Overview

This pages describes some useful commands on how to use token streaming.

Usage

We assume you've received your API key from us in our usual manner.

Usage - Inference with token-stream

Regular token streaming is difficult to read in the console.

The expanded command processes the output into a far more digestible format on the console.

Chat platforms like LibreChat include native mechanisms for handling token streaming or parsing the stream.

The stream: true flag needs to be set.

#Persönlichen Key setzen
STONEY_KEY='sk-...'

#Modell auswählen
MODEL='zai-org/GLM-5.3-Flash'

#Prompt eingeben
PROMPT='Hallo, bitte führe meine Prompt aus'

#Tokenlimit setzen
MAX_TOKENS=2000

#Inference Call
jq -n \
    --arg model "$MODEL" \
    --arg prompt "$PROMPT" \
    --argjson max_tokens "$MAX_TOKENS" '{
      model: $model,
      messages: [
        {
          role: "user",
          content: $prompt
        }
      ],
      stream: true,
      stream_options: {include_usage: true},
      max_tokens: $max_tokens
    }' |
  curl --fail-with-body --no-buffer -sS \
    --connect-timeout 5 \
    --max-time 180 \
    https://llm.stoney-cloud.com/v1/chat/completions \
    -H "Authorization: Bearer ${STONEY_KEY:?Set your production STONEY_KEY}" \
    -H 'Content-Type: application/json' \
    --data-binary @- |
  tee "$STREAM_FILE" |
  python3 -u -c '
import json
import sys

section = None
latest_usage = None
done = False

for line in sys.stdin:
    if not line.startswith("data:"):
        continue

    payload = line[5:].strip()
    if payload == "[DONE]":
        done = True
        continue

    try:
        event = json.loads(payload)
    except json.JSONDecodeError:
        continue

    if not isinstance(event, dict):
        continue

    if "error" in event:
        print("\nAPI error:", event["error"], flush=True)

    for choice in event.get("choices") or []:
        delta = choice.get("delta") or {}
        reasoning = delta.get("reasoning") or delta.get("reasoning_content")

        for label, text in (
            ("Reasoning", reasoning),
            ("Answer", delta.get("content")),
        ):
            if isinstance(text, str) and text:
                if section != label:
                    print("\n\n--- {} ---".format(label), flush=True)
                    section = label
                print(text, end="", flush=True)

        if choice.get("finish_reason"):
            print(
                "\n[Finish reason: {}]".format(choice["finish_reason"]),
                flush=True,
            )

    usage = event.get("usage")
    if isinstance(usage, dict):
        latest_usage = usage

if latest_usage is not None:
    label = "Usage" if done else "Last reported usage (stream incomplete)"
    print("\n\n{}:\n{}".format(
        label, json.dumps(latest_usage, indent=2)
    ), flush=True)

print(
    "\n[Stream complete]" if done else "\n[Stream ended without DONE]",
    flush=True,
)
'