Event Bus Module

Typed publish-subscribe events for agents, tools, and sessions

Subscribe to agent lifecycle and tool events with a typed event bus — zero cost when nothing is listening.

from praisonaiagents import Agent
from praisonaiagents.bus import EventBus, EventType

bus = EventBus()

def on_tool_done(event):
    print(f"Tool finished: {event.data}")

bus.subscribe(on_tool_done, event_types=EventType.TOOL_COMPLETED)

agent = Agent(name="assistant", instructions="Be helpful")
agent.start("Say hello briefly.")

The user runs the agent; tool and lifecycle events publish to subscribers for logging, UI updates, or custom automation.

How It Works

Quick Start

1. Subscribe and publish

   from praisonaiagents.bus import EventBus, EventType
   
   bus = EventBus()
   
   def on_message(event):
       print(f"Received: {event.data}")
   
   bus.subscribe(on_message, event_types=EventType.MESSAGE_CREATED)
   
   bus.publish(
       EventType.MESSAGE_CREATED,
       data={"text": "Hello, World!"},
       source="demo",
   )
   
2. Global bus with an agent

   from praisonaiagents import Agent
   from praisonaiagents.bus import get_default_bus, EventType
   
   bus = get_default_bus()
   
   bus.subscribe(
       lambda e: print(f"Agent started: {e.data}"),
       event_types=EventType.AGENT_STARTED,
   )
   bus.subscribe(
       lambda e: print(f"Tool used: {e.data}"),
       event_types=EventType.TOOL_COMPLETED,
   )
   
   agent = Agent(name="Assistant", instructions="Be helpful")
   agent.start("Hello!")
   

publish() returns immediately when there are no subscribers — no lock, no UUID, no history. Wrap expensive payload construction in a has_subscribers check.

---

Event Types

Event Type	Description
SESSION_CREATED / UPDATED / DELETED / FORKED	Session lifecycle
MESSAGE_CREATED	New message added
TOOL_STARTED / TOOL_COMPLETED	Tool execution
AGENT_STARTED / AGENT_COMPLETED	Agent execution
SUBAGENT_SPAWNED / COMPLETED / ERROR	Sub-agent coordination
SNAPSHOT_CREATED	File snapshot
COMPACTION_COMPLETED	Context compaction
CUSTOM	Application-defined events

---

Common Patterns

Async subscriber:

import asyncio
from praisonaiagents.bus import EventBus, EventType

bus = EventBus()

async def async_handler(event):
    await asyncio.sleep(0.1)
    print(event.data)

bus.subscribe(async_handler, event_types=EventType.TOOL_COMPLETED)
await bus.publish_async(EventType.TOOL_COMPLETED, data={"tool": "bash"})

Guard expensive payloads:

from praisonaiagents.bus import get_default_bus, EventType

bus = get_default_bus()

if bus.has_subscribers:
    bus.publish(
        EventType.CUSTOM,
        data={"snippet": expensive_summarise(text)},
        source="memory",
    )

Event history:

bus = EventBus(history_size=1000)
bus.publish(EventType.CUSTOM, data={"index": 1})
recent = bus.get_history(limit=5)

---

Best Practices

Check has_subscribers before heavy work

Building summaries or embeddings for events nobody listens to wastes CPU — guard with bus.has_subscribers.

Filter by event_types

Pass event_types= to subscribe() so handlers only run for relevant events.

Use get_default_bus for cross-component wiring

The shared default bus lets agents, memory, and hooks emit events without passing a bus instance everywhere.

Prefer sync handlers unless you need async

Sync callbacks run inline; async handlers are awaited during publish_async only.

---

Related

Hook Events

Hook lifecycle events alongside the bus

Spawn & Announce

Sub-agent events and coordination