EventBusPublishHook — bridge tool lifecycle to the EventBus

Orqest exposes two complementary observability primitives: a HookRunner that fires before/after a tool runs, and an EventBus that fans events out to any subscriber. Until now, wiring one to the other was boilerplate — every consumer that wanted "publish a structured event for each tool call" had to re-implement the same await bus.emit(AgentEvent(...)) adapter.

EventBusPublishHook removes that friction. Register it on a HookRunner and every compound tool call yields three events on the bus (tool.before, tool.after, tool.error) with consistent payloads. Subscribers — an SSE sidecar, a metrics exporter, a UI timeline — get a uniform stream without coupling to HookRunner internals.

Minimal example

import asyncio

from orqest.hooks import HookRunner
from orqest.observability import AgentEvent, EventBus, EventBusPublishHook


async def main() -> None:
    bus = EventBus()

    def log(event: AgentEvent) -> None:
        print(f"[{event.event_type}] {event.data}")

    bus.subscribe_all(log)

    runner = HookRunner([EventBusPublishHook(bus, agent_name="demo")])

    await runner.run_before("do_work", {"task": "x"}, state=None)
    await runner.run_after("do_work", {"task": "x"}, result="ok", state=None, duration_ms=12.3)


asyncio.run(main())

Emits:

[tool.before] {'tool_name': 'do_work', 'args': {'task': 'x'}}
[tool.after]  {'tool_name': 'do_work', 'duration_ms': 12.3, 'result_preview': 'ok', 'result_len': 2}

When to reach for this

• You want tool calls to show up in a dashboard, trace viewer, or the Vercel AI frontend without writing a custom hook.
• You have multiple subscribers (metrics, SSE, logging) — the bus lets them fan out without the hook knowing about any of them.
• You want tool events to carry consistent session/project metadata — EventBusPublishHook reads state.session_id / state.project_id via getattr, so any state object that exposes those fields works.

Notes

• Fire-and-forget contract: both HookRunner and EventBus swallow handler errors. A broken subscriber can never break a tool call.
• Payload bounding: long tool results are truncated to result_preview_chars (default 400) with a ... suffix. Full length is reported separately as result_len so consumers can show "result was 12 kB, preview shows first 400 chars".
• Structural hook: the class satisfies the ToolHook protocol structurally — no inheritance required.

Reference

EventBusPublishHook

EventBusPublishHook(
    bus: EventBus,
    *,
    agent_name: str = "unknown",
    result_preview_chars: int = _RESULT_PREVIEW_CHARS,
)

Publish tool lifecycle to an :class:EventBus.

Structural ToolHook — implements before_tool, after_tool, and on_error, so it satisfies :class:orqest.hooks.ToolHook without inheriting from the Protocol.

Wire the hook to bus and tag every published event with agent_name.

Parameters:

Name	Type	Description	Default
bus	EventBus	The :class:EventBus to publish to.	required
agent_name	str	Value used as AgentEvent.agent_name on every emitted event. Consumers usually pass the owning orchestrator's name.	'unknown'
result_preview_chars	int	Max characters from tool results included in tool.after payloads. Long results are truncated to keep event payloads bounded.	_RESULT_PREVIEW_CHARS

Source code in orqest/observability/event_bus_hook.py

def __init__(
    self,
    bus: EventBus,
    *,
    agent_name: str = "unknown",
    result_preview_chars: int = _RESULT_PREVIEW_CHARS,
) -> None:
    """Wire the hook to *bus* and tag every published event with
    ``agent_name``.

    Args:
        bus: The :class:`EventBus` to publish to.
        agent_name: Value used as ``AgentEvent.agent_name`` on every
            emitted event. Consumers usually pass the owning
            orchestrator's name.
        result_preview_chars: Max characters from tool results
            included in ``tool.after`` payloads. Long results are
            truncated to keep event payloads bounded.
    """
    self._bus = bus
    self._agent_name = agent_name
    self._result_preview_chars = result_preview_chars

before_tool async

before_tool(
    tool_name: str, args: dict[str, Any], state: Any
) -> None

Publish tool.before to the bus.

Source code in orqest/observability/event_bus_hook.py

async def before_tool(
    self, tool_name: str, args: dict[str, Any], state: Any
) -> None:
    """Publish ``tool.before`` to the bus."""
    await self._bus.emit(
        AgentEvent(
            event_type="tool.before",
            agent_name=self._agent_name,
            data={
                "tool_name": tool_name,
                "args": _summarize_args(args),
                **_state_meta(state),
            },
        )
    )

after_tool async

after_tool(
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> None

Publish tool.after to the bus.

Source code in orqest/observability/event_bus_hook.py

async def after_tool(
    self,
    tool_name: str,
    args: dict[str, Any],
    result: Any,
    state: Any,
    duration_ms: float,
) -> None:
    """Publish ``tool.after`` to the bus."""
    preview = _truncate(result, self._result_preview_chars)
    await self._bus.emit(
        AgentEvent(
            event_type="tool.after",
            agent_name=self._agent_name,
            data={
                "tool_name": tool_name,
                "duration_ms": round(duration_ms, 2),
                "result_preview": preview,
                "result_len": len(str(result)) if result is not None else 0,
                **_state_meta(state),
            },
        )
    )

on_error async

on_error(
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> None

Publish tool.error to the bus.

Source code in orqest/observability/event_bus_hook.py

async def on_error(
    self,
    tool_name: str,
    args: dict[str, Any],
    error: Exception,
    state: Any,
) -> None:
    """Publish ``tool.error`` to the bus."""
    await self._bus.emit(
        AgentEvent(
            event_type="tool.error",
            agent_name=self._agent_name,
            data={
                "tool_name": tool_name,
                "error_type": type(error).__name__,
                "error_message": str(error),
                **_state_meta(state),
            },
        )
    )