Build a Tiny Streaming Timer Before You Trust a Free Model Endpoint

Naive latency timers lie when a model streams: the first chunk can arrive in 400 ms while the full response takes 3 seconds, and a total-time-only probe hides where the wait happens.

I wanted one small probe I could point at any OpenAI-compatible chat endpoint. This week's agent and model-endpoint discussions kept asking whether a free hosted endpoint is fast enough without separating queue wait from token generation. The learning question: can a tiny wall-clock recorder show the difference between time-to-first-token and total completion time?

Disclosure: This article was prepared as part of MonkeyCode's product outreach. The operator describes MonkeyCode as an open-source project and advertises free model access with a 30 million token allowance and a free server option. I used the free server path as the target for this probe. The timing numbers below are shape examples, not a benchmark.

Prerequisites

• Python 3.10+
• requests 2.31+ (pip install requests)
• An OpenAI-compatible chat completions endpoint and API token
• Comfort reading server-sent event chunks as newline-delimited JSON

No GPU is required. If you do not have a local endpoint, the operator-advertised free server path removes that setup step.

The probe

The script records two useful timestamps in stream mode:

1. start: before the POST
2. first: at the first parsed chunk that contains text
3. last: at the last parsed chunk before [DONE]

Then it prints ttfbt and total. In non-stream mode there is only one finish time, so ttfbt equals total.

Save this as ttfbt.py:

import argparse
import json
import os
import time

import requests


def fetch_chunks(url, token, payload, timeout):
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
    }
    with requests.post(
        url, headers=headers, json=payload, stream=True, timeout=timeout
    ) as resp:
        resp.raise_for_status()
        for raw in resp.iter_lines(decode_unicode=True):
            if not raw:
                continue
            if raw.startswith('data:'):
                raw = raw[5:].strip()
            if raw == '[DONE]':
                break
            if not raw:
                continue
            try:
                obj = json.loads(raw)
            except json.JSONDecodeError:
                continue
            delta = obj.get('choices', [{}])[0].get('delta', {})
            piece = delta.get('content')
            if piece:
                yield time.perf_counter(), piece


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--url', default=os.environ.get('MODEL_URL'))
    parser.add_argument('--token', default=os.environ.get('MODEL_TOKEN'))
    parser.add_argument('--model', default=os.environ.get('MODEL_NAME', 'free-tier-model'))
    parser.add_argument('--stream', action='store_true')
    args = parser.parse_args()

    if not args.url or not args.token:
        raise SystemExit('set MODEL_URL and MODEL_TOKEN, or pass --url and --token')

    payload = {
        'model': args.model,
        'messages': [{'role': 'user', 'content': 'Count from 1 to 10, one number per line.'}],
        'temperature': 0.0,
        'stream': args.stream,
        'max_tokens': 64,
    }

    start = time.perf_counter()
    first = None
    last = start
    text = []

    if args.stream:
        for at, piece in fetch_chunks(args.url, args.token, payload, 60):
            if first is None:
                first = at
            last = at
            text.append(piece)
        if first is None:
            raise SystemExit('No SSE chunks parsed. Check that --stream matches the endpoint response.')
        joined = ''.join(text)
        ttfbt = (first - start) * 1000
        total_ms = (last - start) * 1000
        print(f'stream=true chars={len(joined)} ttfbt={ttfbt:.0f}ms total={total_ms:.0f}ms')
    else:
        headers = {'Authorization': f'Bearer {args.token}', 'Content-Type': 'application/json'}
        resp = requests.post(args.url, headers=headers, json=payload, timeout=60)
        resp.raise_for_status()
        obj = resp.json()
        content = obj['choices'][0]['message']['content']
        total_ms = (time.perf_counter() - start) * 1000
        print(f'stream=false chars={len(content)} ttfbt={total_ms:.0f}ms total={total_ms:.0f}ms')


if __name__ == '__main__':
    main()

Run it

export MODEL_URL=https://your-free-endpoint/v1/chat/completions
export MODEL_TOKEN=your-token
python ttfbt.py --stream

Expected shape:

stream=true chars=53 ttfbt=412ms total=2795ms

Then run the same prompt without streaming:

python ttfbt.py

Expected shape:

stream=false chars=53 ttfbt=2713ms total=2713ms

Use these only as shape examples. A single pair tells you little about a shared free tier.

One failing input

Point the streaming flag at an endpoint that returns one JSON object instead of SSE chunks:

python ttfbt.py --stream

If the response is not newline-delimited SSE, the script exits with:

No SSE chunks parsed. Check that --stream matches the endpoint response.

That is the fixture I check first: the probe assumes the server actually streams. A 200 response is not enough to prove stream framing.

What the learner should understand

• ttfbt includes queue wait, prompt processing, and first-token generation; it is not network latency alone.
• total includes everything after the first token, including per-token generation time and any gaps between chunks.
• Non-streaming total is not equal to streaming total on a cold or shared endpoint. The split matters when you decide whether to add a timeout or retry.
• A short ttfbt with a long total suggests generation dominates; a long ttfbt with a short total suggests queue or server startup.

Common mistakes

• Measuring once and treating it as truth. Shared free tiers vary. Run at least five times and report the median.
• Different prompt lengths between runs. Keep max_tokens and prompt fixed.
• Calling resp.json() on a streamed body. Use iter_lines/SSE parsing.
• Comparing ttfbt across providers without controlling for response length.

Where the free server option fits

The MonkeyCode operator advertises a free server option, which removes local GPU setup for a probe like this. A free endpoint is also the exact environment where naive timing breaks: noisy queues and stream buffering are visible in the ttfbt/total split. I do not treat the 30 million token allowance as verified; if you have access, run this probe and inspect the shape yourself.

Limitations

• The probe is not a load test or benchmark.
• It depends on the server's SSE framing and may miscount chunks that do not contain delta.content.
• It cannot separate prompt processing from queue wait without server-side timings.
• Free-server availability and quotas can change; do not build production features on an unverified free endpoint.

Who should not use this

Skip this if you need a production latency SLO, a concurrency test, or provider-neutral benchmark. Use proper tooling for those. This is for learners who want to see the difference between the first token appearing and the response finishing.

Extension

Add an X-Turn or request ID and send the same prompt at three different times of day. Record ttfbt/total pairs and plot a scatter. The moment two runs with the same prompt produce very different splits, you have found queue noise instead of model speed.

If you run the probe against a free endpoint, share one pair: ttfbt and total. I am curious which one varies first.

Comments

[Marcus Kim]

Using the first parsed delta.content chunk instead of the first SSE line is an important distinction; a 200 response-or even an early metadata event-doesn't prove the user has received useful output. The 412 ms TTFBT versus 2795 ms total example also makes queue delay and generation time easier to reason about, though I'd record [DONE] separately because the last text chunk can understate full stream completion. For an endpoint decision, I'd add output-token count, errors, and p50/p95 across several time windows: a five-run median teaches the concept, but it can hide the cold starts and throttling that shape the actual product experience.