Streaming UI integration

You’ll stream an agent’s events (text, tool calls, results) to a web UI as they happen, using FastAPI and Server-Sent Events. Use this pattern when users should watch the agent work instead of waiting for the final answer, and you want to put the endpoint inside your own FastAPI app.

agent.stream() yields events through a bounded buffer (128 events), so a slow client applies backpressure: once the buffer is full, the agent waits for your handler to catch up instead of piling up events in memory.

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
from loomflow import Agent
 
app = FastAPI()
agent = Agent("...", model="claude-opus-4-7")
 
@app.get("/chat")
async def chat(prompt: str, user_id: str):
    async def event_source():
        async for event in agent.stream(prompt, user_id=user_id):
            yield {
                "event": event.kind.value,
                "data": event.model_dump_json(),
            }
    return EventSourceResponse(event_source())

When the client disconnects, the server stops iterating, and breaking out of the iteration cancels the run cleanly; a tool call that is mid-flight is cancelled too, so abandoned tabs don’t keep spending tokens. In a real app, take user_id from your auth layer rather than the query string.

If you don’t need a custom route, create_app(agent) already serves a POST /stream SSE endpoint with the same cancellation behaviour. See Serving over HTTP.