Skip to main content
The progress compositor folds an agent’s typed StreamEvent stream into a bounded, multi-line status view you can render in any transport — bots today, TUI or web tomorrow. The compositor is pure stdlib — zero third-party dependencies, deterministic, and side-effect free. Every call returns a new list; the input is never mutated. The same three exports power the bot progress_style="feed" view.

​
Quick Start

1

Fold events into lines

from praisonaiagents.streaming import merge_progress_line, render_progress

lines = []
for event in agent_events:          # any typed StreamEvent stream
    lines = merge_progress_line(lines, event)

print(render_progress(lines))
2

Wire it into your own callback

from praisonaiagents import Agent
from praisonaiagents.streaming import (
    ProgressLine, merge_progress_line, render_progress,
)

lines: list[ProgressLine] = []

def my_callback(event) -> None:
    global lines
    lines = merge_progress_line(lines, event)
    print("\033[2J\033[H" + render_progress(lines, max_lines=6))

agent = Agent(name="research-assistant", instructions="Research and summarise.")
agent.stream_emitter.add_callback(my_callback)
agent.start("Research quantum computing")
Any typed StreamEvent stream — even one you produce yourself — folds into the same bounded view.

​
How It Works

A start event and its matching finish event share one correlation id, so merge_progress_line updates the existing line instead of appending a duplicate. Each event type maps to a line kind and state:
StreamEventTypeRendered kindRendered state
TOOL_CALL_START, DELTA_TOOL_CALLtoolrunning (⏳)
TOOL_CALL_END, TOOL_CALL_RESULTtooldone (✓)
TOOL_PROGRESScommand-outputrunning (⏳)
ERRORtoolerror (✗)
anything else—no-op (returns the input list unchanged)

​
Configuration Options

render_progress bounds the output for edit-in-place delivery.
OptionTypeDefaultDescription
max_linesint8Trailing rolling window — only the last N lines are shown
max_line_charsint120Per-line character cap including the glyph prefix; words are truncated at a boundary with an ellipsis
Each ProgressLine carries the state for a single step.
FieldTypeDefaultDescription
idstr—Correlation key (tool call id, plan step id, approval id)
kindstr—"tool", "plan", "approval", or "command-output"
textstr—Human-readable status text
statestr"running""running", "done", or "error"

​
State machine rules

Terminal states are sticky, so the feed never lies about an outcome.
  • Error takes precedence — once a line is error (✗) it stays error, even if a later done event arrives for the same id.
  • Done is not downgraded — a late running event never resets a completed done (✓) line back to running.
  • Non-progress events are no-ops — events outside the mapping table return the input list unchanged, so unrelated stream traffic never adds noise.

​
Common Patterns

Build a live TUI panel straight from the agent’s event emitter. 
from praisonaiagents import Agent
from praisonaiagents.streaming import merge_progress_line, render_progress

agent = Agent(name="assistant", instructions="Use tools and summarise.")
lines = []

def render_panel(event):
    global lines
    lines = merge_progress_line(lines, event)
    print("\033[2J\033[H" + render_progress(lines, max_lines=10))

agent.stream_emitter.add_callback(render_panel)
agent.start("Compare three databases")
Feed the compositor from your own event source — plan steps, approvals, or command output all fold in the same way, since merge_progress_line only cares about the event’s type and correlation id. Override the glyphs by wrapping render_progress with a renderer that walks each ProgressLine.state. 
from praisonaiagents.streaming import ProgressLine

GLYPHS = {"running": "•", "done": "[OK]", "error": "[X]"}

def render_ascii(lines: list[ProgressLine]) -> str:
    return "\n".join(f"{GLYPHS.get(l.state, '•')} {l.text}" for l in lines[-8:])

​
Best Practices

merge_progress_line returns a new list and never mutates its input. Always re-assign — lines = merge_progress_line(lines, event) — so your view reflects the fold.
Message-based channels crop long text. Pass a small max_lines (4–8) to render_progress so the newest steps stay visible while older ones scroll off.
A tool’s start and finish events share one id (from tool_call.id / metadata.id, falling back to tool:<name>), so updates rewrite the same line. You do not need to de-duplicate events yourself.
Don’t post-process error → done. The compositor already guarantees a sticky terminal state, so your renderer can trust ProgressLine.state as-is.

Bot Streaming Replies

Enable the feed style on Telegram, Slack, and Discord

Streaming Tool Events

The typed StreamEvents the compositor folds

Tool Progress Streaming

Emit mid-tool progress updates into the stream