Streaming Responses in Python

The problem streaming solves

With chat.complete(), you wait for the model to generate its entire response before receiving a single character. For a short answer, this delay goes unnoticed. But for a detailed analysis, a long code snippet, or multi-paragraph reasoning, the user stares at a blank screen for several seconds. In a chat interface, even a few seconds of silence can make the application appear frozen.

Streaming changes this perception: the response arrives incrementally, chunk by chunk, and displays progressively just like the chat interfaces you know. The total generation time doesn’t drop by a millisecond, but the first word appears almost immediately. That initial delay is what the user actually feels.

Using chat.stream()

The Mistral SDK V2 exposes the chat.stream() method, which returns an iterator of chunks instead of a single object:

import os
from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {
            "role": "user",
            "content": "Describe the steps to deploy a Flask application on a Linux server."
        }
    ]
)

for chunk in stream:
    content = chunk.data.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

# Final newline
print()

Two details make all the difference in this loop. The if content check is essential because some chunks arrive without text and would otherwise print None. And flush=True forces Python to write immediately to the output rather than waiting for its buffer to fill. Without it, all the text would appear in one block at the end, completely negating the benefit you’re after.

What a chunk contains

Each chunk carries a delta, meaning a fragment of the response under construction. By inspecting it, you access both the text and the end-of-stream metadata:

for chunk in stream:
    delta = chunk.data.choices[0].delta

    # Text content (can be None for some chunks)
    text = delta.content

    # Stop reason (None except for the last chunk)
    end = chunk.data.choices[0].finish_reason

    if text:
        print(text, end="", flush=True)

    if end:
        print(f"\n[Stream end: {end}]")

The flow always follows the same choreography: the first chunk usually carries the assistant role, intermediate chunks deliver the text in small pieces, and the last one provides the finish_reason. You determine whether the response ended normally or was cut off by max_tokens by checking this final chunk.

Displaying and saving simultaneously

Streaming has a downside: the text scrolls by, but you don’t hold onto it anywhere. Yet a real-world application often needs to archive the response in a database for conversation history. The solution is to accumulate fragments while displaying them:

import os
from mistralai.client import Mistral

client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

stream = client.chat.stream(
    model="mistral-small-latest",
    messages=[
        {"role": "user", "content": "List five Python web frameworks."}
    ]
)

full_response = []

for chunk in stream:
    content = chunk.data.choices[0].delta.content
    if content:
        full_response.append(content)
        print(content, end="", flush=True)

print()

# The reconstituted full response
final_text = "".join(full_response)
print(f"\nTotal length: {len(final_text)} characters")

Note the use of a list followed by a final "".join() rather than string concatenation in every loop iteration. For a long response, the latter approach copies all accumulated text for every fragment received, which is inefficient.

The asynchronous variant

In a web server—FastAPI, aiohttp—the synchronous version blocks the thread during the entire generation, freezing other ongoing requests. The SDK therefore provides chat.stream_async():

import asyncio
import os
from mistralai.client import Mistral

async def generate_response():
    client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])

    stream = await client.chat.stream_async(
        model="mistral-small-latest",
        messages=[
            {"role": "user", "content": "Explain the concept of a coroutine in Python."}
        ]
    )

    async for chunk in stream:
        content = chunk.data.choices[0].delta.content
        if content:
            print(content, end="", flush=True)

    print()

asyncio.run(generate_response())

The structure is identical to the synchronous version, aside from the await and async for keywords. As soon as your code runs in a web server, this is the variant you must use.

When to stream and when to skip it

In a chat interface, streaming is always mandatory: reading comfort is at stake. On the backend API side, it’s worth enabling as long as your frontend can consume Server-Sent Events. For any response longer than a few sentences, the gain in perceived latency justifies the added complexity. Conversely, in a batch script processing a thousand documents and writing results to a file, no one is watching the screen: chat.complete() remains simpler and perfectly adequate.

Key takeaways

• chat.stream() returns an iterator of chunks for progressive display
• Content lives in chunk.data.choices[0].delta.content
• Use flush=True in your print() calls for real-time display
• Reconstruct the full response by accumulating chunks if needed
• The asynchronous version chat.stream_async() is recommended for web applications