18_otel_capture.py

"""18: Observability. Run a tiny OTLP/HTTP receiver on localhost, point the agent at it,
and print the spans, metrics and log events the Claude Code binary exported."""

import asyncio
import json
import threading
from collections import Counter
from http.server import BaseHTTPRequestHandler, HTTPServer

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from demo_util import PROJECT

received = {"traces": [], "metrics": [], "logs": []}


class Collector(BaseHTTPRequestHandler):
    """Accepts OTLP over HTTP with JSON encoding and stores the payloads."""

    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        signal = self.path.rsplit("/", 1)[-1]          # /v1/traces -> traces
        if signal in received:
            received[signal].append(json.loads(body))
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b"{}")

    def log_message(self, *args):                    # silence per-request logging
        pass


def start_collector(port=4318):
    server = HTTPServer(("127.0.0.1", port), Collector)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


OTEL_ENV = {
    "CLAUDE_CODE_ENABLE_TELEMETRY": "1",
    "CLAUDE_CODE_ENHANCED_TELEMETRY_BETA": "1",             # traces are beta
    "OTEL_TRACES_EXPORTER": "otlp",
    "OTEL_METRICS_EXPORTER": "otlp",
    "OTEL_LOGS_EXPORTER": "otlp",
    "OTEL_EXPORTER_OTLP_PROTOCOL": "http/json",             # JSON so this demo can read it without protobuf
    "OTEL_EXPORTER_OTLP_ENDPOINT": "http://127.0.0.1:4318",
    "OTEL_SERVICE_NAME": "tutorial-agent",
    "OTEL_RESOURCE_ATTRIBUTES": "deployment.environment=demo,tenant.id=acme",
    "OTEL_METRIC_EXPORT_INTERVAL": "1000",                  # flush fast: this run lasts seconds
    "OTEL_LOGS_EXPORT_INTERVAL": "1000",
    "OTEL_TRACES_EXPORT_INTERVAL": "1000",
    "CLAUDE_CODE_OTEL_DIAG_STDERR": "1",                    # surface exporter errors on stderr
    "NO_PROXY": "127.0.0.1,localhost",                      # don't send localhost traffic through a corporate proxy
}

stderr_lines = []


async def main():
    server = start_collector()
    try:
        async for message in query(
            prompt="Read README.md and reply with its first heading only.",
            options=ClaudeAgentOptions(cwd=PROJECT, allowed_tools=["Read"], max_turns=3,
                                       env=OTEL_ENV, stderr=stderr_lines.append),
        ):
            if isinstance(message, ResultMessage):
                print(f"[result] {message.subtype}: {message.result!r} cost=${message.total_cost_usd:.4f}")
        await asyncio.sleep(8)                               # let the final export intervals fire
    finally:
        server.shutdown()

    # ---- what arrived --------------------------------------------------------------------
    spans = [s for p in received["traces"] for rs in p.get("resourceSpans", [])
             for ss in rs.get("scopeSpans", []) for s in ss.get("spans", [])]
    metrics = [m["name"] for p in received["metrics"] for rm in p.get("resourceMetrics", [])
               for sm in rm.get("scopeMetrics", []) for m in sm.get("metrics", [])]
    logs = [r for p in received["logs"] for rl in p.get("resourceLogs", [])
            for sl in rl.get("scopeLogs", []) for r in sl.get("logRecords", [])]

    def attrs(obj):
        return {a["key"]: list(a["value"].values())[0] for a in obj.get("attributes", [])}

    print(f"\nreceived: {len(received['traces'])} trace batches, {len(received['metrics'])} metric batches, {len(received['logs'])} log batches")
    if received["traces"]:
        res = attrs(received["traces"][0]["resourceSpans"][0]["resource"])
        print("resource attributes:", {k: res[k] for k in res if k in ("service.name", "deployment.environment", "tenant.id")})
    print("\nspans (name -> count):", dict(Counter(s["name"] for s in spans)))
    for s in spans[:6]:
        a = attrs(s)
        extra = {k: a[k] for k in a if k in ("tool.name", "model", "gen_ai.request.model", "session.id")}
        dur_ms = (int(s["endTimeUnixNano"]) - int(s["startTimeUnixNano"])) / 1e6
        print(f"  {s['name']:<32} {dur_ms:8.0f} ms  parent={'yes' if s.get('parentSpanId') else 'root'} {extra}")
    print("\nmetrics:", sorted(set(metrics)))
    print("\nlog events (name -> count):", dict(Counter(attrs(r).get("event.name", r.get('body', {}).get('stringValue', '?')) for r in logs)))
    if stderr_lines:
        print("\nstderr diagnostics (first 3):", stderr_lines[:3])


asyncio.run(main())