Bind Request Context When LLM Workers Dequeue Jobs

Last reviewed: November 4, 2026 (Asia/Shanghai).

Direct answer

Carry the job’s approved correlation metadata in an explicit envelope. After a long-lived LLM worker dequeues that job, bind the envelope to a module-level ContextVar, await the complete handler, and reset the returned token in finally. Bind every job, including the first one. A producer setting a context variable before queue.put() does not update an already running consumer task.

The Python Coroutines and Tasks reference states that a task copies the current context when it is created unless an explicit context is supplied. The asyncio Queues reference defines put() and get() as moving items into and out of a queue. Receiving an item does not create a new task with the producer’s context. The local check below demonstrates that boundary.

For example, two HTTP handlers enqueue summaries for jobs A and B. A worker created during application startup keeps its own task context while it processes both. Without an explicit binding, logs may have no request reference. If the worker sets A’s context but forgets cleanup, a later unbound path can retain A’s reference. Both defects can obscure which job owns a failed provider attempt.

Who this is for

This guide is for Python maintainers using long-lived asyncio workers to process queued LLM operations in one process. The examples require Python 3.11 or later and use only the standard library. They apply to correlation metadata read by logging helpers or adapters below the worker’s handler.

Keep authorization, credential selection, deadlines, and durable job ownership explicit in the application contract. A context variable provides convenient access to metadata within execution; it does not authorize a tenant, fence a stale worker, or prove that a result belongs to a particular customer.

Key takeaways

• Put approved job metadata in the queue item; bind it again in the consumer.
• Start each long-lived worker with a deliberate baseline context.
• Reset with the token returned by set() so nested scopes restore correctly.
• Store immutable values and avoid copying an entire request context into a job.
• Await job-owned child tasks before leaving the binding scope.
• Test reuse of one worker across multiple jobs, plus failure and cancellation.

Contract details to verify

Distinguish task creation from job delivery

Context inheritance happens when a task is created. An existing worker waiting in queue.get() is still the same task when it resumes. Its context is independent of the producer’s later changes. Likewise, directly awaiting another coroutine runs it within the current task; that await does not capture a fresh producer context.

Choose the worker’s startup context deliberately. For the application-owned variable used below, an empty baseline makes accidental missing bindings visible. Python 3.11 added the context argument to asyncio.create_task(). Create each worker with a fresh Context() where that is compatible with the application’s instrumentation:

from contextvars import Context

task = asyncio.create_task(worker(), context=Context())

This expression assumes asyncio and the worker function are already defined. Construct a separate context for every worker. An explicitly supplied context is the task’s execution context; sharing the same mutable context object among workers defeats the intended separation. If startup needs selected service or tracing values, initialize those deliberately rather than inheriting the request that happened to launch a replacement worker.

The site’s trace propagation guide covers causal tracing across service boundaries. Here, the problem is the local execution scope after a queue handoff. A tracing library may need its own supported extract, attach, and detach calls alongside the application binding.

Bind one immutable scope around the whole handler

Save this as job_context.py:

from contextvars import ContextVar
from dataclasses import dataclass


@dataclass(frozen=True)
class JobContext:
    job_ref: str
    request_ref: str
    tenant_ref: str


@dataclass(frozen=True)
class Job:
    scope: JobContext
    input_ref: str


current_job: ContextVar[JobContext] = ContextVar("llm_job")


async def run_bound(job, handle):
    previous_binding = current_job.set(job.scope)
    try:
        return await handle(job)
    finally:
        current_job.reset(previous_binding)

At the worker’s processing boundary, call await run_bound(job, existing_handler). The handler should encompass the job’s locally owned work, including response cleanup and outcome recording. A logging helper inside that scope can read current_job.get() without threading correlation arguments through every helper signature.

The contextvars reference recommends declaring context variables at module level. It documents that set() returns a token and reset(token) restores the previous binding. With no default, get() raises LookupError when no value is bound. Treat that as an instrumentation defect in code that requires a job scope; do not silently substitute a global tenant or the last observed job.

Using the token matters when scopes nest. Setting the value to None on exit would erase an outer job binding. Resetting the token restores that outer value, or restores the absence of a value when none existed. Reset it once, in the same execution context where it was created. The explicit try/finally also avoids relying on token context-manager syntax, which the reference marks as added in Python 3.14.

The dataclasses use strings and a frozen scope to discourage accidental mutation. Context copies are shallow: placing a shared dictionary in a context variable does not make its contents private to each task. Avoid mutable defaults and mutable request objects. The frozen dataclass is a local programming aid; validate types, lengths, and allowed fields at your actual queue ingress.

Keep authority in the job contract

Build these references from trusted application state. If the queue accepts external messages, validate the envelope and establish the permitted job and tenant before binding. A matching tenant_ref in a log is evidence of the selected label, not evidence of membership or permission.

Use references that are safe for the intended diagnostic store. Keep prompts, generated text, credentials, and raw user identifiers out of this context envelope. A helper that automatically exports every context value can otherwise turn convenient metadata access into unintended data retention.

End child work before resetting the parent

A task created while job A is bound normally copies A’s context. Resetting the worker’s binding afterward does not erase the child’s copy. Detached work can therefore continue logging A after the worker moves to B, or continue consuming resources after the parent job has ended.

Keep request-owned children inside an awaited scope. The site’s TaskGroup ownership guide explains the all-required branch case. An independently durable background job needs its own envelope, owner, and recovery policy. Creating it with the parent’s context does not supply those responsibilities.

Do not enqueue copy_context() as a general job format. It captures the current context’s bindings, including unrelated application or instrumentation values, and does not provide a cross-process serialization contract. Carry the allowlisted metadata you need and reconstruct the consumer’s scope explicitly.

Happy-path operator workflow

Authenticate and authorize the submitting action, create the job envelope, and enqueue it under the existing admission policy. On dequeue, validate the job’s execution eligibility, bind its scope, and await the handler. The handler records its terminal local outcome while that scope is active. After it returns, reset the token before the worker accepts its next job.

Observe two successive jobs on the same worker. Require each handler’s events to carry its own job and request references, and require the worker baseline to be restored between jobs. Record worker identity, job reference, fixed outcome category, and binding revision in restricted diagnostics. Use bounded labels in metrics rather than one metric series per job.

Error-path operator workflow

If the handler raises or receives cancellation, let its resource cleanup finish and let run_bound() reset the binding. Propagate cancellation to the owning scope. Map expected failures through the existing job policy after recording the appropriate outcome; avoid a broad catch that converts cancellation into another model attempt.

If metadata is missing or disagrees with the explicit job envelope, stop affected dispatch or result delivery until the binding path is understood. A label mismatch alone does not establish that the wrong credential or output destination was used. Compare those explicit bindings and the durable job record separately. Preserve sanitized identifiers and dispatch state, and reconcile any uncertain upstream work before authorizing a replay.

Queue accounting also needs a separate success rule. The Queues reference ties join() to the count of task_done() calls. Returning that counter to zero does not establish successful inference or durable result persistence. The probe below uses it only to finish its in-memory fixture; production outcome and retry responsibility must be recorded through the real job contract.

Verify the queue boundary offline

Save the following as probe_job_context.py beside the first file and run python3 probe_job_context.py. It moves two synthetic envelopes through the same worker. It then checks restoration after an exception and cancellation. No model API or network transport is involved.

import asyncio
from contextvars import Context

from job_context import Job, JobContext, current_job, run_bound


def fixture(label):
    return Job(
        JobContext(f"job-{label}", f"req-{label}", f"tenant-{label}"),
        f"input-{label}",
    )


async def queue_trial(bind):
    queue = asyncio.Queue(maxsize=1)
    observed = []

    async def handle(job):
        observed.append(current_job.get(None))

    async def worker():
        for _ in range(2):
            job = await queue.get()
            try:
                if bind:
                    await run_bound(job, handle)
                else:
                    await handle(job)
                assert current_job.get(None) is None
            finally:
                queue.task_done()

    task = asyncio.create_task(worker(), context=Context())
    for label in ("A", "B"):
        job = fixture(label)
        previous_binding = current_job.set(job.scope)
        try:
            await queue.put(job)
        finally:
            current_job.reset(previous_binding)
    await queue.join()
    await task
    return observed


async def checks():
    assert await queue_trial(False) == [None, None]
    assert await queue_trial(True) == [fixture("A").scope, fixture("B").scope]

    async def fail(job):
        assert current_job.get() == job.scope
        raise ValueError("fixture_failure")

    outer = fixture("outer").scope
    previous_binding = current_job.set(outer)
    try:
        try:
            await run_bound(fixture("A"), fail)
        except ValueError:
            pass
        else:
            raise AssertionError("failure_was_lost")
        assert current_job.get() == outer
    finally:
        current_job.reset(previous_binding)

    entered = asyncio.Event()

    async def block(job):
        assert current_job.get() == job.scope
        entered.set()
        await asyncio.Event().wait()

    async def cancellable():
        try:
            await run_bound(fixture("A"), block)
        finally:
            assert current_job.get(None) is None

    task = asyncio.create_task(cancellable(), context=Context())
    await entered.wait()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    assert task.cancelled()
    assert current_job.get(None) is None
    print("queue boundary, reset, failure, and cancellation: passed")


async def main():
    async with asyncio.timeout(5):
        await checks()


asyncio.run(main())

A local run on November 4, 2026 in Asia/Shanghai using Python 3.12.3 printed the stated pass line. Without consumer binding, both observations were None; with binding, they matched A and B respectively. The exception check restored the outer scope, and the canceled task observed an empty baseline after unwinding.

These checks establish the local standard-library behavior shown. They do not verify a logging framework, distributed queue, credential isolation, provider cancellation, or result delivery. Before rollout, repeat the two-job and failure cases through the deployed wrappers. Include a rejected envelope, an awaited child reading its parent’s scope, and a worker replacement. Require each event to match the explicit job envelope without exporting sensitive values.

Failure modes

The producer binds context but the consumer never does. An existing worker retains its own baseline. Put the metadata in the item and bind it after dequeue.

A worker forgets to reset after an error. The next unbound helper can read the previous job’s value. Use one finally boundary that covers the entire handler.

Cleanup assigns a default instead of restoring a token. Nested scopes lose their parent binding. Reset the exact token returned by that scope’s set().

A shared dictionary becomes the context value. Separate context mappings still reference the same mutable object. Use immutable metadata or construct independent values.

A child outlives its job. Its copied context remains even after the parent resets. Await owned children and give independent work a new job contract.

FAQ

Does asyncio propagate context automatically?

New tasks normally copy the context present at task creation. A queue delivering another item to an existing worker is a different boundary. Bind the item’s approved metadata in that worker.

Can ContextVar hold the authorization decision?

It can hold a reference for convenience, but it is not an authorization control. Establish permission and credential selection explicitly, and verify them at the boundaries that dispatch work and accept results.

Will resetting the worker clear existing child tasks?

No. Children that copied the earlier context keep their own bindings. Parent reset does not revoke those copies or cancel the children.

Should a distributed queue carry a Python Context object?

Use a validated, bounded envelope with an explicit serialization schema. Reconstruct local context after receipt. Python execution context is not a durable job or a portable tracing protocol.

Reader next step

Trace one queued LLM job from producer to the worker’s first handler call. Identify where its envelope becomes trusted, where context is bound, and where all child work ends. Run the offline probe, then exercise the same two-job sequence through staging. Accept the change when success, failure, and cancellation all restore the worker baseline and every handler event matches its explicit job envelope.

Sources checked

• Python 3.14: contextvars — Context Variables — fetched November 3, 2026 UTC, November 4 in Asia/Shanghai. Supports module-level variables, missing-value behavior, token restoration, shallow context copying, empty contexts, and asyncio integration.
• Python 3.14: Coroutines and Tasks — fetched at the same time. Supports task context copying, explicit task contexts, the Python 3.11 context argument, and cancellation cleanup.
• Python 3.14: Queues — fetched at the same time. Supports item handoff, bounded queues, task_done() accounting, and join() behavior.

These three maintained primary references support an evergreen Python worker task. The article does not depend on a recent release or incident. The envelope, authority boundaries, diagnostic policy, and staging acceptance criteria are engineering recommendations. The reported observations are limited to the local Python 3.12.3 probe.