Module

tools.events

Tool call event bus — async broadcast for real-time dashboards.

When an MCP tool is invoked, aToolCallEventis emitted through the ToolEventBus. SSE routes subscribe to receive events as they happen, enabling live agent-activity dashboards.

Free-threading safety:

  • ToolCallEvent is a frozen dataclass (immutable, safe to share)
  • ToolEventBus uses a Lock to protect the subscriber set
  • Each subscriber gets its own asyncio.Queue (no shared mutable state)
2Classes

Classes

ToolCallEvent 5 ▼
A single tool invocation event. Emitted by the ``ToolRegistry`` after each successful tool call. C…

A single tool invocation event.

Emitted by theToolRegistryafter each successful tool call. Consumed by SSE routes for real-time agent dashboards.

Attributes

Name Type Description
tool_name str —
arguments dict[str, Any] —
result Any —
timestamp float —
call_id str —
ToolEventBus 4 ▼
Async broadcast channel for tool call events. Each call to ``subscribe()`` returns an async iterat…

Async broadcast channel for tool call events.

Each call tosubscribe()returns an async iterator backed by its ownasyncio.Queue. When emit()is called, the event is placed into every active subscriber's queue.

Usage in SSE routes::

async def stream():
    async for event in app.tool_events.subscribe():
        yield Fragment("dashboard.html", "row", event=event)
return EventStream(stream())

Methods

emit 1 ▼
Broadcast an event to all active subscribers.
async
async def emit(self, event: ToolCallEvent) -> None
Parameters
Name Type Description
event —
subscribe 0 AsyncIterator[ToolCallEv… ▼
Subscribe to tool call events. Returns an async iterator that yields events as…
async
async def subscribe(self) -> AsyncIterator[ToolCallEvent]

Subscribe to tool call events.

Returns an async iterator that yields events as they are emitted. The subscription is automatically cleaned up when the iterator exits.

Returns
AsyncIterator[ToolCallEvent]
close 0 ▼
Signal all subscribers to stop. Puts ``None`` into every queue, which causes t…
def close(self) -> None

Signal all subscribers to stop.

PutsNoneinto every queue, which causes the async iterator to break cleanly.

Internal Methods 1 ▼
__init__ 0 ▼
def __init__(self) -> None