bus

AI Agent Event Bus Module for PraisonAI Agents. Provides a typed publish/subscribe event system for real-time communication between components. Extends the existing hooks system with a more general event-driven architecture. Features:

Typed event definitions with dataclass payloads
Sync and async subscribers
Event filtering by type
Global and scoped event buses
SSE-compatible event streaming

Zero Performance Impact:

All imports are lazy loaded
No overhead when not subscribed
Optional dependency for server features

Usage:

from praisonaiagents.bus import EventBus, Event

# Create event bus
bus = EventBus()

# Subscribe to events
@bus.on("session.created")
def handle_session(event):
    print(f"Session created: {event.data}")

# Publish events
bus.publish("session.created", {"session_id": "abc123"})

Import

from praisonaiagents import bus

Constants

Name	Value
`_LAZY_IMPORTS`	`{'EventBus': ('praisonaiagents.bus.bus', 'EventBus'), 'Event': ('praisonaiagents.bus.event', 'Event'), 'EventType': ('praisonaiagents.bus.event', 'EventType'), 'Subscriber': ('praisonaiagents.bus.bus'...`