Skip to content

Propagating Request IDs with contextvars

A customer reports a failed checkout at 14:02. The API gateway log has the request, the orders service log has an error at 14:02, the payments service log has three errors at 14:02, and nothing connects them. Somebody greps timestamps across three dashboards for an hour. A request ID fixes this only if it is present on every log line and every hop: generated or accepted once at the edge, attached to every record the service writes — including those from worker threads, background tasks and third-party libraries — and forwarded on every outbound call and queued message. In an asyncio service, the correct carrier for that ID is a contextvars.ContextVar, because tasks copy context automatically and the event loop keeps each request's value separate while thousands interleave on one thread. This guide builds the full chain with no framework lock-in: ASGI middleware, a logging filter, an outbound HTTP hook, and queue propagation.

Prerequisites

  • Python 3.11+. The middleware and logging code are standard library only; the outbound example uses httpx (pip install httpx).
  • The context model from Context Variables & Request Context: tasks copy context at creation, and changes never flow back to the parent.
  • An ASGI application (Starlette, FastAPI, Quart, or a bare ASGI callable). The same pattern applies to aiohttp middleware.
One ID, every hop 5 stages from edge header to queued jobs. One ID, every hop edge header accept or mint middleware set and reset log records filter stamps it outbound HTTP header hook queued jobs explicit field Tasks carry the ID automatically; network hops and queues need it written down.

1. Declare the variable once, at import time

Create the ContextVar at module level in a small shared module, with a default that is obviously "no request". Creating context variables per request is a leak — each ContextVar object lives forever in every context that references it — and scattering them across modules makes it impossible to find every reader.

# correlation.py
import contextvars
import re
import uuid

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

_VALID = re.compile(r"^[A-Za-z0-9._-]{8,64}$")


def accept_or_generate(incoming: str | None) -> str:
    """Trust a well-formed upstream ID; otherwise mint a new one."""
    if incoming and _VALID.match(incoming):
        return incoming
    return uuid.uuid4().hex

Validation matters because the ID comes from the network. An unvalidated header can inject newlines into log lines, overflow indexes with megabyte-long values, or smuggle characters that break a log shipper's parser. Accept only a bounded, boring alphabet.

Verify: accept_or_generate("abc12345-from-edge") returns the input, and accept_or_generate("bad value!\n") returns a fresh 32-character hex string.

2. Set it in ASGI middleware and echo it back

Middleware is the one place every request passes through, so it is where the ID is set — and, just as importantly, reset. Writing it as pure ASGI middleware keeps it independent of any framework, and echoing the ID in a response header lets clients and support staff quote it.

import asyncio


class RequestIdMiddleware:
    """Pure ASGI middleware: works under Starlette, FastAPI, Quart or a bare ASGI app."""

    header = b"x-request-id"

    def __init__(self, app) -> None:
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        incoming = dict(scope["headers"]).get(self.header, b"").decode("latin-1")
        rid = accept_or_generate(incoming)
        token = request_id.set(rid)

        async def send_with_id(message):
            if message["type"] == "http.response.start":
                message.setdefault("headers", []).append((self.header, rid.encode()))
            await send(message)

        try:
            await self.app(scope, receive, send_with_id)
        finally:
            request_id.reset(token)              # never let the value outlive the request

The ASGI server runs each request in its own task, so request_id.set() affects only that request, and every task the handler creates — including children of a TaskGroup — inherits the value automatically. Register it as the outermost middleware so that authentication, rate limiting and error handlers all log with the ID.

Verify: drive the middleware with three concurrent fake requests — one with a valid header, one without, one with an invalid value. Each response carries its own ID in x-request-id, and after all three complete, request_id.get() in the calling task still returns -.

3. Stamp every log record with a filter

A filter attached to the handler runs for every record that reaches it, from any logger, and reads the variable in the context of the code that made the log call. That covers your code, libraries such as httpx and asyncpg, and — through asyncio.to_thread() — blocking code running in worker threads.

import json
import logging


class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname,
            "logger": record.name,
            "request_id": getattr(record, "request_id", "-"),
            "msg": record.getMessage(),
        })


def configure_logging() -> None:
    handler = logging.StreamHandler()
    handler.addFilter(RequestIdFilter())
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers[:] = [handler]
    root.setLevel(logging.INFO)

Emitting request_id as a structured field rather than a prefix in the message is what makes it queryable: log platforms index the field, and "show everything for this request across all services" becomes a single filter.

Verify: log from the handler, from a child task, and from a function run with asyncio.to_thread(); all three JSON lines carry the same request_id. A line with "request_id": "-" inside a request means a boundary dropped the context — see carrying contextvars across threads and executors.

Where the filter finds the ID 4 stacked layers from Request handler to Library loggers. Where the filter finds the ID Request handler request task context set here TaskGroup children copied context same request_id asyncio.to_thread work context copied in same request_id Library loggers httpx, asyncpg filter on handler Attach the filter to the handler so every logger passes through it.

4. Forward the ID on outbound HTTP calls

Correlation stops at the service boundary unless the ID travels with outbound requests. An httpx event hook reads the variable when each request is sent, so one long-lived client serves every request without per-call plumbing.

# pip install httpx
import httpx


async def add_request_id(request: httpx.Request) -> None:
    rid = request_id.get()
    if rid != "-" and "x-request-id" not in request.headers:
        request.headers["x-request-id"] = rid


def make_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        event_hooks={"request": [add_request_id]},
        timeout=httpx.Timeout(5.0),
    )

The hook runs inside the task that sends the request, so it reads that request's ID even though the client object is shared process-wide, as it should be per reusing a client session across requests. If you use OpenTelemetry, its traceparent propagation works the same way underneath — it also stores the active span in a context variable — and the request ID can ride along as baggage or stay a separate header.

Verify: point the client at an echo endpoint inside a request scope; the echoed headers include x-request-id matching the incoming request. Outside any request, no header is added.

5. Carry the ID through queues and background work

Context variables do not cross a queue: the consumer task was created long before the message arrived and has its own context. Put the ID in the message explicitly at the producer, and restore it around processing at the consumer.

import asyncio
import contextlib
from dataclasses import dataclass, field


@dataclass(frozen=True)
class Job:
    payload: dict
    request_id: str = field(default_factory=request_id.get)   # captured at creation


@contextlib.contextmanager
def restored_request_id(rid: str):
    token = request_id.set(rid)
    try:
        yield
    finally:
        request_id.reset(token)


async def consumer(queue: asyncio.Queue[Job]) -> None:
    while True:
        job = await queue.get()
        try:
            with restored_request_id(job.request_id):
                await process(job.payload)           # logs and outbound calls carry the ID
        finally:
            queue.task_done()

Capturing the ID with default_factory=request_id.get means producers do not have to remember to pass it. The same explicit handoff applies to broker messages — put it in message headers — and to dead-letter queues, where the original request ID is exactly what an engineer needs when inspecting a poison message days later.

Verify: enqueue jobs from two concurrent requests and log inside process; each log line carries the ID of the request that enqueued the job, and after the consumer finishes a job its context is back to -.

Tasks versus queues 2 columns contrasting child task, queue consumer. Tasks versus queues child task context copied for you created inside the request inherits request_id no code needed queue consumer context must be rebuilt created at startup has its own context restore from job field Anything that waits in a queue loses its context unless the message carries it.

Verification

Request correlation is complete when:

  • Every log line inside a request has its ID, including lines from child tasks, worker threads and third-party loggers; the rate of "-" during request handling is near zero.
  • IDs never mix: under concurrent load, no request's log lines contain another request's ID.
  • The ID crosses the edge both ways: it is accepted from upstream when valid, generated otherwise, and returned in the response header.
  • Downstream services receive it on every outbound HTTP call and queued message, so one query returns the full path.
  • Nothing outlives its request: the variable is reset in middleware and consumers, and background tasks do not log stale IDs.

Pitfalls & edge cases

  • Middleware registered too far in. If the ID middleware runs after authentication or exception handling, errors in those layers log without an ID. Make it the outermost layer.
  • Background tasks started inside a request. A long-lived task created lazily during the first request inherits that request's ID for its whole life. Start background tasks at application startup, or create them with context=contextvars.Context().
  • Trusting any incoming header. Unvalidated IDs are a log-injection vector. Validate length and alphabet, and generate a new ID when validation fails rather than rejecting the request.
  • Streaming responses. For a streamed body, the handler returns before the stream finishes sending. Make sure the middleware's reset() happens after the last send, which the pure ASGI version above guarantees because it awaits the whole application call.
  • WebSocket connections. One connection carries many messages over a long time. Set a connection-level ID in the middleware for "websocket" scopes as well, and consider a per-message ID for long-lived WebSocket streams.

Frequently Asked Questions

How do I add a request ID to every log line in an asyncio app?

Store the ID in a module-level contextvars.ContextVar, set it in middleware at the start of each request and reset it at the end, and attach a logging.Filter to your handler that copies request_id.get() onto each record. Because tasks copy context, every coroutine and child task in the request logs with the correct ID.

Why not store the request ID in a global or thread-local variable?

An asyncio server handles many requests on one thread and switches between them at every await, so a global or thread-local variable holds whichever request wrote it last. Log lines would carry the wrong ID. A ContextVar keeps a separate value for each task, which is exactly the per-request isolation needed.

Should I trust an X-Request-ID header sent by the client?

Accept it only after validation. A well-formed ID from an upstream gateway lets you correlate across systems, but an arbitrary header can inject newlines or huge values into logs. Restrict it to a short alphabet and bounded length, and generate a new ID when the incoming value fails the check.

How do I keep the request ID when work goes through an asyncio.Queue?

Context does not travel through a queue, because the consumer task has its own context. Store the ID on the job object when it is created, for example with a dataclass field whose default factory is request_id.get, then set it around processing in the consumer and reset it afterwards.