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

Event Types

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

If an async subscriber raises when publish() is called inside a running event loop, the bus logs it as Error in subscriber <id>: ... (the same prefix sync-subscriber errors use), plus the traceback. Previously it surfaced only as Python’s Task exception was never retrieved warning at GC time.

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

Related