Streaming Events

Distri streams a structured event for everything an agent does: it starts a run, plans, calls a tool, streams a token, hands off, finishes. You use those events two ways. You can listen live and update your UI as things happen, or you can attach to a task that is already running and watch it read-only.

Event types

Run lifecycle

Event	Payload	Notes
run_started	{ thread_id, run_id, agent_id }	A new run began.
plan_started	{ initial_plan }	The planner kicked off.
plan_finished	{ total_steps }	The planner produced a plan.
plan_pruned	{ removed_steps }	The planner removed branches.
run_finished	{ success, total_steps, failed_steps }	The run completed.
run_error	{ message, code? }	Terminal failure.

Step execution

Event	Payload	Notes
step_started	{ step_id, step_index }	A planner step began.
step_completed	{ step_id, success }	A step finished. Use it to spot retries or flaky tools.

Tool execution

Event	Payload	Notes
tool_execution_start	{ step_id, tool_call_id, tool_call_name, input }	A tool call was dispatched.
tool_execution_end	{ step_id, tool_call_id, tool_call_name, success }	A tool finished. Check success.
tool_calls	{ step_id, parent_message_id?, tool_calls: ToolCall[] }	Batched tool-call metadata.
tool_results	{ step_id, parent_message_id?, results: ToolResponse[] }	Batched tool outputs.

Streaming messages

Event	Payload	Notes
text_message_start	{ message_id, step_id, role, is_final? }	Content started streaming.
text_message_content	{ message_id, step_id, delta, stripped_content? }	Token-level updates.
text_message_end	{ message_id, step_id }	The message finished streaming.

Handover and workflows

Event	Payload	Notes
agent_handover	{ from_agent, to_agent, reason? }	Control passed to another agent.
workflow_started	{ workflow_name, total_steps }	A workflow started.
node_started	{ node_id, node_name, step_type }	A workflow node began.
node_completed	{ node_id, node_name, success, error? }	A node finished.
todos_updated	{ formatted_todos, action, todo_count }	The planner's todo list changed.

Listen live

React

useAgentEvents hands you a callback per event type.

import { useAgent, useAgentEvents } from '@distri/react';

function AgentUI() {
  const { agent } = useAgent({ agentIdOrDef: 'my_agent' });

  useAgentEvents({
    onRunStarted: (event) => console.log('run started', event.run_id),
    onToolExecutionStart: (event) => console.log('tool', event.tool_call_name),
    onTextMessageContent: (event) => console.log('delta', event.delta),
    onRunFinished: (event) => console.log('done', event.success),
  });

  return <Chat agent={agent} />;
}

Server-Sent Events

If you are not in React, read the raw SSE stream.

const es = new EventSource('https://api.distri.dev/v1/agents/my_agent/stream');

es.onmessage = (event) => {
  const { type, payload } = JSON.parse(event.data);
  if (type === 'text_message_content') render(payload.delta);
};

Follow a task read-only

<Chat> and useChat own a turn: they send a message and stream the reply. Sometimes you want the opposite. You want to watch a task that something else started, a background job, a task another user kicked off, or the same task shown a second time next to your chat.

<TaskView> and useTaskStreaming do that. They attach to a task, replay its history, and follow the live tail with the same renderers <Chat> uses, but with no composer and no tool interaction. Under the hood this is the A2A tasks/resubscribe stream.

You follow a task through the Agent that owns it (a task id alone is not enough, since A2A streams are per-agent).

<TaskView>

The turnkey option. It exposes the same renderer surface as <Chat>.

import { useAgent, TaskView } from '@distri/react';

function Runner() {
  const { agent } = useAgent({ agentIdOrDef: 'coder' });
  const [taskId, setTaskId] = useState<string | null>(null);

  const start = async () => {
    const task = await agent!.invoke({ message: buildMessage('Summarize the repo') });
    setTaskId((task as { id?: string }).id ?? null);
  };

  return (
    <>
      <button onClick={start}>Run in background</button>
      <TaskView agent={agent} taskId={taskId} rendering="rich" />
    </>
  );
}

Pass its id to follow a pre-existing task instead.

useTaskStreaming

Own the layout, reuse Distri's rendering. Wrap the returned store in ChatStoreContext and drop in <ChatMessageList>, or read messages and render it yourself.

import { useTaskStreaming, ChatMessageList, ChatStoreContext } from '@distri/react';

function FollowPanel({ agent, taskId }) {
  const { store, messages, isStreaming, isTerminal, reconnect } = useTaskStreaming({ agent, taskId });
  return (
    <ChatStoreContext.Provider value={store}>
      <header>
        {isStreaming ? 'Running…' : isTerminal ? 'Done' : 'Idle'}
        {isTerminal && <button onClick={reconnect}>Replay</button>}
      </header>
      <ChatMessageList messages={messages} rendering="rich" />
    </ChatStoreContext.Provider>
  );
}

The hook returns store, messages, isStreaming, isTerminal, error, reconnect, and stop. A few things worth knowing:

• tasks/resubscribe always replays the full log from the start, so every connect clears and replays. Text deltas are never doubled.
• The server closes the stream on the task's terminal event, so isTerminal flips true and reconnection stops.
• A dropped connection on a running task reconnects with a short bounded backoff.
• It never sends and never completes tools. Inline-hook and external-tool events render as read-only status.

Core primitive

The framework-agnostic version lives in @distri/core:

for await (const event of agent.resubscribe(taskId)) {
  // decoded DistriChatMessage, the read-only twin of agent.invokeStream()
}

Custom events

use distri_types::events::{AgentEvent, AgentEventType};

let event = AgentEvent::with_context(
    AgentEventType::RunError {
        message: "verification failed".into(),
        code: Some("INVALID_STATE".into()),
    },
    thread_id.to_string(),
    run_id.to_string(),
    task_id.to_string(),
    agent_id.to_string(),
);