Skip to content

Streaming Events

agent.stream() yields AgentEvent objects as the runtime executes. Under the hood, the agent runs on a background thread and pushes events through a thread-safe queue, so every event reaches your loop the instant it's emitted — no buffering, no batched delivery.

Basic usage

from shipit_agent import Agent
from examples.run_multi_tool_agent import build_llm_from_env

agent = Agent.with_builtins(llm=build_llm_from_env('openai'))

for event in agent.stream("Find today's Bitcoin price in USD."):
    print(f"{event.type:22s} {event.message}")

Event reference

Event type When it fires Key payload fields
run_started Very first event of a run, once per stream()/run() call. prompt
mcp_attached Once per attached MCP server, right after run_started. server
skills_selected Skills activated for this run and tools injected by them. Emitted once without an LLM call. skills, skill_ids, injected_tools, count
planning_started Router policy decided the prompt is complex enough to invoke plan_task. Fires before the first LLM call. prompt
planning_completed Planner returned. Output is injected into history as a user-role context message (Bedrock tool-pairing safe). output
step_started Each iteration of the tool loop, right before calling the LLM. iteration, tool_count
reasoning_started 🧠 LLM response contained a thinking/reasoning block. iteration
reasoning_completed Immediately after reasoning_started, carrying the full reasoning text. iteration, content
agent_decision Concise public intent in the model's own response text. Silent models do not get fabricated decisions. summary, next_action, tools, generated_by_model
tool_called Model decided to call a tool. Fires before execution. tool, call_id, iteration, arguments
tool_output_started A tool output stream opened. It can repeat for retries. tool, call_id, attempt, buffered
tool_output_delta One incremental, provisional output chunk. tool, call_id, chunk, chunk_metadata, sequence, attempt
tool_completed Tool finished; this is the canonical complete result plus model-context telemetry. tool, call_id, iteration, output, output_chars, model_output_chars, model_output_reduced, metadata
agent_observation Concise factual summary composed from the tool contract, arguments, and tool-provided metadata. It is never represented as model reasoning. summary, next_action, generated_by_model
tool_retry Transient tool failure, retry scheduled by RetryPolicy. iteration, attempt, error
tool_failed Non-retryable tool error, or model hallucinated an unregistered tool name (synthetic error result still appended for pairing balance). iteration, error
llm_retry Transient LLM provider error, retry scheduled. attempt, error
interactive_request A tool returned metadata.interactive=True (e.g. ask_user, human review). UI can pause and collect input. kind, payload
run_completed Final event. Fires once the loop exits or hits the iteration cap. output, content, format

Event structure

@dataclass
class AgentEvent:
    type: str                # e.g. "tool_called"
    message: str             # concise human-readable event text
    payload: dict[str, Any]  # event-specific fields

Serialize with event.to_dict() for WebSocket/SSE transport. print(event) and notebook list displays preserve the complete diagnostic representation, including payload and timestamp. Use format_event_line(event) for a compact human-facing stream, or event.to_dict() for serialization.

Typical event trace

A Bedrock gpt-oss-120b run with two tool calls:

1.  run_started          Agent run started
2.  step_started         iteration=1, tool_count=28
3.  reasoning_started    🧠 iteration=1
4.  reasoning_completed  🧠 "The user wants two BTC price sources. I'll start with web_search..."
5.  tool_called          Tool called: web_search
6.  tool_output_started  Tool output started: web_search
7.  tool_output_delta    chunk="..."
8.  tool_completed       Tool completed: web_search
9.  step_started         iteration=2
10. reasoning_completed  🧠 "Now I'll open both URLs to confirm..."
11. tool_called          Tool called: open_url
12. tool_output_delta    chunk="..."
13. tool_completed       Tool completed: open_url
14. step_started         iteration=3
15. run_completed        "**Bitcoin Price — 2026-04-09** ..."

Streaming custom tool output

Existing tools need no changes: they emit one output delta when their result arrives. A custom generator tool can publish output incrementally while still producing one complete canonical ToolResult:

from shipit_agent import FunctionTool, ToolOutputChunk

def follow_build():
    yield ToolOutputChunk("compiling...\n", {"phase": "compile"})
    yield ToolOutputChunk("testing...\n", {"phase": "test"})
    yield "passed\n"

agent.add_tool(FunctionTool.from_callable(follow_build))

Chunks are concatenated without separators, and chunk metadata is merged in arrival order. Use call_id to keep concurrent tool streams separate. If guardrails are enabled, the runtime deliberately buffers the complete result and emits one sanitized delta so raw secrets or indirect prompt injection cannot reach a live client.

Live UI updates in Jupyter

from IPython.display import Markdown, clear_output, display

lines = []
for event in agent.stream(prompt):
    lines.append(f"{event.type} — {event.message}")
    clear_output(wait=True)
    display(Markdown("## Live Stream\n\n" + "\n".join(lines)))

Uses clear_output(wait=True) + display(...) for reliable incremental rendering in Jupyter, VS Code, and JupyterLab.

WebSocket/SSE packet transports

session = agent.chat_session(session_id='demo')

for packet in session.stream_packets("Research Bitcoin", transport='websocket'):
    print(packet)   # serialized AgentEvent dict

for packet in session.stream_packets("Research Bitcoin", transport='sse'):
    print(packet)   # SSE-formatted string

Both transports yield packets incrementally — no buffering.

tool_output_delta is provisional and should be discarded on reconnect. tool_completed is canonical and contains the complete output, so the UI can replace its provisional buffer with the durable result.

agent.run_live() and StreamRenderer display tool-output chunks by default. Pass StreamRenderer(show_tool_output=False) when a terminal should show only the bounded completion preview while other consumers still receive all events.

Error handling

Errors raised during the run (LLM provider exceptions, non-retryable tool failures) are captured on the background worker thread and re-raised on the consumer thread when the stream terminates. Nothing gets silently swallowed.

try:
    for event in agent.stream(prompt):
        ...
except RuntimeError as exc:
    print("Agent run failed:", exc)