SHIPIT

Streaming

Real-time event streaming from shipit_agent.Agent — event types, terminal renderers, SSE and WebSocket transports.

5 min read
14 sections
Edit this page

agent.stream(prompt) is a generator that yields AgentEvents the instant they're emitted. There is no buffering: each tool_called event arrives before the tool runs, each tool_completed arrives the moment the tool returns, and run_completed is the very last event.

Event types

TypeEmitted whenUseful payload fields
run_startedThe agent receives a user promptprompt
step_startedThe runtime begins an LLM iterationiteration, tool_count
reasoning_startedThe model surfaces a thinking blockiteration
reasoning_completedThe thinking block is finalisedcontent
planning_startedAuto-planner is invoked—
planning_completedPlanner output is readyplan
tool_calledA tool is about to runtool_name, arguments
tool_completedA tool returned successfullytool_name, metadata, output
tool_failedA tool raisedtool_name, error
interactive_requestThe agent needs the human to answerquestion, options
mcp_attachedAn MCP server has been wired inserver_name
llm_retryThe LLM call is being retriedattempt, error
tool_retryA tool call is being retriedattempt, error
context_snapshotToken usage updateusage, compaction_ratio
rag_sourcesRAG sources captured during the runsources
final_answerThe answer is ready, just before the run closescontent, format
run_completedThe run is overoutput, iterations, usage

Tokens, tool arguments and groups

TypeEmitted whenUseful payload fields
text_deltaEach chunk of the answer, as it is generatedchunk
tool_input_startedA tool's arguments begin streamingcall_id, tool
tool_input_deltaA fragment of those argumentscall_id, delta
tool_output_startedA tool's output begins streamingcall_id, tool
tool_output_deltaA fragment of that output, as it is producedcall_id, delta
tool_group_startedOne iteration's tool calls begingroup_id, tool_count, tools
tool_group_completedThat iteration's calls are all donegroup_id, completed_count
tool_deniedThe permission gate refused a calltool, call_id, reason
tool_arguments_rejectedA call was refused for invalid arguments (e.g. asking for everything)tool, arguments, iteration
tool_call_healedA text-shaped tool call was promoted to a real onetools, healed_from
run_summaryAn end-of-run narration from the decision model, when enabledsummary, headline
usage_tickRunning token totalsusage

tool_called and tool_completed carry group_id too, so a UI can draw one expandable box per iteration.

Decisions, cards and things the run produced

TypeEmitted whenUseful payload fields
agent_decisionBefore a step, with narration onsummary, next_action, tools, iteration
agent_observationAfter tools return, same settingsummary, iteration
progress_summary_failedA summary call failed; the run continueserror, purpose
action_queuedA side-effecting call was held for approvalaction_id, tool, title, tag
connection_requestedThe agent needs something connectedconnection_id, title, reason, auth
artifact_createdA tool produced a filepath, title, kind, tool
sub_agent_eventA delegated child did somethingagent, task, inner_type, inner
lockdown_engagedSensitive data was read; actions are now deniedreason, tool
guardrail_triggeredA guardrail stopped the runreason
context_compactedOlder turns were condensedreason
run_cancelledThe run was cancelled—

See the Event Types reference for the complete schema.

Minimal example

python
for event in agent.stream("Search the web for SQLite news"):
    print(f"[{event.type}] {event.message}")

Files and images on the turn

stream() takes the same attachments as run() — hand it media directly instead of making the agent read it back with a tool. Text, markdown and code files inline on every provider; PDFs ride as native document blocks where the provider reads them; images need a vision-capable model.

python
for event in agent.stream(
    "What is different between these builds, and does it match the log?",
    images=["before.png", "after.png"],     # url, path, or base64
    files=["report.pdf", "server.py", "trace.log"],
):
    if event.type == "text_delta":
        print(event.chunk, end="", flush=True)

The attachments are part of the first user turn, so the model reads them before its first step — no read_file/read_document round-trip, and one fewer iteration on the clock.

Token by token

This is the branch people leave out, and then wonder why nothing streams:

python
for event in agent.stream("Summarise the inbox"):
    if event.type == "text_delta":
        print(event.payload["chunk"], end="", flush=True)   # ← the tokens
    elif event.type == "tool_called":
        print(f"\n[{event.payload['tool']}] ", end="", flush=True)

A loop that only handles tool_called / tool_completed sits silent through the entire answer. The tokens arrive in text_delta and nowhere else.

Coloured terminal renderer

python
RESET = "\033[0m"
DIM   = "\033[2m"
BOLD  = "\033[1m"
CYAN  = "\033[36m"
GREEN = "\033[32m"
YELL  = "\033[33m"

for event in agent.stream("Find today's BTC price"):
    if event.type == "run_started":
        print(BOLD + "🚀 run started" + RESET)
    elif event.type == "step_started":
        print(DIM + f"  · iter {event.payload.get('iteration')}" + RESET)
    elif event.type == "reasoning_started":
        print(YELL + "  🧠 thinking…" + RESET)
    elif event.type == "reasoning_completed":
        print(YELL + "  🧠 " + event.payload.get('content', '')[:80] + RESET)
    elif event.type == "tool_called":
        print(CYAN + "  ▶ " + event.message + RESET)
    elif event.type == "tool_completed":
        print(GREEN + "  ✓ " + event.message + RESET)
    elif event.type == "rag_sources":
        for s in event.payload.get("sources", []):
            print(DIM + f"    📎 [{s['index']}] {s['source']}" + RESET)
    elif event.type == "run_completed":
        print(BOLD + "✅ done" + RESET)
        print((event.payload.get('output') or '')[:300])

examples/02_streaming_with_reasoning.py ships a more polished version of this you can copy verbatim.

Server-Sent Events (SSE)

For web UIs, every event has a built-in SSE encoder (shipit_agent.packets.sse_event_packet):

python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

from shipit_agent import Agent
from shipit_agent.packets import sse_event_packet, sse_result_packet

app = FastAPI()
agent = Agent.with_builtins(llm=llm)

@app.get("/stream")
async def stream(q: str):
    def gen():
        for event in agent.stream(q):
            yield sse_event_packet(event)
        # Final marker — useful for clients that watch for `event: done`
        yield "event: done\ndata: {}\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

The browser side reads it with EventSource("/stream?q=…") and renders events as they arrive.

WebSocket

python
from shipit_agent.packets import websocket_event_packet

@app.websocket("/ws")
async def ws(websocket):
    await websocket.accept()
    user_msg = await websocket.receive_text()
    for event in agent.stream(user_msg):
        await websocket.send_json(websocket_event_packet(event))

websocket_event_packet returns a JSON-friendly dict; pair it with send_json for a clean transport.

Streaming inside a chat session

AgentChatSession.stream mirrors Agent.stream but also persists each turn to the session store:

python
session = agent.chat_session(session_id="user-42")

for event in session.stream("Hi, what can you do?"):
    print(event.message)

# Next turn — same session, same history
for event in session.stream("Search the web for SQLite news"):
    print(event.message)

Subscribe to events programmatically with session.add_event_callback or session.add_packet_callback if you want a callback API instead of a generator.

Stopping a stream

A for event in agent.stream(...): loop can be exited with break — the runtime cleans up the background thread automatically. For explicit cancellation from another thread, raise StopIteration or close the generator (stream.close()).

One loop, everything in it

Every feature reports through this one generator. Nothing reaches the renderers that does not reach you:

python
for event in agent.stream(prompt):
    p = event.payload
    match event.type:
        case "text_delta":           ui.append_token(p["chunk"])
        case "agent_decision":       ui.add_step(p["summary"])
        case "agent_observation":    ui.add_result(p["summary"])
        case "tool_group_started":   ui.open_group(p["group_id"], p["tool_count"])
        case "tool_called":          ui.add_call(p["tool"], p["arguments"], p.get("group_id"))
        case "tool_completed":       ui.settle_call(p["call_id"], p["duration_ms"])
        case "tool_denied":          ui.blocked(p["tool"], p["reason"])
        case "tool_group_completed": ui.close_group(p["group_id"])
        case "artifact_created":     ui.add_card(p["title"], p["kind"], p["path"])
        case "action_queued":        ui.approval(p["title"], p["tag"])
        case "connection_requested": ui.connect_card(p["title"], p["reason"])
        case "final_answer":         ui.answer(p["content"])

If you would rather not switch on raw events, stream_timeline gives the same run pre-shaped for a UI, and watch renders it for you.

Durability — what a reconnecting client keeps

agent.stream_sse() frames every event for text/event-stream and labels it:

python
@app.get("/run")
def run(prompt: str):
    return StreamingResponse(agent.stream_sse(prompt),
                             media_type="text/event-stream")
text
id: 0
event: stream_hello
data: {"type": "stream_hello", "durability": "control",
       "generation": "62298-1786013279396", "sequence": 0,
       "payload": {"generation": "62298-1786013279396",
                   "discard_provisional": true}}

id: 1
event: run_started
data: {"type": "run_started", "durability": "canonical", "sequence": 1, …}
DurabilityMeaningOn reconnect
canonicalSettled fact — a tool ran, an answer landedReplayed
provisionalIn flight — a token, a half-written argumentDiscarded and re-streamed
controlStream plumbingActed on, not drawn

The opening stream_hello carries a per-process generation, so a client can tell a network blip (same generation — keep what you drew) from a server restart (new generation — start again). The stream closes with event: done.

See also