Stop Losing Request Context in Async Python

Use contextvars deliberately across tasks, logging, and thread boundaries in Python 3.14.

One request boundary, three lanes, and each lane reads its own binding.

Request IDs, tenant IDs, trace IDs, and authenticated principals often need to reach code several calls below an API handler. Passing each value through every function is explicit, but it can turn infrastructure metadata into parameters across an entire call graph. A module-level global looks simpler until two requests overlap. Thread-local storage works for synchronous threads, but an event loop runs many tasks on one thread.

Python’s contextvars module is built for this problem. A ContextVar stores a value in the current execution context, and asyncio tasks preserve separate contexts while they interleave. That solves the basic isolation problem, but it does not remove the need to design boundaries. A task captures context at a specific time. asyncio.to_thread() propagates it, while lower-level executor calls require an explicit choice. A value that is never reset can also leak into later work in the same task.

This article targets CPython 3.14 and uses only the standard library. The central rule is simple: treat context as scoped execution metadata, not as hidden application state.

A global fails as soon as requests overlap

The failure is easy to reproduce with one thread and two coroutines:

import asyncio
 
current_request = "-"
 
 
async def handle(request_id: str, delay: float) -> None:
    global current_request
    current_request = request_id
    await asyncio.sleep(delay)
    print(f"{request_id} logged as {current_request}")
 
 
async def main() -> None:
    await asyncio.gather(
        handle("request-alpha", 0.02),
        handle("request-beta", 0.01),
    )
 
 
if __name__ == "__main__":
    asyncio.run(main())

Both handlers write the same variable before suspending. The beta request wakes first and prints its own value. The alpha request wakes later but also reads beta, because beta was the last writer. No operating-system thread race is required. Cooperative scheduling is enough to corrupt the association.

Adding a lock would serialize the handlers and defeat normal concurrency. Passing request_id explicitly through every function would be correct, and it remains the right choice when the value affects business behavior. For cross-cutting metadata used by logging, tracing, or diagnostics, a context variable gives each execution path its own binding without changing every function signature.

With a module global, the last writer wins and alpha reads beta. With a ContextVar, each task keeps its own binding across await.

Bind metadata at the request boundary

Declare a ContextVar once at module scope, then set it at the boundary that knows the value. In Python 3.14, the token returned by ContextVar.set() supports the context manager protocol, so leaving the with block restores the previous binding automatically.

import asyncio
import logging
from contextvars import ContextVar
 
request_id: ContextVar[str] = ContextVar("request_id", default="-")
 
 
class RequestContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True
 
 
def configure_logging() -> logging.Logger:
    handler = logging.StreamHandler()
    handler.addFilter(RequestContextFilter())
    handler.setFormatter(
        logging.Formatter("request=%(request_id)s %(message)s")
    )
 
    logger = logging.getLogger("context-demo")
    logger.handlers.clear()
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False
    return logger
 
 
logger = configure_logging()
 
 
async def store_record() -> None:
    await asyncio.sleep(0.01)
    logger.info("record stored")
 
 
async def handle(value: str) -> None:
    with request_id.set(value):
        logger.info("request started")
        await store_record()
        logger.info("request finished")
 
 
async def main() -> None:
    await asyncio.gather(
        handle("request-alpha"),
        handle("request-beta"),
    )
 
 
if __name__ == "__main__":
    asyncio.run(main())

The two handlers interleave, but every log record gets the request ID from the task that emitted it. store_record() does not need a metadata parameter. The filter injects request_id before the formatter reads it, and the default "-" keeps unrelated log calls format-safe when no request scope exists.

The with block matters as much as the variable. Long-lived tasks, test runners, and worker loops can process more than one operation sequentially. If a handler sets context and never restores it, the next operation in that task may inherit stale metadata. On Python 3.13 and earlier, keep the token from set() and call reset(token) in a finally block. The lifecycle is the same even though the syntax is longer.

A logging filter is one integration option, not a requirement. A LoggerAdapter, a custom LogRecord factory, or a tracing library can read the same context variable. Keep the binding independent of the output mechanism. That lets tests inspect context without parsing logs and prevents the logging layer from becoming the owner of request lifecycle.

Do not put mutable application objects into context merely to avoid parameters. A database session, shopping cart, or permissions object has a lifecycle and behavior that callers should see. Context is best for small, immutable metadata whose meaning is tied to the current operation.

A task receives a snapshot, not a live view

Asyncio copies the current context when create_task() is called unless an explicit context is supplied. The timing is important: creating a coroutine object does not capture task context. Creating the task does.

import asyncio
import contextvars
from contextvars import ContextVar
 
tenant: ContextVar[str] = ContextVar("tenant", default="public")
 
 
async def report(label: str) -> None:
    await asyncio.sleep(0)
    print(f"{label}: {tenant.get()}")
 
 
async def main() -> None:
    tenant.set("alpha")
    coroutine = report("task snapshot")
 
    tenant.set("beta")
    task = asyncio.create_task(coroutine)
 
    tenant.set("gamma")
    await task
 
    isolated = asyncio.create_task(
        report("isolated task"),
        context=contextvars.Context(),
    )
    await isolated
 
 
if __name__ == "__main__":
    asyncio.run(main())

The first task prints beta. That was the binding at task creation, even though the coroutine object was created under alpha and the parent changed to gamma before the child ran. The second task receives an empty Context, so tenant.get() returns its declared default, public. The unscoped tenant.set() calls in main() are acceptable here because main() runs as its own task under asyncio.run(), so those bindings are discarded when it ends.

This behavior is useful for normal child work. A task created inside a request usually should inherit that request’s trace metadata. It is less suitable for detached work that outlives the request. A cache refresher, periodic monitor, or process-wide cleanup task should not accidentally carry a customer or authorization context just because it was spawned while handling one request.

The coroutine object captures nothing. create_task() snapshots beta, and an explicit empty Context() falls back to the default.

For detached tasks, choose the context deliberately. Pass context=contextvars.Context() for a clean context, or construct a copied context and set only the values the task is allowed to inherit. Context is not a security boundary by itself. It is a propagation mechanism, so sensitive authorization decisions should still use explicit validated inputs.

Task ownership remains a separate concern. Context propagation does not make fire-and-forget tasks safe. Keep a strong reference, decide how exceptions are observed, and give long-running tasks an application scope that can cancel and await them during shutdown.

Thread handoffs are not all equivalent

Blocking code often needs a thread so it does not stall the event loop. asyncio.to_thread() copies the current context into that worker call. The lower-level run_in_executor() API does not promise that propagation, so use copy_context().run when the executor call should inherit metadata.

import asyncio
import contextvars
from contextvars import ContextVar
 
trace_id: ContextVar[str] = ContextVar("trace_id", default="-")
 
 
def read_context(label: str) -> None:
    print(f"{label}: {trace_id.get()}")
 
 
async def main() -> None:
    trace_id.set("trace-42")
 
    await asyncio.to_thread(read_context, "to_thread")
 
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, read_context, "plain executor")
 
    context = contextvars.copy_context()
    await loop.run_in_executor(
        None,
        context.run,
        read_context,
        "copied executor",
    )
 
 
if __name__ == "__main__":
    asyncio.run(main())

The to_thread call sees trace-42. The plain executor call sees the default. That result is for the default build. On free-threaded 3.14 builds, or with -X thread_inherit_context=1, executor worker threads start with a copy of the context that was current when the thread was created, so a plain call may see an old value instead of the default. Either way, it is not the submitter's current context. Use copy_context().run when the call must see the caller's metadata. Running the function through a copied context restores the trace value for that submission. This is a propagation difference, not a general reason to prefer one API. run_in_executor() still matters when code needs a specific executor or existing executor-oriented interface.

Create a fresh context copy for each executor submission, concurrent or not. A Context cannot be entered by two threads at the same time. Reusing one copied context concurrently can raise RuntimeError. Sequential reuse is legal but carries any values the previous call set. Calling copy_context() for each submission gives every call an independent snapshot.

Get Udit Rawat’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The copied context also does not merge changes back into its parent. If worker code sets trace_id, that change remains in the worker's context. Return business results explicitly instead of treating context as a hidden output channel.

If your service submits to executors in more than one place, put the copy in one small helper so every call site gets a fresh snapshot:

import asyncio
import contextvars
from collections.abc import Callable
from concurrent.futures import Executor
 
 
async def run_with_context[T](
    executor: Executor | None,
    func: Callable[..., T],
    *args: object,
) -> T:
    loop = asyncio.get_running_loop()
    context = contextvars.copy_context()
    return await loop.run_in_executor(executor, context.run, func, *args)

Process boundaries need another design. Context objects are not a portable cross-process request envelope. Send the required fields as explicit, serialized input, validate them in the child, and bind a new local context there if downstream logging needs it.

create_task() and to_thread() carry context. A plain executor call does not. A per-submission copy_context().run does. Processes need explicit fields.

Test isolation and boundary behavior

Tests should verify which metadata is present, not only that a function returned successfully. Run two operations concurrently with different IDs and capture their records. Assert that each record carries the expected ID. Then test sequential reuse of one long-lived worker task to prove that the first binding is reset before the second operation starts.

Add focused tests for every handoff your service uses:

1. A child task created inside a scope inherits the intended values.
2. A detached task starts with an empty or deliberately reduced context.
3. to_thread() sees the caller's diagnostic metadata.
4. A custom executor wrapper copies context once per submission.
5. Code outside a request scope produces a safe default or a deliberate LookupError.

That last choice deserves attention. A default such as "-" works for optional diagnostics. For a value that must always be established, declare the ContextVar without a default. get() then raises LookupError when the boundary forgot to bind it, which exposes the lifecycle bug instead of quietly assigning work to a fake tenant or user.

Avoid tests that inspect private asyncio task internals. Exercise the public behavior through task creation, logging records, and thread calls. Also reset context in test fixtures or use scoped bindings so one test cannot influence the next when they run in the same task.

Keep the contract narrow

Before adding a context variable, I use this checklist:

1. Is the value execution metadata rather than business input or mutable state?
2. Which boundary sets it, and which scope guarantees restoration?
3. Should ordinary child tasks inherit it?
4. Which detached tasks must start clean?
5. Does work cross through to_thread, a custom executor, or a process?
6. Should missing context use a safe diagnostic default or fail immediately?
7. Do concurrency tests prove isolation between overlapping operations?

Use ContextVar for request IDs, trace identifiers, locale hints, and similarly narrow metadata that follows one logical operation. Pass values explicitly when they determine business behavior, authorization, persistence, or public API results. Explicit parameters make those dependencies visible to readers, type checkers, and tests.

The hard part is not calling set() and get(). It is deciding where context begins, when it is copied, where it must stop, and how it is restored. Once those boundaries are explicit, context variables remove repetitive plumbing without turning important state into an invisible global.

Sources

• Python 3.14 contextvars documentation
• Python 3.14 asyncio task and thread documentation
• Python 3.14 logging documentation
• PEP 567: Context Variables

Udit Rawat is a Senior Software Engineer with more than 10 years of experience and co-founder of Moosky AI.