Consuming Streams with the Python SDK
Use the OpenAI async client with async for to consume streaming completions, accumulate the full response, and handle mid-stream errors without losing partial output.
Consuming Streams with the Python SDK is a free AI Engineering Academy lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Engineering Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
Sync vs Async Streaming Clients
The OpenAI Python SDK provides both a synchronous OpenAI client and an asynchronous AsyncOpenAI client. For command-line scripts and simple applications, the synchronous client is easier to use. For web servers, APIs, and applications that handle multiple concurrent requests, the async client is essential — it does not block the event loop while waiting for tokens, allowing other requests to be served concurrently.
# Synchronous client (simple scripts)
from openai import OpenAI
client = OpenAI()
# Asynchronous client (web servers, concurrent workloads)
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
# The async client has the same API surface as the sync client
# but all methods are coroutines that must be awaitedAsync Streaming with AsyncOpenAI
With the AsyncOpenAI client, the streaming call becomes a coroutine. You use async for to iterate over chunks instead of a regular for loop. The event loop can schedule other coroutines between each chunk arrival, enabling your server to handle other requests while waiting for the next token from the LLM — this is the key advantage over synchronous streaming in a web context.
import asyncio
from openai import AsyncOpenAI
async_client = AsyncOpenAI()
async def async_stream_completion(prompt: str) -> str:
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
full_text = ''
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end='', flush=True)
full_text += delta
print()
return full_text
# Run the coroutine
asyncio.run(async_stream_completion('Explain what async/await does in Python'))Using the Stream Context Manager
The OpenAI SDK also provides a stream context manager via client.chat.completions.stream(). This approach automatically closes the stream when the context exits and provides convenience methods like stream.text_stream that yield only non-None text deltas and stream.get_final_completion() for post-stream usage statistics without manually accumulating them.
from openai import AsyncOpenAI
import asyncio
async def stream_with_context_manager(prompt: str):
async with async_client.chat.completions.stream(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
) as stream:
# text_stream filters None deltas automatically
async for text in stream.text_stream:
print(text, end='', flush=True)
# Access final completion after stream ends
completion = await stream.get_final_completion()
print(f'\nUsage: {completion.usage}')
return completion
asyncio.run(stream_with_context_manager('What are the benefits of async I/O?'))Handling Mid-Stream Errors Gracefully
Errors can occur at any point during a stream: during the initial connection, after the first token, or near the end of a long response. Wrap your stream iteration in try/except blocks and handle openai.APIConnectionError, openai.RateLimitError, and openai.APIStatusError separately, as each requires a different recovery strategy (retry, backoff, or user notification).
import openai
async def resilient_stream(prompt: str):
try:
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
accumulated = ''
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
accumulated += delta
yield delta # async generator
except openai.RateLimitError:
yield '[Rate limit reached — please wait and retry]'
except openai.APIConnectionError:
yield '[Connection error — check your network]'
except openai.APIStatusError as e:
yield f'[API error {e.status_code}]'
except Exception as e:
yield f'[Unexpected error: {type(e).__name__}]'Async Generator for Streaming
The cleanest async pattern for streaming is an async generator function that yields tokens. Consumers iterate over it with async for. This keeps the streaming logic separate from how the output is used — a FastAPI endpoint, a WebSocket handler, and a test all consume the same generator without knowing about each other.
from typing import AsyncGenerator
async def token_stream(
messages: list[dict],
model: str = 'gpt-4o-mini',
) -> AsyncGenerator[str, None]:
stream = await async_client.chat.completions.create(
model=model,
messages=messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
# Consumer 1: print to terminal
async def print_stream(messages):
async for token in token_stream(messages):
print(token, end='', flush=True)
# Consumer 2: collect to string
async def collect_stream(messages) -> str:
return ''.join([t async for t in token_stream(messages)])Concurrent Streaming Requests
A major benefit of async streaming is the ability to run multiple streams concurrently within a single process. Using asyncio.gather, you can start several LLM streaming requests simultaneously and process their tokens as they arrive. This is useful for fan-out patterns where you want to compare multiple prompt variations or run parallel sub-tasks.
import asyncio
async def run_parallel_streams(queries: list[str]) -> list[str]:
async def collect(query):
messages = [{'role': 'user', 'content': query}]
return ''.join([t async for t in token_stream(messages)])
results = await asyncio.gather(*[collect(q) for q in queries])
return results
queries = [
'What is RAG?',
'What is a vector database?',
'What is BM25?',
]
async def main():
answers = await run_parallel_streams(queries)
for q, a in zip(queries, answers):
print(f'Q: {q}\nA: {a[:100]}\n')
asyncio.run(main())Timeout and Cancellation
Long-running streams should have timeouts to prevent indefinite blocking. Use asyncio.wait_for to apply a coroutine-level timeout or httpx.Timeout to set connection and read timeouts at the HTTP client level. Both approaches ensure that a stalled stream does not hold a request indefinitely. Always cancel streams explicitly when the user disconnects to avoid wasting GPU compute.
import asyncio
async def stream_with_timeout(messages, timeout_seconds: float = 30.0):
try:
async with asyncio.timeout(timeout_seconds):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=messages,
stream=True,
timeout=timeout_seconds, # HTTP-level timeout
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
except asyncio.TimeoutError:
yield '\n[Stream timed out after {:.0f}s]'.format(timeout_seconds)Buffering Partial Lines
When streaming to a client that processes complete lines (like a CLI that renders markdown), you may want to buffer tokens until a newline or sentence boundary before forwarding them. This avoids flickering renders of partial sentences. Accumulate tokens in a buffer, flush the buffer to the consumer when you detect a sentence-ending punctuation or a newline character, and always flush the remaining buffer at the end of the stream.
async def buffered_line_stream(messages):
buffer = ''
flush_on = {'.', '!', '?', '\n'}
async for token in token_stream(messages):
buffer += token
if any(c in buffer for c in flush_on):
# Find the last sentence-ending position
for i, c in enumerate(reversed(buffer)):
if c in flush_on:
split_pos = len(buffer) - i
yield buffer[:split_pos]
buffer = buffer[split_pos:]
break
if buffer: # flush remainder
yield bufferRecording Stream Latency in Production
In production, instrument every stream to record TTFT and total generation time for monitoring. Store these metrics in a time-series database and alert when TTFT exceeds your SLA threshold (typically 1-2 seconds for interactive applications). Correlate TTFT spikes with prompt length, model load, and time of day to identify root causes of latency degradation.
import time
from dataclasses import dataclass
@dataclass
class StreamMetrics:
prompt_chars: int
ttft_ms: float
total_ms: float
token_count: int
async def instrumented_stream(messages) -> tuple[str, StreamMetrics]:
t_start = time.perf_counter()
t_first = None
token_count = 0
full_text = ''
stream = await async_client.chat.completions.create(
model='gpt-4o-mini', messages=messages, stream=True
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
if t_first is None:
t_first = time.perf_counter()
token_count += 1
full_text += delta
t_end = time.perf_counter()
prompt_len = sum(len(m.get('content', '')) for m in messages)
metrics = StreamMetrics(
prompt_chars=prompt_len,
ttft_ms=(t_first - t_start) * 1000 if t_first else 0,
total_ms=(t_end - t_start) * 1000,
token_count=token_count,
)
return full_text, metricsTesting Async Streaming Code
Testing async streaming requires special care. Use pytest-asyncio to run async test functions, and mock the OpenAI client to avoid real API calls in unit tests. Create a fake stream that yields predefined chunks with configurable delays to test both happy-path token processing and error-handling paths without spending API budget.
# pip install pytest pytest-asyncio
import pytest
from unittest.mock import AsyncMock, MagicMock
async def fake_stream(tokens: list[str]):
for token in tokens:
chunk = MagicMock()
chunk.choices[0].delta.content = token
yield chunk
@pytest.mark.asyncio
async def test_stream_accumulates_correctly(monkeypatch):
mock_create = AsyncMock(return_value=fake_stream(['Hello', ', ', 'world', '!']))
monkeypatch.setattr(async_client.chat.completions, 'create', mock_create)
result = await collect_stream([{'role': 'user', 'content': 'Hi'}])
assert result == 'Hello, world!'SDK Helpers: stream.text and stream.final_message
The OpenAI Python SDK's stream context manager provides helper attributes that avoid manual accumulation. stream.text_stream is an async iterable that yields only non-None content strings. After the stream completes, await stream.get_final_message() returns a full ChatCompletionMessage with the complete text and usage data. These helpers reduce boilerplate and handle edge cases like empty deltas automatically.
async def clean_streaming_example(prompt: str):
async with async_client.chat.completions.stream(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
) as stream:
# Iterate only over text tokens, None deltas filtered automatically
async for text in stream.text_stream:
print(text, end='', flush=True)
# After context exit, get accumulated result
final = await stream.get_final_completion()
return final.choices[0].message.contentQuick Check
Test your understanding of async streaming with the OpenAI Python SDK from this lesson.
Lesson Recap
In this lesson you learned: AsyncOpenAI enables non-blocking streaming that lets servers handle concurrent requests, async generators are the cleanest pattern for yielding streaming tokens to downstream consumers, and asyncio.wait_for and timeout parameters prevent indefinitely stalled streams from blocking your server. The stream context manager provides convenience helpers like text_stream and get_final_completion. Next up we expose LLM streaming to browser clients via FastAPI and Server-Sent Events.
Frequently asked questions
Is the “Consuming Streams with the Python SDK” lesson free?
Yes — the full text of “Consuming Streams with the Python SDK” is free to read here on the web, and the AI Engineering Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Engineering Academy course, upgrade to CoddyKit PRO.
What will I learn in “Consuming Streams with the Python SDK”?
Use the OpenAI async client with async for to consume streaming completions, accumulate the full response, and handle mid-stream errors without losing partial output. You practise AI Engineering Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Engineering Academy?
No prior experience is required. AI Engineering Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Consuming Streams with the Python SDK” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Engineering Academy lesson?
Yes. Every AI Engineering Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Understanding Token Streaming
- Consuming Streams with the Python SDK
- Streaming in FastAPI with Server-Sent Events
- Handling Tool Calls in Streamed Responses