Python contextvars Across ASGI Threadpools: Debugging Token Context Mismatches in Streaming Generators and Async Workers

Python's standard contextvars module is the backbone of request-scoped state in modern ASGI frameworks like FastAPI and Starlette—powering distributed tracing IDs, multi-tenant database routing, and per-client rate-limiting telemetry without polluting function signatures. However, when you combine ContextVar.set() and ContextVar.reset(token) inside a synchronous generator passed to Starlette's StreamingResponse (such as an SSE LLM token stream), production logs can suddenly erupt with a cryptic crash: ValueError: <Token var=<ContextVar name='client_id' ...>> was created in a different Context.

While scaling real-time legal AI streaming for KortiAI on KortiBundle (kortibundle.com), we encountered and eliminated this exact edge case. In this deep dive, we unpack how Starlette's iterate_in_threadpool manages execution contexts across anyio worker threads and share the battle-tested pattern for context-safe streaming generators.

1. Anatomy of the Bug: How `iterate_in_threadpool` Copies Contexts Per Iteration

Suppose you define a synchronous generator event_stream() for a FastAPI StreamingResponse that sets a ContextVar on entry and resets its Token inside a finally block:

# ANTI-PATTERN: Raises ValueError in Starlette StreamingResponse!
from contextvars import ContextVar
from fastapi.responses import StreamingResponse

_CURRENT_CLIENT_ID: ContextVar[str | None] = ContextVar("client_id", default=None)

def broken_llm_stream(client_id: str):
    def event_stream():
        # Step 1: Runs inside Context Copy #1 (first next() call in threadpool)
        cid_token = _CURRENT_CLIENT_ID.set(client_id)
        try:
            for chunk in sync_ollama_client.stream():
                yield f"data: {chunk}\n\n"
        finally:
            # Step N: Runs inside Context Copy #N (final next() / close() call)!
            # CRASHES with ValueError: Token was created in a different Context
            _CURRENT_CLIENT_ID.reset(cid_token)

    return StreamingResponse(event_stream(), media_type="text/event-stream")

Why does _CURRENT_CLIENT_ID.reset(cid_token) fail when both .set() and .reset() are inside the exact same generator function? Because event_stream() is a synchronous generator (`def`, not `async def`). When Starlette iterates over a sync generator inside an async event loop, it wraps the iterator in starlette.concurrency.iterate_in_threadpool, which calls anyio.to_thread.run_sync(_next, iterator) on every single `yield` step.

Under the hood, each call to to_thread.run_sync() captures a brand-new shallow copy of the parent asyncio task's contextvars.Context via copy_context(). Consequently:

1. On iteration 1, cid_token = _CURRENT_CLIENT_ID.set(client_id) executes inside Context Copy #1 (which is discarded as soon as the first yield returns to the event loop!).
2. On the final iteration (when the generator finishes or closes), the finally block executes inside Context Copy #N.
3. Because Python's ContextVar.reset(token) enforces strict identity equality between the current running Context and the Context that minted the Token, calling .reset(cid_token) inside Context Copy #N immediately throws ValueError!

2. The Two Production Fixes: Parent Context Binding & Defensive Reset

To guarantee both accurate context propagation across every yielded chunk and zero teardown exceptions, apply two complementary rules:

1. Set the `ContextVar` in the outer `async def` endpoint (or convert the stream to an `async def` generator) before Starlette forks per-step threadpool contexts, so every `copy_context()` snapshot inherits the bound value automatically.
2. Wrap any `ContextVar.reset(token)` call in a helper that catches `ValueError` and falls back to `.set(default)` if invoked across a context boundary:

# PRODUCTION PATTERN: Context-Safe Streaming Generator in FastAPI
from contextvars import ContextVar, Token

_CURRENT_CLIENT_ID: ContextVar[str | None] = ContextVar("client_id", default=None)

def safe_context_reset(var: ContextVar, token: Token, default_val=None) -> None:
    try:
        var.reset(token)
    except ValueError:
        # Generator teardown occurred in a copied threadpool Context
        var.set(default_val)

@app.post("/api/ollama/chat")
async def stream_chat_endpoint(request: Request):
    client_id = request.headers.get("X-Client-Id", "anon")
    # Bind in parent async context so every threadpool copy_context() inherits it
    parent_token = _CURRENT_CLIENT_ID.set(client_id)

    async def async_event_stream():
        inner_token = _CURRENT_CLIENT_ID.set(client_id)
        try:
            for chunk in await run_in_threadpool(fetch_stream_batches):
                yield f"data: {chunk}\n\n"
        finally:
            safe_context_reset(_CURRENT_CLIENT_ID, inner_token, None)
            safe_context_reset(_CURRENT_CLIENT_ID, parent_token, None)

    return StreamingResponse(async_event_stream(), media_type="text/event-stream")