Tool Progress Streaming

Stream incremental progress from inside a running tool without changing its signature

Call emit_tool_progress() from inside any tool to push partial output while it runs — zero overhead when no consumer is listening.

from praisonaiagents import Agent
from praisonaiagents.streaming.events import emit_tool_progress

def compress_file(path: str) -> str:
    """Compress a large file, streaming progress."""
    chunks = list(range(10))
    total = len(chunks)
    for i, chunk in enumerate(chunks):
        emit_tool_progress(f"Compressed chunk {i}", progress=(i + 1) / total)
    return f"Compressed {path}"

agent = Agent(
    name="File Manager",
    instructions="Compress files on request.",
    tools=[compress_file],
    stream=True,
)
agent.start("Compress data.log")

The user runs a long tool with streaming enabled; partial progress events surface in the UI while the tool still runs.

Quick Start

1. Emit progress from your tool

   Call emit_tool_progress anywhere inside a tool function:

   from praisonaiagents.streaming.events import emit_tool_progress
   
   def process_records(file_path: str) -> str:
       """Process records from a file."""
       records = ["record_1", "record_2", "record_3", "record_4", "record_5"]
       total = len(records)
   
       for i, record in enumerate(records):
           emit_tool_progress(
               f"Processing {record}",
               progress=(i + 1) / total,
           )
   
       return f"Processed {total} records from {file_path}"

2. Connect a streaming consumer

   Subscribe a callback that listens for TOOL_PROGRESS events:

   from praisonaiagents import Agent
   from praisonaiagents.streaming.events import StreamEvent, StreamEventType
   
   def on_event(event: StreamEvent) -> None:
       if event.type == StreamEventType.TOOL_PROGRESS:
           pct = (event.metadata or {}).get("progress")
           line = event.content or ""
           if pct is not None:
               print(f"[{pct:.0%}] {line}")
           else:
               print(line)
   
   agent = Agent(
       name="File Manager",
       instructions="Process files on request.",
       tools=[process_records],
       stream=True,
   )
   agent.stream_emitter.add_callback(on_event)
   agent.start("Process data.csv")

How It Works

sequenceDiagram
    participant Agent
    participant Loop as Tool Execution Loop
    participant Tool
    participant Channel as tool_progress_channel
    participant CB as Callback

    Agent->>Loop: run tool call
    Loop->>Channel: activate sink (tool_progress_channel(sink))
    Loop->>Tool: call tool function
    Tool->>Channel: emit_tool_progress("chunk 1", progress=0.25)
    Channel->>CB: StreamEvent(type=TOOL_PROGRESS, ...)
    Tool->>Channel: emit_tool_progress("chunk 2", progress=0.5)
    Channel->>CB: StreamEvent(type=TOOL_PROGRESS, ...)
    Tool-->>Loop: return result
    Loop->>Channel: deactivate sink

The channel is thread-local via contextvars.ContextVar — safe in multi-agent scenarios with concurrent tool calls.

API Reference

emit_tool_progress

from praisonaiagents.streaming.events import emit_tool_progress

emit_tool_progress(
    output="Line of stdout",     # partial text chunk
    progress=0.5,                # completion fraction 0.0–1.0
    metadata={"stream": "stderr"},  # optional structured metadata
)

Returns True if forwarded to an active sink, False if no sink is active (the common case outside an agent run — safe no-op).

StreamEventType.TOOL_PROGRESS

The event type emitted by emit_tool_progress. Check event.type == StreamEventType.TOOL_PROGRESS in a stream callback to filter these events.

Common Patterns

Percentage progress bar

from praisonaiagents.streaming.events import emit_tool_progress

def download_file(url: str) -> str:
    """Download a file with progress reporting."""
    total_chunks = 20
    for i in range(total_chunks):
        emit_tool_progress(
            f"Downloaded {(i + 1) * 5}%",
            progress=(i + 1) / total_chunks,
        )
    return f"Downloaded {url}"

Separate stdout and stderr streams

from praisonaiagents.streaming.events import emit_tool_progress

def run_build(command: str) -> str:
    """Run a build command with separated stdout/stderr."""
    emit_tool_progress("Build started", metadata={"stream": "stdout"})
    emit_tool_progress("Warning: deprecated API", metadata={"stream": "stderr"})
    emit_tool_progress("Build complete", progress=1.0, metadata={"stream": "stdout"})
    return "Build succeeded"

Long-running shell command

import subprocess
from praisonaiagents.streaming.events import emit_tool_progress

def run_tests(test_path: str) -> str:
    """Run tests and stream output."""
    result = subprocess.run(
        ["pytest", test_path, "-v"],
        capture_output=True, text=True
    )
    lines = result.stdout.splitlines()
    total = len(lines) or 1
    for i, line in enumerate(lines):
        emit_tool_progress(line, progress=(i + 1) / total)
    return f"Tests complete: {'passed' if result.returncode == 0 else 'failed'}"

Best Practices

In AG-UI

Over the AG-UI wire, a TOOL_PROGRESS event reaches the client as a custom event named praisonai.tool_progress with value { content, tool_call, metadata }. See the AGUI API reference.

Related