Implementation:Mistralai Client python EventStream

Knowledge Sources	Mistral Client Python
Domains	Streaming, Event_Processing
Last Updated	2026-02-15 14:00 GMT

Overview

Concrete tool for parsing and iterating over Server-Sent Events from streaming API responses provided by the mistralai SDK.

Description

EventStream and EventStreamAsync are context manager classes that wrap an httpx HTTP response and parse the SSE event stream. They implement __iter__ / __aiter__ to yield typed CompletionEvent objects. Each event contains a data field with a CompletionChunk Pydantic model. The classes handle SSE parsing (splitting by newlines, extracting data: fields), JSON deserialization, and proper HTTP connection lifecycle management.

Usage

These classes are returned by chat.stream() and chat.stream_async(). Use them within a with / async with context manager and iterate with for / async for to process chunks.

Code Reference

Source Location

Repository: client-python
File: src/mistralai/client/utils/eventstreaming.py
Lines: L19-48 (EventStream), L50-78 (EventStreamAsync)

Signature

class EventStream(Generic[T]):
    def __init__(self, response: httpx.Response, event_type: Type[T]):
        ...

    def __enter__(self) -> "EventStream[T]":
        return self

    def __exit__(self, *args) -> None:
        self.response.close()

    def __iter__(self) -> Iterator[T]:
        ...

class EventStreamAsync(Generic[T]):
    def __init__(self, response: httpx.Response, event_type: Type[T]):
        ...

    async def __aenter__(self) -> "EventStreamAsync[T]":
        return self

    async def __aexit__(self, *args) -> None:
        await self.response.aclose()

    async def __aiter__(self) -> AsyncIterator[T]:
        ...

Import

# Not typically imported directly; returned by client.chat.stream()
from mistralai.utils.eventstreaming import EventStream, EventStreamAsync

I/O Contract

Inputs

Name	Type	Required	Description
response	httpx.Response	Yes	Open HTTP response with SSE content
event_type	Type[T]	Yes	Pydantic model class for event deserialization

Outputs

Name	Type	Description
events	Iterator[CompletionEvent]	Yields typed SSE events with deserialized data
event.data	CompletionChunk	Contains choices[0].delta.content for text tokens

Usage Examples

Iterating a Sync Stream

# EventStream is returned by client.chat.stream()
with client.chat.stream(
    model="mistral-large-latest",
    messages=[UserMessage(content="Hello")],
) as stream:
    full_response = ""
    for chunk in stream:
        delta = chunk.data.choices[0].delta.content
        if delta:
            full_response += delta
            print(delta, end="")
    print(f"\n\nFull response: {full_response}")

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment