When building command-line tools, interactive chatbots, or backend automation that interacts with Large Language Models (LLMs), waiting for the entire text
When building command-line tools, interactive chatbots, or backend automation that interacts with Large Language Models (LLMs), waiting for the entire text generation to finish creates noticeable latency. High-latency blocking calls hurt user experience and fail to provide real-time feedback during long operations like generating scripts, documentation, or code refactoring.
Standard SDK wrappers abstract away the transport layer, but engineers often need to route requests through custom API gateways, reverse proxies, internal LLM endpoints (like vLLM, Ollama, or LocalAI), or custom middleware. In these environments, higher-level SDKs can fail due to non-standard headers or strict SSE frame formatting. Implementing a custom Server-Sent Events (SSE) parser using httpx gives you complete control over streaming HTTP responses, custom frame parsing, timeout handling, and transport failovers while keeping full compatibility with standard LLM provider SDKs like openai.
The script operates on standard streaming protocol mechanisms defined by the W3C Server-Sent Events (SSE) specification. When an HTTP request is made to an OpenAI-compatible endpoint with "stream": true, the server sets Content-Type: text/event-stream and returns chunks using HTTP Transfer-Encoding: chunked.
httpx.Client.stream() opens a persistent HTTP connection and yields raw text lines as they arrive over the wire.SSEDecoder): The parser accumulates lines into a buffer, identifies key-value pairs (e.g., data: {...}), handles multi-line payloads, and emits structured SSEEvent objects when encountering an empty line separator (\n\n).data field. The parser extracts the token delta (choices[0].delta.content), handles stream completion signals (like [DONE]), and yields tokens directly to sys.stdout with immediate flushing for instant terminal rendering.#!/usr/bin/env python3
"""
Streaming AI Responses with SSE Parser
A production-ready CLI utility for streaming LLM completions using a custom
Server-Sent Events (SSE) parser with httpx, alongside standard OpenAI SDK support.
"""
import argparse
import json
import os
import sys
from dataclasses import dataclass
from typing import Generator, Optional
import httpx
from openai import OpenAI
# Environment-based defaults
DEFAULT_MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
DEFAULT_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1")
DEFAULT_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "30.0"))
@dataclass
class SSEEvent:
"""Represents a single parsed Server-Sent Event frame."""
event: str = "message"
data: str = ""
id: Optional[str] = None
retry: Optional[int] = None
class SSEDecoder:
"""
Decodes an incoming line-buffered text stream into W3C-compliant SSE events.
"""
def __init__(self) -> None:
self._event = "message"
self._data: list[str] = []
self._last_id: Optional[str] = None
self._retry: Optional[int] = None
def decode_line(self, line: str) -> Optional[SSEEvent]:
"""
Process a single line from the stream and return an SSEEvent if a complete
event boundary (empty line) is hit.
"""
# Strip trailing end-of-line characters
line = line.rstrip("\r\n")
# Empty line indicates end of an event block
if not line:
if not self._data:
return None
event = SSEEvent(
event=self._event,
data="\n".join(self._data),
id=self._last_id,
retry=self._retry,
)
# Reset event accumulator state
self._event = "message"
self._data = []
return event
# Ignore SSE comments
if line.startswith(":"):
return None
# Parse field name and field value
if ":" in line:
field, value = line.split(":", 1)
if value.startswith(" "):
value = value[1:]
else:
field = line
value = ""
if field == "event":
self._event = value
elif field == "data":
self._data.append(value)
elif field == "id":
self._last_id = value
elif field == "retry":
if value.isdigit():
self._retry = int(value)
return None
def stream_httpx_sse(
prompt: str,
system_prompt: str,
model: str,
base_url: str,
api_key: str,
temperature: float,
max_tokens: int,
) -> Generator[str, None, None]:
"""
Streams completions using direct HTTP POST requests with custom SSE line decoding.
"""
endpoint = f"{base_url.rstrip('/')}/chat/completions"
headers = {
"Content-Type": "application/json",
"Accept": "text/event-stream",
}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
],
"temperature": temperature,
"max_tokens": max_tokens,
"stream": True,
}
decoder = SSEDecoder()
timeout_config = httpx.Timeout(DEFAULT_TIMEOUT, connect=10.0)
with httpx.Client(timeout=timeout_config) as client:
with client.stream("POST", endpoint, headers=headers, json=payload) as response:
if response.status_code != 200:
response.read()
raise httpx.HTTPStatusError(
f"API Request Failed [{response.status_code}]: {response.text}",
request=response.request,
response=response,
)
for line in response.iter_lines():
event = decoder.decode_line(line)
if event and event.data:
# OpenAI SSE stream signals completion with [DONE]
if event.data.strip() == "[DONE]":
break
try:
chunk = json.loads(event.data)
choices = chunk.get("choices", [])
if choices:
delta = choices[0].get("delta", {})
content = delta.get("content", "")
if content:
yield content
except json.JSONDecodeError:
# Skip malformed JSON chunks gracefully
continue
def stream_openai_sdk(
prompt: str,
system_prompt: str,
model: str,
base_url: str,
api_key: str,
temperature: float,
max_tokens: int,
) -> Generator[str, None, None]:
"""
Streams completions using the official OpenAI Python SDK.
"""
client = OpenAI(api_key=api_key or "dummy", base_url=base_url)
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt},
],
temperature=temperature,
max_tokens=max_tokens,
stream=True,
)
for chunk in response:
if chunk.choices:
content = chunk.choices[0].delta.content
if content:
yield content
def main() -> None:
parser = argparse.ArgumentParser(
description="Stream AI responses using low-level SSE parser or OpenAI SDK."
)
parser.add_argument(
"--prompt", "-p", required=True, help="Prompt text to send to the AI model"
)
parser.add_argument(
"--system",
"-s",
default="You are a helpful, precise technical assistant.",
help="System instructions for the model",
)
parser.add_argument(
"--model", "-m", default=DEFAULT_MODEL, help=f"Model name (default: {DEFAULT_MODEL})"
)
parser.add_argument(
"--temperature",
"-t",
type=float,
default=0.7,
help="Sampling temperature (0.0 - 2.0)",
)
parser.add_argument(
"--max-tokens",
type=int,
default=1024,
help="Maximum tokens to generate",
)
parser.add_argument(
"--use-sdk",
action="store_true",
help="Use OpenAI SDK instead of custom httpx SSE parser",
)
args = parser.parse_args()
api_key = os.environ.get("OPENAI_API_KEY", "")
base_url = DEFAULT_BASE_URL
# Validate API Key requirement for cloud OpenAI endpoints
if "api.openai.com" in base_url and not api_key:
sys.stderr.write("Error: OPENAI_API_KEY environment variable is required.\n")
sys.exit(1)
mode_label = "OpenAI SDK" if args.use-sdk else "httpx SSE Parser"
sys.stderr.write(f"Connecting to {base_url} using model '{args.model}' ({mode_label})...\n")
sys.stderr.flush()
try:
if args.use_sdk:
token_stream = stream_openai_sdk(
prompt=args.prompt,
system_prompt=args.system,
model=args.model,
base_url=base_url,
api_key=api_key,
temperature=args.temperature,
max_tokens=args.max_tokens,
)
else:
token_stream = stream_httpx_sse(
prompt=args.prompt,
system_prompt=args.system,
model=args.model,
base_url=base_url,
api_key=api_key,
temperature=args.temperature,
max_tokens=args.max_tokens,
)
for token in token_stream:
sys.stdout.write(token)
sys.stdout.flush()
sys.stdout.write("\n")
sys.stdout.flush()
except KeyboardInterrupt:
sys.stderr.write("\n\nOperation cancelled by user.\n")
sys.exit(130)
except httpx.HTTPStatusError as e:
sys.stderr.write(f"\nHTTP Error: {e}\n")
sys.exit(2)
except httpx.RequestError as e:
sys.stderr.write(f"\nNetwork Error: {e}\n")
sys.exit(3)
except Exception as e:
sys.stderr.write(f"\nUnexpected Error: {e}\n")
sys.exit(1)
if __name__ == "__main__":
main()
The script uses environment variables for operational configuration:
| Variable | Default Value | Description |
|---|