Correlation IDs and Trace Context in Async Python

A request enters your service, fans out to two services, kicks off a Celery task, ends up in a Kafka consumer. Connecting all that activity to “this user clicked X” requires correlation IDs propagated through every async boundary. Python’s contextvars is the primitive; propagation across boundaries is where it gets tricky.

The problem

request → service A
         ├─> HTTP call to service B    [different process]
         ├─> publish to Kafka          [different process eventually]
         └─> Celery task               [different worker]
                  └─> publish to Kafka  [different process]

To debug “what happened in this user’s flow”, every log line, span, and metric across all five hops needs the same correlation ID.

The primitive: contextvars

Python’s contextvars module (3.7+) gives per-task-local storage that survives await correctly:

import contextvars

trace_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("trace_id")

async def handler():
    trace_id_var.set("abc-123")
    await call_downstream()       # trace_id is still "abc-123" inside this call

async def call_downstream():
    trace_id = trace_id_var.get()  # "abc-123"

Unlike threading.local, contextvars propagates across await automatically. Each asyncio Task gets its own context (forked from the spawner). Bind once at request entry, read anywhere on the chain.

Wiring it in FastAPI / Starlette

ASGI middleware that sets the trace_id from the incoming header (or generates one):

import uuid
from starlette.middleware.base import BaseHTTPMiddleware

trace_id_var = contextvars.ContextVar("trace_id", default=None)

class TraceIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        tid = request.headers.get("x-trace-id") or str(uuid.uuid4())
        token = trace_id_var.set(tid)
        try:
            response = await call_next(request)
            response.headers["x-trace-id"] = tid
            return response
        finally:
            trace_id_var.reset(token)

app.add_middleware(TraceIDMiddleware)

Every coroutine inside the request handler can trace_id_var.get().

Putting it in logs (structlog)

import structlog

def add_trace_id(logger, method, event_dict):
    tid = trace_id_var.get()
    if tid:
        event_dict["trace_id"] = tid
    return event_dict

structlog.configure(
    processors=[
        add_trace_id,
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger()

async def handler():
    log.info("processing", user_id=42)
    # {"event": "processing", "user_id": 42, "trace_id": "abc-123", "level": "info"}

Every log line in the request flow has the trace_id without code changes.

Propagation: outbound HTTP (httpx)

import httpx

class TraceIDTransport(httpx.AsyncHTTPTransport):
    async def handle_async_request(self, request):
        tid = trace_id_var.get()
        if tid:
            request.headers["x-trace-id"] = tid
        return await super().handle_async_request(request)

client = httpx.AsyncClient(transport=TraceIDTransport())

Or just set it on each call manually. The middleware-on-transport pattern is cleaner.

For full distributed tracing, use OpenTelemetry’s HTTPX instrumentation — it injects traceparent automatically. Custom trace_id is supplementary.

Propagation: into Celery

Celery doesn’t propagate contextvars across the broker. Three options:

Option 1: Send in the message

@before_task_publish.connect
def add_trace_id(headers=None, **kwargs):
    tid = trace_id_var.get()
    if tid:
        headers["trace_id"] = tid

@task_prerun.connect
def restore_trace_id(task_id, task, **kwargs):
    request = task.request
    tid = (request.headers or {}).get("trace_id")
    if tid:
        trace_id_var.set(tid)

Option 2: OpenTelemetry Celery instrumentation

Auto-propagates trace context for you:

from opentelemetry.instrumentation.celery import CeleryInstrumentor
CeleryInstrumentor().instrument()

If you’re already using OTel, this is the right answer. Trace context flows via the Celery message headers.

Option 3: Pass as an explicit argument

@app.task
def process(payload, trace_id):
    trace_id_var.set(trace_id)
    do_work(payload)

process.delay(payload, trace_id_var.get())

Less elegant; works.

Propagation: into Kafka

Same pattern — set the trace_id as a Kafka message header:

def publish(topic, value, key=None):
    tid = trace_id_var.get()
    headers = []
    if tid:
        headers.append(("trace_id", tid.encode()))
    producer.send(topic, value=value, key=key, headers=headers)

Consumer side:

for msg in consumer:
    headers = dict(msg.headers or [])
    tid = headers.get("trace_id")
    if tid:
        trace_id_var.set(tid.decode() if isinstance(tid, bytes) else tid)
    process(msg.value)

OpenTelemetry’s Kafka instrumentation does this automatically for traceparent.

Threads and run_in_executor

# WRONG — contextvars don't transfer to threads automatically
result = await loop.run_in_executor(None, sync_function)

asyncio.to_thread propagates contextvars (3.9+); raw run_in_executor does not. For threads:

ctx = contextvars.copy_context()
result = await loop.run_in_executor(None, ctx.run, sync_function)

copy_context() snapshots the current contextvars; ctx.run(fn) executes fn inside that snapshot.

OpenTelemetry trace IDs

If you’re using OTel, the trace ID is already in OTel Span Context:

from opentelemetry import trace

def get_trace_id():
    span = trace.get_current_span()
    if span and span.get_span_context().is_valid:
        return f"{span.get_span_context().trace_id:032x}"
    return None

Use OTel’s trace_id rather than a separate UUID — same ID flows through all OTel instrumentations and your logs, automatically connecting traces ↔ logs ↔ metrics.

For logs, OTel’s logging instrumentation auto-injects trace_id and span_id.

What to actually correlate

Stack them in your log lines. The trace_id is the one that bridges services.

Common gotchas

• contextvars set in middleware, lost after response. ContextVar lives for the duration of the coroutine that set it (or its tasks). Once the request handler returns, downstream tasks (background, etc.) lose the binding. Pass explicitly or use contextvars.copy_context() when spawning.
• asyncio.create_task without context copy. The new task gets a copy of the spawning context — that works. But if you spawn from a sync callback or run_in_executor without context, the new task starts empty.
• Celery tasks lose contextvars across the broker. Always do explicit propagation via headers + restore in task_prerun.
• Background threads (not asyncio.to_thread) lose contextvars. Use copy_context().run(fn) to bridge.
• Trace ID conflicts with OTel trace ID. Pick one. Either use OTel and use its trace_id, or roll your own and don’t use OTel. Mixing creates “which one matters” confusion.

Interview angle

• “How do you propagate a request ID through async Python?” — contextvars.ContextVar. Set in middleware at request entry; reads anywhere downstream in coroutines. Propagates across await and asyncio.Task automatically.
• “How do you propagate a trace ID into a Celery task?” — Celery doesn’t propagate contextvars across the broker. Use Celery signals (before_task_publish to inject into headers, task_prerun to restore in the worker), or OpenTelemetry’s Celery instrumentation which does it for you.
• “How does contextvars differ from threading.local?” — threading.local is per-thread; loses values across await (because asyncio runs many coroutines on one thread). contextvars is per-task; properly propagates across await, forked correctly when spawning child tasks.
• “Why don’t contextvars propagate into loop.run_in_executor?” — threads don’t inherit asyncio task context automatically. Workaround: contextvars.copy_context().run(fn) snapshots the current context. asyncio.to_thread does this for you.
• “What’s the relationship between OpenTelemetry trace IDs and your application correlation ID?” — they can be the same thing if you commit to OTel. OTel’s trace_id is auto-generated, propagated via traceparent, available everywhere via trace.get_current_span(). Custom correlation IDs duplicate that work; prefer OTel when feasible.
• “You have a Celery worker running a task that runs in run_in_executor. Trace ID is missing in the thread’s logs. Why?” — three boundary crossings: Celery (broker) lost it; worker started fresh; run_in_executor doesn’t copy contextvars to the thread. Need explicit propagation at each layer.