Observability and Tracing with Claude Code: Complete Guide

Introduction to Observability with Claude Code

As AI-assisted development becomes mainstream, understanding what your AI coding agent is doing behind the scenes is no longer optional. Claude Code, Anthropic's command-line AI coding tool, can autonomously read files, edit code, run shell commands, and make architectural decisions. Without proper observability, you're flying blind — trusting an autonomous agent to modify your codebase with zero visibility into its reasoning, actions, or performance.

Observability and tracing for Claude Code means instrumenting your workflows so you can answer critical questions: What files did Claude read? What commands did it execute? How long did each step take? Where did the reasoning go wrong? How many tokens were consumed? This guide walks you through everything you need to know to build a robust observability layer around your Claude Code usage.

What Is Observability and Tracing in the Context of Claude Code?

Observability is the ability to understand the internal state of a system based on its external outputs. In the context of Claude Code, it refers to your ability to monitor, measure, and analyze the behavior of Claude as it interacts with your codebase. Tracing is a specific observability technique that records the causal journey of a request through various components — in this case, the journey of a prompt through Claude's reasoning, tool calls, file operations, and final output.

Key Concepts

• Spans: Individual units of work, such as a single tool call, a file read, or a model inference.
• Traces: A collection of spans that represent the full lifecycle of a task from prompt to completion.
• Metrics: Quantitative measurements like token count, latency, cost per request, and success rate.
• Logs: Discrete event records capturing what happened at a specific point in time.
• Context propagation: Passing trace identifiers across process boundaries so you can correlate actions.

Why Observability Matters for Claude Code

Claude Code operates with a high degree of autonomy. It can chain together dozens of tool calls, make decisions about which files to modify, and execute potentially destructive shell commands. Without observability, you face several serious risks:

Debugging Failed Tasks

When Claude Code produces an incorrect result or fails mid-task, you need to know exactly where things went wrong. Was it a misinterpretation of your prompt? A bad file read? A shell command that returned unexpected output? Tracing lets you replay the exact sequence of events and pinpoint the failure.

Cost Management

Claude Code consumes tokens with every interaction, and autonomous agents can burn through tokens quickly when stuck in loops. Observability helps you identify wasteful patterns, optimize your prompts, and set budget limits.

Security and Compliance

In enterprise environments, you need audit trails of what files were accessed, what commands were run, and what changes were made. Tracing provides this audit trail automatically.

Performance Optimization

Understanding latency breakdowns — how much time is spent on model inference versus file I/O versus shell execution — helps you optimize your workflows and set realistic expectations.

Setting Up Basic Logging

The simplest form of observability is structured logging. Claude Code provides hooks that let you intercept and log every action. Let's start with a basic logging setup.

Configuring Claude Code Hooks

Claude Code supports a hooks system defined in your settings file. You can configure hooks to fire on various events such as tool execution, file edits, and command completions. Create or edit your .claude/settings.json file:

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": ".*",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /home/dev/claude-observers/pre_tool.py"
          }
        ]
      }
    ],
    "PostToolUse": [
      {
        "matcher": ".*",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /home/dev/claude-observers/post_tool.py"
          }
        ]
      }
    ],
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "python3 /home/dev/claude-observers/on_stop.py"
          }
        ]
      }
    ]
  }
}

The matcher field uses a regex pattern to filter which tools trigger the hook. The pattern .* matches all tools. Hook scripts receive JSON data via stdin containing context about the event.

Writing a Logging Hook Script

Here's a Python script that logs every tool use event to a structured JSON log file:

#!/usr/bin/env python3
import json
import sys
import os
from datetime import datetime, timezone

LOG_FILE = os.path.expanduser("~/.claude/logs/tool_events.jsonl")

def ensure_log_dir():
    log_dir = os.path.dirname(LOG_FILE)
    os.makedirs(log_dir, exist_ok=True)

def write_log_entry(entry):
    ensure_log_dir()
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

def main():
    # Read the hook input from stdin
    raw_input = sys.stdin.read()
    try:
        hook_data = json.loads(raw_input)
    except json.JSONDecodeError:
        hook_data = {"raw": raw_input}

    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "session_id": hook_data.get("session_id", "unknown"),
        "tool_name": hook_data.get("tool_name", "unknown"),
        "tool_input": hook_data.get("tool_input", {}),
        "hook_event": hook_data.get("hook_event_name", "unknown"),
        "cwd": hook_data.get("cwd", ""),
    }

    write_log_entry(entry)
    
    # Output empty JSON to allow the tool to proceed
    print(json.dumps({}))

if __name__ == "__main__":
    main()

This script reads the JSON payload from stdin, enriches it with a timestamp, and appends it to a JSONL log file. Each line in the file is a self-contained JSON object, making it easy to parse with tools like jq.

Querying Your Logs

Once you have logs accumulating, you can query them to understand Claude's behavior:

# Count tool usage by type
cat ~/.claude/logs/tool_events.jsonl | jq -r '.tool_name' | sort | uniq -c | sort -rn

# Find all file writes in a session
cat ~/.claude/logs/tool_events.jsonl | jq 'select(.tool_name == "Write" or .tool_name == "Edit")'

# Extract all shell commands executed
cat ~/.claude/logs/tool_events.jsonl | jq 'select(.tool_name == "Bash") | .tool_input.command'

Implementing Distributed Tracing

While logging gives you individual events, tracing connects them into a coherent narrative. Let's implement a proper tracing system using OpenTelemetry concepts adapted for Claude Code.

Building a Trace Context

First, let's create a trace context module that manages trace IDs, span IDs, and parent-child relationships:

# claude_tracing/trace_context.py
import uuid
import json
import os
import threading
from datetime import datetime, timezone
from contextlib import contextmanager

TRACE_FILE = os.path.expanduser("~/.claude/logs/traces.jsonl")
_local = threading.local()

def ensure_log_dir():
    log_dir = os.path.dirname(TRACE_FILE)
    os.makedirs(log_dir, exist_ok=True)

def generate_trace_id():
    return str(uuid.uuid4())

def generate_span_id():
    return uuid.uuid4().hex[:16]

def get_current_span():
    return getattr(_local, "current_span", None)

@contextmanager
def start_span(name, trace_id=None, parent_span_id=None, attributes=None):
    """Start a new span and record it in the trace log."""
    if trace_id is None:
        existing = get_current_span()
        if existing:
            trace_id = existing["trace_id"]
            parent_span_id = existing["span_id"]
        else:
            trace_id = generate_trace_id()

    span_id = generate_span_id()
    start_time = datetime.now(timezone.utc)

    span = {
        "trace_id": trace_id,
        "span_id": span_id,
        "parent_span_id": parent_span_id,
        "name": name,
        "start_time": start_time.isoformat(),
        "attributes": attributes or {},
        "status": "in_progress",
    }

    # Save previous span to restore later
    previous_span = getattr(_local, "current_span", None)
    _local.current_span = span

    try:
        yield span
        span["status"] = "ok"
    except Exception as e:
        span["status"] = "error"
        span["error"] = str(e)
        raise
    finally:
        span["end_time"] = datetime.now(timezone.utc).isoformat()
        span["duration_ms"] = (
            datetime.fromisoformat(span["end_time"]) - start_time
        ).total_seconds() * 1000

        ensure_log_dir()
        with open(TRACE_FILE, "a") as f:
            f.write(json.dumps(span) + "\n")

        _local.current_span = previous_span

Integrating Tracing with Claude Code Hooks

Now let's update our hook scripts to use the tracing system. The PreToolUse hook starts a span, and the PostToolUse hook ends it. Since hooks run as separate processes, we need to pass trace context through a temporary state file:

# claude_tracing/pre_tool_hook.py
#!/usr/bin/env python3
import json
import sys
import os
from claude_tracing.trace_context import start_span, generate_trace_id, generate_span_id

STATE_FILE = os.path.expanduser("~/.claude/logs/current_spans.json")

def main():
    raw_input = sys.stdin.read()
    hook_data = json.loads(raw_input)

    tool_name = hook_data.get("tool_name", "unknown")
    tool_input = hook_data.get("tool_input", {})
    session_id = hook_data.get("session_id", "unknown")

    # Check if there's an active trace for this session
    active_traces = {}
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            try:
                active_traces = json.load(f)
            except json.JSONDecodeError:
                active_traces = {}

    session_trace = active_traces.get(session_id, {})
    trace_id = session_trace.get("trace_id") or generate_trace_id()
    parent_span_id = session_trace.get("current_span_id")

    span_id = generate_span_id()

    # Record span start
    span = {
        "trace_id": trace_id,
        "span_id": span_id,
        "parent_span_id": parent_span_id,
        "name": f"tool:{tool_name}",
        "session_id": session_id,
        "tool_name": tool_name,
        "tool_input": tool_input,
        "start_time": datetime.now(timezone.utc).isoformat(),
        "status": "in_progress",
    }

    from datetime import datetime, timezone
    # Update state
    active_traces[session_id] = {
        "trace_id": trace_id,
        "current_span_id": span_id,
    }
    os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
    with open(STATE_FILE, "w") as f:
        json.dump(active_traces, f)

    # Write span to trace log
    trace_file = os.path.expanduser("~/.claude/logs/traces.jsonl")
    os.makedirs(os.path.dirname(trace_file), exist_ok=True)
    with open(trace_file, "a") as f:
        f.write(json.dumps(span) + "\n")

    print(json.dumps({}))

if __name__ == "__main__":
    main()

# claude_tracing/post_tool_hook.py
#!/usr/bin/env python3
import json
import sys
import os
from datetime import datetime, timezone

STATE_FILE = os.path.expanduser("~/.claude/logs/current_spans.json")
TRACE_FILE = os.path.expanduser("~/.claude/logs/traces.jsonl")

def main():
    raw_input = sys.stdin.read()
    hook_data = json.loads(raw_input)

    tool_name = hook_data.get("tool_name", "unknown")
    session_id = hook_data.get("session_id", "unknown")
    tool_response = hook_data.get("tool_response", {})

    # Load active traces
    if not os.path.exists(STATE_FILE):
        print(json.dumps({}))
        return

    with open(STATE_FILE, "r") as f:
        active_traces = json.load(f)

    session_trace = active_traces.get(session_id)
    if not session_trace:
        print(json.dumps({}))
        return

    span_id = session_trace.get("current_span_id")
    trace_id = session_trace.get("trace_id")

    # Write completion span
    completion = {
        "trace_id": trace_id,
        "span_id": span_id,
        "name": f"tool:{tool_name}:complete",
        "end_time": datetime.now(timezone.utc).isoformat(),
        "tool_response_summary": str(tool_response)[:500],
        "status": "ok",
    }

    with open(TRACE_FILE, "a") as f:
        f.write(json.dumps(completion) + "\n")

    # Reset current span to parent (or clear it)
    active_traces[session_id] = {
        "trace_id": trace_id,
        "current_span_id": None,
    }
    with open(STATE_FILE, "w") as f:
        json.dump(active_traces, f)

    print(json.dumps({}))

if __name__ == "__main__":
    main()

Visualizing Traces

Raw JSONL logs are hard to read. Let's build a trace visualization tool that reconstructs the span hierarchy and displays it in a readable format.

# claude_tracing/visualize.py
#!/usr/bin/env python3
import json
import sys
import os
from collections import defaultdict
from datetime import datetime

TRACE_FILE = os.path.expanduser("~/.claude/logs/traces.jsonl")

def load_traces():
    traces = defaultdict(list)
    if not os.path.exists(TRACE_FILE):
        return traces

    with open(TRACE_FILE, "r") as f:
        for line in f:
            try:
                span = json.loads(line.strip())
                traces[span.get("trace_id")].append(span)
            except json.JSONDecodeError:
                continue
    return traces

def parse_time(time_str):
    if not time_str:
        return None
    try:
        return datetime.fromisoformat(time_str)
    except (ValueError, TypeError):
        return None

def build_tree(spans):
    """Build a parent-child tree from flat span list."""
    span_map = {}
    for span in spans:
        span_id = span.get("span_id")
        if span_id:
            span_map[span_id] = {**span, "children": []}

    roots = []
    for span in spans:
        span_id = span.get("span_id")
        parent_id = span.get("parent_span_id")
        if span_id not in span_map:
            continue
        if parent_id and parent_id in span_map:
            span_map[parent_id]["children"].append(span_map[span_id])
        else:
            roots.append(span_map[span_id])

    return roots

def print_tree(node, indent=0):
    prefix = "  " * indent
    name = node.get("name", "unknown")
    status = node.get("status", "?")
    duration = node.get("duration_ms")
    tool_input = node.get("tool_input", {})

    duration_str = f" [{duration:.1f}ms]" if duration else ""
    status_icon = "✓" if status == "ok" else "✗" if status == "error" else "→"

    print(f"{prefix}{status_icon} {name}{duration_str}")

    # Show relevant input for tool spans
    if tool_input and indent > 0:
        if "command" in tool_input:
            print(f"{prefix}  └─ cmd: {tool_input['command'][:80]}")
        elif "file_path" in tool_input:
            print(f"{prefix}  └─ file: {tool_input['file_path']}")

    for child in node.get("children", []):
        print_tree(child, indent + 1)

def main():
    traces = load_traces()
    if not traces:
        print("No traces found.")
        return

    trace_id = sys.argv[1] if len(sys.argv) > 1 else None

    if trace_id:
        if trace_id in traces:
            roots = build_tree(traces[trace_id])
            print(f"\n=== Trace: {trace_id} ===\n")
            for root in roots:
                print_tree(root)
        else:
            print(f"Trace {trace_id} not found.")
    else:
        # List all traces
        print(f"\nFound {len(traces)} trace(s):\n")
        for tid, spans in traces.items():
            roots = build_tree(spans)
            total_spans = len(spans)
            print(f"  {tid} ({total_spans} spans)")
            for root in roots[:1]:  # Show first root as preview
                print_tree(root, indent=1)
            print()

if __name__ == "__main__":
    main()

Run the visualizer to see your traces:

python3 claude_tracing/visualize.py

# Or view a specific trace
python3 claude_tracing/visualize.py <trace-id>

Exporting to OpenTelemetry-Compatible Backends

For production-grade observability, you'll want to export traces to a dedicated backend like Jaeger, Zipkin, Datadog, or Honeycomb. Here's how to convert your Claude Code traces into OpenTelemetry format and export them:

# claude_tracing/otel_exporter.py
#!/usr/bin/env python3
import json
import os
import sys
import requests
from datetime import datetime, timezone

TRACE_FILE = os.path.expanduser("~/.claude/logs/traces.jsonl")

# Configure your OTLP endpoint
OTLP_ENDPOINT = os.environ.get("OTLP_ENDPOINT", "http://localhost:4318/v1/traces")

def load_spans():
    spans = []
    if not os.path.exists(TRACE_FILE):
        return spans

    with open(TRACE_FILE, "r") as f:
        for line in f:
            try:
                spans.append(json.loads(line.strip()))
            except json.JSONDecodeError:
                continue
    return spans

def to_otel_span(span):
    """Convert our internal span format to OTLP JSON format."""
    start_time = span.get("start_time")
    end_time = span.get("end_time")

    start_ts_ns = 0
    end_ts_ns = 0
    if start_time:
        dt = datetime.fromisoformat(start_time)
        start_ts_ns = int(dt.timestamp() * 1_000_000_000)
    if end_time:
        dt = datetime.fromisoformat(end_time)
        end_ts_ns = int(dt.timestamp() * 1_000_000_000)

    attributes = []
    for key, value in span.get("attributes", {}).items():
        attr_value = {"stringValue": str(value)}
        attributes.append({"key": key, "value": attr_value})

    if span.get("tool_name"):
        attributes.append({
            "key": "tool.name",
            "value": {"stringValue": span["tool_name"]}
        })
    if span.get("session_id"):
        attributes.append({
            "key": "session.id",
            "value": {"stringValue": span["session_id"]}
        })

    status_code = "STATUS_CODE_ERROR" if span.get("status") == "error" else "STATUS_CODE_OK"

    return {
        "traceId": span.get("trace_id", "").replace("-", "").ljust(32, "0")[:32],
        "spanId": span.get("span_id", "").ljust(16, "0")[:16],
        "parentSpanId": span.get("parent_span_id", "").ljust(16, "0")[:16] if span.get("parent_span_id") else None,
        "name": span.get("name", "claude-code-span"),
        "kind": "SPAN_KIND_INTERNAL",
        "startTimeUnixNano": str(start_ts_ns),
        "endTimeUnixNano": str(end_ts_ns),
        "attributes": attributes,
        "status": {"code": status_code},
    }

def export_traces():
    spans = load_spans()
    if not spans:
        print("No spans to export.")
        return

    otel_spans = [to_otel_span(s) for s in spans]

    payload = {
        "resourceSpans": [{
            "resource": {
                "attributes": [{
                    "key": "service.name",
                    "value": {"stringValue": "claude-code"}
                }]
            },
            "scopeSpans": [{
                "scope": {"name": "claude-code-hooks"},
                "spans": otel_spans,
            }]
        }]
    }

    try:
        response = requests.post(
            OTLP_ENDPOINT,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        print(f"Exported {len(otel_spans)} spans. Status: {response.status_code}")
    except requests.RequestException as e:
        print(f"Export failed: {e}")

if __name__ == "__main__":
    export_traces()

Building a Metrics Dashboard

Beyond traces, you need aggregate metrics to track trends over time. Let's build a metrics collector that computes key statistics from your trace data:

# claude_tracing/metrics.py
#!/usr/bin/env python3
import json
import os
from collections import defaultdict, Counter
from datetime import datetime, timedelta

TRACE_FILE = os.path.expanduser("~/.claude/logs/traces.jsonl")
TOOL_LOG_FILE = os.path.expanduser("~/.claude/logs/tool_events.jsonl")

def load_jsonl(filepath):
    records = []
    if not os.path.exists(filepath):
        return records
    with open(filepath, "r") as f:
        for line in f:
            try:
                records.append(json.loads(line.strip()))
            except json.JSONDecodeError:
                continue
    return records

def compute_metrics():
    tool_events = load_jsonl(TOOL_LOG_FILE)
    traces = load_jsonl(TRACE_FILE)

    metrics = {
        "summary": {},
        "tool_usage": {},
        "errors": [],
        "session_stats": {},
    }

    # Tool usage counts
    tool_counter = Counter(e.get("tool_name", "unknown") for e in tool_events)
    metrics["tool_usage"] = dict(tool_counter.most_common())

    # Total events
    metrics["summary"]["total_tool_calls"] = len(tool_events)
    metrics["summary"]["total_traces"] = len(set(t.get("trace_id") for t in traces))
    metrics["summary"]["total_spans"] = len(traces)

    # Unique sessions
    sessions = set(e.get("session_id", "unknown") for e in tool_events)
    metrics["summary"]["unique_sessions"] = len(sessions)

    # Error spans
    error_spans = [t for t in traces if t.get("status") == "error"]
    metrics["summary"]["error_count"] = len(error_spans)
    for err in error_spans[-10:]:  # Last 10 errors
        metrics["errors"].append({
            "trace_id": err.get("trace_id"),
            "span_name": err.get("name"),
            "error": err.get("error", "unknown"),
        })

    # Per-session statistics
    session_spans = defaultdict(list)
    for t in traces:
        sid = t.get("session_id", "unknown")
        session_spans[sid].append(t)

    for sid, spans in session_spans.items():
        durations = [s.get("duration_ms", 0) for s in spans if s.get("duration_ms")]
        metrics["session_stats"][sid] = {
            "span_count": len(spans),
            "total_duration_ms": sum(durations),
            "avg_duration_ms": sum(durations) / len(durations) if durations else 0,
            "max_duration_ms": max(durations) if durations else 0,
        }

    return metrics

def print_dashboard():
    metrics = compute_metrics()

    print("=" * 60)
    print("  Claude Code Observability Dashboard")
    print("=" * 60)

    print("\n📊 Summary")
    print("-" * 40)
    for key, value in metrics["summary"].items():
        print(f"  {key:.<30} {value}")

    print("\n🔧 Tool Usage")
    print("-" * 40)
    for tool, count in metrics["tool_usage"].items():
        bar = "█" * min(count, 30)
        print(f"  {tool:.<20} {count:>4} {bar}")

    print("\n📁 Session Statistics")
    print("-" * 40)
    for sid, stats in list(metrics["session_stats"].items())[:5]:
        print(f"  Session: {sid[:12]}...")
        print(f"    Spans: {stats['span_count']}")
        print(f"    Total: {stats['total_duration_ms']:.0f}ms")
        print(f"    Avg:   {stats['avg_duration_ms']:.0f}ms")
        print()

    if metrics["errors"]:
        print("\n❌ Recent Errors")
        print("-" * 40)
        for err in metrics["errors"]:
            print(f"  [{err['span_name']}] {err['error'][:60]}")

    print("\n" + "=" * 60)

if __name__ == "__main__":
    print_dashboard()

Adding Token and Cost Tracking

One of the most important metrics for Claude Code is token consumption and associated costs. You can capture this by hooking into Claude Code's API interaction events. Here's an approach using a wrapper script:

# claude_tracing/token_tracker.py
#!/usr/bin/env python3
import json
import os
from datetime import datetime, timezone

COST_LOG = os.path.expanduser("~/.claude/logs/token_costs.jsonl")

# Pricing per 1M tokens (update with current rates)
PRICING = {
    "claude-sonnet-4-20250514": {
        "input": 3.00,
        "output": 15.00,
    },
    "claude-opus-4-1-20250805": {
        "input": 15.00,
        "output": 75.00,
    },
    "claude-haiku-3-5": {
        "input": 0.80,
        "output": 4.00,
    },
}

DEFAULT_PRICING = {"input": 3.00, "output": 15.00}

def calculate_cost(model, input_tokens, output_tokens):
    pricing = PRICING.get(model, DEFAULT_PRICING)
    input_cost = (input_tokens / 1_000_000) * pricing["input"]
    output_cost = (output_tokens / 1_000_000) * pricing["output"]
    return round(input_cost + output_cost, 6)

def log_token_usage(session_id, model, input_tokens, output_tokens, 
                    cache_read_tokens=0, cache_write_tokens=0):
    cost = calculate_cost(model, input_tokens, output_tokens)
    
    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "session_id": session_id,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "cache_read_tokens": cache_read_tokens,
        "cache_write_tokens": cache_write_tokens,
        "total_tokens": input_tokens + output_tokens,
        "estimated_cost_usd": cost,
    }

    os.makedirs(os.path.dirname(COST_LOG), exist_ok=True)
    with open(COST_LOG, "a") as f:
        f.write(json.dumps(entry) + "\n")

    return entry

def get_cost_summary():
    """Summarize total costs from the log."""
    if not os.path.exists(COST_LOG):
        return {"total_cost": 0, "total_tokens": 0, "by_model": {}}

    total_cost = 0
    total_tokens = 0
    by_model = {}

    with open(COST_LOG, "r") as f:
        for line in f:
            try:
                entry = json.loads(line.strip())
                total_cost += entry["estimated_cost_usd"]
                total_tokens += entry["total_tokens"]

                model = entry["model"]
                if model not in by_model:
                    by_model[model] = {"cost": 0, "tokens": 0, "calls": 0}
                by_model[model]["cost"] += entry["estimated_cost_usd"]
                by_model[model]["tokens"] += entry["total_tokens"]
                by_model[model]["calls"] += 1
            except (json.JSONDecodeError, KeyError):
                continue

    return {
        "total_cost": round(total_cost, 4),
        "total_tokens": total_tokens,
        "by_model": by_model,
    }

if __name__ == "__main__":
    summary = get_cost_summary()
    print(f"\n💰 Total Cost: ${summary['total_cost']:.4f}")
    print(f"📊 Total Tokens: {summary['total_tokens']:,}")
    print("\nBy Model:")
    for model, stats in summary["by_model"].items():
        print(f"  {model}: ${stats['cost']:.4f} ({stats['tokens']:,} tokens, {stats['calls']} calls)")

Real-Time Monitoring with a Watch Script

For live observability during active Claude Code sessions, a watch script that tails your logs and displays updates in real time is invaluable:

# claude_tracing/live_monitor.py
#!/usr/bin/env python3
import json
import os
import time
import sys
from datetime import datetime

TOOL_LOG = os.path.expanduser("~/.claude/logs/tool_events.jsonl")

def tail_file(filepath):
    """Generator that yields new lines as they're appended."""
    with open(filepath, "r") as f:
        # Start from end of file
        f.seek(0, 2)
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.2)
                continue
            yield line

def format_event(event):
    ts = event.get("timestamp", "")
    try:
        dt = datetime.fromisoformat(ts)
        time_str = dt.strftime("%H:%M:%S")
    except (ValueError, TypeError):
        time_str = "???:??:???"

    tool = event.get("tool_name", "unknown")
    
    details = ""
    tool_input = event.get("tool_input", {})
    if tool == "Bash" and "command" in tool_input:
        details = tool_input["command"][:60]
    elif "file_path" in tool_input:
        details = tool_input["file_path"]
    elif "pattern" in tool_input:
        details = f"pattern: {tool_input['pattern']}"
    elif "query" in tool_input:
        details = tool_input["query"][:60]

    return f"[{time_str}] {tool:.<20} {details}"

def main():
    print("🔍 Claude Code Live Monitor")
    print("=" * 60)
    print("Watching for tool events... (Ctrl+C to stop)\n")

    if not os.path.exists(TOOL_LOG):
        print(f"Log file not found: {TOOL_LOG}")
        print("Run some Claude Code tasks first to generate events.")
        return

    try:
        for line in tail_file(TOOL_LOG):
            try:
                event = json.loads(line.strip())
                print(format_event(event))
                sys.stdout.flush()
            except json.JSONDecodeError:
                continue
    except KeyboardInterrupt:
        print("\n\nMonitor stopped.")

if __name__ == "__main__":
    main()

Run this in a separate terminal while Claude Code is working:

python3 claude_tracing/live_monitor.py

Best Practices for Claude Code Observability

1. Always Log Full Context

Capture not just what tool was called, but the full input and a summary of the output. When debugging a failed task months later, you'll need the complete picture. However, be mindful of sensitive data — consider adding a redaction layer for secrets, API keys, and credentials before writing to logs.

2. Use Consistent Session Identifiers

Claude Code provides a session ID in hook payloads. Use this consistently across all your observability tooling to correlate events from the same conversation. This is your primary key for grouping spans into traces.

3. Set Up Alerting on Anomalies

Configure alerts for scenarios that indicate problems:

• Token usage exceeding a budget threshold per session
• Repeated failed tool calls (indicating Claude is stuck in a loop)
• Shell commands that modify critical files without explicit approval
• Unusually long spans that might indicate timeouts or hangs

Here's a simple alerting hook example:

# claude_tracing/alert_hook.py
#!/usr/bin/env python3
import json
import sys
import os
from collections import defaultdict

STATE_FILE = os.path.expanduser("~/.claude/logs/tool_counts.json")
MAX_RETRIES = 5  # Alert if same tool called more than this in a session

def main():
    raw = sys.stdin.read()
    data = json.loads(raw)

    session_id = data.get("session_id", "unknown")
    tool_name = data.get("tool_name", "unknown")

    # Load state
    counts = {}
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            try:
                counts = json.load(f)
            except json.JSONDecodeError:
                counts = {}

    key = f"{session_id}:{tool_name}"
    counts[key] = counts.get(key, 0) + 1

    if counts[key] > MAX_RETRIES:
        # Write alert to stderr (visible in Claude Code output)
        print(
            f"⚠️  ALERT: Tool '{tool_name}' called {counts[key]} times "
            f"in session {session_id[:8]}. Possible loop detected.",
            file=sys.stderr,
        )

    # Save state
    os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
    with open(STATE_FILE, "w") as f:
        json.dump(counts, f)

    print(json.dumps({}))

if __name__ == "__main__":
    main()

4. Rotate and Archive Logs

Log files grow continuously. Implement log rotation to prevent disk space issues. A simple approach:

# Add to crontab - runs daily
# 0 2 * * * python3 ~/claude_tracing/rotate_logs.py

#!/usr/bin/env python3
import os
import gzip
import shutil
from datetime import datetime, timedelta

LOG_DIR = os.path.expanduser("~/.claude/logs")
ARCHIVE_DIR = os.path.expanduser("~/.claude/logs/archive")
RETENTION_DAYS = 30

def rotate_logs():
    os.makedirs(ARCHIVE_DIR, exist_ok=True)
    date_str = datetime.now().strftime("%Y%m%d")

    for filename in ["tool_events.jsonl", "traces.jsonl", "token_costs.jsonl"]:
        filepath = os.path.join(LOG_DIR, filename)
        if not os.path.exists(filepath):
            continue

        # Compress and archive
        archive_name = f"{filename}.{date_str}.gz"
        archive_path = os.path.join(ARCHIVE_DIR, archive_name)

        with open(filepath, "rb") as f_in:
            with gzip.open(archive_path, "wb") as f_out:
                shutil.copyfileobj(f_in, f_out)

        # Clear the active log
        open(filepath, "w").close()
        print(f"Rotated {filename} -> {archive_name}")

    # Clean old archives
    cutoff = datetime.now() - timedelta(days=RETENTION_DAYS)
    for archive in os.listdir(ARCHIVE_DIR):
        archive_path = os.path.join(ARCHIVE_DIR, archive)
        if os.path.getmtime(archive_path) < cutoff.timestamp():
            os.remove