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)

StreamEvent

At the model layer, streaming yields Bedrock-style chunks typed as StreamEvent. Use this public type when calling Model.stream directly or type-hinting low-level stream output:

from elsai.types.streaming import StreamEvent
from elsai_model.bedrock import BedrockModel

model = BedrockModel(...)

async for event in model.stream(messages=[...]):
    # event is a StreamEvent TypedDict
    if "contentBlockDelta" in event:
        delta = event["contentBlockDelta"]["delta"]
        if "text" in delta:
            print(delta["text"], end="", flush=True)

Do not import from elsai.types._events — those are internal framework wrappers, not part of the public API.

Agent.stream_async wraps model streaming into the simpler event dicts above (data, complete, etc.) for application use.

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