Observability API

observability

Observability primitives for orqest agents.

Provides tracing (Span, Tracer, JSONTracer) and event-driven observability (AgentEvent, EventBus) with zero external dependencies.

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

AgentEvent dataclass

AgentEvent(
    event_type: str,
    agent_name: str,
    timestamp: datetime = (lambda: datetime.now(tz=UTC))(),
    data: dict[str, Any] = dict(),
    span_id: str | None = None,
    trace_id: str | None = None,
)

Event emitted during agent execution.

Carries the event type, originating agent, and an arbitrary data payload. Optionally linked to a trace span for correlation.

EventBus

EventBus()

In-process pub/sub for AgentEvents.

Fire-and-forget: handler errors are logged at WARNING level and never re-raised, so a broken handler cannot disrupt agent execution. Supports both sync and async handlers.

Initialize with empty handler registries.

Source code in orqest/observability/events.py

def __init__(self) -> None:
    """Initialize with empty handler registries."""
    self._handlers: dict[str, list[EventHandler]] = {}
    self._global_handlers: list[EventHandler] = []

subscribe

subscribe(event_type: str, handler: EventHandler) -> None

Register a handler for a specific event type.

Source code in orqest/observability/events.py

def subscribe(self, event_type: str, handler: EventHandler) -> None:
    """Register a handler for a specific event type."""
    self._handlers.setdefault(event_type, []).append(handler)

subscribe_all

subscribe_all(handler: EventHandler) -> None

Register a handler that receives every event regardless of type.

Source code in orqest/observability/events.py

def subscribe_all(self, handler: EventHandler) -> None:
    """Register a handler that receives every event regardless of type."""
    self._global_handlers.append(handler)

unsubscribe

unsubscribe(event_type: str, handler: EventHandler) -> None

Remove a handler for a specific event type.

Silently ignores handlers that are not registered.

Source code in orqest/observability/events.py

def unsubscribe(self, event_type: str, handler: EventHandler) -> None:
    """Remove a handler for a specific event type.

    Silently ignores handlers that are not registered.
    """
    handlers = self._handlers.get(event_type)
    if handlers is None:
        return
    try:
        handlers.remove(handler)
    except ValueError:
        pass

unsubscribe_all

unsubscribe_all(handler: EventHandler) -> None

Remove a handler registered via :meth:subscribe_all.

Silently ignores handlers that are not registered.

Source code in orqest/observability/events.py

def unsubscribe_all(self, handler: EventHandler) -> None:
    """Remove a handler registered via :meth:`subscribe_all`.

    Silently ignores handlers that are not registered.
    """
    try:
        self._global_handlers.remove(handler)
    except ValueError:
        pass

emit async

emit(event: AgentEvent) -> None

Dispatch an event to all matching handlers.

Calls type-specific handlers first, then global handlers. Each handler is invoked independently — a failure in one does not prevent others from running.

Source code in orqest/observability/events.py

async def emit(self, event: AgentEvent) -> None:
    """Dispatch an event to all matching handlers.

    Calls type-specific handlers first, then global handlers.
    Each handler is invoked independently — a failure in one does not
    prevent others from running.
    """
    targets = list(self._handlers.get(event.event_type, []))
    targets.extend(self._global_handlers)
    for handler in targets:
        await self._safe_call(handler, event)

JSONTracer

JSONTracer()

Default tracer — stores spans in memory, exports to JSON.

Thread-safe for single-writer scenarios (typical async usage). No external dependencies.

Initialize with an empty span store.

Source code in orqest/observability/tracer.py

def __init__(self) -> None:
    """Initialize with an empty span store."""
    self._spans: list[Span] = []

start_span

start_span(
    name: str,
    *,
    agent_name: str = "",
    parent: Span | None = None,
) -> Span

Create and register a new span.

If a parent is provided, the child inherits its trace_id. Otherwise a new trace_id is generated (root span).

Source code in orqest/observability/tracer.py

def start_span(
    self, name: str, *, agent_name: str = "", parent: Span | None = None
) -> Span:
    """Create and register a new span.

    If a parent is provided, the child inherits its trace_id.
    Otherwise a new trace_id is generated (root span).
    """
    trace_id = parent.trace_id if parent else uuid.uuid4().hex
    span = Span(
        trace_id=trace_id,
        span_id=uuid.uuid4().hex,
        parent_span_id=parent.span_id if parent else None,
        name=name,
        agent_name=agent_name,
        started_at=datetime.now(tz=UTC),
    )
    self._spans.append(span)
    return span

end_span

end_span(
    span: Span,
    *,
    status: str = "ok",
    attributes: dict[str, Any] | None = None,
) -> None

Mark a span as finished, computing its duration.

Merges any extra attributes into the span.

Source code in orqest/observability/tracer.py

def end_span(
    self,
    span: Span,
    *,
    status: str = "ok",
    attributes: dict[str, Any] | None = None,
) -> None:
    """Mark a span as finished, computing its duration.

    Merges any extra attributes into the span.
    """
    span.ended_at = datetime.now(tz=UTC)
    span.duration_ms = (
        (span.ended_at - span.started_at).total_seconds() * 1000.0
    )
    span.status = status  # type: ignore[assignment]
    if attributes:
        span.attributes.update(attributes)

get_spans

get_spans() -> list[Span]

Return all recorded spans in insertion order.

Source code in orqest/observability/tracer.py

def get_spans(self) -> list[Span]:
    """Return all recorded spans in insertion order."""
    return list(self._spans)

export_json

export_json() -> list[dict[str, Any]]

Serialize all spans to JSON-safe dicts.

Source code in orqest/observability/tracer.py

def export_json(self) -> list[dict[str, Any]]:
    """Serialize all spans to JSON-safe dicts."""
    out: list[dict[str, Any]] = []
    for s in self._spans:
        out.append({
            "trace_id": s.trace_id,
            "span_id": s.span_id,
            "parent_span_id": s.parent_span_id,
            "name": s.name,
            "agent_name": s.agent_name,
            "started_at": s.started_at.isoformat(),
            "ended_at": s.ended_at.isoformat() if s.ended_at else None,
            "duration_ms": s.duration_ms,
            "status": s.status,
            "attributes": s.attributes,
            "events": s.events,
        })
    return out

clear

clear() -> None

Remove all recorded spans.

Source code in orqest/observability/tracer.py

def clear(self) -> None:
    """Remove all recorded spans."""
    self._spans.clear()

Span dataclass

Span(
    trace_id: str,
    span_id: str,
    parent_span_id: str | None,
    name: str,
    agent_name: str,
    started_at: datetime,
    ended_at: datetime | None = None,
    duration_ms: float | None = None,
    status: Literal["ok", "error"] = "ok",
    attributes: dict[str, Any] = dict(),
    events: list[dict[str, Any]] = list(),
)

A single unit of work within a trace.

Spans form a tree via parent_span_id. A root span has parent_span_id=None and defines the trace_id that child spans inherit.

Tracer

Protocol for trace collection backends.

start_span

start_span(
    name: str,
    *,
    agent_name: str = "",
    parent: Span | None = None,
) -> Span

Create and register a new span.

Source code in orqest/observability/tracer.py

def start_span(
    self,
    name: str,
    *,
    agent_name: str = "",
    parent: Span | None = None,
) -> Span:
    """Create and register a new span."""
    ...

end_span

end_span(
    span: Span,
    *,
    status: str = "ok",
    attributes: dict[str, Any] | None = None,
) -> None

Mark a span as finished.

Source code in orqest/observability/tracer.py

def end_span(
    self,
    span: Span,
    *,
    status: str = "ok",
    attributes: dict[str, Any] | None = None,
) -> None:
    """Mark a span as finished."""
    ...

get_spans

get_spans() -> list[Span]

Return all recorded spans.

Source code in orqest/observability/tracer.py

def get_spans(self) -> list[Span]:
    """Return all recorded spans."""
    ...