Skip to main content
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 TypeDescription
SESSION_CREATED / UPDATED / DELETED / FORKEDSession lifecycle
MESSAGE_CREATEDNew message added
TOOL_STARTED / TOOL_COMPLETEDTool execution
AGENT_STARTED / AGENT_COMPLETEDAgent execution
SUBAGENT_SPAWNED / COMPLETED / ERRORSub-agent coordination
SNAPSHOT_CREATEDFile snapshot
COMPACTION_COMPLETEDContext compaction
CUSTOMApplication-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

Building summaries or embeddings for events nobody listens to wastes CPU — guard with bus.has_subscribers.
Pass event_types= to subscribe() so handlers only run for relevant events.
The shared default bus lets agents, memory, and hooks emit events without passing a bus instance everywhere.
Sync callbacks run inline; async handlers are awaited during publish_async only.

Hook Events

Hook lifecycle events alongside the bus

Spawn & Announce

Sub-agent events and coordination