Skip to main content

monitoring.graph_telemetry

Graph-level telemetry capability for pydantic-graph execution.

Instruments pydantic-graph execution to emit fine-grained OTEL telemetry for graph-level processing: steps, joins, decisions, and parallel execution.

Unlike AbstractCapability hooks (which fire for pydantic-ai Agent runs), graph telemetry tracks node-level execution within a pydantic-graph Graph. It produces:

  • Per-node OTEL counters/histograms (start, complete, error, duration)
  • A graph execution trace (list of node events with timestamps and edges)
  • A graph topology snapshot (nodes + edges from the Graph structure)

The trace is stored in a process-global registry keyed by agent_id so the monitoring snapshot builder can include it in WebSocket payloads.

Examples​

Wrap your graph execution with :func:run_graph_with_telemetry::

from agent_runtimes.monitoring.graph_telemetry import run_graph_with_telemetry

result = await run_graph_with_telemetry(
    graph=my_graph,
    agent_id="my-agent",
    state=my_state,
    start_node=MyStartNode(),
)

For the beta graph API, use :func:run_beta_graph_with_telemetry.

GraphNodeEvent Objects​

@dataclass
class GraphNodeEvent()

A single node execution event in the graph trace.

node_type​

e.g. "step", "join", "decision", "end", "start"

status​

"started", "completed", "error"

inputs​

Stringified summary of inputs

output​

Stringified summary of output

parent_node_id​

Previous node in the execution chain

GraphEdge Objects​

@dataclass
class GraphEdge()

An edge in the graph topology.

edge_type​

"normal", "parallel", "decision", "join"

GraphTelemetryData Objects​

@dataclass
class GraphTelemetryData()

Full graph telemetry data for an agent.

get_graph_telemetry​

def get_graph_telemetry(agent_id: str) -> GraphTelemetryData | None

Retrieve the latest graph telemetry for an agent.

get_graph_telemetry_dict​

def get_graph_telemetry_dict(agent_id: str) -> dict[str, Any] | None

Retrieve graph telemetry as a plain dict (for JSON serialisation).

clear_graph_telemetry​

def clear_graph_telemetry(agent_id: str) -> None

Clear stored graph telemetry for an agent.

run_graph_with_telemetry​

async def run_graph_with_telemetry(graph: Any,
                                   agent_id: str,
                                   *,
                                   start_node: Any,
                                   state: Any = None,
                                   deps: Any = None,
                                   service_name: str = "agent-runtimes",
                                   **kwargs: Any) -> Any

Run a classic pydantic-graph Graph with OTEL telemetry.

Uses graph.iter() to step through execution node-by-node, emitting OTEL metrics and building a trace of node events.

Returns the graph result.

run_beta_graph_with_telemetry​

async def run_beta_graph_with_telemetry(graph: Any,
                                        agent_id: str,
                                        *,
                                        state: Any = None,
                                        deps: Any = None,
                                        inputs: Any = None,
                                        service_name: str = "agent-runtimes",
                                        **kwargs: Any) -> Any

Run a beta pydantic-graph Graph with OTEL telemetry.

Uses graph.iter() to step through execution, emitting OTEL metrics and building a trace.

Returns the graph output.