FastAPI Async Context Vars for Multi-Tenant Request Isolation: Why asyncio.create_task() Breaks Your Tenant Context and How to Fix It

I learned this lesson the hard way at 3 AM debugging a production incident where a customer's invoice generation task landed in another customer's database. The context variable was None. The tenant was leaked. The CEO was not happy.

The problem is subtle: Python's asyncio.create_task() creates a new task that does NOT inherit your request's context by default. In a multi-tenant FastAPI app, this means background jobs, webhooks, and child coroutines will lose your carefully propagated tenant ID unless you explicitly copy the context forward. Let me show you exactly why and how to fix it.

Why This Matters: Silent Data Leaks in Concurrent Systems

When you run FastAPI on Uvicorn with multiple workers, you're handling dozens of concurrent requests. Each request arrives with its own tenant ID (from JWT claims, subdomain, or header). You inject that tenant ID into a context variable so your database queries, logging, and business logic can access it without threading it through every function signature.

Then you spawn a background task:

# ❌ WRONG: Context is lost
asyncio.create_task(send_invoice_email(user_id, invoice_id))

The new task runs in a different async context. Your tenant context variable reads as None. The email service can't filter by tenant. Chaos ensues.

I prefer context variables over manually threading tenant IDs because it's cleaner and reduces parameter pollution. But you have to respect how async contexts work—they're copied at task creation time, not inherited by magic.

The Correct Pattern: Copy Context Explicitly

Here's the solution. Use contextvars.copy_context() and context.run():

# FastAPI setup
from contextvars import ContextVar
from typing import Optional

TENANT_ID: ContextVar[Optional[str]] = ContextVar("tenant_id", default=None)
USER_ID: ContextVar[Optional[str]] = ContextVar("user_id", default=None)

@app.middleware("http")
async def tenant_middleware(request: Request, call_next):
    # Extract tenant from JWT or header
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        tenant_id = payload.get("tenant_id")
        user_id = payload.get("user_id")
    except:
        tenant_id = None
        user_id = None

    # Set context variables
    token_tenant = TENANT_ID.set(tenant_id)
    token_user = USER_ID.set(user_id)

    try:
        response = await call_next(request)
    finally:
        TENANT_ID.reset(token_tenant)
        USER_ID.reset(token_user)

    return response

Now, when spawning a background task, copy the current context:

import asyncio
from contextvars import copy_context

# ✅ CORRECT: Context is preserved
def spawn_task(coro):
    ctx = copy_context()
    return asyncio.create_task(ctx.run(asyncio.create_task, coro))

# Or simpler: use a helper that wraps the coroutine
def spawn_task(coro):
    ctx = copy_context()
    async def wrapped():
        # This runs inside the copied context
        return await coro
    return asyncio.create_task(ctx.run(wrapped))

Actually, I prefer this cleaner version using asyncio.TaskGroup (Python 3.11+):

# Python 3.11+ with TaskGroup (implicit context copying)
async def endpoint():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(send_invoice_email(user_id, invoice_id))
    return {"status": "queued"}

TaskGroup copies context automatically. But if you're stuck on older Python or need more control, stick with the explicit copy_context() pattern.

Real Example: Multi-Tenant Invoice Processing

Here's a production-grade example from CitizenApp:

# models.py
from sqlalchemy.orm import Session
from contextlib import asynccontextmanager

class InvoiceService:
    @staticmethod
    async def queue_generation(invoice_id: str, db: Session):
        # This runs in the request context—tenant ID is available
        tenant_id = TENANT_ID.get()
        if not tenant_id:
            raise ValueError("Tenant context missing")

        # Copy context and spawn background task
        ctx = copy_context()
        asyncio.create_task(ctx.run(
            InvoiceService._generate_async,
            invoice_id=invoice_id,
            tenant_id=tenant_id
        ))

        return {"status": "queued"}

    @staticmethod
    async def _generate_async(invoice_id: str, tenant_id: str):
        # Set context explicitly since we're in a spawned task
        TENANT_ID.set(tenant_id)

        try:
            # Now database queries automatically filter by tenant
            async with get_async_session() as db:
                invoice = await db.execute(
                    select(Invoice).where(
                        Invoice.id == invoice_id,
                        Invoice.tenant_id == TENANT_ID.get()  # ✅ Safe
                    )
                )
                invoice = invoice.scalar_one_or_none()
                if not invoice:
                    return  # Silent fail—prevents cross-tenant access

                # Generate PDF, send email, etc.
                await send_email(invoice.customer_email)
        except Exception as e:
            logger.error(f"Invoice generation failed: {e}", extra={
                "tenant_id": TENANT_ID.get()
            })

# endpoints.py
@router.post("/invoices/{invoice_id}/generate")
async def generate_invoice(invoice_id: str, db: Session = Depends(get_db)):
    await InvoiceService.queue_generation(invoice_id, db)
    return {"status": "queued"}

The key insight: when you spawn a task, either copy the context explicitly or set it again inside the task. I prefer copying because it's automatic and reduces boilerplate.

Gotcha: Uvicorn Worker Reloading and Context Leaks

This burned me hard: if you're using --reload in development, context variables sometimes persist across reloads if they're module-level singletons. Force reset in your middleware's finally block (I showed this above with TENANT_ID.reset(token)). In production, you won't hit this because workers don't reload mid-request, but in dev it's maddening.

Also, don't store mutable objects in context variables. Store strings or IDs only. Context is meant to be immutable and copyable. If you store a list or dict, mutations in one task bleed to others.

Why Explicit Beats Implicit Here

I could use a decorator pattern to auto-wrap all background tasks, but I prefer explicit copy_context() calls because:

1. Readability: anyone reading the code knows context is being managed
2. Debuggability: you can print the context before spawning to verify it's correct
3. Flexibility: some tasks legitimately shouldn't inherit context (system jobs, webhooks from external services)

One decorator I do use:

def preserve_context(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        ctx = copy_context()
        return await ctx.run(func, *args, **kwargs)
    return wrapper

@preserve_context
async def send_invoice_email(user_id, invoice_id):
    tenant_id = TENANT_ID.get()  # ✅ Will work
    ...

But be selective. Explicit context copying at the spawn point is usually better for multi-tenant systems.

The Takeaway

In high-throughput multi-tenant systems, context variables are your safety rail, but only if you respect how async contexts are created. Copy context when spawning tasks, verify tenant ID is set inside background jobs, and test with concurrent requests in staging. The 3 AM debugging session is not worth the 5 minutes of setup now.