Events

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}")

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

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())

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())

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())

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