Skip to content

kaboo_workflows.renderers

Terminal renderers for streamed agent activity.

kaboo_workflows.renderers

Renderers for displaying agent output.

AnsiRenderer 

AnsiRenderer(
    *, file=None, separator_width=None, typewriter_delay=0.0
)

Bases: EventRenderer

Renders events using raw ANSI escape codes.

Initialize the AnsiRenderer.

No third-party dependencies. Colour codes are automatically suppressed when stdout is not a TTY (piped / redirected output).

Token and reasoning events are written inline (no trailing newline) so they appear as a continuous stream. A separator line is printed when the mode changes between reasoning and responding, or when the active agent changes.

Parameters:

Name Type Description Default
file Any | None

Output stream (defaults to sys.stdout).

 None
separator_width int | None

Width of separator lines. Defaults to 70 or the current terminal width, whichever is available.

 None
typewriter_delay float

Seconds to sleep after each printable character in TOKEN and REASONING events. 0.0 (default) disables the effect entirely with no overhead.

 0.0
Source code in src/kaboo_workflows/renderers/ansi.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    *,
    file: Any | None = None,
    separator_width: int | None = None,
    typewriter_delay: float = 0.0,
) -> None:
    """Initialize the AnsiRenderer.

    No third-party dependencies.  Colour codes are automatically
    suppressed when stdout is not a TTY (piped / redirected output).

    Token and reasoning events are written inline (no trailing newline)
    so they appear as a continuous stream.  A separator line is printed
    when the mode changes between reasoning and responding, or when the
    active agent changes.

    Args:
        file: Output stream (defaults to ``sys.stdout``).
        separator_width: Width of separator lines. Defaults to 70 or the
            current terminal width, whichever is available.
        typewriter_delay: Seconds to sleep after each printable character
            in TOKEN and REASONING events.  ``0.0`` (default) disables
            the effect entirely with no overhead.
    """
    self._out = file or sys.stdout
    self._in_stream = False
    self._mode: str | None = None  # "reasoning" | "responding" | None
    self._active_agent: str | None = None
    self._typewriter_delay = typewriter_delay

    # Cache TTY state and pre-compute ANSI escape strings.
    is_tty = hasattr(self._out, "isatty") and self._out.isatty()
    self._dim = "\033[2m" if is_tty else ""
    self._bold = "\033[1m" if is_tty else ""
    self._cyan = "\033[36m" if is_tty else ""
    self._green = "\033[32m" if is_tty else ""
    self._red = "\033[31m" if is_tty else ""
    self._yellow = "\033[33m" if is_tty else ""
    self._magenta = "\033[95m" if is_tty else ""
    self._reset = "\033[0m" if is_tty else ""

    if separator_width is not None:
        self._separator_width = separator_width
    else:
        self._separator_width = shutil.get_terminal_size((70, 24)).columns

    self._handlers: dict[str, Callable[[StreamEvent], None]] = {
        EventType.SESSION_START: self._handle_session_start,
        EventType.SESSION_END: self._handle_session_end,
        EventType.TOKEN: self._handle_token,
        EventType.AGENT_START: self._handle_agent_start,
        EventType.TOOL_START: self._handle_tool_start,
        EventType.TOOL_END: self._handle_tool_end,
        EventType.AGENT_COMPLETE: self._handle_agent_complete,
        EventType.ERROR: self._handle_error,
        EventType.NODE_START: self._handle_node_start,
        EventType.NODE_STOP: self._handle_node_stop,
        EventType.HANDOFF: self._handle_handoff,
        EventType.MULTIAGENT_START: self._handle_multiagent_start,
        EventType.MULTIAGENT_COMPLETE: self._handle_multiagent_complete,
        EventType.REASONING: self._handle_reasoning,
    }