Back to blog

API & SDKGuide

Streaming responses in Python

Print the model's answer as it is generated with the official openai package — sync and async examples, reasoning output, token usage and error handling.

By CheapTokens TeamPublished 3 min read

Python code with a for loop over stream chunks and text appearing token by token
On this page
  1. Setup
  2. Minimal example
  3. Reasoning models
  4. Token usage
  5. asyncio
  6. Stopping early and timeouts
  7. Errors
  8. What's on the wire

With streaming, the answer arrives piece by piece while the model is still generating it, so your users see text right away instead of waiting for the whole response. The API uses the standard OpenAI streaming format (server-sent events), and the official openai Python package handles it for you.

Every example below is complete: copy it into a file and run it. The examples were checked with Python 3.10.12 and openai 3.31.0 against a local test server that reproduces the API's streaming format (syntax and execution).

Setup#

bash
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"

Minimal example#

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a haiku about latency"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:  # service chunk without text, e.g. the final usage chunk
        continue
    text = chunk.choices[0].delta.content
    if text:
        print(text, end="", flush=True)
print()

Two details matter:

  • Skip chunks without choices. Some chunks carry no text — for example the final chunk with token usage.
  • delta.content can be empty. The first chunk often carries only the role, and the last one only finish_reason. Print text only when it's there.

Reasoning models#

zai-org/GLM-5.3-Flash reasons before it answers. The reasoning comes in a separate field of the delta, reasoning_content, and the answer itself in content. The SDK keeps unknown fields, so you can read it with getattr. This is a complete example with its own client and its own stream:

python
import os
import sys
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="zai-org/GLM-5.3-Flash",
    messages=[{"role": "user", "content": "Is 1013 a prime number?"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    thinking = getattr(delta, "reasoning_content", None)
    if thinking:
        print(thinking, end="", flush=True, file=sys.stderr)  # reasoning → stderr
    if delta.content:
        print(delta.content, end="", flush=True)              # answer → stdout
print()

Reasoning can take a while before the first word of the answer appears — another reason to stream.

Token usage#

Ask for usage with stream_options; it comes in the last chunk, which has no choices:

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph"}],
    stream=True,
    stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

if usage:
    print(f"\ninput: {usage.prompt_tokens}, output: {usage.completion_tokens}")

Multiply these numbers by the model's rates to get the cost of the request — see How to estimate the cost of an AI agent.

asyncio#

In async code use AsyncOpenAI and async for. The client has its own name, async_client, so it doesn't clash with the synchronous client from the other examples:

python
import asyncio
import os
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

async def main() -> None:
    stream = await async_client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Name three uses of a message queue"}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

asyncio.run(main())

Stopping early and timeouts#

Leave the loop to stop generation. Opening the stream with with closes the connection when the block ends. The client timeout is set explicitly; the SDK also retries some failed requests on its own (max_retries, 2 by default).

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
    timeout=120,  # seconds; long answers take time
)

received = ""
with client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a long story about a lighthouse"}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            received += chunk.choices[0].delta.content
        if len(received) > 300:  # stop early: enough text received
            break
print(received)

Errors#

Errors that happen before the first chunk — wrong key, empty balance, unknown model — are raised as exceptions when you create the stream:

python
import os
import openai
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

try:
    stream = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Hello"}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()
except openai.AuthenticationError:          # 401: wrong or rotated key
    print("Check the API key")
except openai.BadRequestError as e:         # 400: e.g. unknown model ID
    print("Bad request:", e.message)
except openai.APIStatusError as e:          # other statuses, e.g. 402 when the balance is used up
    print(e.status_code, e.message)

A full list of status codes and fixes is in Common connection errors and how to fix them.

What's on the wire#

To debug, look at the raw stream with curl (-N turns off buffering):

bash
curl -N https://api.cheaptokens.dev/v1/chat/completions \
  -H "Authorization: Bearer $CHEAPTOKENS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-ai/DeepSeek-V4-Flash-0731", "stream": true,
       "messages": [{"role": "user", "content": "Hi"}]}'

Each event is a data: {…} line with a JSON chunk; the stream ends with data: [DONE]. Lines that start with a colon are keep-alive comments sent during long pauses — clients ignore them.

Tags: