Observability

Hooks and telemetry sinks. See the Observability guide.

agentflow.observability

Observability: lifecycle hooks and basic run metrics.

Attach a :class:Hooks implementation at compile(hooks=...) to observe a run without changing its behavior. All callbacks are optional; subclass :class:Hooks and override only what you need. Hooks are awaited, so they can do async I/O (emit to a log sink, push a metric), but they must not raise — an exception in a hook is swallowed so instrumentation can never break a run.

:class:RunMetrics is a ready-made :class:Hooks that records per-node durations, counts, and errors, plus step and run totals.

Hooks

No-op base for run lifecycle callbacks. Override selectively.

Every method is async and must not raise. Timing is the runtime's concern; hooks receive already-measured values where relevant.

Source code in agentflow/observability.py

class Hooks:
    """No-op base for run lifecycle callbacks. Override selectively.

    Every method is async and must not raise. Timing is the runtime's concern;
    hooks receive already-measured values where relevant.
    """

    async def on_run_start(self, thread: str, step: int) -> None: ...
    async def on_node_start(self, thread: str, step: int, node: str) -> None: ...
    async def on_node_end(self, thread: str, step: int, node: str, seconds: float) -> None: ...
    async def on_node_error(
        self, thread: str, step: int, node: str, exc: BaseException
    ) -> None: ...
    async def on_step_end(self, thread: str, step: int, ran: tuple[str, ...]) -> None: ...
    async def on_run_end(self, thread: str, step: int, completed: bool) -> None: ...
    #: A backend event a node surfaced via ``ctx.emit`` (TextChunk, ToolCall,
    #: ToolResult, ...). Fired best-effort; ordering relative to node
    #: callbacks is not guaranteed since emit is synchronous.
    async def on_event(self, thread: str, step: int, node: str, event: Any) -> None: ...

RunMetrics dataclass

Bases: Hooks

A Hooks implementation that accumulates per-node and per-run metrics.

Source code in agentflow/observability.py

@dataclass
class RunMetrics(Hooks):
    """A Hooks implementation that accumulates per-node and per-run metrics."""

    nodes: dict[str, NodeStat] = field(default_factory=lambda: defaultdict(NodeStat))
    steps: int = 0
    started: bool = False
    completed: bool = False

    async def on_run_start(self, thread: str, step: int) -> None:
        self.started = True

    async def on_node_end(self, thread: str, step: int, node: str, seconds: float) -> None:
        stat = self.nodes[node]
        stat.calls += 1
        stat.total_seconds += seconds
        stat.max_seconds = max(stat.max_seconds, seconds)

    async def on_node_error(self, thread: str, step: int, node: str, exc: BaseException) -> None:
        self.nodes[node].errors += 1

    async def on_step_end(self, thread: str, step: int, ran: tuple[str, ...]) -> None:
        self.steps += 1

    async def on_run_end(self, thread: str, step: int, completed: bool) -> None:
        self.completed = completed

    def summary(self) -> dict[str, Any]:
        return {
            "steps": self.steps,
            "completed": self.completed,
            "nodes": {
                name: {
                    "calls": s.calls,
                    "errors": s.errors,
                    "avg_seconds": round(s.avg_seconds, 6),
                    "max_seconds": round(s.max_seconds, 6),
                }
                for name, s in self.nodes.items()
            },
        }

NodeStat dataclass

Aggregated stats for one node across a run.

Source code in agentflow/observability.py

@dataclass
class NodeStat:
    """Aggregated stats for one node across a run."""

    calls: int = 0
    errors: int = 0
    total_seconds: float = 0.0
    max_seconds: float = 0.0

    @property
    def avg_seconds(self) -> float:
        return self.total_seconds / self.calls if self.calls else 0.0

agentflow.telemetry

Persistent telemetry as a :class:~agentflow.observability.Hooks.

Attach at compile(hooks=...) to durably record a run. Two pieces:

• :class:MultiHooks composes several Hooks so a run can, say, collect in-memory metrics AND write a JSONL log AND push a custom sink at once.
• :class:JsonlTelemetry writes one JSON-per-line event stream to a file: run_start, node_start, node_end, node_error, event (backend events surfaced via ctx.emit), step, run_end. Each line is a flat object with a UTC timestamp, thread, and step.

Nothing here is required by the engine; it is one more optional listener.

MultiHooks

Bases: Hooks

Fan every lifecycle callback out to several :class:Hooks.

Each child is invoked through the same swallow-exceptions guarantee as a single hook, so one misbehaving listener cannot break the others or the run.

Source code in agentflow/telemetry.py

class MultiHooks(Hooks):
    """Fan every lifecycle callback out to several :class:`Hooks`.

    Each child is invoked through the same swallow-exceptions guarantee as a
    single hook, so one misbehaving listener cannot break the others or the run.
    """

    def __init__(self, *hooks: Hooks) -> None:
        self._hooks: tuple[Hooks, ...] = tuple(hooks)

    def add(self, hook: Hooks) -> MultiHooks:
        self._hooks += (hook,)
        return self

    async def on_run_start(self, thread, step):
        for h in self._hooks:
            await _safe(h.on_run_start(thread, step))

    async def on_node_start(self, thread, step, node):
        for h in self._hooks:
            await _safe(h.on_node_start(thread, step, node))

    async def on_node_end(self, thread, step, node, seconds):
        for h in self._hooks:
            await _safe(h.on_node_end(thread, step, node, seconds))

    async def on_node_error(self, thread, step, node, exc):
        for h in self._hooks:
            await _safe(h.on_node_error(thread, step, node, exc))

    async def on_step_end(self, thread, step, ran):
        for h in self._hooks:
            await _safe(h.on_step_end(thread, step, ran))

    async def on_run_end(self, thread, step, completed):
        for h in self._hooks:
            await _safe(h.on_run_end(thread, step, completed))

    async def on_event(self, thread, step, node, event):
        for h in self._hooks:
            await _safe(h.on_event(thread, step, node, event))

JsonlTelemetry

Bases: Hooks

Append run events to a JSON Lines file.

Pass a directory to get <dir>/<thread>.jsonl per run thread, or a file path to write everything to one file. Writes are appended under an async lock so concurrent nodes in a super-step don't interleave partial lines.

Source code in agentflow/telemetry.py

class JsonlTelemetry(Hooks):
    """Append run events to a JSON Lines file.

    Pass a directory to get ``<dir>/<thread>.jsonl`` per run thread, or a file
    path to write everything to one file. Writes are appended under an async
    lock so concurrent nodes in a super-step don't interleave partial lines.
    """

    def __init__(
        self,
        path: Path | str,
        *,
        per_thread: bool | None = None,
        redact: Redactor | None = None,
        secure_permissions: bool = True,
    ) -> None:
        p = Path(path)
        # Heuristic: a path without a .jsonl suffix is treated as a directory.
        self._is_dir = per_thread if per_thread is not None else (p.suffix == "")
        self.path = p
        self._redact = redact or redact_none
        self._secure = secure_permissions
        self._lock = asyncio.Lock()
        if self._is_dir:
            self.path.mkdir(parents=True, exist_ok=True)
            self._chmod_dir(self.path)
        else:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            self._chmod_dir(self.path.parent)

    def _chmod_dir(self, d: Path) -> None:
        # best-effort; some filesystems (e.g. Windows) ignore this
        if self._secure:
            with contextlib.suppress(OSError):
                d.chmod(0o700)

    def _file_for(self, thread: str) -> Path:
        if self._is_dir:
            safe = thread.replace("/", "_").replace("\\", "_")
            return self.path / f"{safe}.jsonl"
        return self.path

    async def _write(self, thread: str, event: str, **data: Any) -> None:
        entry = {"ts": _now(), "thread": thread, "event": event, **data}
        entry = self._redact(entry)
        line = json.dumps(entry, ensure_ascii=False, default=_json_default) + "\n"
        async with self._lock:
            path = self._file_for(thread)
            new_file = not path.exists()
            # Local file append is fast; running it inline keeps ordering
            # deterministic within the lock.
            with path.open("a", encoding="utf-8") as fh:
                fh.write(line)
            if new_file and self._secure:
                with contextlib.suppress(OSError):
                    path.chmod(0o600)

    async def on_run_start(self, thread, step):
        await self._write(thread, "run_start", step=step)

    async def on_node_start(self, thread, step, node):
        await self._write(thread, "node_start", step=step, node=node)

    async def on_node_end(self, thread, step, node, seconds):
        await self._write(thread, "node_end", step=step, node=node, seconds=round(seconds, 6))

    async def on_node_error(self, thread, step, node, exc):
        await self._write(
            thread, "node_error", step=step, node=node, error=f"{type(exc).__name__}: {exc}"
        )

    async def on_step_end(self, thread, step, ran):
        await self._write(thread, "step", step=step, ran=list(ran))

    async def on_run_end(self, thread, step, completed):
        await self._write(thread, "run_end", step=step, completed=completed)

    async def on_event(self, thread, step, node, event):
        await self._write(
            thread,
            "event",
            step=step,
            node=node,
            eventType=type(event).__name__,
            payload=_summarize(event),
        )