Welcome to assistant-stream

assistant-stream is a small library for building typed assistant streaming responses in Python. It is designed to work well with:

• FastAPI / Starlette responses (via assistant_stream_ce.serialization.assistant_stream_response.AssistantStreamResponse)

• assistant-ui transports (via assistant_stream_ce.serialization.data_stream.DataStreamResponse or assistant_stream_ce.serialization.assistant_transport.AssistantTransportResponse)

At the core is assistant_stream_ce.create_run.create_run(), which turns an async callback into an async generator that yields structured stream chunks.

User Guide

• Getting started
  • Installation
  • Core concepts
  • A first stream
  • State updates
  • Tool calls
• FastAPI / Starlette integration
  • Pick a transport
  • Example: assistant-ui DataStreamResponse
  • Example: SSE JSON events (AssistantTransportResponse)
  • CORS and streaming
  • Error handling
• Streaming formats
  • Chunk stream (source of truth)
  • Encoding options
  • Custom encoders
• State management
  • Initialize state
  • Common operations
  • Batching
  • Caveats
• Tool calls
  • Streaming a tool call
  • Tool call IDs
  • Returning artifacts
  • Nested streams
• Sources (citations)
  • Emit a source
• LangGraph integration
  • Append LangGraph events
  • Tool subgraph state helper
  • Typical usage

API Reference

• API reference
  • assistant_stream_ce package
  • serialization
  • modules

Quick example

A minimal FastAPI endpoint streaming assistant output:

from fastapi import FastAPI
from assistant_stream_ce import create_run
from assistant_stream_ce.serialization.data_stream import DataStreamResponse

app = FastAPI()

@app.get("/chat")
async def chat():
    async def run(controller):
        controller.append_text("Hello ")
        controller.append_text("world!")

    return DataStreamResponse(create_run(run))

Next steps:

• Read Getting started for a tour of the features.

• See FastAPI / Starlette integration for practical server patterns and tips.