Rich terminal display consumer for agent events.
This module provides a dashboard-style event consumer focused on tool execution status rather than streaming text. The display is fixed-height and never grows — it shows what's running, what finished, and phase progress.
Everything domain-specific is injected: the panel title, an optional mode label, the phase specification (count + names, via :class:AgentDisplayState), and a discovery-extractor callable. The framework hardcodes no /5, no product name, and no codebase-specific discovery keys.
RichEventConsumer(
state,
*,
title,
mode_label=None,
discovery_extractor=None,
console=None,
)
Bases: EventConsumer
Dashboard-style consumer for agent execution.
Shows a fixed-height panel with: - Timer and progress bar - Tool category breakdown with mini bars - Active tool with spinner - Recent tool history with timing and status - Phase indicator [i/N] where N comes from the injected phase spec - Discovery feed populated by the injected extractor
No streaming text display. The agent's reasoning is not shown — only tool execution status matters for a long-running analysis agent.
When teams have been dispatched the coordinator layout is rendered (adding a team-status table); otherwise the single-agent layout is used.
Initialize the dashboard consumer.
Parameters:
| Name | Type | Description | Default |
state | AgentDisplayState | Display state carrying the injected phase spec. | required |
title | str | Panel title (replaces the hardcoded product name). | required |
mode_label | str | None | Optional badge appended to the title (e.g. "FULL"). | None |
discovery_extractor | Callable[[str, Any], DiscoveryEvent | None] | None | Callable mapping (tool_name, result) to a :class:DiscoveryEvent or None. Defaults to a no-op. The consumer guards the call so a raising extractor never crashes the display. | None |
console | Console | None | Optional Rich Console instance. | None |
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| def __init__(
self,
state: AgentDisplayState,
*,
title: str,
mode_label: str | None = None,
discovery_extractor: Callable[[str, Any], DiscoveryEvent | None] | None = None,
console: Console | None = None,
) -> None:
"""Initialize the dashboard consumer.
Args:
state: Display state carrying the injected phase spec.
title: Panel title (replaces the hardcoded product name).
mode_label: Optional badge appended to the title (e.g. ``"FULL"``).
discovery_extractor: Callable mapping ``(tool_name, result)`` to a
:class:`DiscoveryEvent` or ``None``. Defaults to a no-op. The
consumer guards the call so a raising extractor never crashes
the display.
console: Optional Rich Console instance.
"""
self.console = console or Console()
self.state = state
self._live: Live | None = None
self._title = title
self._mode_label = mode_label
self._discovery_extractor = discovery_extractor or _no_discovery
|
async
Start the dashboard display.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def start(self) -> None:
"""Start the dashboard display."""
self.state.reset()
self._live = Live(
self._build_display(),
console=self.console,
refresh_per_second=2,
transient=True,
vertical_overflow="ellipsis",
)
bind_live_renderable(self._live, self._build_display)
self._live.start()
|
async
Stop the dashboard and print final summary.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def stop(self) -> None:
"""Stop the dashboard and print final summary."""
if self._live:
self._live.stop()
self._live = None
elapsed = self.state.get_elapsed_seconds()
total = len(self.state.completed_tools)
errors = self.state.tool_errors
self.console.print(
f"[dim] {total} tools in {self._format_time(elapsed)} ✓ {total - errors} ✗ {errors}[/dim]",
)
|
async
on_run_started(thread_id, run_id)
Handle run started.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_run_started(self, thread_id: str, run_id: str) -> None:
"""Handle run started."""
self.state.thread_id = thread_id
self.state.run_id = run_id
self.state.start_time = time.monotonic()
|
async
on_text_start(message_id, role)
Handle text start — no display action.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_start(self, message_id: str, role: str) -> None:
"""Handle text start — no display action."""
self.state.active_message_id = message_id
|
async
on_text_content(message_id, delta)
Handle text delta — silently accumulate (not displayed).
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_content(self, message_id: str, delta: str) -> None:
"""Handle text delta — silently accumulate (not displayed)."""
self.state.text_buffer += delta
|
async
Handle text end — increment turn counter.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_end(self, message_id: str) -> None:
"""Handle text end — increment turn counter."""
self.state.active_message_id = None
self.state.turn_count += 1
self.state.text_buffer = ""
|
on_tool_start(tool_call_id, tool_name)
Handle tool start — show in active spinner.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_start(self, tool_call_id: str, tool_name: str) -> None:
"""Handle tool start — show in active spinner."""
self.state.active_tool = ToolCallState(
tool_call_id=tool_call_id,
tool_name=tool_name,
)
self.state.tool_start_time = time.monotonic()
self._detect_phase(tool_name)
|
on_tool_args(tool_call_id, args_delta)
Handle tool args — silently accumulate.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_args(self, tool_call_id: str, args_delta: str) -> None:
"""Handle tool args — silently accumulate."""
if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
self.state.active_tool.args_buffer += args_delta
|
on_tool_result(tool_call_id, result)
Handle tool result — check for errors and extract discoveries.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_result(self, tool_call_id: str, result: Any) -> None:
"""Handle tool result — check for errors and extract discoveries."""
if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
self.state.active_tool.result = result
if (isinstance(result, str) and "error" in result.lower()) or (
isinstance(result, dict) and result.get("error")
):
self.state.tool_errors += 1
discovery = self._extract_discovery(self.state.active_tool.tool_name, result)
if discovery:
self.state.add_discovery(discovery)
|
on_tool_end(tool_call_id)
Handle tool end — move to completed list.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_end(self, tool_call_id: str) -> None:
"""Handle tool end — move to completed list."""
if self.state.active_tool and self.state.active_tool.tool_call_id == tool_call_id:
self.state.complete_active_tool()
|
async
on_state_snapshot(snapshot)
Handle state snapshot.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_state_snapshot(self, snapshot: dict[str, Any]) -> None:
"""Handle state snapshot."""
self.state.state_snapshot = snapshot
if "phase" in snapshot:
self.state.current_phase = str(snapshot["phase"])
|
async
on_run_finished(thread_id, run_id)
Handle run finished.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_run_finished(self, thread_id: str, run_id: str) -> None:
"""Handle run finished."""
self.state.completed = True
|
async
on_error(message, code=None)
Handle error.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_error(self, message: str, code: str | None = None) -> None:
"""Handle error."""
error_text = f"[{code}] {message}" if code else message
self.state.error = error_text
|
Bases: EventConsumer
Silent consumer that only writes errors to stderr.
Initialize quiet consumer.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| def __init__(self) -> None:
"""Initialize quiet consumer."""
self._stderr = Console(stderr=True, no_color=True, highlight=False)
|
async
on_run_started(thread_id, run_id)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_run_started(self, thread_id: str, run_id: str) -> None:
"""No output."""
|
async
on_text_start(message_id, role)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_start(self, message_id: str, role: str) -> None:
"""No output."""
|
async
on_text_content(message_id, delta)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_content(self, message_id: str, delta: str) -> None:
"""No output."""
|
async
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_text_end(self, message_id: str) -> None:
"""No output."""
|
on_tool_start(tool_call_id, tool_name)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_start(self, tool_call_id: str, tool_name: str) -> None:
"""No output."""
|
on_tool_args(tool_call_id, args_delta)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_args(self, tool_call_id: str, args_delta: str) -> None:
"""No output."""
|
on_tool_result(tool_call_id, result)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_result(self, tool_call_id: str, result: Any) -> None:
"""No output."""
|
on_tool_end(tool_call_id)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_tool_end(self, tool_call_id: str) -> None:
"""No output."""
|
async
on_state_snapshot(snapshot)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_state_snapshot(self, snapshot: dict[str, Any]) -> None:
"""No output."""
|
async
on_run_finished(thread_id, run_id)
No output.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_run_finished(self, thread_id: str, run_id: str) -> None:
"""No output."""
|
async
on_error(message, code=None)
Print error to stderr.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| async def on_error(self, message: str, code: str | None = None) -> None:
"""Print error to stderr."""
error_text = f"[{code}] {message}" if code else message
self._stderr.print(f"Error: {error_text}")
|
async
Initialize the consumer (optional).
Override to perform setup before events start streaming. Called before the first event is received.
Source code in packages/strands-teams/src/strands_teams/display/base.py
| async def start(self) -> None:
"""Initialize the consumer (optional).
Override to perform setup before events start streaming.
Called before the first event is received.
"""
|
async
Cleanup the consumer (optional).
Override to perform cleanup after events stop streaming. Called after the last event is received or on error.
Source code in packages/strands-teams/src/strands_teams/display/base.py
| async def stop(self) -> None:
"""Cleanup the consumer (optional).
Override to perform cleanup after events stop streaming.
Called after the last event is received or on error.
"""
|
bind_live_renderable(live, builder)
Point Rich Live's auto-refresh at a custom display builder.
Rich calls get_renderable() on each refresh tick. Replacing it with builder lets the dashboard show fresh state (timer, tool elapsed, etc.) without manually calling live.update() on every change.
Source code in packages/strands-teams/src/strands_teams/display/rich.py
| def bind_live_renderable(live: Live, builder: Callable[[], RenderableType]) -> None:
"""Point Rich Live's auto-refresh at a custom display builder.
Rich calls ``get_renderable()`` on each refresh tick. Replacing it with
*builder* lets the dashboard show fresh state (timer, tool elapsed, etc.)
without manually calling ``live.update()`` on every change.
"""
live.get_renderable = builder # type: ignore[assignment] # ty: ignore[invalid-assignment]
|