Agent Observability Basics

9 examples to get you started with Agent Observability - 6 basic and 3 intermediate.

These sketches are provider-agnostic. They teach the data model of a run, not a single vendor UI.

Prerequisites

Comfort with a simple agent loop: model call, optional tool call, observe result.
No vendor account required for this page.
Optional: skim What Agent Observability Actually Needs to Capture for the field list.

Basic Examples

1. Give Every Run a Stable Id

Without a run_id, steps cannot be joined across workers or log lines.

from uuid import uuid4
 
def new_run(goal: str) -> dict:
    return {"run_id": str(uuid4()), "goal": goal, "steps": []}

Create the id in the host at run start, not inside the model.
Propagate the same id into tools, workers, and HTTP clients.
Prefer UUID or ULID strings that sort and search cleanly.

Related: What Agent Observability Actually Needs to Capture - required fields

2. Record Model Steps as Spans

Capture model id, latency, and a short result summary - not only the final answer.

import time
 
def model_step(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text = call_model(messages)
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "preview": text[:200],
    })
    return text

Latency per model step finds slow providers fast.
Previews help humans; full bodies can live behind retention policy.
Always name the model version you actually called.

Related: LangSmith for LangChain/LangGraph-Native Tracing - automatic model nesting

3. Record Tool Calls With Args and Outcome

Tools are where side effects hide.

def tool_step(run: dict, name: str, args: dict, fn) -> dict:
    t0 = time.perf_counter()
    try:
        result = fn(**args)
        status, err = "ok", None
    except Exception as e:
        result, status, err = None, "error", str(e)
    run["steps"].append({
        "type": "tool",
        "name": name,
        "args": args,
        "status": status,
        "error": err,
        "ms": int((time.perf_counter() - t0) * 1000),
        "result_preview": None if result is None else str(result)[:200],
    })
    if status != "ok":
        raise RuntimeError(err)
    return result

Log name, args (redacted), status, and duration every time.
Separate status from exception handling so partial runs stay visible.
Cap result size so one PDF does not explode storage.

Related: Building Custom Trace Logging Without a Third-Party Platform - production-shaped logger

4. Trace One Bounded ReAct Loop

Nest model and tool steps under one run with a turn counter.

def run_agent(goal: str, tools: dict, call_model, max_turns: int = 5) -> dict:
    run = new_run(goal)
    messages = [{"role": "user", "content": goal}]
    for turn in range(max_turns):
        decision = model_step(run, "demo-model", messages, call_model)
        # decision is either final text or "TOOL name {...}" in a real system
        if decision.startswith("FINAL:"):
            run["stop_reason"] = "goal_met"
            run["answer"] = decision.removeprefix("FINAL:").strip()
            return run
        name, args = parse_tool(decision)  # your parser
        obs = tool_step(run, name, args, tools[name])
        messages.append({"role": "assistant", "content": decision})
        messages.append({"role": "tool", "content": str(obs)})
        run["steps"][-1]["turn"] = turn
    run["stop_reason"] = "max_turns"
    return run

Always emit a stop_reason.
Bound turns in the host so thrash is visible and finite.
Attach turn numbers so UI timelines stay readable.

Related: Stopping Conditions: How an Agent Knows When It's Done - stop rules

5. Redact Secrets Before You Write

Traces are a high-value exfiltration target if they store tokens.

SECRET_KEYS = ("password", "token", "secret", "authorization", "api_key")
 
def redact(value):
    if isinstance(value, dict):
        out = {}
        for k, v in value.items():
            out[k] = "***" if any(s in k.lower() for s in SECRET_KEYS) else redact(v)
        return out
    if isinstance(value, list):
        return [redact(v) for v in value]
    return value
 
# when appending a tool step:
# "args": redact(args)

Redact at emission time, not only in the UI filter.
Extend the key list for your domain (ssn, card_number, etc.).
Prefer fingerprints for highly sensitive business ids when policy requires it.

Related: Audit Trails: Logging Every Agent Action for Later Review - stricter integrity rules

6. Print a Human Reconstruction

Practice reading a run the way on-call will.

def explain(run: dict) -> None:
    print("run", run["run_id"], "goal=", run["goal"])
    for i, step in enumerate(run["steps"]):
        if step["type"] == "model":
            print(f"{i}. model {step['model_id']} {step['ms']}ms :: {step['preview']}")
        else:
            print(f"{i}. tool  {step['name']} {step['status']} {step['ms']}ms args={step['args']}")
    print("stop_reason=", run.get("stop_reason"), "answer=", run.get("answer"))

If you cannot explain the side effects from this output, capture more fields.
Use the same shape later when exporting JSON to a platform.
Keep one "golden" explained run in your runbook.

Related: Agent Observability Best Practices - review habits

Intermediate Examples

7. Attach Cost and Token Attributes

Cost regressions show up as attributes before finance invoices do.

def model_step_with_usage(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text, usage = call_model(messages)  # usage: {prompt, completion}
    tokens = usage["prompt"] + usage["completion"]
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "tokens": tokens,
        "usage": usage,
        "preview": text[:200],
    })
    run["tokens_total"] = run.get("tokens_total", 0) + tokens
    return text

Aggregate tokens per run for budget alerts.
Keep raw usage on the step for provider disputes.
Pair with setting SLOs when you go live.

Related: Pairing Agent Observability with Infrastructure Monitoring - whole-stack view

8. Nest Child Spans for Multi-Agent Handoffs

Specialists need their own subtree under the parent run.

def handoff_step(run: dict, specialist: str, packet: dict, run_specialist) -> dict:
    child = {
        "type": "handoff",
        "specialist": specialist,
        "packet": redact(packet),
        "child_run_id": str(uuid4()),
        "steps": [],
    }
    result = run_specialist(packet, sink=child["steps"])
    child["status"] = result.get("status", "ok")
    run["steps"].append(child)
    return result

Parent run_id plus child_run_id keeps ownership clear.
Log the handoff packet contract, not the entire chat dump by default.
Failures inside specialists must bubble status to the parent.

Related: Multi-Agent Handoff: Specialist Agents Passing Work to Each Other - orchestration model

9. Export One Run as Structured JSON Lines

Platforms and warehouses all accept structured events.

import json
import time
 
def export_jsonl(run: dict, path: str) -> None:
    base = {"run_id": run["run_id"], "ts": time.time(), "goal": run["goal"]}
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps({**base, "event": "run_started"}) + "\n")
        for step in run["steps"]:
            f.write(json.dumps({**base, "event": "step", "step": redact(step)}) + "\n")
        f.write(json.dumps({
            **base,
            "event": "run_finished",
            "stop_reason": run.get("stop_reason"),
            "tokens_total": run.get("tokens_total"),
        }) + "\n")

JSONL is easy to ship to files, queues, or OpenTelemetry collectors later.
Keep event names stable so dashboards do not break weekly.
This shape maps cleanly onto Langfuse, Phoenix, or a custom stack.

Related: Langfuse: OpenTelemetry-Based Tracing for Any Framework - OTEL export path

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Agent Observability Basics

9 examples to get you started with Agent Observability - 6 basic and 3 intermediate.

These sketches are provider-agnostic. They teach the data model of a run, not a single vendor UI.

Prerequisites

Comfort with a simple agent loop: model call, optional tool call, observe result.
No vendor account required for this page.
Optional: skim What Agent Observability Actually Needs to Capture for the field list.

Basic Examples

1. Give Every Run a Stable Id

Without a run_id, steps cannot be joined across workers or log lines.

from uuid import uuid4
 
def new_run(goal: str) -> dict:
    return {"run_id": str(uuid4()), "goal": goal, "steps": []}

Create the id in the host at run start, not inside the model.
Propagate the same id into tools, workers, and HTTP clients.
Prefer UUID or ULID strings that sort and search cleanly.

Related: What Agent Observability Actually Needs to Capture - required fields

2. Record Model Steps as Spans

Capture model id, latency, and a short result summary - not only the final answer.

import time
 
def model_step(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text = call_model(messages)
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "preview": text[:200],
    })
    return text

Latency per model step finds slow providers fast.
Previews help humans; full bodies can live behind retention policy.
Always name the model version you actually called.

Related: LangSmith for LangChain/LangGraph-Native Tracing - automatic model nesting

3. Record Tool Calls With Args and Outcome

Tools are where side effects hide.

def tool_step(run: dict, name: str, args: dict, fn) -> dict:
    t0 = time.perf_counter()
    try:
        result = fn(**args)
        status, err = "ok", None
    except Exception as e:
        result, status, err = None, "error", str(e)
    run["steps"].append({
        "type": "tool",
        "name": name,
        "args": args,
        "status": status,
        "error": err,
        "ms": int((time.perf_counter() - t0) * 1000),
        "result_preview": None if result is None else str(result)[:200],
    })
    if status != "ok":
        raise RuntimeError(err)
    return result

Log name, args (redacted), status, and duration every time.
Separate status from exception handling so partial runs stay visible.
Cap result size so one PDF does not explode storage.

Related: Building Custom Trace Logging Without a Third-Party Platform - production-shaped logger

4. Trace One Bounded ReAct Loop

Nest model and tool steps under one run with a turn counter.

def run_agent(goal: str, tools: dict, call_model, max_turns: int = 5) -> dict:
    run = new_run(goal)
    messages = [{"role": "user", "content": goal}]
    for turn in range(max_turns):
        decision = model_step(run, "demo-model", messages, call_model)
        # decision is either final text or "TOOL name {...}" in a real system
        if decision.startswith("FINAL:"):
            run["stop_reason"] = "goal_met"
            run["answer"] = decision.removeprefix("FINAL:").strip()
            return run
        name, args = parse_tool(decision)  # your parser
        obs = tool_step(run, name, args, tools[name])
        messages.append({"role": "assistant", "content": decision})
        messages.append({"role": "tool", "content": str(obs)})
        run["steps"][-1]["turn"] = turn
    run["stop_reason"] = "max_turns"
    return run

Always emit a stop_reason.
Bound turns in the host so thrash is visible and finite.
Attach turn numbers so UI timelines stay readable.

Related: Stopping Conditions: How an Agent Knows When It's Done - stop rules

5. Redact Secrets Before You Write

Traces are a high-value exfiltration target if they store tokens.

SECRET_KEYS = ("password", "token", "secret", "authorization", "api_key")
 
def redact(value):
    if isinstance(value, dict):
        out = {}
        for k, v in value.items():
            out[k] = "***" if any(s in k.lower() for s in SECRET_KEYS) else redact(v)
        return out
    if isinstance(value, list):
        return [redact(v) for v in value]
    return value
 
# when appending a tool step:
# "args": redact(args)

Redact at emission time, not only in the UI filter.
Extend the key list for your domain (ssn, card_number, etc.).
Prefer fingerprints for highly sensitive business ids when policy requires it.

Related: Audit Trails: Logging Every Agent Action for Later Review - stricter integrity rules

6. Print a Human Reconstruction

Practice reading a run the way on-call will.

def explain(run: dict) -> None:
    print("run", run["run_id"], "goal=", run["goal"])
    for i, step in enumerate(run["steps"]):
        if step["type"] == "model":
            print(f"{i}. model {step['model_id']} {step['ms']}ms :: {step['preview']}")
        else:
            print(f"{i}. tool  {step['name']} {step['status']} {step['ms']}ms args={step['args']}")
    print("stop_reason=", run.get("stop_reason"), "answer=", run.get("answer"))

If you cannot explain the side effects from this output, capture more fields.
Use the same shape later when exporting JSON to a platform.
Keep one "golden" explained run in your runbook.

Related: Agent Observability Best Practices - review habits

Intermediate Examples

7. Attach Cost and Token Attributes

Cost regressions show up as attributes before finance invoices do.

def model_step_with_usage(run: dict, model_id: str, messages: list, call_model) -> str:
    t0 = time.perf_counter()
    text, usage = call_model(messages)  # usage: {prompt, completion}
    tokens = usage["prompt"] + usage["completion"]
    run["steps"].append({
        "type": "model",
        "model_id": model_id,
        "ms": int((time.perf_counter() - t0) * 1000),
        "tokens": tokens,
        "usage": usage,
        "preview": text[:200],
    })
    run["tokens_total"] = run.get("tokens_total", 0) + tokens
    return text

Aggregate tokens per run for budget alerts.
Keep raw usage on the step for provider disputes.
Pair with setting SLOs when you go live.

Related: Pairing Agent Observability with Infrastructure Monitoring - whole-stack view

8. Nest Child Spans for Multi-Agent Handoffs

Specialists need their own subtree under the parent run.

def handoff_step(run: dict, specialist: str, packet: dict, run_specialist) -> dict:
    child = {
        "type": "handoff",
        "specialist": specialist,
        "packet": redact(packet),
        "child_run_id": str(uuid4()),
        "steps": [],
    }
    result = run_specialist(packet, sink=child["steps"])
    child["status"] = result.get("status", "ok")
    run["steps"].append(child)
    return result

Parent run_id plus child_run_id keeps ownership clear.
Log the handoff packet contract, not the entire chat dump by default.
Failures inside specialists must bubble status to the parent.

Related: Multi-Agent Handoff: Specialist Agents Passing Work to Each Other - orchestration model

9. Export One Run as Structured JSON Lines

Platforms and warehouses all accept structured events.

import json
import time
 
def export_jsonl(run: dict, path: str) -> None:
    base = {"run_id": run["run_id"], "ts": time.time(), "goal": run["goal"]}
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps({**base, "event": "run_started"}) + "\n")
        for step in run["steps"]:
            f.write(json.dumps({**base, "event": "step", "step": redact(step)}) + "\n")
        f.write(json.dumps({
            **base,
            "event": "run_finished",
            "stop_reason": run.get("stop_reason"),
            "tokens_total": run.get("tokens_total"),
        }) + "\n")

JSONL is easy to ship to files, queues, or OpenTelemetry collectors later.
Keep event names stable so dashboards do not break weekly.
This shape maps cleanly onto Langfuse, Phoenix, or a custom stack.

Related: Langfuse: OpenTelemetry-Based Tracing for Any Framework - OTEL export path

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Agent Observability Basics

Search Documentation

Agent Observability Basics