Event Bus Module

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

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

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

Both publish() / publish_event() and publish_async() always record into history, whether or not subscribers are attached — get_history() returns every event you publish (capped at the 1000 most recent). The fast path still skips the subscriber walk when nothing is listening, so publishing to a bare bus stays cheap. Wrap expensive payload construction in a has_subscribers check, since that work is genuinely skipped when nothing is listening. Changed for sync in PR #4958; the async parity fix landed in PR #5258.

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

Durable Persistence

Attach a durable sink to persist every published event to SQLite across processes. In-memory get_history() already records every event; a durable sink adds cross-process persistence on top. See Durable Event Log for the full story.

from praisonaiagents.bus import get_default_bus, SqliteEventLog

bus = get_default_bus()
bus.attach_sink(SqliteEventLog())

EventBus.attach_sink(sink) / detach_sink(sink) -> bool / has_sinks manage sinks; a sink is any object implementing EventLogProtocol.append(event).

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)
# Even with no subscribers attached, this is still recorded to get_history()
# (was silently dropped on the async path before PR #5258).
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: Every published event is recorded, whether or not subscribers are registered — no need to subscribe first. History is capped at the 1000 most recent events.

from praisonaiagents.bus import EventBus, EventType

bus = EventBus()                                     # no subscribers needed

bus.publish(EventType.CUSTOM, data={"index": 1})
bus.publish(EventType.CUSTOM, data={"index": 2})

recent = bus.get_history(limit=5)                    # -> two events

For cross-process persistence, attach a durable sink as well (see Durable Event Log).

Best Practices

Check has_subscribers before heavy work

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

get_history() works without subscribers

Every published event lands in history regardless of subscribers, so get_history() is reliable for after-the-fact debugging on a bare bus. Since PR #4958 you no longer need a no-op subscriber to capture a record on the sync path; PR #5258 closed the same gap for publish_async(), so async lifecycle events fired before an observability subscriber attaches are no longer lost. For cross-process persistence, attach a durable sink (see Durable Event Log).

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.

Durable Event Log

Persist events to SQLite and query per session

Hook Events

Hook lifecycle events alongside the bus

Spawn & Announce

Sub-agent events and coordination

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

How It Works

Quick Start

Event Types

Durable Persistence

Common Patterns

Best Practices

Durable Event Log

Hook Events

Spawn & Announce

​How It Works

​Quick Start

​Event Types

​Durable Persistence

​Common Patterns

​Best Practices

​Related

Durable Event Log

Hook Events

Spawn & Announce

How It Works

Quick Start

Event Types

Durable Persistence

Common Patterns

Best Practices

Related