Streaming Tool Events

Track tool execution in real time with streaming events — use TOOL_CALL_END as the completion marker for UI spinners and progress bars.

from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEventType

agent = Agent(name="Assistant")

def on_stream_event(event):
    if event.type == StreamEventType.TOOL_CALL_END:
        print(f"Tool '{event.tool_call['name']}' completed")

agent.stream_emitter.add_callback(on_stream_event)
agent.start("Use the search tool to find information about Python")

The user runs a tool-heavy prompt; streaming events mark when each tool finishes.

How It Works

Quick Start

Subscribe to Tool Events

Listen for tool execution completion events:

from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEvent, StreamEventType

agent = Agent(name="Assistant")

def on_stream_event(event):
    if event.type == StreamEventType.TOOL_CALL_END:
        print(f"Tool '{event.tool_call['name']}' completed")
        print(f"Duration: {event.metadata.get('duration_ms')}ms")

agent.stream_emitter.add_callback(on_stream_event)
agent.start("Use the search tool to find information about Python")

UI Integration

Use tool events for progress indicators in your UI:

class ToolProgressUI:
    def __init__(self):
        self.active_tools = {}
    
    def handle_stream_event(self, event):
        if event.type == StreamEventType.TOOL_CALL_START:
            self.active_tools[event.tool_call['id']] = {
                'name': event.tool_call['name'],
                'status': 'running'
            }
            self.update_progress_ui()
            
        elif event.type == StreamEventType.TOOL_CALL_END:
            tool_id = event.tool_call['id']
            if tool_id in self.active_tools:
                self.active_tools[tool_id]['status'] = 'completed'
                self.update_progress_ui()

Tool Event Types

TOOL_CALL_START

Emitted when tool execution begins with parsed arguments:

{
    "type": "tool_call_start",
    "timestamp": 1704153600.123,
    "tool_call": {
        "id": "tc_abc123",
        "name": "search_web", 
        "arguments": {"query": "Python tutorials"}
    },
    "metadata": {
        "agent_name": "Assistant"
    }
}

TOOL_CALL_END

The stream end marker - signals tool execution completion:

{
    "type": "tool_call_end",
    "timestamp": 1704153602.456,
    "tool_call": {
        "id": "tc_abc123",
        "name": "search_web",
        "result": "Found 10 results about Python tutorials"
    },
    "metadata": {
        "duration_ms": 2333.0,
        "success": True
    }
}

TOOL_CALL_RESULT

Final tool output after execution:

{
    "type": "tool_call_result", 
    "timestamp": 1704153602.460,
    "tool_call": {
        "id": "tc_abc123",
        "name": "search_web",
        "result": "Formatted search results..."
    }
}

Event Flow Sequence

Key Marker: TOOL_CALL_END is the completion signal for UI consumers. Use this to:

Hide loading spinners
Show completion checkmarks
Enable follow-up actions
Update progress bars

Multiple Tool Handling

When agents use multiple tools simultaneously:

class MultiToolTracker:
    def __init__(self):
        self.tools = {}
    
    def handle_event(self, event):
        if event.type == StreamEventType.TOOL_CALL_START:
            self.tools[event.tool_call['id']] = {
                'name': event.tool_call['name'],
                'start_time': event.timestamp,
                'status': 'running'
            }
            
        elif event.type == StreamEventType.TOOL_CALL_END:
            tool_id = event.tool_call['id']
            if tool_id in self.tools:
                self.tools[tool_id].update({
                    'end_time': event.timestamp,
                    'status': 'completed',
                    'duration': event.metadata.get('duration_ms', 0)
                })
                self.on_tool_completed(tool_id)
    
    def on_tool_completed(self, tool_id):
        tool = self.tools[tool_id]
        print(f"✅ {tool['name']} completed in {tool['duration']}ms")

Error Handling

Tools can fail during execution:

def handle_tool_events(event):
    if event.type == StreamEventType.TOOL_CALL_END:
        success = event.metadata.get('success', True)
        
        if success:
            print(f"✅ Tool {event.tool_call['name']} succeeded")
        else:
            error = event.metadata.get('error', 'Unknown error')
            print(f"❌ Tool {event.tool_call['name']} failed: {error}")
            
    elif event.type == StreamEventType.ERROR:
        if 'tool_call' in event.metadata:
            print(f"🚨 Tool execution error: {event.content}")

UI Integration Patterns

React Component

function ToolProgress({ agent }) {
  const [tools, setTools] = useState({});
  
  useEffect(() => {
    const handleEvent = (event) => {
      if (event.type === 'tool_call_start') {
        setTools(prev => ({
          ...prev,
          [event.tool_call.id]: {
            name: event.tool_call.name,
            status: 'running'
          }
        }));
      } else if (event.type === 'tool_call_end') {
        setTools(prev => ({
          ...prev, 
          [event.tool_call.id]: {
            ...prev[event.tool_call.id],
            status: 'completed'
          }
        }));
      }
    };
    
    agent.stream_emitter.add_callback(handleEvent);
    return () => agent.stream_emitter.remove_callback(handleEvent);
  }, [agent]);
  
  return (
    <div className="tool-progress">
      {Object.values(tools).map(tool => (
        <div key={tool.id} className={`tool ${tool.status}`}>
          {tool.status === 'running' && <Spinner />}
          {tool.status === 'completed' && <CheckIcon />}
          {tool.name}
        </div>
      ))}
    </div>
  );
}

Terminal Progress

import time
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn

console = Console()

class TerminalToolProgress:
    def __init__(self):
        self.progress = Progress(
            SpinnerColumn(),
            TextColumn("[bold blue]{task.description}"),
            console=console
        )
        self.tasks = {}
        self.progress.start()
    
    def handle_event(self, event):
        if event.type == StreamEventType.TOOL_CALL_START:
            task_id = self.progress.add_task(
                f"Running {event.tool_call['name']}", 
                total=None
            )
            self.tasks[event.tool_call['id']] = task_id
            
        elif event.type == StreamEventType.TOOL_CALL_END:
            tool_id = event.tool_call['id']
            if tool_id in self.tasks:
                task_id = self.tasks[tool_id]
                self.progress.update(
                    task_id,
                    description=f"✅ {event.tool_call['name']} completed",
                    completed=True
                )
                del self.tasks[tool_id]

Best Practices

Use TOOL_CALL_END as the completion signal

Don’t rely on TOOL_CALL_RESULT for completion detection. TOOL_CALL_END is the definitive marker:

# Good - reliable completion detection
if event.type == StreamEventType.TOOL_CALL_END:
    mark_tool_completed(event.tool_call['id'])

# Avoid - TOOL_CALL_RESULT may not always fire
if event.type == StreamEventType.TOOL_CALL_RESULT:
    mark_tool_completed(event.tool_call['id'])

Handle tool ID mapping

Track tool calls by their unique IDs, not just names:

# Tools may have same name but different IDs
tool_tracker = {}

def handle_event(event):
    tool_id = event.tool_call['id']  # Always use ID
    tool_name = event.tool_call['name']
    
    if event.type == StreamEventType.TOOL_CALL_START:
        tool_tracker[tool_id] = {'name': tool_name, 'start': time.time()}

Gracefully handle missing events

Network issues may cause missed events. Implement timeouts:

import time

class RobustToolTracker:
    def __init__(self, timeout=30):
        self.tools = {}
        self.timeout = timeout
    
    def cleanup_stale_tools(self):
        now = time.time()
        stale = [
            tool_id for tool_id, tool in self.tools.items()
            if tool['status'] == 'running' 
            and now - tool['start_time'] > self.timeout
        ]
        for tool_id in stale:
            self.tools[tool_id]['status'] = 'timeout'
            self.on_tool_timeout(tool_id)

Event Data Reference

TOOL_CALL_START Event

Field	Type	Description
`type`	`string`	Always `"tool_call_start"`
`timestamp`	`float`	High-precision timestamp
`tool_call.id`	`string`	Unique tool call identifier
`tool_call.name`	`string`	Tool function name
`tool_call.arguments`	`dict`	Parsed tool arguments

TOOL_CALL_END Event

Field	Type	Description
`type`	`string`	Always `"tool_call_end"`
`timestamp`	`float`	High-precision timestamp
`tool_call.id`	`string`	Tool call identifier
`tool_call.name`	`string`	Tool function name
`metadata.duration_ms`	`float`	Execution duration in milliseconds
`metadata.success`	`bool`	Whether execution succeeded
`metadata.error`	`string`	Error message if failed

TOOL_PROGRESS Event

TOOL_PROGRESS events carry incremental output emitted by a tool while it runs — before it returns a result. Tools call emit_tool_progress() to surface partial output without changing their signature.

Field	Type	Description
`type`	`string`	Always `"tool_progress"`
`timestamp`	`float`	High-precision timestamp
`content`	`string \| None`	Partial text output chunk (e.g. a stdout line)
`metadata.progress`	`float \| None`	Completion fraction 0.0–1.0
`metadata.stream`	`string \| None`	Optional stream tag, e.g. `"stderr"`

from praisonaiagents.streaming.events import StreamEvent, StreamEventType, emit_tool_progress

def on_event(event: StreamEvent) -> None:
    if event.type == StreamEventType.TOOL_PROGRESS:
        pct = (event.metadata or {}).get("progress")
        line = event.content or ""
        print(f"[{pct:.0%}] {line}" if pct is not None else line)

See Tool Progress Streaming for the full guide.

STREAM_UNAVAILABLE Event

The agent or model adapter emits STREAM_UNAVAILABLE when token streaming cannot run in the current configuration — for example when a provider falls back to a non-streaming response instead of failing silently. Treat it as an observable fallback signal for UIs and loggers.

Field	Type	Description
`type`	`string`	Always `"stream_unavailable"`
`timestamp`	`float`	High-precision timestamp
`error`	`string`	Human-readable summary (often `Streaming unavailable: …`)
`metadata.reason`	`string`	Provider-specific reason code or message

from praisonaiagents.streaming import StreamEventType

def on_event(event):
    if event.type == StreamEventType.STREAM_UNAVAILABLE:
        reason = (event.metadata or {}).get("reason") or event.error
        print(f"Streaming fallback: {reason}")
        # Continue with non-streaming UI or wait for DELTA_TEXT / final result

Agent Streaming

Core streaming concepts

Tool Progress Streaming

Stream incremental output from inside running tools

Tool Development

Creating custom tools

​How It Works

​Quick Start

​Tool Event Types

​TOOL_CALL_START

​TOOL_CALL_END

​TOOL_CALL_RESULT

​Event Flow Sequence

​Multiple Tool Handling

​Error Handling

​UI Integration Patterns

​React Component

​Terminal Progress

​Best Practices

​Event Data Reference

​TOOL_CALL_START Event

​TOOL_CALL_END Event

​TOOL_PROGRESS Event

​STREAM_UNAVAILABLE Event

​Related