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.

This post matches PyPI genkit==0.8.1 and genkit-google-genai==0.8.1.

---

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. On 0.8.1 it carries:

class ActionRunContext:
    context: dict[str, object]       # auth, request metadata, user data
    streaming_callback: ...          # for streaming responses
    # plus is_streaming / send_chunk helpers

There is no abort_signal on ActionRunContext in the PyPI package. Cancellation, when you need it, is your application’s concern.

The context dict is the important one. On PyPI there is no Flask or FastAPI handler package that fills it for you. You pass context= yourself at the HTTP boundary: on flow.run(...) or on ai.generate(...).

from fastapi import FastAPI, Header, HTTPException
from genkit import Genkit, ActionRunContext
from genkit_google_genai import GoogleAI

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

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

@app.post("/chat")
async def chat(question: str, authorization: str | None = Header(default=None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing auth token")

    token = authorization.removeprefix("Bearer ")
    uid = verify_token(token)  # your auth logic

    result = await answer_question.run(
        question,
        context={
            "auth": {"uid": uid, "token": token},
            "request_id": "",  # or pull from X-Request-ID
        },
    )
    return {"text": result.response}

The dict you pass as context= becomes ctx.context inside the flow. It 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 you call flow.run(..., context=...) or ai.generate(..., context=...):

1. Your HTTP handler builds a dict like {"auth": {...}, "request_id": "..."}
2. That dict is passed into the action as context=
3. The action stores it in a Python ContextVar
4. Nested ai.generate() calls in that request read from the same ContextVar unless you override context=

That last point matters. You don’t have to pass context to every nested ai.generate() yourself. It picks up the current ContextVar 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 once the outer run(..., context=...) (or top-level generate(..., context=...)) set it.

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 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; did the HTTP handler pass context=?")

    # 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: easy to misuse if you reach for ContextVar inside workers
    tasks = [
        asyncio.create_task(fetch_source(s))
        for s in get_sources(topic)
    ]
    results = await asyncio.gather(*tasks)
    return summarize(results)

async def fetch_source(url: str) -> str:
    # Prefer not to depend on ContextVar inheritance here
    context = Genkit.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 the action 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:

from genkit import ActionRunContext

@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, ActionRunContext
from genkit_google_genai import GoogleAI

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

@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

# HTTP boundary: pass context= yourself
async def handle(question: str, uid: str, token: str, request_id: str | None = None):
    return await multi_step_research.run(
        question,
        context={
            "auth": {"uid": uid, "token": token},
            "request_id": request_id or str(uuid.uuid4()),
        },
    )

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_google_genai import GoogleAI
from pydantic import BaseModel

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

import uuid
from fastapi import FastAPI, Header, HTTPException
from genkit import Genkit, ActionRunContext, ToolRunContext
from genkit_google_genai import GoogleAI
from pydantic import BaseModel

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

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)

@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

@app.post("/support")
async def support(
    question: str,
    authorization: str | None = Header(default=None),
    x_request_id: str | None = Header(default=None),
):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing auth token")

    token = authorization.removeprefix("Bearer ")
    uid = verify_token(token)  # raises if invalid

    result = await customer_support.run(
        question,
        context={
            "auth": {"uid": uid, "token": token},
            "request_id": x_request_id or str(uuid.uuid4()),
        },
    )
    return {"text": result.response}

The differences:

• Context comes from the HTTP handler via flow.run(..., context=...), not from shared globals. Your flow doesn’t invent auth state; it reads ctx.context.
• Tools accept ctx: ToolRunContext, so they get per-request auth without shared state.
• Nested ai.generate() propagates context automatically. You pass tools=[get_customer_data] and the framework handles the rest.
• No global variables. _request_user_id is gone.

You can also pass context= directly to ai.generate(...) when you are not going through a flow.

---

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. Pass context= at the HTTP boundary (flow.run(..., context=...) or ai.generate(..., context=...)). On PyPI 0.8.1 there is no genkit_flask / genkit_fastapi helper that does this for you.

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

4. Trust nested ai.generate() to propagate context automatically once the outer call set it. You don’t need to plumb ctx.context through every nested generate.

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/genkit-ai/genkit.

Install: uv add genkit==0.8.1 genkit-google-genai==0.8.1