Logging Utilities

stirrup.utils.logging

Rich logging for agent workflows with visual hierarchy.

__all__ module-attribute

__all__ = ['AgentLogger', 'AgentLoggerBase']

console module-attribute

console = Console()

SUBAGENT_INDENT_SPACES module-attribute

SUBAGENT_INDENT_SPACES: int = 8

AssistantMessage

Bases: BaseModel

LLM response message with optional tool calls and token usage tracking.

ToolMessage

Bases: BaseModel

Tool execution result returned to the LLM.

UserMessage

Bases: BaseModel

User input message to the LLM.

AgentLoggerBase

Bases: ABC

Abstract base class for agent loggers.

Defines the interface that Agent uses for logging. Implement this to create custom loggers (e.g., for testing, file output, or monitoring services).

Properties are set by Agent after construction: - name, model, max_turns, depth: Agent configuration - finish_params, run_metadata, output_dir: Set before exit for final stats

__enter__ abstractmethod

__enter__() -> Self

Enter logging context. Called when agent session starts.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def __enter__(self) -> Self:
    """Enter logging context. Called when agent session starts."""
    ...

__exit__ abstractmethod

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Exit logging context. Called when agent session ends.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit logging context. Called when agent session ends."""
    ...

on_step abstractmethod

on_step(
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None

Report step progress and stats during agent execution.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def on_step(
    self,
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None:
    """Report step progress and stats during agent execution."""
    ...

assistant_message abstractmethod

assistant_message(
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None

Log an assistant message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def assistant_message(
    self,
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None:
    """Log an assistant message."""
    ...

user_message abstractmethod

user_message(user_message: UserMessage) -> None

Log a user message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def user_message(self, user_message: UserMessage) -> None:
    """Log a user message."""
    ...

task_message abstractmethod

task_message(task: str | list[Any]) -> None

Log the initial task/prompt at the start of a run.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def task_message(self, task: str | list[Any]) -> None:
    """Log the initial task/prompt at the start of a run."""
    ...

tool_result abstractmethod

tool_result(tool_message: ToolMessage) -> None

Log a tool execution result.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def tool_result(self, tool_message: ToolMessage) -> None:
    """Log a tool execution result."""
    ...

context_summarization_start abstractmethod

context_summarization_start(
    pct_used: float, cutoff: float
) -> None

Log that context summarization is starting.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def context_summarization_start(self, pct_used: float, cutoff: float) -> None:
    """Log that context summarization is starting."""
    ...

context_summarization_complete abstractmethod

context_summarization_complete(
    summary: str, bridge: str
) -> None

Log completed context summarization.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def context_summarization_complete(self, summary: str, bridge: str) -> None:
    """Log completed context summarization."""
    ...

debug abstractmethod

debug(message: str, *args: object) -> None

Log a debug message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def debug(self, message: str, *args: object) -> None:
    """Log a debug message."""
    ...

info abstractmethod

info(message: str, *args: object) -> None

Log an info message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def info(self, message: str, *args: object) -> None:
    """Log an info message."""
    ...

warning abstractmethod

warning(message: str, *args: object) -> None

Log a warning message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def warning(self, message: str, *args: object) -> None:
    """Log a warning message."""
    ...

error abstractmethod

error(message: str, *args: object) -> None

Log an error message.

Source code in src/stirrup/utils/logging.py

@abstractmethod
def error(self, message: str, *args: object) -> None:
    """Log an error message."""
    ...

AgentLogger

AgentLogger(
    *, show_spinner: bool = True, level: int = INFO
)

Bases: AgentLoggerBase

Rich console logger for agent workflows.

Implements AgentLoggerBase with rich formatting, spinners, and visual hierarchy. Each agent (including sub-agents) should have its own logger instance.

Usage

from stirrup.clients.chat_completions_client import ChatCompletionsClient

Agent creates logger internally by default

client = ChatCompletionsClient(model="gpt-4") agent = Agent(client=client, name="assistant")

Or pass a pre-configured logger

logger = AgentLogger(show_spinner=False) agent = Agent(client=client, name="assistant", logger=logger)

Agent sets these properties before calling enter:

logger.name, logger.model, logger.max_turns, logger.depth

Agent sets these before calling exit:

logger.finish_params, logger.run_metadata, logger.output_dir

Initialize the agent logger.

Parameters:

Name	Type	Description	Default
show_spinner	bool	Whether to show a spinner while agent runs (only for depth=0)	True
level	int	Logging level (default: INFO)	INFO

Source code in src/stirrup/utils/logging.py

def __init__(
    self,
    *,
    show_spinner: bool = True,
    level: int = logging.INFO,
) -> None:
    """Initialize the agent logger.

    Args:
        show_spinner: Whether to show a spinner while agent runs (only for depth=0)
        level: Logging level (default: INFO)
    """
    # Properties set by Agent before __enter__
    self.name: str = "agent"
    self.model: str | None = None
    self.max_turns: int | None = None
    self.depth: int = 0

    # State set by Agent before __exit__
    self.finish_params: BaseModel | None = None
    self.run_metadata: dict[str, list[Any]] | None = None
    self.output_dir: str | None = None

    # Configuration
    self._show_spinner = show_spinner
    self._level = level

    # Spinner state (only used when depth == 0 and show_spinner is True)
    self._current_step = 0
    self._tool_calls = 0
    self._input_tokens = 0
    self._output_tokens = 0
    self._live: Live | None = None

    # Configure rich logging on first logger creation
    self._configure_logging()

__enter__

__enter__() -> Self

Enter logging context. Logs agent start and starts spinner if depth=0.

Source code in src/stirrup/utils/logging.py

def __enter__(self) -> Self:
    """Enter logging context. Logs agent start and starts spinner if depth=0."""
    # Log agent start (rule + system prompt display)
    indent_spaces = self.depth * SUBAGENT_INDENT_SPACES

    # Build title with optional model info
    model_str = f" ({self.model})" if self.model else ""
    if self.depth == 0:
        title = f"▶ {self.name}{model_str}"
        console.rule(f"[bold cyan]{title}[/]", style="cyan")
    else:
        title = f"▶ {self.name}: Level {self.depth}{model_str}"
        rule = Rule(f"[bold cyan]{title}[/]", style="cyan")
        self._print_indented(rule, indent_spaces)
    console.print()

    # Start spinner only for top-level agent
    if self.depth == 0 and self._show_spinner:
        self._live = Live(self._make_spinner(), console=console, refresh_per_second=10)
        self._live.start()

    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Exit logging context. Stops spinner and logs completion stats.

Source code in src/stirrup/utils/logging.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Exit logging context. Stops spinner and logs completion stats."""
    # Stop spinner first
    if self._live:
        self._live.stop()
        self._live = None

    error = str(exc_val) if exc_type is not None else None
    self._log_finish(error=error)

on_step

on_step(
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None

Report step progress and stats during agent execution.

Source code in src/stirrup/utils/logging.py

def on_step(
    self,
    step: int,
    tool_calls: int = 0,
    input_tokens: int = 0,
    output_tokens: int = 0,
) -> None:
    """Report step progress and stats during agent execution."""
    self._current_step = step
    self._tool_calls = tool_calls
    self._input_tokens = input_tokens
    self._output_tokens = output_tokens
    if self._live:
        self._live.update(self._make_spinner())

set_level

set_level(level: int) -> None

Set the logging level.

Source code in src/stirrup/utils/logging.py

def set_level(self, level: int) -> None:
    """Set the logging level."""
    self._level = level
    # Also update root logger level
    logging.getLogger().setLevel(level)

is_enabled_for

is_enabled_for(level: int) -> bool

Check if a given log level is enabled.

Source code in src/stirrup/utils/logging.py

def is_enabled_for(self, level: int) -> bool:
    """Check if a given log level is enabled."""
    return level >= self._level

debug

debug(message: str, *args: object) -> None

Log a debug message (dim style).

Source code in src/stirrup/utils/logging.py

def debug(self, message: str, *args: object) -> None:
    """Log a debug message (dim style)."""
    if self._level <= logging.DEBUG:
        formatted = message % args if args else message
        console.print(f"[dim]{formatted}[/]")

info

info(message: str, *args: object) -> None

Log an info message.

Source code in src/stirrup/utils/logging.py

def info(self, message: str, *args: object) -> None:
    """Log an info message."""
    if self._level <= logging.INFO:
        formatted = message % args if args else message
        console.print(formatted)

warning

warning(message: str, *args: object) -> None

Log a warning message (yellow style).

Source code in src/stirrup/utils/logging.py

def warning(self, message: str, *args: object) -> None:
    """Log a warning message (yellow style)."""
    if self._level <= logging.WARNING:
        formatted = message % args if args else message
        console.print(f"[yellow]⚠ {formatted}[/]")

error

error(message: str, *args: object) -> None

Log an error message (red style).

Source code in src/stirrup/utils/logging.py

def error(self, message: str, *args: object) -> None:
    """Log an error message (red style)."""
    if self._level <= logging.ERROR:
        formatted = message % args if args else message
        console.print(f"[red]✗ {formatted}[/]")

critical

critical(message: str, *args: object) -> None

Log a critical message (bold red style).

Source code in src/stirrup/utils/logging.py

def critical(self, message: str, *args: object) -> None:
    """Log a critical message (bold red style)."""
    if self._level <= logging.CRITICAL:
        formatted = message % args if args else message
        console.print(f"[bold red]✗ CRITICAL: {formatted}[/]")

exception

exception(message: str, *args: object) -> None

Log an error message with exception traceback (red style with traceback).

Source code in src/stirrup/utils/logging.py

def exception(self, message: str, *args: object) -> None:
    """Log an error message with exception traceback (red style with traceback)."""
    if self._level <= logging.ERROR:
        formatted = message % args if args else message
        console.print(f"[red]✗ {formatted}[/]")
        console.print_exception()

assistant_message

assistant_message(
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None

Log an assistant message with content and tool calls in a panel.

Parameters:

Name	Type	Description	Default
turn	int	Current turn number (1-indexed)	required
max_turns	int	Maximum number of turns	required
assistant_message	AssistantMessage	The assistant's response message	required

Source code in src/stirrup/utils/logging.py

def assistant_message(
    self,
    turn: int,
    max_turns: int,
    assistant_message: AssistantMessage,
) -> None:
    """Log an assistant message with content and tool calls in a panel.

    Args:
        turn: Current turn number (1-indexed)
        max_turns: Maximum number of turns
        assistant_message: The assistant's response message
    """
    if self._level > logging.INFO:
        return

    # Build panel content
    content = Text()

    # Add assistant content if present
    if assistant_message.content:
        text = assistant_message.content
        if isinstance(text, list):
            text = "\n".join(str(block) for block in text)
        # Truncate long content
        if len(text) > 500:
            text = text[:500] + "..."
        content.append(text, style="white")

    # Add tool calls if present
    if assistant_message.tool_calls:
        if assistant_message.content:
            content.append("\n\n")
        content.append("Tool Calls:\n", style="bold magenta")
        for tc in assistant_message.tool_calls:
            args_parsed = json.loads(tc.arguments)
            args_formatted = json.dumps(args_parsed, indent=2, ensure_ascii=False)
            args_preview = args_formatted[:1000] + "..." if len(args_formatted) > 1000 else args_formatted
            content.append(f"  🔧 {tc.name}", style="magenta")
            content.append(args_preview, style="dim")

    # Create and print panel with agent name in title
    title = f"[bold]AssistantMessage[/bold] │ {self.name} │ Turn {turn}/{max_turns}"
    panel = Panel(content, title=title, title_align="left", border_style="yellow", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

user_message

user_message(user_message: UserMessage) -> None

Log a user message in a panel.

Parameters:

Name	Type	Description	Default
user_message	UserMessage	The user's message	required

Source code in src/stirrup/utils/logging.py

def user_message(self, user_message: UserMessage) -> None:
    """Log a user message in a panel.

    Args:
        user_message: The user's message
    """
    if self._level > logging.INFO:
        return

    # Build panel content
    content = Text()

    # Add user content
    if user_message.content:
        text = user_message.content
        if isinstance(text, list):
            text = "\n".join(str(block) for block in text)
        # Truncate long content
        if len(text) > 500:
            text = text[:500] + "..."
        content.append(text, style="white")

    # Create and print panel with agent name in title
    title = f"[bold]UserMessage[/bold] │ {self.name}"
    panel = Panel(content, title=title, title_align="left", border_style="blue", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

task_message

task_message(task: str | list[Any]) -> None

Log the initial task/prompt at the start of a run.

Source code in src/stirrup/utils/logging.py

def task_message(self, task: str | list[Any]) -> None:
    """Log the initial task/prompt at the start of a run."""
    if self._level > logging.INFO:
        return

    # Convert list content to string
    if isinstance(task, list):
        task = "\n".join(str(block) for block in task)

    # Clean up whitespace from multi-line strings
    # Normalize each line by stripping leading/trailing whitespace and rejoining
    lines = [line.strip() for line in task.split("\n")]
    task = " ".join(line for line in lines if line)

    # Use "Sub Agent" prefix for nested agents
    prefix = "Sub Agent" if self.depth > 0 else "Agent"

    if self.depth > 0:
        indent = " " * (self.depth * SUBAGENT_INDENT_SPACES)
        console.print(f"{indent}[bold]{prefix} Task:[/bold]")
        console.print()
        for line in task.split("\n"):
            console.print(f"{indent}{line}")
    else:
        console.print(f"[bold]{prefix} Task:[/bold]")
        console.print()
        console.print(task)

    console.print()  # Add gap after task section

warnings_message

warnings_message(warnings: list[str]) -> None

Display warnings at run start as simple text.

Source code in src/stirrup/utils/logging.py

def warnings_message(self, warnings: list[str]) -> None:
    """Display warnings at run start as simple text."""
    if self._level > logging.INFO or not warnings:
        return

    console.print("[bold orange1]Warnings[/bold orange1]")
    console.print()
    for warning in warnings:
        console.print(f"[orange1]⚠ {warning}[/orange1]")
        console.print()  # Add gap between warnings

tool_result

tool_result(tool_message: ToolMessage) -> None

Log a single tool execution result in a panel with XML syntax highlighting.

Parameters:

Name	Type	Description	Default
tool_message	ToolMessage	The tool execution result	required

Source code in src/stirrup/utils/logging.py

def tool_result(self, tool_message: ToolMessage) -> None:
    """Log a single tool execution result in a panel with XML syntax highlighting.

    Args:
        tool_message: The tool execution result
    """
    if self._level > logging.INFO:
        return

    tool_name = tool_message.name or "unknown"

    # Get result content
    result_text = tool_message.content
    if isinstance(result_text, list):
        result_text = "\n".join(str(block) for block in result_text)

    # Unescape HTML entities (e.g., &lt; -> <, &gt; -> >, &amp; -> &)
    result_text = html.unescape(result_text)

    # Truncate long results
    if len(result_text) > 1000:
        result_text = result_text[:1000] + "..."

    # Format as XML with syntax highlighting
    content = Syntax(result_text, "xml", theme="monokai", word_wrap=True)

    # Status indicator in title with agent name
    status = "✓" if tool_message.args_was_valid else "✗"
    status_style = "green" if tool_message.args_was_valid else "red"
    title = f"[{status_style}]{status}[/{status_style}] [bold]ToolResult[/bold] │ {self.name} │ [green]{tool_name}[/green]"

    panel = Panel(content, title=title, title_align="left", border_style="green", padding=(0, 1))

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

context_summarization_start

context_summarization_start(
    pct_used: float, cutoff: float
) -> None

Log context window summarization starting in an orange panel.

Parameters:

Name	Type	Description	Default
pct_used	float	Percentage of context window currently used (0.0-1.0)	required
cutoff	float	The threshold that triggered summarization (0.0-1.0)	required

Source code in src/stirrup/utils/logging.py

def context_summarization_start(self, pct_used: float, cutoff: float) -> None:
    """Log context window summarization starting in an orange panel.

    Args:
        pct_used: Percentage of context window currently used (0.0-1.0)
        cutoff: The threshold that triggered summarization (0.0-1.0)
    """
    # Build panel content
    content = Text()
    content.append("Context window limit reached\n\n", style="bold")
    content.append("Used: ", style="dim")
    content.append(f"{pct_used:.1%}", style="bold orange1")
    content.append("  │  ", style="dim")
    content.append("Threshold: ", style="dim")
    content.append(f"{cutoff:.1%}", style="bold")
    content.append("\n\n", style="dim")
    content.append("Summarizing conversation history...", style="italic")

    panel = Panel(
        content,
        title="[bold orange1]📝 Context Summarization[/]",
        title_align="left",
        border_style="orange1",
        padding=(0, 1),
    )

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

context_summarization_complete

context_summarization_complete(
    summary: str, bridge: str
) -> None

Log the completed context summarization with summary content.

Parameters:

Name	Type	Description	Default
summary	str	The generated summary of the conversation	required
bridge	str	The bridge message that will be used to continue the conversation	required

Source code in src/stirrup/utils/logging.py

def context_summarization_complete(self, summary: str, bridge: str) -> None:
    """Log the completed context summarization with summary content.

    Args:
        summary: The generated summary of the conversation
        bridge: The bridge message that will be used to continue the conversation
    """
    # Truncate long summaries for display
    summary_display = summary
    if len(summary_display) > 800:
        summary_display = summary_display[:800] + "..."

    # Build panel content
    content = Text()
    content.append("Summary:\n", style="bold")
    content.append(summary_display, style="white")

    if self._level > logging.INFO:
        bridge_display = bridge
        if len(bridge_display) > 200:
            bridge_display = bridge_display[:200] + "..."
        content.append("\n\n")
        content.append("Bridge Message:\n", style="bold dim")
        content.append(bridge_display, style="dim italic")

    panel = Panel(
        content,
        title="[bold green]✓ Summary Generated[/]",
        title_align="left",
        border_style="green",
        padding=(0, 1),
    )

    if self.depth > 0:
        self._print_indented(panel, self.depth * SUBAGENT_INDENT_SPACES)
    else:
        console.print(panel)

_aggregate_list

_aggregate_list(metadata_list: list[T]) -> T | None

Aggregate a list of metadata using add.

Source code in src/stirrup/core/models.py

def _aggregate_list[T: Addable](metadata_list: list[T]) -> T | None:
    """Aggregate a list of metadata using __add__."""
    if not metadata_list:
        return None
    aggregated = metadata_list[0]
    for m in metadata_list[1:]:
        aggregated = aggregated + m
    return aggregated

aggregate_metadata

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: Literal[True] = True,
) -> object

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: Literal[False] = ...,
) -> dict

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: bool = True,
) -> dict | object

Aggregate metadata lists and flatten sub-agents into a single-level dict with hierarchical keys.

For entries with nested run_metadata (e.g., SubAgentMetadata), flattens sub-agents using dot notation. Each sub-agent's value is a dict mapping its direct tool names to their aggregated metadata (excluding nested sub-agent data, which gets its own top-level key).

At the root level, token_usage is rolled up to include all sub-agent token usage.

Parameters:

Name	Type	Description	Default
metadata_dict	dict[str, list[Any]]	Dict mapping names (tools or agents) to lists of metadata instances	required
prefix	str	Key prefix for nested calls (used internally for recursion)	''

Returns:

Name	Type	Description
	dict | object	Flat dict with dot-notation keys for sub-agents.
Example	dict | object	{ "token_usage": , "web_browsing_sub_agent": {"web_search": , "token_usage": }, "web_browsing_sub_agent.web_fetch_sub_agent": {"fetch_web_page": , "token_usage": }
	dict | object	}

Source code in src/stirrup/core/models.py

def aggregate_metadata(
    metadata_dict: dict[str, list[Any]], prefix: str = "", return_json_serializable: bool = True
) -> dict | object:
    """Aggregate metadata lists and flatten sub-agents into a single-level dict with hierarchical keys.

    For entries with nested run_metadata (e.g., SubAgentMetadata), flattens sub-agents using dot notation.
    Each sub-agent's value is a dict mapping its direct tool names to their aggregated metadata
    (excluding nested sub-agent data, which gets its own top-level key).

    At the root level, token_usage is rolled up to include all sub-agent token usage.

    Args:
        metadata_dict: Dict mapping names (tools or agents) to lists of metadata instances
        prefix: Key prefix for nested calls (used internally for recursion)

    Returns:
        Flat dict with dot-notation keys for sub-agents.
        Example: {
            "token_usage": <combined from all agents>,
            "web_browsing_sub_agent": {"web_search": <aggregated>, "token_usage": <aggregated>},
            "web_browsing_sub_agent.web_fetch_sub_agent": {"fetch_web_page": <aggregated>, "token_usage": <aggregated>}
        }
    """
    result: dict = {}

    # First pass: aggregate all entries in this level
    aggregated_level: dict = {}
    for name, metadata_list in metadata_dict.items():
        if not metadata_list:
            continue
        aggregated_level[name] = _aggregate_list(metadata_list)

    # Second pass: separate nested sub-agents from direct tools, and recurse
    direct_tools: dict = {}
    for name, aggregated in aggregated_level.items():
        if hasattr(aggregated, "run_metadata") and isinstance(aggregated.run_metadata, dict):
            # This is a sub-agent - recurse into it
            full_key = f"{prefix}.{name}" if prefix else name
            nested = aggregate_metadata(aggregated.run_metadata, prefix=full_key, return_json_serializable=False)
            result.update(nested)
        else:
            # This is a direct tool/metadata - keep it at this level
            direct_tools[name] = aggregated

    # Store direct tools under the current prefix
    if prefix:
        result[prefix] = direct_tools
    else:
        # At root level, merge direct tools into result
        result.update(direct_tools)

    # At root level, roll up all token_usage from sub-agents
    if not prefix:
        total_token_usage = _collect_all_token_usage(result)
        if total_token_usage.total > 0:
            result["token_usage"] = [total_token_usage]

    if return_json_serializable:
        # Convert all Pydantic models to JSON-serializable dicts
        return to_json_serializable(result)
    return result

_is_subagent_metadata

_is_subagent_metadata(data: object) -> bool

Check if data represents sub-agent metadata.

Sub-agent metadata can be: - A Pydantic SubAgentMetadata object with run_metadata attribute - A dict where all values are dicts/objects (from aggregate_metadata flattening)

Source code in src/stirrup/utils/logging.py

def _is_subagent_metadata(data: object) -> bool:
    """Check if data represents sub-agent metadata.

    Sub-agent metadata can be:
    - A Pydantic SubAgentMetadata object with run_metadata attribute
    - A dict where all values are dicts/objects (from aggregate_metadata flattening)
    """
    # Check for Pydantic SubAgentMetadata object
    if hasattr(data, "run_metadata") and isinstance(data.run_metadata, dict):
        return True
    # Check for flattened dict of dicts (from aggregate_metadata)
    if isinstance(data, dict) and data:
        return all(isinstance(v, dict) or hasattr(v, "model_dump") for v in data.values())
    return False

_format_token_usage

_format_token_usage(data: object) -> str

Format token_usage (dict or TokenUsage object) as a human-readable string.

Source code in src/stirrup/utils/logging.py

def _format_token_usage(data: object) -> str:
    """Format token_usage (dict or TokenUsage object) as a human-readable string."""
    if isinstance(data, dict):
        # Dict representation
        data_dict = cast(dict[str, Any], data)
        input_tokens: int = data_dict.get("input", 0)
        output_tokens: int = data_dict.get("output", 0)
        reasoning_tokens: int = data_dict.get("reasoning", 0)
    elif hasattr(data, "input") and hasattr(data, "output"):
        # Pydantic TokenUsage object - use getattr for type safety
        input_tokens = int(getattr(data, "input", 0))
        output_tokens = int(getattr(data, "output", 0))
        reasoning_tokens = int(getattr(data, "reasoning", 0))
    else:
        return str(data)
    total = input_tokens + output_tokens + reasoning_tokens
    return f"{total:,} tokens"

_get_nested_tools

_get_nested_tools(data: object) -> dict[str, object]

Extract nested tools dict from sub-agent metadata.

Source code in src/stirrup/utils/logging.py

def _get_nested_tools(data: object) -> dict[str, object]:
    """Extract nested tools dict from sub-agent metadata."""
    if hasattr(data, "run_metadata"):
        # Pydantic SubAgentMetadata - return its run_metadata
        run_metadata = data.run_metadata
        if isinstance(run_metadata, dict):
            return cast(dict[str, object], run_metadata)
    if isinstance(data, dict):
        # Already a dict
        return cast(dict[str, object], data)
    return {}

_add_tool_branch

_add_tool_branch(
    parent: Tree,
    tool_name: str,
    tool_data: object,
    skip_fields: set[str],
) -> None

Add a tool entry to the tree, handling nested sub-agent data recursively.

Parameters:

Name	Type	Description	Default
parent	Tree	The tree or branch to add to	required
tool_name	str	Name of the tool or sub-agent	required
tool_data	object	The tool's metadata (dict, Pydantic model, list, or scalar)	required
skip_fields	set[str]	Fields to skip when displaying dict contents	required

Source code in src/stirrup/utils/logging.py

def _add_tool_branch(
    parent: Tree,
    tool_name: str,
    tool_data: object,
    skip_fields: set[str],
) -> None:
    """Add a tool entry to the tree, handling nested sub-agent data recursively.

    Args:
        parent: The tree or branch to add to
        tool_name: Name of the tool or sub-agent
        tool_data: The tool's metadata (dict, Pydantic model, list, or scalar)
        skip_fields: Fields to skip when displaying dict contents
    """
    # Special case: token_usage formatted as total tokens
    if tool_name == "token_usage":
        if isinstance(tool_data, list) and tool_data:
            parent.add(f"[dim]token_usage:[/] {_format_token_usage(tool_data[0])}")
        else:
            parent.add(f"[dim]token_usage:[/] {_format_token_usage(tool_data)}")
        return

    # Case 1: List → aggregate using __add__, then recurse
    if isinstance(tool_data, list) and tool_data:
        aggregated = _aggregate_list(tool_data)
        if aggregated is not None:
            _add_tool_branch(parent, tool_name, aggregated, skip_fields)
        return

    # Case 2: SubAgentMetadata → recurse into run_metadata only
    if _is_subagent_metadata(tool_data):
        branch = parent.add(f"[magenta]{tool_name}[/]")
        for nested_name, nested_data in sorted(_get_nested_tools(tool_data).items()):
            _add_tool_branch(branch, nested_name, nested_data, skip_fields)
        return

    # Case 3: Leaf node - display fields as branches
    # Convert to dict if Pydantic model
    if hasattr(tool_data, "model_dump"):
        data_dict = cast(Callable[[], dict[str, Any]], tool_data.model_dump)()
    elif isinstance(tool_data, dict):
        data_dict = cast(dict[str, Any], tool_data)
    else:
        # Scalar value - just show it inline
        parent.add(f"[magenta]{tool_name}[/]: {tool_data}")
        return

    # Show num_uses inline with the tool name if present
    num_uses = data_dict.get("num_uses")
    if num_uses is not None:
        branch = parent.add(f"[magenta]{tool_name}[/]: {num_uses} call(s)")
    else:
        branch = parent.add(f"[magenta]{tool_name}[/]")

    for k, v in data_dict.items():
        if k not in skip_fields and v is not None:
            branch.add(f"[dim]{k}:[/] {v}")