How to Write Expression Policies to Manage Flow Context Keys in authentik: A Complete Guide

Expression policies in authentik let you read and write flow context keys by modifying request.context inside Python code, which automatically merges into the flow's shared state for downstream stages and blueprints.

Expression policies are one of authentik's most powerful tools for dynamic authentication flows. They execute arbitrary Python code during flow execution and give you direct access to the flow context—a shared dictionary that persists across all stages in a flow. Understanding how to manipulate this context unlocks advanced use cases like conditional branching, data passing between stages, and dynamic blueprint configuration.

What Is Flow Context in authentik?

The flow context is a dictionary that accumulates data throughout a flow's lifecycle. It starts when a flow begins execution and is continually updated as each stage and policy runs. Expression policies interact with this context through the PolicyRequest object, specifically its context attribute.

According to the authentik source code in authentik/policies/types.py, the PolicyRequest dataclass encapsulates everything a policy needs to evaluate a request, including the current context dictionary. When you modify request.context, those changes propagate back into the broader flow context before the next stage executes.

How Expression Policies Access Flow Context

The ExpressionPolicy class defined in authentik/policies/expression/models.py (lines 27-32) handles policy execution. Its passes method invokes the PolicyEvaluator from authentik/policies/expression/evaluator.py, which injects the PolicyRequest into the evaluation environment via set_policy_request(request). This makes the request variable available in your Python expressions.

Key capabilities available inside any expression policy:

• Read a key: request.context.get("my_key") — returns None if absent, or use request.context["my_key"] for strict access
• Write a key: request.context["my_key"] = value — immediately adds to flow context
• Delete a key: request.context.pop("my_key", None) — safe removal with default
• Check existence: "my_key" in request.context — boolean check

Writing Expression Policies to Set Flow Context Keys

Basic Pattern: Storing Values for Later Stages

The most common use case is storing computed values that downstream stages need. In authentik/core/sources/flow_manager.py (lines 239-285), the flow manager merges request.context modifications into the overall flow context before progressing to the next step.

# Store user attributes for later consumption

request.context["department"] = request.user.attributes.get("department", "unknown")
request.context["login_timestamp"] = datetime.now().isoformat()
return True

This makes department and login_timestamp available to subsequent stages via flow_context in Jinja templates or directly in other expression policies.

Conditional Logic with Context Dependencies

Expression policies can read context keys set by previous stages or policies to make branching decisions:

# Check if email verification was completed in a prior stage

if not request.context.get("email_verified"):
    # Policy fails—this can trigger a deny message or redirect

    return False

# Additional account checks based on risk score from earlier

risk_score = request.context.get("risk_score", 0)
if risk_score > 75:
    request.context["requires_mfa"] = True

return True

Reading from request.context uses .get() for safe access with defaults, preventing KeyError exceptions that would fail the policy.

Generating and Storing Temporary Data

Generate ephemeral values that survive only for the current flow execution:

import secrets
import hashlib

# Create a single-use verification code

code = secrets.randbelow(1_000_000)
request.context["verification_code"] = f"{code:06d}"
request.context["code_expiry"] = datetime.now() + timedelta(minutes=10)

# Hash for secure comparison later

request.context["code_hash"] = hashlib.sha256(
    str(code).encode()
).hexdigest()

return True

Later stages can reference {{ flow_context.verification_code }} in email templates or validate against code_hash in another expression policy.

Advanced Flow Context Management

Clearing Keys to Control Data Lifecycle

Explicitly remove sensitive or unnecessary data when no longer needed:

# After successful MFA verification, clean up temporary state

if request.context.get("mfa_verified"):
    request.context.pop("mfa_challenge_secret", None)
    request.context.pop("mfa_attempt_count", None)
    # Preserve verification status but remove implementation details

    return True

return False

Using .pop() with a default of None ensures safe deletion even if the key doesn't exist.

Coordinating Multiple Policies

When multiple expression policies execute in sequence, each sees the accumulated context modifications from previous policies:

# First policy: Set initial state

request.context["policy_stage"] = 1
request.context["collected_data"] = {}
return True

# Second policy: Build on previous work

stage = request.context.get("policy_stage", 0)
request.context["collected_data"][f"stage_{stage}"] = "processed"
request.context["policy_stage"] = stage + 1
return True

This pattern enables progressive data collection and validation across complex multi-stage flows.

Integration with Blueprint Context

Values written to request.context become available in blueprint-generated objects through the flow_context variable. This bridges Python expression logic with declarative configuration:

# Set dynamic configuration for a blueprint-created application

request.context["app_slug"] = f"temp-{request.user.username}"
request.context["redirect_uri"] = f"https://{request.http_request.get_host()}/callback"
return True

In a blueprint, reference these as {{ flow_context.app_slug }} and {{ flow_context.redirect_uri }}.

Complete Working Examples

Example 1: Role-Based Flow Branching

# Determine user type and set appropriate next steps

groups = list(request.user.groups.values_list("name", flat=True))

if "administrators" in groups:
    request.context["dashboard_url"] = "/admin/dashboard"
    request.context["requires_2fa"] = True
    request.context["session_duration_hours"] = 8
elif "contractors" in groups:
    request.context["dashboard_url"] = "/contractor/portal"
    request.context["requires_2fa"] = True
    request.context["session_duration_hours"] = 4
    request.context["max_concurrent_sessions"] = 1
else:
    request.context["dashboard_url"] = "/user/home"
    request.context["requires_2fa"] = False
    request.context["session_duration_hours"] = 12

return True

Example 2: Risk-Based Step-Up Authentication

from datetime import datetime, timedelta

# Calculate risk indicators

failed_attempts = request.context.get("failed_login_count", 0)
last_success = request.context.get("last_successful_login")
device_trusted = request.context.get("device_trusted", False)

risk_factors = 0

if failed_attempts > 2:
    risk_factors += 1
if last_success and (datetime.now() - last_success) > timedelta(days=30):
    risk_factors += 1
if not device_trusted:
    risk_factors += 1

# Set authentication requirements based on risk

request.context["risk_level"] = "high" if risk_factors >= 2 else "medium" if risk_factors == 1 else "low"
request.context["step_up_required"] = risk_factors >= 2
request.context["allowed_methods"] = ["totp", "webauthn"] if risk_factors >= 2 else ["any"]

return True

Example 3: External System Integration Results

import json

# Assume previous stage made API call and stored raw response

api_response = request.context.get("external_api_response")

if api_response:
    try:
        data = json.loads(api_response)
        request.context["external_user_valid"] = data.get("active", False)
        request.context["external_permissions"] = data.get("permissions", [])
        
        # Transform external format to authentik-compatible

        request.context["entitlements"] = [
            f"ext:{p}" for p in data.get("permissions", [])
        ]
    except json.JSONDecodeError:
        request.context["external_user_valid"] = False
        request.context["api_error"] = "Invalid response format"

return request.context.get("external_user_valid", False)

Key Source Files Reference

File	Purpose
authentik/policies/expression/models.py	ExpressionPolicy class with passes method (lines 27-32)
authentik/policies/expression/evaluator.py	PolicyEvaluator that injects PolicyRequest into expression scope
authentik/policies/types.py	PolicyRequest dataclass defining the context attribute
authentik/core/sources/flow_manager.py	Flow context merging and lifecycle management (lines 239-285)

Summary

• Expression policies modify flow context by reading from and writing to request.context inside Python code
• Changes are automatically merged into the flow's shared state before the next stage executes
• Use .get() for safe reads, direct assignment for writes, and .pop() for cleanup
• Context keys become flow_context variables accessible in Jinja templates and blueprints
• Multiple policies compose sequentially, with each seeing accumulated changes from previous policies

Mastering request.context manipulation turns expression policies from simple boolean checks into sophisticated flow orchestration tools.

Frequently Asked Questions

How do I access flow context keys set by a previous stage?

Use request.context.get("key_name") inside your expression policy. The PolicyRequest object automatically contains all context accumulated from earlier stages and policies. For required values, use request.context["key_name"] which raises KeyError if absent—handle this with try/except or validate presence first.

Can expression policies delete or modify existing flow context keys?

Yes. Delete with request.context.pop("key_name", None) or modify by direct reassignment: request.context["key_name"] = new_value. These changes propagate to subsequent stages. Be cautious when modifying keys set by system components, as this may affect flow behavior in unexpected ways.

What's the difference between request.context and flow_context in templates?

They reference the same underlying dictionary. Inside expression policies, you interact with request.context. In Jinja templates (email stages, blueprints), you access the same data through flow_context. The naming difference reflects the evaluation environment—Python code versus template rendering.

How long do flow context keys persist?

Flow context exists only for the duration of a single flow execution. It initializes when a flow starts and disappears when the flow completes (successfully or with failure). For persistence across sessions, store data in user attributes, system settings, or external systems—not flow context.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}