Writing

Genkit Action Context & Request Propagation: The Hidden Production Bug

The root cause, in nearly every case I've seen: misuse of ActionRunContext.

Genkit Action Context & Request Propagation: The Hidden Production Bug

Your Genkit app works fine in dev. You deploy. Three days in, user support tickets start arriving: “I got someone else’s data.” Or worse—you don’t get tickets. Users just silently receive the wrong results.

The root cause, in nearly every case I’ve seen: misuse of ActionRunContext.

This isn’t a rare edge case. It’s a category of bug that’s easy to introduce, invisible in local testing, and only surfaces under concurrent production load. Let’s name it, understand why it happens, and kill it.

The Bug in Three Lines

Here’s the pattern that causes the production failure:

# global variable, set per-request
_current_user_token: str = ""

@ai.flow()
async def answer_question(question: str) -> str:
    # Bug: sets a shared module-level variable
    _current_user_token = get_token_from_somewhere()
    return await ai.generate(prompt=question)

@ai.tool()
async def fetch_user_data(input: FetchInput) -> str:
    # Bug: reads from the same shared variable
    return call_api(input.query, auth=_current_user_token)

Under a single request, this works. Under two concurrent requests, it doesn’t: request A sets _current_user_token = "alice_token", then request B sets _current_user_token = "bob_token", then request A’s tool call reads bob_token. Alice’s query executes with Bob’s authorization. Data leaks. Auth bypasses.

The fix is ActionRunContext. But to use it correctly, you need to understand what it actually is.

What ActionRunContext Actually Is

ActionRunContext is the execution context for a single action invocation. Every flow, every tool, every model call runs inside one. It carries three things:

class ActionRunContext:
    context: dict[str, object]       # auth, request metadata, user data
    streaming_callback: ...          # for streaming responses
    abort_signal: asyncio.Event      # for cooperative cancellation

The context dict is the important one. It’s populated by your HTTP middleware before a request ever reaches your flow function. Here’s a complete Flask example:

from flask import Flask, Request
from genkit import Genkit
from genkit._core._action import ActionRunContext
from genkit.plugin_api import RequestData
from genkit.plugins.flask import genkit_flask_handler

app = Flask(__name__)
ai = Genkit(plugins=[...])

async def auth_context_provider(request_data: RequestData[Request]) -> dict:
    """Runs before every flow invocation. Populate the context dict here."""
    auth_header = request_data.request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise ValueError("Missing auth token")
    
    token = auth_header[len("Bearer "):]
    # Validate token, extract claims...
    uid = verify_token(token)  # your auth logic
    
    return {
        "auth": {
            "uid": uid,
            "token": token,
        },
        "request_id": request_data.request.headers.get("X-Request-ID", ""),
    }

@app.post("/chat")
@genkit_flask_handler(ai, context_provider=auth_context_provider)
@ai.flow()
async def answer_question(question: str, ctx: ActionRunContext) -> str:
    uid = ctx.context.get("auth", {}).get("uid")
    # uid is now the authenticated user for THIS request
    return await ai.generate(prompt=f"[User {uid}] {question}")

The context_provider function runs for every HTTP request. It receives the raw request and returns a dict that becomes ctx.context inside your flow. This dict is per-request and isolated—it lives in Python’s contextvars machinery, not in a shared module-level variable.

How Context Propagates Through the Execution Chain

Here’s what’s happening under the hood. When Genkit processes an HTTP request:

  1. Your context_provider runs and returns {"auth": {...}, "request_id": "..."}
  2. This dict gets passed to Action.run(input, context=your_dict)
  3. Action.run() stores it in a Python ContextVar called _action_context
  4. Every subsequent ai.generate() call in that request automatically reads from this ContextVar

That last point is the key. You don’t have to pass context to ai.generate() yourself—it picks it up automatically:

# In genkit/_ai/_aio.py, simplified:
async def generate(self, ..., context=None):
    # If you don't pass context, it reads from the ContextVar automatically
    resolved_context = context if context else get_current_context()
    return await generate_action(registry, gen_options, context=resolved_context)

This means for simple flows that just call ai.generate(), context propagation is automatic. Your code doesn’t have to do anything.

But tools are different. Tools that call external APIs need to extract auth from the context explicitly—and this is where the bug lives.

The Tool Bug: Auth That Doesn’t Follow Requests

Here’s the most common production failure pattern:

# ❌ BROKEN: ignores ctx, hardcodes credentials
@ai.tool()
async def search_user_documents(input: SearchInput) -> str:
    """Search documents for the current user."""
    # BUG: Which user? This uses the same API key for everyone.
    results = await httpx.get(
        f"https://api.example.com/search?q={input.query}",
        headers={"Authorization": f"Bearer {os.getenv('SERVICE_API_KEY')}"}
    )
    return results.text

This tool works fine in single-user testing. In production with concurrent users, it returns any user’s documents to any other user, because there’s no per-request auth.

Tools receive a ToolRunContext (a subclass of ActionRunContext) as their second argument. Accept it, and you get the same context dict that your flow has:

# ✅ CORRECT: extracts user-specific auth from context
from pydantic import BaseModel
from genkit._ai._tools import ToolRunContext

class SearchInput(BaseModel):
    query: str

@ai.tool()
async def search_user_documents(input: SearchInput, ctx: ToolRunContext) -> str:
    """Search documents for the authenticated user."""
    auth = ctx.context.get("auth", {})
    user_token = auth.get("token")
    if not user_token:
        raise ValueError("No auth token in context—is context_provider configured?")
    
    # Now each user's tool call uses their own token
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.example.com/search?q={input.query}",
            headers={"Authorization": f"Bearer {user_token}"}
        )
        return response.text

The ctx parameter in a tool is optional—if you don’t declare it, Genkit won’t pass it. That’s the trap. Forgetting to declare ctx means forgetting to isolate per-request state.

The Second Bug: asyncio.create_task() Breaks Context Isolation

There’s a subtler variant that trips up more experienced Python developers. If you spawn concurrent subtasks inside a flow using asyncio.create_task(), the context behavior depends on when you create the task relative to when context was set.

@ai.flow()
async def parallel_research(topic: str, ctx: ActionRunContext) -> str:
    # ❌ BUG: tasks created before context is accessible in this scope
    tasks = [
        asyncio.create_task(fetch_source(s))  # ctx.context may not be set here
        for s in get_sources(topic)
    ]
    results = await asyncio.gather(*tasks)
    return summarize(results)

async def fetch_source(url: str) -> str:
    # get_current_context() may return None here
    context = get_current_context()
    auth_token = (context or {}).get("auth", {}).get("token")
    ...

Python’s asyncio copies the current contextvars context when you call create_task(). If the ContextVar was already set (which it is, since Action.run() sets it before calling your flow function), the tasks inherit it correctly.

But if you do anything that creates tasks before the context var is set—like at module import time, or in __init__—those tasks won’t have the context.

The safe pattern: create tasks from inside a flow function (where context is already set), and if you need the context inside a subtask, pass it explicitly:

@ai.flow()
async def parallel_research(topic: str, ctx: ActionRunContext) -> str:
    # ✅ CORRECT: pass context explicitly to subtasks
    auth_token = ctx.context.get("auth", {}).get("token", "")
    request_id = ctx.context.get("request_id", "")
    
    tasks = [
        asyncio.create_task(fetch_source(s, auth_token, request_id))
        for s in get_sources(topic)
    ]
    results = await asyncio.gather(*tasks)
    return summarize(results)

async def fetch_source(url: str, auth_token: str, request_id: str) -> str:
    # Context is explicit, not from ContextVar
    async with httpx.AsyncClient() as client:
        response = await client.get(
            url,
            headers={
                "Authorization": f"Bearer {auth_token}",
                "X-Request-ID": request_id,
            }
        )
        return response.text

Explicit is better than implicit—especially when debugging production incidents at 2 AM.

Request ID Tracing: Correlating One User Request Across Many Model Calls

One under-used pattern: propagating a request_id through your entire execution chain so you can correlate traces across multiple model calls and tool executions.

import uuid
from genkit import Genkit
from genkit._core._action import ActionRunContext
from genkit.plugins.google_genai import GoogleAI

ai = Genkit(
    plugins=[GoogleAI()],
    model="googleai/gemini-2.0-flash",
)

# context_provider (e.g. in Flask or FastAPI)
async def context_provider(request_data) -> dict:
    return {
        "auth": {"uid": extract_uid(request_data)},
        "request_id": request_data.request.headers.get(
            "X-Request-ID", str(uuid.uuid4())
        ),
    }

@ai.flow()
async def multi_step_research(question: str, ctx: ActionRunContext) -> str:
    request_id = ctx.context.get("request_id", "unknown")
    uid = ctx.context.get("auth", {}).get("uid", "anon")
    
    # Log with request_id for correlation
    print(f"[{request_id}] Starting research for user {uid}: {question}")
    
    # Step 1: decompose question
    decomposition = await ai.generate(
        prompt=f"Break this into 3 sub-questions: {question}",
        # context propagates automatically — no need to pass it
    )
    print(f"[{request_id}] Decomposed into sub-questions")
    
    # Step 2: answer each sub-question
    sub_questions = decomposition.text.split("\n")
    answers = []
    for i, sub_q in enumerate(sub_questions[:3]):
        answer = await ai.generate(prompt=f"Answer concisely: {sub_q}")
        print(f"[{request_id}] Answered sub-question {i+1}")
        answers.append(answer.text)
    
    # Step 3: synthesize
    synthesis = await ai.generate(
        prompt=f"Synthesize: {chr(10).join(answers)}"
    )
    print(f"[{request_id}] Research complete")
    
    return synthesis.text

Every print line above gets the same request_id. Every external API call your tools make should include that ID as a header. Now when something goes wrong, you can grep your logs for one ID and see exactly what happened across every LLM call, tool execution, and API request for that user’s session.

Genkit also generates its own trace IDs via OpenTelemetry. The request_id from your context lets you correlate Genkit traces with your application logs and external service logs—something Genkit’s built-in tracing can’t do for you automatically.

The Complete Before/After

Here’s a realistic production flow, showing the broken pattern and the fix side by side.

Before: The Broken Pattern

# ❌ This will cause data leaks under concurrent load

from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
from pydantic import BaseModel
import os

ai = Genkit(plugins=[GoogleAI()], model="googleai/gemini-2.0-flash")

# DANGER: module-level state
_request_user_id: str = ""

class CustomerDataInput(BaseModel):
    query: str

@ai.tool()
async def get_customer_data(input: CustomerDataInput) -> str:
    """Fetch customer data from CRM."""
    # BUG: reads shared state — wrong user under concurrency
    return await crm_api.search(input.query, user_id=_request_user_id)

@ai.flow()
async def customer_support(question: str) -> str:
    global _request_user_id
    # BUG: sets shared state — race condition waiting to happen
    _request_user_id = extract_uid_from_somewhere()
    
    response = await ai.generate(
        prompt=question,
        tools=[get_customer_data],
    )
    return response.text

After: The Correct Pattern

# ✅ Per-request context, no shared state

from genkit import Genkit
from genkit.plugins.google_genai import GoogleAI
from genkit._core._action import ActionRunContext
from genkit._ai._tools import ToolRunContext
from genkit.plugin_api import RequestData
from pydantic import BaseModel
from flask import Flask, Request
from genkit.plugins.flask import genkit_flask_handler

ai = Genkit(plugins=[GoogleAI()], model="googleai/gemini-2.0-flash")
app = Flask(__name__)

class CustomerDataInput(BaseModel):
    query: str

@ai.tool()
async def get_customer_data(input: CustomerDataInput, ctx: ToolRunContext) -> str:
    """Fetch customer data for the authenticated user."""
    auth = ctx.context.get("auth", {})
    uid = auth.get("uid")
    token = auth.get("token")
    
    if not uid or not token:
        return "Error: not authenticated"
    
    # Each request uses its own uid and token
    return await crm_api.search(input.query, user_id=uid, token=token)

@app.post("/support")
@genkit_flask_handler(ai, context_provider=auth_context_provider)
@ai.flow()
async def customer_support(question: str, ctx: ActionRunContext) -> str:
    uid = ctx.context.get("auth", {}).get("uid")
    request_id = ctx.context.get("request_id", "unknown")
    
    print(f"[{request_id}] Support request from {uid}: {question}")
    
    response = await ai.generate(
        prompt=question,
        tools=[get_customer_data],
        # ai.generate auto-propagates ctx.context to tools via ContextVar
    )
    return response.text

async def auth_context_provider(request_data: RequestData[Request]) -> dict:
    token = request_data.request.headers.get("Authorization", "").removeprefix("Bearer ")
    uid = verify_token(token)  # raises if invalid
    return {
        "auth": {"uid": uid, "token": token},
        "request_id": request_data.request.headers.get("X-Request-ID", str(uuid.uuid4())),
    }

The differences:

  • Context comes from HTTP middleware, not from flow logic. Your flow doesn’t set auth state; it reads it.
  • Tools accept ctx, so they get per-request auth without shared state.
  • ai.generate() propagates context automatically—you pass tools=[get_customer_data] and the framework handles the rest.
  • No global variables. None. _request_user_id is gone.

Summary: The Rules

  1. Never store per-request state in module-level variables. This is the original sin. It works perfectly in dev and fails in production.

  2. Always configure a context_provider if your flows need auth or user context. The framework can’t extract request data for you automatically.

  3. Accept ctx in tools that call external APIs. If you don’t declare it, you can’t access per-request auth.

  4. Trust ai.generate() to propagate context automatically. You don’t need to plumb ctx.context through every ai.generate() call—just make sure you have a context_provider configured.

  5. Pass context explicitly to asyncio.create_task() workers if those tasks need auth. Don’t rely on ContextVar inheritance—be explicit.

  6. Include request_id in every external API call. You’ll thank yourself when something goes wrong in production.

The framework gives you the tools to build multi-user production systems. But it can’t protect you from shared mutable state—that’s your job.

Jeff Huang is Tech Lead for Genkit Python at Google. The Genkit Python SDK is open source at github.com/firebase/genkit.