Streaming formats

assistant-stream-ce separates what you produce (typed chunks) from how you encode those chunks for HTTP streaming.

Chunk stream (source of truth)

Your application produces an async generator of assistant_stream_ce.assistant_stream_chunk.AssistantStreamChunk.

Encoding options

Data stream (assistant-ui)

assistant_stream_ce.serialization.data_stream.DataStreamEncoder converts chunks to a newline-delimited protocol used by assistant-ui.

Response class: assistant_stream_ce.serialization.data_stream.DataStreamResponse

SSE JSON transport

assistant_stream_ce.serialization.assistant_transport.AssistantTransportEncoder converts each chunk to a JSON object and emits it as a :sse:Server-Sent Event payload.

Response class: assistant_stream_ce.serialization.assistant_transport.AssistantTransportResponse

OpenAI-style SSE

assistant_stream_ce.serialization.openai_stream.OpenAIStreamEncoder emits OpenAI-like chat.completion.chunk events. At the moment it only maps text-delta chunks to delta.content.

Response class: assistant_stream_ce.serialization.openai_stream.OpenAIStreamResponse

Custom encoders

To implement your own encoder, subclass assistant_stream_ce.serialization.stream_encoder.StreamEncoder:

from assistant_stream_ce.serialization.stream_encoder import StreamEncoder

class MyEncoder(StreamEncoder):
    def get_media_type(self) -> str:
        return "text/plain"

    async def encode_stream(self, stream):
        async for chunk in stream:
            yield f"{chunk.type}\n"

Then use assistant_stream_ce.serialization.assistant_stream_response.AssistantStreamResponse:

from assistant_stream_ce.serialization.assistant_stream_response import AssistantStreamResponse

return AssistantStreamResponse(create_run(run), MyEncoder())