Streaming

Elsai supports token-level streaming so your application can display responses as they're generated rather than waiting for the full response.

Async streaming

Use stream_async to get an async iterator of events:

import asyncio
from elsai import Agent

agent = Agent()

async def main():
    async for event in agent.stream_async("Write a haiku about the ocean"):
        if "data" in event:
            print(event["data"], end="", flush=True)
    print()  # newline after stream ends

asyncio.run(main())

Event types

Key in event dict	Description
data	A chunk of text being generated
complete	True when the text generation is complete
current_tool_use	Dict with name and input while a tool is being called
result	Final AgentResult (last event)

Full streaming example

import asyncio
from elsai import Agent, tool

@tool
def get_fact(topic: str) -> str:
    """Get a fact about a topic.

    Args:
        topic: What to get a fact about.
    """
    return f"Interesting fact about {topic}: It's fascinating!"

agent = Agent(tools=[get_fact])

async def stream_with_tools():
    async for event in agent.stream_async("Tell me a fact about astronomy"):
        if "data" in event:
            # Text chunk
            print(event["data"], end="", flush=True)
        elif "current_tool_use" in event:
            # Tool call in progress
            tool_info = event["current_tool_use"]
            print(f"\n[Calling tool: {tool_info.get('name')}]")
        elif "result" in event:
            # Final result
            result = event["result"]
            print(f"\n\nDone. Stop reason: {result.stop_reason}")

asyncio.run(stream_with_tools())

Streaming with FastAPI

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from elsai import Agent

app = FastAPI()
agent = Agent()

@app.get("/chat")
async def chat(prompt: str):
    async def generate():
        async for event in agent.stream_async(prompt):
            if "data" in event:
                yield f"data: {event['data']}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Disabling the callback handler

By default, agents print events to stdout via PrintingCallbackHandler. When building APIs or streaming manually, disable it:

agent = Agent(callback_handler=None)  # Silent — handle events yourself

Bidirectional streaming (experimental)

Real-time voice agents use bidirectional audio streaming, shown below.

import asyncio
from elsai.experimental.bidi import BidiAgent
from elsai.experimental.bidi.models import BidiNovaSonicModel
from elsai.experimental.bidi.io import BidiAudioIO, BidiTextIO

async def main():
    model = BidiNovaSonicModel()
    agent = BidiAgent(model=model)

    audio_io = BidiAudioIO()
    text_io = BidiTextIO()

    await agent.run(
        inputs=[audio_io.input()],
        outputs=[audio_io.output(), text_io.output()]
    )

asyncio.run(main())