RS
Ringside
Back to cookbook

Streaming chat completion

Tokens appear as the model writes them rather than after a ten-second stare at a spinner, and Ctrl-C closes the connection instead of leaking it. You also get an exact token count at the end, which most streaming clients drop, because the usage frame only arrives if you ask for it.

What you need

  • An FC API key (from /app/api-keys) with scope api:chat
  • pip install openai>=1.40
  • Python 3.10+

Full code

python
# stream_chat.py
import os, signal, sys
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["FC_API_KEY"],
    base_url="https://api.fightclub.pro/v1",
)

# Graceful cancel. Hit Ctrl-C once to close the HTTP stream.
_current_stream = None

def _handle_sigint(*_):
    if _current_stream is not None:
        _current_stream.close()
    sys.exit(130)

signal.signal(signal.SIGINT, _handle_sigint)


def stream_chat(user_prompt: str, customer: str):
    """Stream one assistant turn. Returns (full_text, usage_dict)."""
    global _current_stream
    buf = []
    usage = None

    _current_stream = client.chat.completions.create(
        model="fc:openai/gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are helpful and terse."},
            {"role": "user", "content": user_prompt},
        ],
        stream=True,
        stream_options={"include_usage": True},
        extra_headers={
            "FC-Customer": customer,
            "FC-Tag": "cli.stream_chat",
        },
    )

    for chunk in _current_stream:
        # Content deltas arrive until the last chunk, which carries usage.
        if chunk.choices and chunk.choices[0].delta.content:
            delta = chunk.choices[0].delta.content
            buf.append(delta)
            sys.stdout.write(delta)
            sys.stdout.flush()
        if chunk.usage is not None:
            usage = {
                "prompt": chunk.usage.prompt_tokens,
                "completion": chunk.usage.completion_tokens,
                "total": chunk.usage.total_tokens,
            }

    _current_stream = None
    sys.stdout.write("\n")
    return "".join(buf), usage


if __name__ == "__main__":
    text, usage = stream_chat(
        user_prompt="Explain Ringside Client Tokens in 2 sentences.",
        customer="cus_demo",
    )
    print(f"\n[usage] {usage}")

Walkthrough

The trick for a clean Ctrl-C is to keep the iterator in a module-level _current_stream handle and call .close() from the SIGINT handler. Without that, Python's stdlib prints a traceback and the HTTP connection lingers until TCP keepalive times out on Ringside's side.

stream_options={"include_usage": True} tells Ringside to emit one extra SSE frame at the end containing usage.prompt_tokens / completion_tokens / total_tokens. The usage is computed server-side from canonical token counters, trust this over any client-side tokenizer.

FC-Tag attaches a free-form label to the request. You'll see it in /app/logs and can filter by it in GET /v1/usage?group_by=tag.

Run it

bash
export FC_API_KEY=ko_0d7f2a91c4e35b86af10d2c7e94b6f3a5d81c02e7b4936af18d5c60e2a7f9b34
python stream_chat.py

You should see tokens stream inline, then a final [usage] {'prompt': 22, 'completion': 41, 'total': 63} line.

What's next