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 on its owning event loop
- Cross-thread emitters schedule queue mutation with call_soon_threadsafe
tools.events
| Name | Type | Default | Description |
|---|---|---|---|
type
|
|
— | |
qualified_name
|
|
— | |
element_type
|
|
— | |
description
|
|
— | |
source_file
|
|
— | |
line_number
|
|
— | |
is_autodoc
|
|
— | |
autodoc_element
|
|
— | |
_autodoc_template
|
|
— | |
_autodoc_url_path
|
|
— | |
_autodoc_page_type
|
|
— | |
title
|
|
— | |
doc_content_hash
|
|
— |
Symbols on this page
A single tool invocation event.
Emitted by theToolRegistryafter each successful tool call.
Consumed by SSE routes for real-time agent dashboards.
Async broadcast channel for tool call events.
Each call tosubscribe()returns an async iterator backed by its
ownasyncio.Queue. When emit()is…
ToolCallEvent
class
A single tool invocation event.
Emitted by theToolRegistryafter each successful tool call.
Consumed by SSE routes for real-time agent dashboards.
ToolEventBus
class
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())
View source · /home/runner/work/chirp/chirp/site/../src/chirp/tools/events.py:1