Stream vLLM responses with Python and detect incomplete answers

TL;DR: Build a Python client for a text-only vLLM chat stream, keep a record of the first visible text and final completion, and deliberately break the connection to check the failure path. Start with the included CPU fixture, then change the endpoint to your model. A successful HTTP connection, an assistant-role event and a partial sentence are three different observations; this client requires nonempty text, a stop finish reason and the terminal [DONE] event before marking the run complete.

Your model answers correctly in a non-streaming request. After you enable streaming, the interface sometimes displays half an answer and treats the closed connection as success. Increasing the output limit does not repair that contract. The client needs to know which bytes form an event, which events carry visible text, and which event ends the response.

This project produces stream_client.py, a deliberately small synthetic server and executable protocol tests. It uses Python 3.12 and its standard library on Linux or macOS. Windows users can follow the commands inside WSL. The local fixture uses no model, account or GPU. The real-model extension assumes an already running server from the Qwen deployment guide or the Unsloth adapter guide, which requires a free account.

The portable skill is reading a streamed response correctly. The model name and serving release are replaceable inputs. Sources were checked on 26 September 2026; the companion GPU recipe uses vLLM 0.29.0. Actual GPU streaming performance has not been measured for this guide.

1. Get a complete stream running locally

Create a directory and download the three files from this page's Project files links into it: stream_client.py, fixture_server.py and test_stream.py. Keep their names unchanged because the tests import the other two modules.

mkdir -p streaming-lab
cd streaming-lab
python3 --version
python3 -m unittest -v test_stream.py

Use Python 3.12. You should see five tests pass. The tests open a temporary loopback port; a machine policy that blocks local listeners must be resolved before continuing. They do not contact an external inference provider.

In terminal A, from streaming-lab, start the fixture:

python3 fixture_server.py --port 18080 --mode ok

Leave that terminal running. In terminal B, change to the same directory and send one request:

python3 stream_client.py \
  --url http://127.0.0.1:18080/v1/chat/completions \
  --model synthetic-fixture \
  --output fixture-ok.json
python3 -m json.tool fixture-ok.json

The saved record must contain "ok": true, "text": "READY ✓", "finish_reason": "stop" and "done": true. It also contains two content events and a fixture-supplied completion-token count of three. Those numbers intentionally differ: an event is a transport envelope and can contain more than one token. The fixture's token count is synthetic data, not a tokenizer measurement.

Timing values vary with your machine. Check their ordering: first_text_ms is positive and smaller than duration_ms. Do not copy a fixture duration into a GPU performance comparison. The server deliberately sleeps between writes, which makes the client behavior inspectable without pretending to model inference.

A second invocation with the same output path fails before sending a request. Choose a new filename for every attempt. This protects the evidence when a later run fails.

2. Understand the three boundaries in a stream

An HTTP response body arrives in arbitrary byte fragments. Server-sent events, abbreviated SSE, put a text format inside that body. A JSON completion chunk is then carried inside an SSE data: field. These boundaries need not align.

The parser handles UTF-8 incrementally, joins multiple data: lines within one event and emits an event only at an empty line. It accepts the line-ending forms defined by the WHATWG event-stream rules. Comments beginning with : do not become completion data. A final line without an event terminator is not a complete event.

One visible character can cross reads. In the fixture, the check mark is encoded as three UTF-8 bytes. Decoding each received byte fragment independently would sometimes raise a decoding error or corrupt the character. codecs.getincrementaldecoder retains incomplete character bytes until the next fragment arrives.

self.decoder = codecs.getincrementaldecoder('utf-8-sig')()
for char in self.decoder.decode(raw):
    # Line and event state survive the next network read.
    ...

This excerpt locates the mechanism; the complete executable implementation appears at the end. The tests feed the same stream at every chunk size from one byte through the entire message. That catches a parser that works only when the operating system happens to return one whole event per read.

The client is scoped to one text completion (n=1). It rejects tool calls, multiple choices and unexpected content types. Tool arguments need their own assembly and validation rules before they can trigger an action. A reasoning field also does not count as user-visible answer text in this client. Review your model's chat template and reasoning settings before assuming that an initial silent interval is a network problem.

3. Start the latency clock at the right observation

The diagram shows event order. Its vertical distances are schematic. The timer starts before connection setup and stops its first-text measurement on the first nonempty delta.content string. It therefore includes client-observed connection, queueing and delivery delay. It is not the engine's internal time-to-first-token metric.

The implementation uses a monotonic performance clock. Wall-clock corrections cannot make a later event appear earlier. Python's timing reference explains the clock's role.

For example, suppose an explicitly hypothetical trace has request start at 0 ms, headers at 40 ms, an assistant-role event at 55 ms, first text at 120 ms and terminal completion at 420 ms. The first-visible-text latency is 120 − 0 = 120 ms. The full request takes 420 − 0 = 420 ms. Reporting 40 ms would describe the headers; reporting 55 ms would describe an event with no answer text.

event_gaps_ms measures spacing between content-bearing events observed by the client. A single read can contain several events, giving nearly zero gaps even when generation happened at different times on the server. These values help diagnose delivery bursts. They are not token-level latency samples, and their percentile must not be labelled inter-token latency. For the underlying terminology, see TTFT, TPOT and goodput.

4. Prove that partial output does not pass

Stop terminal A with Ctrl+C, then restart the fixture in broken mode:

python3 fixture_server.py --port 18080 --mode broken

In terminal B:

python3 stream_client.py \
  --url http://127.0.0.1:18080/v1/chat/completions \
  --model synthetic-fixture \
  --output fixture-broken.json

The process exits with status 1. The JSON retains READY ✓, but ok is false, done is false and the error identifies EOF before [DONE]. An interface can display retained text as an interrupted answer; it must not silently save it as a completed response.

Repeat with --mode length on the server and a new output filename on the client. This time the fixture sends [DONE] after finish_reason=length. The client still fails the run because the output budget ended the response. Applications that deliberately accept truncated output may choose another policy, but that policy must be explicit in their result schema.

The distinction also explains why a transport marker is insufficient. [DONE] tells this endpoint's client that the stream has ended. It does not establish factual correctness, task success or an acceptable finish reason. The vLLM server reference describes the compatible endpoints; the client adds a deliberately strict text-completion policy on top.

The fixture's malformed and http-error modes exercise invalid JSON and HTTP 503. A 503 produces no answer, while malformed JSON preserves the text received before the invalid event. Keeping both outcomes makes operational diagnosis much easier than one generic “empty response” counter.

5. Bound waiting and memory

Two time limits answer different questions. --idle 10 sets the socket wait used for network operations. --deadline 60 limits elapsed time checked during body reads, including time already spent establishing the connection. Receiving periodic bytes can avoid an idle timeout, so the read loop also checks the overall elapsed budget.

python3 stream_client.py \
  --url http://127.0.0.1:18080/v1/chat/completions \
  --model synthetic-fixture \
  --idle 2 --deadline 5 \
  --output bounded-run.json

This teaching client is synchronous. Operating-system hostname resolution and an unusually slow header exchange are not guaranteed to obey a strict wall-clock cancellation deadline. If your service requires a hard end-to-end deadline across those stages, use a client/runtime with explicit cancellation of the entire request and test it. The saved duration exposes what actually happened; the guide does not describe the socket timeout as a universal five-second guarantee. See Python's HTTP connection behavior.

The parser rejects a line larger than 65,536 decoded characters, a multiline event larger than 1,048,576 characters, and an assembled answer larger than the same answer limit. These are lab guardrails, not recommended production payload limits. Raise them only after defining the payloads your application accepts. Do not remove every bound to accommodate one broken stream.

An idle timeout closes the client connection. That is a cancellation signal to the server; this client cannot prove the GPU stopped computing. Test cancellation against the serving engine's actual request lifecycle before claiming saved GPU time. Likewise, this client performs no automatic retry. Retrying after text has been shown can generate a second, different answer and can charge for another request.

6. Connect the same client to your model

First establish that your existing model responds without streaming. On the GPU host, use the smoke check from the deployment guide. The companion servers require the API key you chose during deployment. In the client terminal, reuse that value through the client's environment variable, then inspect the model-list endpoint:

export MODEL_API_KEY="${VLLM_API_KEY:?Set the key used by your serving process}"
curl --fail --silent --show-error \
  --header "Authorization: Bearer $MODEL_API_KEY" \
  http://127.0.0.1:8000/v1/models

For the companion Qwen3.5 deployment, use hands-on-qwen. For the saved adapter, use incident-labeler and port 8001. The identifier in a client request selects the served model or adapter; it is not an instruction to download another repository.

Run the client on the same host, or reach its loopback listener through an SSH tunnel. Set --url to the actual chat-completions endpoint and --model to the returned identifier:

python3 stream_client.py \
  --url http://127.0.0.1:8001/v1/chat/completions \
  --model incident-labeler \
  --prompt 'A worker was killed after exhausting its available RAM.' \
  --max-tokens 16 \
  --output adapter-stream.json

Treat the answer as an observation, not an expected benchmark result. Compare it with the incident classification contract in the fine-tuning guide. This client does not apply that guide's exact-match quality evaluator automatically.

For a remote authenticated HTTPS endpoint, put the token in MODEL_API_KEY; do not put it into the URL or paste it into shared command history. The client uses normal certificate and hostname verification. --ca-file can provide a trusted private CA for a lab endpoint. It has no switch to disable certificate verification. Plain HTTP is accepted only on loopback addresses. An SSH tunnel ending on your local loopback is one way to retain that property.

The client's saved JSON contains the generated answer. Keep the file private if the output can reveal prompt data. It does not save the API key or request headers. Review Python's TLS client defaults before adapting the connection code.

7. Interpret failures in the order they occur

Start with the first failing boundary. Changing scheduler flags cannot repair a client that is talking to the wrong TLS hostname. Conversely, opening more firewall ports cannot repair a JSON event that your parser discarded.

8. Check your understanding and keep the evidence

A response sends an empty role event, two text events, finish_reason=length, usage and [DONE]. Should it pass? Under this guide's contract it fails. The stream terminated, but the generation budget ended the answer.

A second response sends one text event containing ten tokens. How many token gaps did the client measure? None. It observed one content event and cannot recover the ten token timestamps from that envelope.

Keep the client source, Python version, output JSON and the model/container revision from the serving guide together. If you compare direct and proxied endpoints, use the same prompt, output limit and model configuration and record which path each file represents. Never mix the synthetic fixture results with model results in one performance table.

Stop the fixture with Ctrl+C. Stop any SSH tunnel separately. If you started a cloud GPU machine for the extension, stop or delete it through that provider after preserving the required artifacts; closing a terminal does not stop instance billing.

For a broader service failure contract, read Streaming tokens to clients: what can go wrong?. For non-streaming quality and release checks, continue to Benchmark a vLLM service and build a release gate, which requires Premium.

Complete project files

The files below are the same bytes served by this page's download links. The tests exercise a real local HTTP connection and parser fragmentation. They establish client behavior; they do not execute vLLM or measure a GPU.

stream_client.py

"""Bounded text-only SSE client. No automatic retries or token-gap claims."""
import argparse
import codecs
import http.client
import ipaddress
import json
import math
import os
from pathlib import Path
import ssl
import time
from urllib.parse import urlsplit


class Events:
    def __init__(self):
        self.decoder = codecs.getincrementaldecoder('utf-8-sig')()
        self.line = ''
        self.data = []
        self.size = 0
        self.after_cr = False

    def feed(self, raw):
        events = []
        for char in self.decoder.decode(raw):
            if self.after_cr:
                self.after_cr = False
                if char == '\n':
                    continue
            if char in '\r\n':
                line, self.line = self.line, ''
                self.after_cr = char == '\r'
                if not line:
                    if self.data:
                        events.append('\n'.join(self.data))
                    self.data, self.size = [], 0
                elif line.startswith('data:') or line == 'data':
                    value = line.partition(':')[2]
                    value = value[1:] if value.startswith(' ') else value
                    self.size += len(value)
                    if self.size > 1_048_576:
                        raise ValueError('SSE event exceeds 1 MiB character limit')
                    self.data.append(value)
            else:
                self.line += char
                if len(self.line) > 65_536:
                    raise ValueError('SSE line exceeds 64 KiB character limit')
        return events

    def finish(self):
        self.decoder.decode(b'', final=True)  # reject an incomplete UTF-8 character
        # An unterminated final event is not dispatched.


def stream(url, model, prompt, *, max_tokens=128, idle=10.0, deadline=60.0,
           ca_file=None, token=None, on_text=None):
    parsed = urlsplit(url)
    if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError('Use a complete HTTP(S) endpoint without credentials, query or fragment')
    if parsed.scheme == 'http':
        try:
            local = ipaddress.ip_address(parsed.hostname).is_loopback
        except ValueError:
            local = parsed.hostname == 'localhost'
        if not local:
            raise ValueError('Plain HTTP is limited to loopback; use HTTPS or an SSH tunnel')
    if not all(math.isfinite(v) and v > 0 for v in (idle, deadline)) or not 1 <= max_tokens <= 32768:
        raise ValueError('Timeouts must be positive and finite; max_tokens must be 1..32768')
    headers = {'Content-Type': 'application/json', 'Accept': 'text/event-stream'}
    if token:
        headers['Authorization'] = 'Bearer ' + token
    payload = json.dumps({'model': model, 'messages': [{'role': 'user', 'content': prompt}],
                          'stream': True, 'stream_options': {'include_usage': True},
                          'max_tokens': max_tokens, 'temperature': 0, 'n': 1}).encode()
    kwargs = {'host': parsed.hostname, 'port': parsed.port, 'timeout': min(idle, deadline)}
    conn = (http.client.HTTPSConnection(**kwargs, context=ssl.create_default_context(cafile=ca_file))
            if parsed.scheme == 'https' else http.client.HTTPConnection(**kwargs))
    start = time.perf_counter()
    record = {'ok': False, 'model': model, 'text': '', 'first_text_ms': None,
              'duration_ms': None, 'content_events': 0, 'event_gaps_ms': [],
              'finish_reason': None, 'done': False, 'completion_tokens': None,
              'error': None}
    last = None
    parser = Events()
    try:
        conn.connect()
        sock = conn.sock
        conn.request('POST', parsed.path or '/', body=payload, headers=headers)
        response = conn.getresponse()
        if response.status != 200:
            raise ValueError('HTTP status ' + str(response.status))
        if response.getheader('Content-Type', '').split(';')[0].strip().lower() != 'text/event-stream':
            raise ValueError('Expected text/event-stream')
        while not record['done']:
            remaining = deadline - (time.perf_counter() - start)
            if remaining <= 0:
                raise TimeoutError('Overall read deadline exceeded')
            sock.settimeout(min(idle, remaining))
            raw = response.read1(4096)
            if not raw:
                parser.finish()
                raise ValueError('EOF before [DONE]')
            for event in parser.feed(raw):
                now = time.perf_counter()
                if now - start > deadline:
                    raise TimeoutError('Overall read deadline exceeded')
                if record['done']:
                    raise ValueError('Unexpected event after [DONE]')
                if event == '[DONE]':
                    record['done'] = True
                    continue
                value = json.loads(event)
                if not isinstance(value, dict) or 'error' in value:
                    raise ValueError('Invalid event or server error event')
                usage = value.get('usage')
                if usage is not None:
                    count = usage.get('completion_tokens')
                    if type(count) is not int or count < 0:
                        raise ValueError('Invalid completion_tokens')
                    record['completion_tokens'] = count
                choices = value.get('choices', [])
                if not isinstance(choices, list) or len(choices) > 1:
                    raise ValueError('Only one completion choice is supported')
                for choice in choices:
                    if choice.get('index') != 0:
                        raise ValueError('Unexpected choice index')
                    delta = choice.get('delta', {})
                    if delta.get('tool_calls') or delta.get('function_call'):
                        raise ValueError('Tool calls require a separate tool-aware client')
                    content = delta.get('content')
                    if content is not None and not isinstance(content, str):
                        raise ValueError('Text delta must be a string')
                    if content:
                        if record['finish_reason'] is not None:
                            raise ValueError('Content arrived after completion')
                        elapsed = (now - start) * 1000
                        if record['first_text_ms'] is None:
                            record['first_text_ms'] = elapsed
                        if last is not None:
                            record['event_gaps_ms'].append((now - last) * 1000)
                        last = now
                        record['text'] += content
                        record['content_events'] += 1
                        if len(record['text']) > 1_048_576:
                            raise ValueError('Text exceeds 1 MiB character limit')
                        if on_text:
                            on_text(content)
                    reason = choice.get('finish_reason')
                    if reason is not None:
                        if record['finish_reason'] is not None:
                            raise ValueError('Duplicate completion event')
                        record['finish_reason'] = reason
        record['ok'] = record['finish_reason'] == 'stop' and bool(record['text'].strip())
        if not record['ok']:
            record['error'] = 'Need nonempty text, finish_reason=stop and [DONE]'
    except (OSError, ValueError, TypeError, KeyError, AttributeError, http.client.HTTPException) as exc:
        record['error'] = str(exc) if isinstance(exc, ValueError) else type(exc).__name__
    finally:
        conn.close()
        record['duration_ms'] = (time.perf_counter() - start) * 1000
    return record


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--url', default='http://127.0.0.1:8000/v1/chat/completions')
    p.add_argument('--model', required=True)
    p.add_argument('--prompt', default='Reply with exactly the word READY.')
    p.add_argument('--max-tokens', type=int, default=128)
    p.add_argument('--idle', type=float, default=10)
    p.add_argument('--deadline', type=float, default=60)
    p.add_argument('--ca-file')
    p.add_argument('--output', required=True)
    a = p.parse_args()
    # Reserve the evidence file before sending any request. Never overwrite a run.
    with Path(a.output).open('x') as output:
        record = stream(a.url, a.model, a.prompt, max_tokens=a.max_tokens,
                        idle=a.idle, deadline=a.deadline, ca_file=a.ca_file,
                        token=os.environ.get('MODEL_API_KEY'))
        json.dump(record, output, indent=2)
        output.write('\n')
    print(json.dumps({k: record[k] for k in ('ok','first_text_ms','duration_ms','finish_reason','error')}))
    raise SystemExit(0 if record['ok'] else 1)


if __name__ == '__main__':
    main()

fixture_server.py

"""Synthetic protocol fixture: this process contains no model or GPU code."""
import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import time


def frames(mode):
    def event(delta=None, reason=None):
        return ('data: ' + json.dumps({'choices': [{'index': 0, 'delta': delta or {},
                                                   'finish_reason': reason}]}, ensure_ascii=False) + '\r\n\r\n').encode()
    output = [b': synthetic keepalive\r\n\r\n', event({'role': 'assistant'}),
              event({'content': 'READY '}), event({'content': '✓'})]
    if mode == 'broken':
        return output
    if mode == 'malformed':
        return output + [b'data: {invalid}\n\n']
    output.append(event(reason='length' if mode == 'length' else 'stop'))
    output.append(b'data: {"choices": [], "usage": {"completion_tokens": 3}}\n\n')
    output.append(b'data: [DONE]\n\n')
    return output


class Handler(BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def log_message(self, *_):
        pass

    def do_POST(self):
        if self.path != '/v1/chat/completions':
            self.send_error(404)
            return
        size = int(self.headers.get('Content-Length', 0))
        if not 0 < size <= 65536:
            self.send_error(413)
            return
        request = json.loads(self.rfile.read(size))
        if request.get('stream') is not True:
            self.send_error(400)
            return
        if self.server.mode == 'http-error':
            self.send_error(503)
            return
        chunks = frames(self.server.mode)
        self.send_response(200)
        self.send_header('Content-Type', 'text/event-stream; charset=utf-8')
        self.send_header('Cache-Control', 'no-store')
        self.send_header('Content-Length', str(sum(map(len, chunks))))
        self.end_headers()
        try:
            for chunk in chunks:
                time.sleep(self.server.delay)
                self.wfile.write(chunk)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError):
            pass


def server(port=18080, mode='ok', delay=0.05):
    instance = ThreadingHTTPServer(('127.0.0.1', port), Handler)
    instance.mode, instance.delay = mode, delay
    return instance


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--port', type=int, default=18080)
    p.add_argument('--mode', choices=['ok','broken','length','malformed','http-error'], default='ok')
    p.add_argument('--delay', type=float, default=0.05)
    a = p.parse_args()
    if a.delay < 0:
        p.error('delay must be nonnegative')
    with server(a.port, a.mode, a.delay) as instance:
        print('Synthetic fixture listening on 127.0.0.1:' + str(instance.server_port), flush=True)
        try:
            instance.serve_forever()
        except KeyboardInterrupt:
            pass

test_stream.py

"""Executable protocol tests; run without a model or third-party package."""
import threading
import unittest
from fixture_server import server
from stream_client import Events, stream


class StreamingTests(unittest.TestCase):
    def run_fixture(self, mode='ok', **kwargs):
        with server(0, mode, 0.002) as instance:
            thread = threading.Thread(target=instance.serve_forever, daemon=True)
            thread.start()
            try:
                return stream(f'http://127.0.0.1:{instance.server_port}/v1/chat/completions',
                              'synthetic-fixture', 'READY', **kwargs)
            finally:
                instance.shutdown()
                thread.join()

    def test_utf8_and_all_newlines_survive_every_byte_boundary(self):
        raw = '\ufeff: comment\rdata: one ✓\rdata: two\r\rdata: three\n\ndata: four\r\n\r\n'.encode()
        for size in range(1, len(raw) + 1):
            parser = Events()
            result = []
            for i in range(0, len(raw), size):
                result.extend(parser.feed(raw[i:i+size]))
            parser.finish()
            self.assertEqual(result, ['one ✓\ntwo', 'three', 'four'])

    def test_pending_event_is_not_dispatched_and_bounds_are_enforced(self):
        parser = Events()
        self.assertEqual(parser.feed(b'data: unfinished'), [])
        parser.finish()
        with self.assertRaises(ValueError):
            Events().feed(b'x' * 65537)
        parser = Events()
        parser.feed(b'data: \xe2')
        with self.assertRaises(UnicodeDecodeError):
            parser.finish()

    def test_finished_stream_retains_text_usage_and_clock_boundaries(self):
        result = self.run_fixture()
        self.assertTrue(result['ok'])
        self.assertEqual(result['text'], 'READY ✓')
        self.assertEqual(result['completion_tokens'], 3)
        self.assertEqual(result['content_events'], 2)
        self.assertEqual(len(result['event_gaps_ms']), 1)
        self.assertGreater(result['first_text_ms'], 0)
        self.assertLess(result['first_text_ms'], result['duration_ms'])

    def test_failure_modes_do_not_become_success(self):
        for mode in ('broken', 'length', 'malformed', 'http-error'):
            with self.subTest(mode=mode):
                result = self.run_fixture(mode)
                self.assertFalse(result['ok'])
                self.assertTrue(result['error'])
                if mode != 'http-error':
                    self.assertEqual(result['text'], 'READY ✓')

    def test_deadline_and_remote_plaintext_are_rejected(self):
        self.assertFalse(self.run_fixture(deadline=0.001)['ok'])
        with self.assertRaises(ValueError):
            stream('http://192.0.2.1:8000/v1/chat/completions', 'demo', 'hi')
        with self.assertRaises(ValueError):
            stream('https://user:[email protected]/v1/chat/completions', 'demo', 'hi')


if __name__ == '__main__':
    unittest.main()