What are Events?

Events are real-time notifications streamed asynchronously from the remote autonomous general AI agent as it executes tasks. They provide complete transparency into the general AI agent’s workflow, allowing you to monitor progress, debug issues, and build interactive applications that respond to the general AI agent’s activities in real-time. Unlike traditional polling-based systems, PandaAGI events are pushed from the remote agent immediately as they occur, ensuring minimal latency and maximum responsiveness.

Event Architecture

The PandaAGI SDK streams events from a remote autonomous general AI agent that operates independently. Events are generated as the general AI agent:
  • Connects to its execution environment
  • Performs web searches and navigation
  • Creates, reads, and modifies files
  • Executes shell commands and scripts
  • Provides progress updates and notifications
  • Completes tasks and reports results
from panda_agi.handlers import LogsHandler

# Events are processed by event handlers as the remote general AI agent works
handlers = [LogsHandler(use_colors=True, show_timestamps=True)]

response = await agent.run(
    "Research AI trends and create a report",
    event_handlers=handlers
)
print(f"Task completed: {response.output}")

Core Event Types

The PandaAGI SDK uses a comprehensive set of event types defined in the EventType enum. Here are all available event types:

Connection & Environment Events

Web Research Events

File System Events

Command Execution Events

Communication Events

Creative & Generation Events

Complete EventType Reference

from panda_agi import EventType

# All available event types:
EventType.AGENT_CONNECTION_SUCCESS  # Agent connected and workspace initialized
EventType.USER_NOTIFICATION        # Progress updates and notifications
EventType.USER_QUESTION            # Agent asking for user input/clarification
EventType.COMPLETED_TASK           # Task completion notification
EventType.WEB_SEARCH               # Web search initiated
EventType.WEB_SEARCH_RESULT        # Web search results received
EventType.WEB_NAVIGATION           # Navigating to specific webpage
EventType.WEB_NAVIGATION_RESULT    # Webpage content extracted
EventType.FILE_READ                # Reading file contents
EventType.FILE_WRITE               # Writing/creating files
EventType.FILE_REPLACE             # Replacing content in files
EventType.FILE_FIND                # Searching for files/content
EventType.FILE_EXPLORE             # Exploring directory structure
EventType.IMAGE_GENERATION         # Generating images/visual content
EventType.SHELL_EXEC               # Executing shell commands
EventType.SHELL_VIEW               # Viewing shell command output
EventType.SHELL_WRITE              # Writing shell scripts

Event Streaming

Using Event Handlers

Process events with custom handlers for better organization and reusability:
from panda_agi import Agent, BaseHandler
from panda_agi.handlers import LogsHandler
from panda_agi.envs import DockerEnv
import asyncio

async def main():
    agent_env = DockerEnv("./workspace")
    agent = Agent(environment=agent_env)

    # Use the built-in logs handler
    handlers = [LogsHandler(
        use_colors=True,
        show_timestamps=True,
        compact_mode=False
    )]

    # Run with event handler
    response = await agent.run(
        "Research machine learning trends and create a report",
        event_handlers=handlers
    )

    print(f"Task completed: {response.output}")
    await agent.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

Basic Streaming with Type Safety

Stream events in real-time as your remote general AI agent works through complex tasks:
from panda_agi import Agent
from panda_agi.envs import DockerEnv
from panda_agi.client.models import (
    # Type guard functions for type safety
    is_agent_connection_success_event,
    is_web_search_event,
    is_web_navigation_event,
    is_file_write_event,
    is_shell_exec_event,
    is_user_notification_event,
    is_completed_task_event,
    BaseStreamEvent,
)
from panda_agi.handlers import BaseHandler
import asyncio

async def main():
    # Initialize the remote general AI agent environment
    agent_env = DockerEnv("./workspace")
    agent = Agent(environment=agent_env)

    # Create custom handler to track different categories of activities with type safety
    class AdvancedMonitorHandler(BaseHandler):
        def process(self, event: BaseStreamEvent) -> None:
            # Type checker knows exactly what type each event is!
            
            if is_agent_connection_success_event(event):
                # Type checker knows event is AgentConnectionSuccessEvent here
                file_system = event.data.get("file_system", {})
                directory = file_system.get("directory", "unknown")
                print(f"🔗 Agent connected: {directory}")
            
            elif is_web_search_event(event):
                # Type checker knows event is WebSearchEvent here
                query = event.data.get("query", "")
                print(f"🔍 Searching: {query}")
            
            elif is_web_navigation_event(event):
                # Type checker knows event is WebNavigationEvent here
                url = event.data.get("url", "")
                print(f"🌐 Reading: {url}")
            
            elif is_file_write_event(event):
                # Type checker knows event is FileWriteEvent here
                filename = event.data.get("file", "")
                print(f"📝 Created: {filename}")
            
            elif is_shell_exec_event(event):
                # Type checker knows event is ShellExecEvent here
                command = event.data.get("command", "")
                print(f"⚙️ Executing: {command}")
            
            elif is_user_notification_event(event):
                # Type checker knows event is UserNotificationEvent here
                text = event.data.get("text", "")
                attachments = event.data.get("attachments", [])
                print(f"📢 Update: {text}")
                if attachments:
                    print(f"📎 Files: {', '.join(attachments)}")
            
            elif is_completed_task_event(event):
                # Type checker knows event is CompletedTaskEvent here
                print("🎉 Task completed!")

    # Use the custom handler
    handlers = [AdvancedMonitorHandler("AdvancedMonitor")]
    response = await agent.run(
        "Analyze codebase and generate documentation",
        event_handlers=handlers
    )
    print(f"Documentation generation completed: {response.output}")
    
    await agent.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

Custom Event Handler Patterns

Create specialized handlers for different use cases:
from panda_agi import Agent, BaseHandler, FilterHandler, CompositeHandler
from panda_agi.client.models import BaseStreamEvent
from panda_agi.handlers import LogsHandler
import asyncio

# 1. Statistics Handler - Track event metrics
class StatsHandler(BaseHandler):
    def __init__(self, name: str = "StatsHandler"):
        super().__init__(name)
        self.event_counts = {}

    def process(self, event: BaseStreamEvent) -> None:
        event_type = event.type
        self.event_counts[event_type] = self.event_counts.get(event_type, 0) + 1

    def print_summary(self):
        print("\n📊 Event Statistics:")
        for event_type, count in self.event_counts.items():
            print(f"  {event_type}: {count}")

# 2. File-only Handler - Only process file events
class FileEventsHandler(FilterHandler):
    def should_process(self, event: BaseStreamEvent) -> bool:
        return event.type.startswith('file_')

    def process_filtered_event(self, event: BaseStreamEvent) -> None:
        print(f"📁 File activity: {event.type} - {event.data.get('file', 'N/A')}")

# 3. Multiple Handlers - Multiple handlers working together
async def demo_handlers():
    agent = Agent()

    # Create multiple handlers
    stats_handler = StatsHandler("EventStats")
    handlers = [
        LogsHandler("EventLogger"),
        stats_handler,
        FileEventsHandler("FileTracker")
    ]

    response = await agent.run(
        "Analyze project files and create documentation",
        event_handlers=handlers
    )

    # Access individual handler results
    stats_handler.print_summary()

    await agent.disconnect()

if __name__ == "__main__":
    asyncio.run(demo_handlers())

Creating Events Programmatically

You can also create typed events programmatically for testing or custom workflows:
from panda_agi.client.models import (
    UserNotificationEvent,
    FileWriteEvent,
    WebSearchEvent,
    StreamEvent,
    is_user_notification_event,
    is_file_write_event,
)

# Create a typed notification event
notification = UserNotificationEvent(
    data={
        "text": "Analysis complete!",
        "attachments": ["report.md", "charts.png"]
    }
)

# Create a typed file event
file_event = FileWriteEvent(
    data={
        "file": "analysis.md",
        "content": "# Analysis Results\n\nFindings...",
        "append": False
    }
)

# Events are fully typed and can be used in unions
events: list[StreamEvent] = [notification, file_event]

# Type-safe event processing
for event in events:
    if is_user_notification_event(event):
        print(f"Notification: {event.data['text']}")
    elif is_file_write_event(event):
        print(f"File written: {event.data['file']}")

Next Steps