The 1-line observability hook that fits any agent

You already pay for the agent calls. You do not yet pay for the data to know whether they were worth it. The fix is one wrapper that intercepts every model call, captures six useful fields, and writes one JSONL line.

This is the smallest observability hook that is actually useful. Smaller and you are flying blind. Larger and you stop installing it. The OpenTelemetry zero-code Python docs describe the production-grade version of this idea, and the OpenTelemetry Python contrib repo provides the per-framework auto-instrumentations (OpenAI, Anthropic, LangChain) once your scale justifies the upgrade.

Where this comes from: Marvin’s 2026-07-20 original. Sources verified 2026-07-22 against the linked Python, OpenTelemetry, and OpenLLMetry pages. One command (@obs.track("name")) is the whole API.

Before you start

You need:

• A Python function that calls your model of choice (any framework — raw openai, raw anthropic, LangChain, or your own).
• A writable directory for the JSONL ledger (any path your agent can append to).
• A way to inspect the ledger (wc -l, awk, or your editor of choice).

No cloud account, no paid service, no special library.

🤖 Let your agent do it (Hermes path): ask Hermes to scaffold this for you.

write an obs.py module with a @track(name) decorator that wraps any model call, captures latency_ms, tokens_in, tokens_out, model, and ok, and appends one JSONL line per call to ~/.cache/agent-obs.jsonl

Hermes has Python + the shell tool. Verify by running python -c "import obs; help(obs.track)" — the docstring should match.

The wrapper

# obs.py — drop-in observability
import time, json, uuid, pathlib, functools, os

LEDGER = pathlib.Path(os.environ.get("OBS_LEDGER", "~/.cache/agent-obs.jsonl")).expanduser()
LEDGER.parent.mkdir(parents=True, exist_ok=True)

def _record(event: dict):
    event["ts"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    with LEDGER.open("a") as f:
        f.write(json.dumps(event) + "\n")

def track(name: str | None = None):
    """Decorator: wrap any model call. Captures latency, tokens, errors."""
    def deco(fn):
        @functools.wraps(fn)   # preserves __name__, __doc__, signature
        def wrap(*args, **kwargs):
            run_id = kwargs.pop("__run_id", None) or uuid.uuid4().hex[:8]
            started = time.time()
            try:
                result = fn(*args, **kwargs)
                usage = getattr(result, "usage", None) or {}
                _record({
                    "kind": "model",
                    "name": name or fn.__name__,
                    "run_id": run_id,
                    "ok": True,
                    "latency_ms": int((time.time() - started) * 1000),
                    "tokens_in":  usage.get("input_tokens") or usage.get("prompt_tokens"),
                    "tokens_out": usage.get("output_tokens") or usage.get("completion_tokens"),
                    "model":  getattr(result, "model", None) or kwargs.get("model"),
                })
                return result
            except Exception as e:
                _record({
                    "kind": "error",
                    "name": name or fn.__name__,
                    "run_id": run_id,
                    "ok": False,
                    "latency_ms": int((time.time() - started) * 1000),
                    "error": f"{type(e).__name__}: {e}",
                })
                raise
        return wrap
    return deco

That is the entire library. ~30 lines. The functools.wraps pattern is straight from the official Python docs; the per-call attribute shape (latency_ms, tokens_in, tokens_out, model) matches the OpenTelemetry semantic conventions used by OpenLLMetry.

How you install it

Three patterns cover the common cases. Drop the relevant one next to your agent code and import obs.

Pattern	What you wrap	Verify
@obs.track("openai.complete") on a thin function that calls openai.chat.completions.create()	Raw OpenAI SDK calls	One JSONL line per call in ~/.cache/agent-obs.jsonl.
@obs.track("claude.complete") on a thin function that calls client.messages.create()	Raw Anthropic SDK calls	Same — one line per call.
class ObservedChat(obs.track("langchain.openai")): ...	LangChain chat model wrapper	Same — one line per call.

Raw OpenAI

import openai, obs

@obs.track("openai.complete")
def complete(prompt: str, model: str = "gpt-4o-mini") -> str:
    r = openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    return r.choices[0].message.content

Raw Anthropic

import anthropic, obs

client = anthropic.Anthropic()

@obs.track("claude.complete")
def complete(prompt: str, model: str = "claude-sonnet-4-5") -> str:
    r = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return r.content[0].text

LangChain

from langchain_openai import ChatOpenAI
import obs

class ObservedChat(obs.track("langchain.openai")):
    def __init__(self, **kw):
        self._inner = ChatOpenAI(**kw)
    def __call__(self, *a, **kw):
        return self._inner(*a, **kw)

The wrapper is a decorator, not an opinion. It works with whatever you are calling. Once your call volume warrants it, the OpenTelemetry Python contrib has zero-code instrumentations for OpenAI, Anthropic, and LangChain that do the same thing at scale.

Command	Job	Success state
pip install obs.py (or copy the 30 lines into obs.py)	Install the local observability module.	python -c "import obs; print(obs.track.__doc__)" prints the docstring.
Decorate one model-call function with @obs.track("name")	Wrap that call.	Calling the function appends one JSONL line to the ledger.
tail -f ~/.cache/agent-obs.jsonl	Live-watch what your agent is doing.	Lines stream in as calls happen.

What the JSONL looks like

{"ts":"2026-07-20T13:11:04Z","kind":"model","name":"openai.complete","run_id":"7c2a","ok":true,"latency_ms":1340,"tokens_in":482,"tokens_out":124,"model":"gpt-4o-mini"}
{"ts":"2026-07-20T13:11:09Z","kind":"model","name":"claude.complete","run_id":"7c2a","ok":true,"latency_ms":1820,"tokens_in":211,"tokens_out":38,"model":"claude-sonnet-4-5"}
{"ts":"2026-07-20T13:11:14Z","kind":"error","name":"openai.complete","run_id":"7c2a","ok":false,"latency_ms":540,"error":"RateLimitError: 429"}

Three lines, one run, full picture.

The four questions this answers immediately

# 1. What did I spend today?
awk -F'"tokens_in":' '/"kind":"model"/{split($2,a,","); in+=a[1]} END{print "tokens_in: " in}' ~/.cache/agent-obs.jsonl

# 2. What was my slowest call?
awk -F'"latency_ms":' '/"kind":"model"/{if ($2+0 > max) {max=$2+0; line=$0}} END{print line}' ~/.cache/agent-obs.jsonl

# 3. What's failing?
grep '"ok":false' ~/.cache/agent-obs.jsonl | tail

# 4. What was the agent actually doing at run X?
grep '"run_id":"7c2a"' ~/.cache/agent-obs.jsonl

No dashboard. No SaaS. No vendor lock-in. Plain text + awk / jq.

When to upgrade

The 1-line hook is for getting started. It is not the final observability layer. Graduate to a real system when:

1. You have > 1 GB of JSONL per day and grep is slow. Switch to DuckDB or sqlite. (See A3 — JSONL to dashboard with zero infra.)
2. You want distributed tracing across multiple processes. Switch to OpenTelemetry — the zero-code Python instrumentation docs cover the upgrade path.
3. You want dashboards non-engineers can read. Add a Grafana / Metabase layer.
4. You want cost attribution per project. Add a project_id field at the decorator site. (See A4 — cost-aware model routing.)

Until then, the 1-line hook wins on cost, installability, and “actually have it installed.”

Per-platform service mechanism

This guide’s wrapper is a pure Python module — no service, no daemon. The only platform-conditional concern is where the JSONL ledger lives.

Platform	Default ledger path	How to override
macOS	~/.cache/agent-obs.jsonl	export OBS_LEDGER=/path/to/ledger.jsonl in your shell rc or your agent’s env.
Linux	~/.cache/agent-obs.jsonl	Same. For systemd units, set Environment=OBS_LEDGER=... in the unit file.
Docker	/var/log/agent-obs.jsonl (mount a volume)	Bake OBS_LEDGER into the container env, mount the volume.
Termux	~/.cache/agent-obs.jsonl	Same as macOS.
Native Windows	%USERPROFILE%\.cache\agent-obs.jsonl	set OBS_LEDGER=... in the cmd/PowerShell env, or set in the agent config.
WSL2	Treat as Linux	Same.

Security

• The ledger may contain user content. Anything the model saw in the prompt or produced in the output is captured in run_id traces. If your agent handles personal data or secrets, redact before storing or set up a separate ledger for sensitive flows.
• Token counts leak billing information. A competitor with your ledger knows your prompt/response sizes.
• File permissions. The JSONL is plain text on disk. Default to chmod 600 on the ledger directory: mkdir -p ~/.cache && chmod 700 ~/.cache.
• No automatic upload. This guide does not call any network service. If you add a remote sink later, set explicit consent and rate limiting.

How to know it is working

The hook is working when:

1. Calling any decorated function appends a JSONL line within ~1 second.
2. tail -f ~/.cache/agent-obs.jsonl shows live writes while the agent runs.
3. The four one-liners above return useful answers in <1 second.
4. latency_ms is populated for every call (your wrapper passed time.time() correctly).

The hook is not working if ok: true is missing for every call (your function signature doesn’t expose usage or model) or if run_id is the same for every line (you’re not passing distinct run_ids per logical agent run).

Common failures and fixes

Failure	Cause	Fix
The ledger file is missing.	OBS_LEDGER env not set; parent dir doesn’t exist.	The wrapper creates the parent dir on import. Verify with ls -la $(python -c "import obs; print(obs.LEDGER)").expanduser().
tokens_in / tokens_out are always null.	The wrapped function returns an object without a .usage attribute.	Set them via kwargs instead, or wrap a function that returns a Usage object (raw OpenAI / Anthropic do).
Every line has latency_ms: 0.	The wrapped function returns synchronously before time.time() records the second reading.	Check that your call actually hits the network (e.g. print(latency_ms) before _record).
run_id is the same on every line.	You’re calling the function without __run_id= kwargs and the auto-generated id collides.	Pass an explicit __run_id="my-run-name" per logical agent run.
The decorator loses the wrapped function’s signature.	Forgot functools.wraps(fn).	Add it back — see the official functools.wraps docs.
The OpenTelemetry upgrade path isn’t a 1:1 replacement.	The decorator captures six fields; OpenTelemetry semconv has more.	Map your existing JSONL into OpenTelemetry spans when you graduate. The contrib repo has one for your framework.

When not to use this pattern

• You already have OpenTelemetry. Adding a parallel decorator is duplicate work; delete this and instrument the SDK calls directly.
• Your agent is a managed product with its own telemetry. Don’t bypass it; pipe through their hook instead.
• You need real-time alerting. This hook is post-hoc. Wire up the OpenTelemetry exporter or send JSONL lines to a tail-based alerting daemon.
• You need cross-process tracing. A decorator inside one process cannot see calls in another. Use OpenTelemetry with a context propagator.

Updates and breaking changes

Re-run the smoke test after any of these:

• Python ≥ 3.11 → 3.12 transition. The time.strftime and pathlib calls are stable, but functools.wraps semantics on *args, **kwargs can shift on Python 3.12+.
• OpenAI Python SDK major version bumps — r.choices[0].message.content and r.usage.input_tokens have moved between SDK versions.
• Anthropic Python SDK major version bumps — r.content[0].text and r.usage.input_tokens may move.
• LangChain refactor (langchain.chat_models → langchain_openai.ChatOpenAI). The wrapper still works; the import path is what changes.

What you will have at the end

• A 30-line observability library.
• A drop-in decorator that works with any agent framework.
• Four shell one-liners for the questions you will actually ask.
• A clear list of when to graduate to a bigger system.

Where to get unstuck

• Python functools.wraps not behaving: official docs.
• OpenTelemetry zero-code Python instrumentation when you graduate: docs, contrib repo.
• OpenLLMetry for the per-framework OpenTelemetry instrumentation libraries: GitHub, docs.
• Open issues in this repo if the wrapper fails on a specific framework — include the SDK version and a minimal repro.

Sources

• Python — functools.wraps — used for: the decorator helper that preserves __name__, __doc__, signature. Verified 2026-07-22 via web_extract (canonical URL returned 200).
• OpenTelemetry — Zero-code Python instrumentation — used for: the production-grade version of this idea, the upgrade path. Verified 2026-07-22.
• OpenTelemetry Python contrib repo — used for: per-framework auto-instrumentation (OpenAI, Anthropic, LangChain) when scale justifies it. Verified 2026-07-22.
• OpenLLMetry — used for: the per-call JSONL/spans schema (latency_ms, tokens_in, tokens_out, model) we mirror. Verified 2026-07-22.
• OpenLLMetry docs — used for: vendor docs explaining the per-call capture pattern. Verified 2026-07-22.

Last verified: 2026-07-22. Adapted from Marvin’s original “A2. The 1-Line Observability Hook That Fits Any Agent” (2026-07-20). Drafter and source-notes author are the same model (MiniMax-M3). Review lineage: pending independent subagent review per the v2.4.0 skill rule for Setup-category guides — review transcript will be added to the source-notes page when complete.