Skip to main content

monitoring.core

Monitoring capability for pydantic-ai lifecycle hooks.

Pushes an agent.snapshot message over the WebSocket pub/sub infrastructure after every agent run completes (or fails). This replaces the need for callers to manually invoke publish_stream_event after each adapter call — the capability fires automatically from the pydantic-ai lifecycle.

The snapshot includes:

  • Context usage (tokens, messages, tool definitions)
  • Cost usage (from the shared CostStore)
  • MCP server status and enabled tools
  • Codemode / skills status
  • Pending tool-approval records
  • Full context (detailed tool schemas, message history)

MonitoringCapability Objects​

@dataclass
class MonitoringCapability(AbstractCapability[Any])

Broadcast a monitoring snapshot after every agent run.

The snapshot is assembled by the existing :func:~agent_runtimes.streams.loop.build_monitoring_snapshot_payload helper and pushed to all WebSocket subscribers of agent_id.

Parameters​

agent_id : str Agent identifier — subscribers keyed on this id receive updates. enabled : bool Master switch — when False the hooks are no-ops. list_approvals : callable or None Optional async callable (agent_id, status) -> list[Record] that returns pending tool approvals for inclusion in the snapshot.

list_approvals​

Optional async callable