Adding Trace IDs and Context to Python CLI Logs

A user reports that "the sync failed last night". The log file has four thousand lines from a dozen runs, two of them overlapping because a cron job and a person ran the tool at the same time. Which lines belong to the failed run? Which item was being processed when it failed? Which release of the tool was it? And what did the server see at that moment? Every one of those questions is answered by the same technique: attach a small set of context fields — a run ID, the command, the version, and the current item — to every log record, and send the run ID to the services the CLI talks to. This guide builds that with nothing but the standard library's contextvars and logging, shows how to carry the context into worker threads, and tests it. It belongs to the structured logging for CLI apps topic.

Prerequisites

• Python 3.10+ and a Typer or Click CLI that already logs through logging; see structured JSON logging in Python CLIs for the basic setup.
• Optionally, an HTTP client such as httpx if the CLI calls APIs.

One ID from the command to the server

The design has four parts. At startup the CLI generates a run ID — a short random string, unique per invocation. It stores the ID in a context variable, together with other fields that stay constant for the run. A logging filter copies those fields onto every record, so formatters can print them without any call site mentioning them. And the HTTP client sends the run ID as a request header, so the server's logs for the same operation can be found by searching for the ID the user quotes in their bug report.

Which fields are worth adding

Keep the set small and stable. run_id groups the lines of one invocation. command says what the user was doing, version says which release did it — invaluable once several versions are in the wild. Target fields such as the profile or environment say where the command was pointed. And a per-unit field such as item says which file, record or host was being processed when something went wrong. Everything else belongs in the message itself.

The recipe

The context lives in one module. A ContextVar holds a dictionary of fields; bind() adds fields for the rest of the run, bound() adds them for a block, and a filter and two formatters put them on the output:

# src/mytool/context.py
from __future__ import annotations

import contextvars
import json
import logging
import uuid
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Any

_fields: contextvars.ContextVar[dict[str, Any]] = contextvars.ContextVar("log_fields", default={})


def new_run_id() -> str:
    return uuid.uuid4().hex[:12]


def bind(**fields: Any) -> None:
    """Add fields to every log record for the rest of this context."""
    _fields.set({**_fields.get(), **fields})


@contextmanager
def bound(**fields: Any) -> Iterator[None]:
    """Add fields for the duration of a block, e.g. one item of work."""
    token = _fields.set({**_fields.get(), **fields})
    try:
        yield
    finally:
        _fields.reset(token)


def current() -> dict[str, Any]:
    return dict(_fields.get())


class ContextFilter(logging.Filter):
    """Copy the bound fields onto each record as attributes."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.ctx = current()
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        event = {"level": record.levelname.lower(), "msg": record.getMessage(),
                 "logger": record.name, **getattr(record, "ctx", {})}
        if record.exc_info:
            event["exc"] = self.formatException(record.exc_info)
        return json.dumps(event, default=str)


class TextFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        ctx = " ".join(f"{k}={v}" for k, v in getattr(record, "ctx", {}).items())
        return f"{record.levelname:<7} {record.getMessage()}  {ctx}".rstrip()

The CLI binds the run-level fields once, in the callback that runs before every command, and binds the item inside the unit of work:

# src/mytool/cli.py
from __future__ import annotations

import logging
import sys
from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context
from importlib.metadata import PackageNotFoundError, version

import typer

from mytool.context import ContextFilter, JsonFormatter, TextFormatter, bind, bound, current, new_run_id

app = typer.Typer()
log = logging.getLogger("mytool")

try:
    VERSION = version("mytool")
except PackageNotFoundError:
    VERSION = "dev"


def setup_logging(json_logs: bool) -> None:
    handler = logging.StreamHandler(sys.stderr)
    handler.addFilter(ContextFilter())
    handler.setFormatter(JsonFormatter() if json_logs else TextFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(logging.INFO)


def upload(name: str) -> None:
    with bound(item=name):
        if name.startswith("b"):
            log.error("upload failed")
        else:
            log.info("uploaded")


@app.callback()
def main(ctx: typer.Context, json_logs: bool = typer.Option(False, "--json-logs")) -> None:
    """Sync tool with correlated logs."""
    setup_logging(json_logs)
    bind(run_id=new_run_id(), version=VERSION, command=ctx.invoked_subcommand)


@app.command()
def sync(items: list[str]) -> None:
    """Upload ITEMS in parallel."""
    log.info("sync started")
    with ThreadPoolExecutor(max_workers=4) as pool:
        # Threads do not inherit context variables: run each task in a copy of ours.
        futures = [pool.submit(copy_context().run, upload, item) for item in items]
        for f in futures:
            f.result()
    log.info("sync finished")
    typer.echo(f"run {current()['run_id']} done", err=True)


def request_headers() -> dict[str, str]:
    """Send the run ID to APIs so their logs can be joined with ours."""
    return {"X-Request-ID": current().get("run_id", "")}

Running mytool --json-logs sync a.csv b.csv produces lines that can be grouped and filtered with any log tool:

Why it is built this way

Context variables, not globals. A module-level dictionary would work for a single-threaded command, but context variables are what asyncio tasks and copy_context() understand. Each task sees its own item, while all of them share the run's run_id, and nothing needs to pass a logger adapter through every function signature.

The filter goes on the handler. A filter attached to the handler sees every record that reaches it, including records from libraries' loggers that propagate to the root. Attaching it to your own logger only would leave third-party lines without a run ID — exactly the lines you need when a library is misbehaving.

Copy the context into threads. ThreadPoolExecutor does not carry the submitting thread's context variables into worker threads by default. Wrapping each task in copy_context().run gives the worker a snapshot of the current fields; any bound() inside the worker affects only that copy. The same concern appears in parallelising CLI work with thread pools. Asyncio tasks copy the context automatically when they are created.

The ID is short and printed once. Twelve hex characters are unique enough for correlating one user's runs and short enough to read aloud. Printing it at the end (on stderr, so pipelines stay clean) gives the user something to quote: "run 76c4371f9fea failed".

Sending the ID to APIs

The run ID becomes much more useful when the server logs it too. With httpx, an event hook adds the header to every request the client makes, so no call site can forget it:

import httpx

from mytool.context import current


def add_request_id(request: httpx.Request) -> None:
    request.headers.setdefault("X-Request-ID", current().get("run_id", ""))


client = httpx.Client(base_url="https://api.example.com", event_hooks={"request": [add_request_id]})

X-Request-ID is the de facto header most proxies and frameworks recognise; if your platform uses W3C trace context (traceparent), send that instead, generated from the same run ID. The client setup this plugs into is described in building an API client CLI with httpx.

UX considerations

• Show the ID when it helps. Print it in error messages ("error: upload failed (run 76c4371f9fea)") and in the --verbose footer, not on every successful run. Combined with friendly error messages and tracebacks, it turns a vague report into a searchable one.
• Keep text mode readable. Human-readable output puts context after the message, so the message stays the first thing the eye reads; JSON mode puts everything in fields for machines.
• Never bind secrets. Context fields end up in every line, so a token bound by mistake is copied thousands of times. Bind identifiers, not credentials, and keep the rules from redacting secrets from CLI output and logs in mind.
• Accept an ID from outside. When a CI job or a wrapper script already has a correlation ID, let MYTOOL_RUN_ID override the generated one so the CLI's lines join the caller's.

Testing the behaviour

Tests parse the JSON output and check that every record from one run has the same ID, that the command is recorded, and that the failing line names its item:

# tests/test_context.py
import contextvars
import json

from typer.testing import CliRunner

from mytool.cli import app
from mytool.context import bind, bound, current

runner = CliRunner()


def test_every_record_carries_the_run_id():
    result = runner.invoke(app, ["--json-logs", "sync", "a.csv", "b.csv", "c.csv"])
    events = [json.loads(line) for line in result.output.splitlines() if line.startswith("{")]
    run_ids = {e["run_id"] for e in events}
    assert len(run_ids) == 1 and len(events) == 5
    assert all(e["command"] == "sync" for e in events)
    failed = [e for e in events if e["level"] == "error"]
    assert failed == [{**failed[0], "item": "b.csv"}]


def test_bound_fields_are_scoped():
    def body():
        bind(run_id="r1")
        with bound(item="x"):
            assert current() == {"run_id": "r1", "item": "x"}
        assert current() == {"run_id": "r1"}
    contextvars.Context().run(body)      # a fresh, empty context: no leakage between tests


def test_text_format_is_readable():
    result = runner.invoke(app, ["sync", "a.csv"])
    assert "uploaded  run_id=" in result.output and "item=a.csv" in result.output

The second test runs inside a fresh contextvars.Context() for a reason worth knowing: CliRunner invokes the app in the test's own thread, so fields bound by one invocation are still set when the next test starts. In a real process that never matters — each run is a new process — but unit tests of context helpers should isolate themselves the same way.

Conclusion

Correlated logs cost a module of about fifty lines: a context variable holding a dictionary, bind() for run-level fields and bound() for per-item fields, a filter on the handler that copies them onto every record, and formatters that print them. Generate a short run ID at startup, bind the command and version with it, copy the context into worker threads, send the ID to servers as X-Request-ID, and show it in error messages. After that, "the sync failed last night" becomes one grep on each side.

Frequently asked questions

Related

• Up: Structured logging for CLI apps
• Structured JSON logging in Python CLIs
• Writing rotating log files from a CLI
• Adding verbose and quiet logging flags
• Building an API client CLI with httpx