Implement streaming in Python
On this page
From prototype to production
The previous lesson covered the concepts of SSE streaming. The harder part remains: turning a demo loop into a client that holds up under load, survives network errors, and feeds a web interface. This lesson follows the real trajectory of a project—skeleton first, then error handling, then async, then HTTP exposure, and finally measurement.
The basic pattern
Everything starts with this ten-line skeleton, which you will adapt for every project. It does two things at once: displays tokens as they arrive and accumulates them for the function return.
from mistralai.client import Mistral
import os
client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))
def stream_chat(messages: list, model: str = "mistral-large-latest") -> str:
"""Sends a streaming request and displays tokens in real time."""
stream = client.chat.stream(
model=model,
messages=messages
)
full_text = ""
for chunk in stream:
content = chunk.data.choices[0].delta.content
if content:
full_text += content
print(content, end="", flush=True)
print() # New line
return full_textThis version works for a local script. It has a fatal flaw in production: it prints to the terminal, so it can do nothing but terminal work.
Comprehensive error handling
Network errors, timeouts, and rate limits are not hypotheticals: they are certainties at real traffic scale. The client below addresses both problems. It automatically retries 429 errors with exponential backoff, and it no longer prints anything: it emits typed events via yield, leaving it to the caller to decide what to do with them—display them, push them to a WebSocket, or write them to a database.
from mistralai.client import Mistral
from mistralai.client.errors import SDKError
import os
import time
client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))
def stream_with_retry(
messages: list,
model: str = "mistral-large-latest",
max_retries: int = 3,
max_tokens: int = 2000
) -> dict:
"""Streaming with automatic retry and metrics."""
for attempt in range(max_retries):
try:
stream = client.chat.stream(
model=model,
messages=messages,
max_tokens=max_tokens
)
full_text = ""
finish_reason = None
chunk_count = 0
for chunk in stream:
delta = chunk.data.choices[0].delta
if delta.content:
full_text += delta.content
chunk_count += 1
yield {"type": "token", "content": delta.content}
if chunk.data.choices[0].finish_reason:
finish_reason = chunk.data.choices[0].finish_reason
yield {
"type": "done",
"full_text": full_text,
"finish_reason": finish_reason,
"chunks": chunk_count
}
return
except SDKError as e:
if e.status_code == 429 and attempt < max_retries - 1:
wait_time = 2 ** attempt # Exponential backoff
yield {"type": "retry", "wait": wait_time, "attempt": attempt + 1}
time.sleep(wait_time)
else:
yield {"type": "error", "message": str(e), "status": e.status_code}
return
except Exception as e:
yield {"type": "error", "message": str(e), "status": None}
returnThe exponential backoff 2 ** attempt waits 1 second, then 2, then 4. This progression is not decorative: if all your clients retried at the same fixed rhythm, they would hit the API in synchronized bursts and prolong the very saturation they are suffering. Note also that only error 429 triggers a retry—retrying a 401 due to an invalid key would only delay diagnosis.
On the caller side, consumption becomes a simple event read:
messages = [
{"role": "system", "content": "You are a technical assistant."},
{"role": "user", "content": "Explain design patterns in Python."}
]
for event in stream_with_retry(messages):
if event["type"] == "token":
print(event["content"], end="", flush=True)
elif event["type"] == "retry":
print(f"\n[Retry {event['attempt']}, waiting {event['wait']}s...]")
elif event["type"] == "done":
print(f"\n\n--- Done ({event['chunks']} chunks, {event['finish_reason']})")
elif event["type"] == "error":
print(f"\n[Error: {event['message']}]")Async streaming with asyncio
The synchronous client blocks its thread for the entire generation. On a web server that must handle fifty simultaneous conversations, this means immobilizing fifty threads just to wait. The async client lifts this constraint:
from mistralai.client import Mistral
import asyncio
import os
client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))
async def async_stream_chat(messages: list) -> str:
"""Async streaming for concurrent applications."""
stream = await client.chat.stream_async(
model="mistral-large-latest",
messages=messages
)
full_text = ""
async for chunk in stream:
content = chunk.data.choices[0].delta.content
if content:
full_text += content
print(content, end="", flush=True)
print()
return full_text
# Execution
async def main():
messages = [{"role": "user", "content": "Hello, introduce yourself."}]
result = await async_stream_chat(messages)
print(f"\nTotal: {len(result)} characters")
asyncio.run(main())The logic is identical to the synchronous version; only stream_async, the await on the call, and the async for on the loop change.
Integration with FastAPI
Next, you need to get this stream out of your Python process and into the browser. FastAPI handles this with StreamingResponse, provided you reformat each fragment to the SSE format seen in the previous lesson:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from mistralai.client import Mistral
import os
import json
app = FastAPI()
client = Mistral(api_key=os.getenv("MISTRAL_API_KEY"))
async def generate_stream(prompt: str):
"""SSE generator for FastAPI."""
stream = await client.chat.stream_async(
model="mistral-large-latest",
messages=[{"role": "user", "content": prompt}]
)
async for chunk in stream:
content = chunk.data.choices[0].delta.content
if content:
data = json.dumps({"content": content})
yield f"data: {data}\n\n"
yield "data: [DONE]\n\n"
@app.get("/chat/stream")
async def chat_stream(prompt: str):
return StreamingResponse(
generate_stream(prompt),
media_type="text/event-stream"
)Two details determine whether this endpoint works correctly. The double newline \n\n that ends each yield is what separates events in the SSE protocol: without it, the browser waits indefinitely for an event it considers unfinished. And media_type="text/event-stream" is what allows the JavaScript client to consume the response with an EventSource rather than a standard fetch.
Measuring performance
Once the stream is in place, you will want to know how it performs. Two metrics are enough to drive optimization:
import time
start = time.perf_counter()
first_token_time = None
token_count = 0
stream = client.chat.stream(
model="mistral-large-latest",
messages=[{"role": "user", "content": "Write a paragraph about Paris."}]
)
for chunk in stream:
if chunk.data.choices[0].delta.content:
if first_token_time is None:
first_token_time = time.perf_counter() - start
token_count += 1
total_time = time.perf_counter() - start
print(f"Time to First Token (TTFT): {first_token_time:.3f}s")
print(f"Tokens generated: {token_count}")
print(f"Total time: {total_time:.3f}s")
print(f"Tokens/second: {token_count / total_time:.1f}")TTFT (Time to First Token)—the time between sending the request and displaying the first character—is the most important metric for user experience. It takes precedence over total time: a fifteen-second generation whose first word appears in three-tenths of a second will feel fast, while a five-second generation delivered in one block will feel slow. Measure it on your real prompts, system message included—a two-page system message lengthens TTFT before the first token is even generated.
Key takeaways
- Use a generator pattern (
yield) to propagate streaming tokens - Implement exponential backoff for 429 errors (rate limit)
- Prefer
stream_asyncfor concurrent web applications - Measure TTFT and tokens/second to optimize performance
- With FastAPI,
StreamingResponse+text/event-streampropagates SSE to the browser