Streaming tokens from a language model changes how your application feels—users see output immediately instead of staring at a spinner. The openai python sdk streaming interface hides most of the SSE plumbing, but you still need to handle async iteration, partial deltas, and completion metadata correctly. This tutorial builds a runnable streaming client step by step, shows exact output at each checkpoint, and covers the edge cases that matter in production.
Prerequisites
- Python 3.10 or newer
openaiPython package >= 1.30.0 (pip install -U openai)- An API key from an OpenAI-compatible provider. Export it as
OPENAI_API_KEY. - (Optional) Any OpenAI-compatible endpoint URL if you are not using OpenAI directly.
If you point the SDK at an OpenAI-compatible gateway like n4n.ai, the same openai python sdk streaming code works across 240+ models and inherits automatic fallback when a provider is rate-limited.
Initialize the client
The modern SDK is explicit about base URLs and timeouts. Create a single client instance and reuse it.
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["OPENAI_API_KEY"],
# base_url="https://api.n4n.ai/v1", # optional: any compatible gateway
timeout=30.0,
max_retries=2,
)Reusing the client avoids socket churn. The timeout applies to the initial connection; the stream itself can run longer. The max_retries parameter does not replay a half-consumed stream, but it will retry the initial HTTP request if it fails before the first byte.
Synchronous streaming, token by token
The simplest correct pattern is a for loop over the stream. Pass stream=True and iterate chunk.choices[0].delta.content.
def stream_basic(prompt: str) -> None:
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
print()
stream_basic("Explain TCP slow start in one paragraph.")Expected output (truncated, printed incrementally):
TCP slow start is a congestion control algorithm that gradually increases the
amount of data sent over a network connection. It begins with a small congestion
window and doubles it every round-trip time until packet loss is detected...Each print fires as a chunk arrives. The flush=True forces stdout to update immediately—without it, many terminals buffer line-by-line and you lose the streaming effect.
What a chunk actually contains
Inspect one chunk to see the shape. The SDK exposes Pydantic models, but the underlying protocol is Server-Sent Events with JSON payloads.
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Say hi."}],
stream=True,
)
first = next(iter(stream))
print(first.model_dump_json())Typical first chunk:
{
"id": "chatcmpl-123",
"object": "chat.completion.chunk",
"created": 1700000000,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"delta": {"role": "assistant", "content": ""},
"finish_reason": null
}
]
}Note the role appears only in the first delta. Subsequent chunks carry content only. The final chunk sets finish_reason to "stop" (or "length", "tool_calls", etc.) and delta is empty. The openai python sdk streaming protocol guarantees that choices is always a single-element list for non-parallel requests.
Async streaming with AsyncOpenAI
For server code (FastAPI, asyncio), use AsyncOpenAI. The iteration becomes async for.
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
async def stream_async(prompt: str) -> None:
stream = await async_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for chunk in stream:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
print()
asyncio.run(stream_async("Write a haiku about latency."))Expected output prints token-by-token as before, but the event loop stays free to handle other connections. Do not wrap the async for in a blocking time.sleep or similar; that defeats the purpose.
Streaming tool calls
When you enable tools, deltas arrive as fragmented tool_calls objects. You must accumulate them yourself; the SDK does not merge across chunks.
def stream_tools(prompt: str) -> None:
tools = [{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
}]
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
tools=tools,
stream=True,
)
for chunk in stream:
tc = chunk.choices[0].delta.tool_calls
if tc:
for call in tc:
# call.function.name and call.function.arguments arrive in pieces
if call.function.arguments:
print(call.function.arguments, end="", flush=True)
print()
stream_tools("What's the weather in Oslo?")You will see JSON fragments like {"city": "Os then lo"}. Concatenate and json.loads only after finish_reason is non-null. Attempting to parse mid-stream raises JSONDecodeError because the arguments string is incomplete.
Capturing usage metadata
By default, streaming responses omit token counts. Request them with stream_options.
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Count to 5."}],
stream=True,
stream_options={"include_usage": True},
)
for chunk in stream:
if chunk.usage:
print("\nUSAGE:", chunk.usage.model_dump())The usage chunk arrives last, after the content deltas. It reports prompt_tokens, completion_tokens, and total_tokens. Per-token metering on gateways such as n4n.ai relies on this same field, so your billing logic should wait for the terminal chunk rather than summing deltas.
Error handling and cancellation
Network drops mid-stream raise APIConnectionError. Wrap the loop and close the stream on cancellation.
from openai import APIConnectionError
def stream_safe(prompt: str) -> None:
stream = None
try:
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
except APIConnectionError:
print("\n[stream interrupted]")
finally:
if stream is not None:
stream.close()If the caller hits Ctrl-C, the generator’s close() triggers a clean shutdown of the HTTP connection. Without it, the underlying requests or httpx connection may linger.
Streaming from a FastAPI endpoint
In a web server, yield chunks directly to the response body.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/chat")
def chat(q: str):
def event_stream():
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": q}],
stream=True,
)
for chunk in stream:
if content := chunk.choices[0].delta.content:
yield content
return StreamingResponse(event_stream(), media_type="text/plain")The StreamingResponse consumes the generator and writes each string to the wire. Set media_type="text/event-stream" if you want strict SSE, but text/plain works for simple token forwarding.
Production checklist
- Set
timeoutandmax_retrieson the client; streams fail differently than request/response calls. - Always flush output or use a proper async writer; buffered stdout hides the streaming benefit.
- Accumulate
tool_callsarguments across chunks before parsing JSON. - Request
include_usageif you bill per token. - For multi-model routing, point
base_urlat a gateway and keep the openai python sdk streaming code unchanged—fallback and cache hints pass through transparently. - Close the stream in a
finallyblock; leaked connections accumulate under load.
That is a complete, runnable path from zero to a resilient streaming client.