Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.praison.ai/llms.txt

Use this file to discover all available pages before exploring further.

Track tool execution in real-time with streaming events that provide completion markers for UI rendering.

​
Quick Start

1

Subscribe to Tool Events

Listen for tool execution completion events:
from praisonaiagents import Agent
from praisonaiagents.streaming.events 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")
        print(f"Duration: {event.metadata.get('duration_ms')}ms")

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

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.setStreamCallback(handleEvent);
    return () => agent.removeStreamCallback(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

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'])
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()}
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

FieldTypeDescription
typestringAlways "tool_call_start"
timestampfloatHigh-precision timestamp
tool_call.idstringUnique tool call identifier
tool_call.namestringTool function name
tool_call.argumentsdictParsed tool arguments

​
TOOL_CALL_END Event

FieldTypeDescription
typestringAlways "tool_call_end"
timestampfloatHigh-precision timestamp
tool_call.idstringTool call identifier
tool_call.namestringTool function name
metadata.duration_msfloatExecution duration in milliseconds
metadata.successboolWhether execution succeeded
metadata.errorstringError message if failed

Agent Streaming

Core streaming concepts

Tool Development

Creating custom tools