Authentication

Aegra supports flexible authentication through configurable auth handlers. You write a Python function that verifies credentials and returns user data — Aegra handles the rest.

Quick setup

Create an auth file

Create a file (e.g., my_auth.py) in your project root:

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        raise Exception("Authentication required")

    # Your verification logic here (JWT decode, OAuth check, etc.)
    return {
        "identity": "user123",
        "display_name": "Jane Doe",
        "permissions": ["read", "write"],
        "is_authenticated": True,
    }

Add auth to aegra.json

{
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "auth": {
    "path": "./my_auth.py:auth"
  }
}

Start the server

aegra dev

All API endpoints now require authentication.

The authenticate handler

The @auth.authenticate decorator registers your authentication function. It receives the request headers and must return a dictionary with user data.

Required fields

Field	Type	Description
`identity`	`string`	Unique user identifier

Optional fields

Field	Type	Default	Description
`display_name`	`string`	Same as `identity`	Display name
`permissions`	`list[str]`	`[]`	Permission strings
`is_authenticated`	`bool`	`True`	Whether this is a signed-in user. Passed to your handlers; Aegra never rejects a request because of it (see who can use your server)

Any additional fields you return (like role, team_id, etc.) are preserved and accessible via ctx.user in authorization handlers and via the User model in custom routes.

Who can use your server

Your authenticate handler is the only gate. For each request it does one of two things:

Your handler	What happens
Raises an exception	The request is rejected with `401` and never reaches your graphs
Returns a user	The request runs as that user

Aegra does not look at is_authenticated to decide. It is information for your own code: authorization handlers, graphs and custom routes. A user returned with "is_authenticated": False still gets in. Pick one of the two setups below.

Signed-in users only

Raise for every request you don’t trust. Most deployments want this.

from langgraph_sdk import Auth

auth = Auth()


@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("authorization", "").removeprefix("Bearer ")
    if not token:
        raise Auth.exceptions.HTTPException(status_code=401, detail="Missing token")
    user = await verify_token(token)  # your JWT / OAuth / API key check
    if user is None:
        raise Auth.exceptions.HTTPException(status_code=401, detail="Invalid token")
    return {"identity": user.id, "display_name": user.name}

Don’t return a guest or anonymous user to mean “no”. Returning any user lets the request in. To reject a request, raise.

Signed-in users and guests

Return a guest user when there is no token, marked with "is_authenticated": False. Then decide in an authorization handler what guests may do.

@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("authorization", "").removeprefix("Bearer ")
    if not token:
        return {"identity": "guest", "is_authenticated": False}
    user = await verify_token(token)
    if user is None:
        raise Auth.exceptions.HTTPException(status_code=401, detail="Invalid token")
    return {"identity": user.id, "display_name": user.name}


READ_ACTIONS = {"read", "search", "get", "list_namespaces"}  # get / list_namespaces are store reads


@auth.on
async def guests_can_only_read(ctx, value):
    if not ctx.user.is_authenticated and ctx.action not in READ_ACTIONS:
        return False  # 403 Forbidden
    return None

A guest can now list and read, but can’t create threads, start runs or change anything. Keep in mind:

All guests share the guest identity, so they see each other’s threads and store items. To keep guests apart, give each one its own identity, for example from a session ID your frontend sends.
@auth.on is the fallback handler. A more specific handler such as @auth.on.threads.create runs instead of it, so repeat the guest check there.
A guest that can start runs spends your model budget. Allow that only on purpose.

Authorization handlers

Authorization handlers give you fine-grained access control for specific resources and actions.

Handler types

Handlers can:

Allow — Return None or True (default behavior)
Deny — Return False (returns 403 Forbidden)
Filter — Return a dictionary with filters to apply to queries
Modify — Modify the value dict (e.g., inject metadata)

Resolution priority

Handlers are matched from most specific to least specific:

@auth.on.threads.create — Resource + action
@auth.on.threads — Resource only
@auth.on.*.create — Action only
@auth.on — Global fallback

Examples

Restrict deletion to admins:

@auth.on.assistants.delete
async def restrict_deletion(ctx, value):
    if ctx.user.role != "admin":
        return False
    return None

Inject metadata on thread creation:

@auth.on.threads.create
async def inject_team_id(ctx, value):
    if "metadata" not in value:
        value["metadata"] = {}
    value["metadata"]["team_id"] = ctx.user.team_id
    return value

Filter threads by team:

@auth.on.threads.search
async def filter_by_team(ctx, value):
    return {"metadata": {"team_id": ctx.user.team_id}}

Handler context

The ctx parameter provides:

ctx.user — Authenticated user object (with all fields you returned from authenticate)
ctx.resource — Resource name ("threads", "assistants", etc.)
ctx.action — Action name ("create", "read", "update", "delete", "search")
ctx.permissions — User permissions list

Return values

Return	Behavior
`None` or `True`	Allow the request
`False`	Deny with 403 Forbidden
`dict`	Allow with filters (e.g., `{"metadata": {"team_id": "123"}}`)
Modified `value`	Allow with modified request data

Which routes a handler covers

Every Agent Protocol route dispatches to your handlers. Sub-resources authorize as their parent, so one rule covers a whole family of endpoints:

Handler	Also covers
`@auth.on.threads.read`	`GET /threads/{id}`, `/state`, `/state/{checkpoint_id}`, `/history`, `GET` a run, its `/join` and `/stream`
`@auth.on.threads.update`	`PATCH /threads/{id}`, `POST /threads/{id}/state`, `PATCH` a run, `POST .../runs/{id}/cancel`, `POST /runs/cancel`
`@auth.on.threads.delete`	`DELETE /threads/{id}`, `DELETE` a run
`@auth.on.threads.create_run`	`POST /threads/{id}/runs`, `/runs/stream`, `/runs/wait`, `/threads/{id}/commands`, and the stateless `POST /runs`, `/runs/stream`, `/runs/wait`
`@auth.on.threads.search`	`GET /threads`, `POST /threads/search`, `GET /threads/{id}/runs`

There is no runs resource: the Agent Protocol authorizes run operations under the thread that owns them, and the SDK’s @auth.on has no runs attribute. Stateless runs create an ephemeral thread, so they authorize as threads.create_run. Omitting the thread_id does not bypass your thread rules. A filter dict returned from a threads handler is applied to thread queries, including by-id reads and deletes, not only to search. Your handler runs once per event. A request that legitimately touches several resources still checks each one: creating a cron authorizes crons.create, then assistants.read, then the thread, so you can deny at any layer. Creating a run likewise authorizes threads.create_run, then assistants.read for the assistant it uses. Handlers may also inject data by mutating value in place. Setting value["metadata"]["team_id"] in an @auth.on.threads.create handler stores that metadata on the new thread.

Route-to-handler mapping lives in core/auth_registry.py. A route that is not registered there fails the test suite, so a new endpoint cannot silently skip your handlers.

Accessing user data in your graph

When auth is enabled, Aegra automatically injects the authenticated user’s data into config["configurable"]["langgraph_auth_user"] before graph execution. This happens server-side, so the client cannot tamper with it. This works with all graph types: custom StateGraph graphs, create_react_agent, create_agent, or any compiled graph.

From tools

InjectedToolArg (recommended)
ToolRuntime

from typing import Annotated
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool, InjectedToolArg

@tool
async def get_user_data(
    query: str,
    config: Annotated[RunnableConfig, InjectedToolArg],
) -> dict:
    """Tool that accesses the authenticated user."""
    auth_user = config["configurable"]["langgraph_auth_user"]
    user_id = auth_user["identity"]
    permissions = auth_user.get("permissions", [])

    if not any(p == "write" or p.endswith(":write") for p in permissions):
        return {"error": "Insufficient permissions"}

    return {"user_id": user_id, "result": f"Data for {query}"}

InjectedToolArg tells the LLM to ignore this parameter (it won’t try to fill it). Aegra injects the config automatically at runtime.

from langchain_core.tools import tool
from langgraph.prebuilt import ToolRuntime

@tool
async def get_user_data(
    query: str,
    tool_runtime: ToolRuntime,
) -> dict:
    """Tool that accesses the authenticated user via ToolRuntime."""
    auth_user = tool_runtime.config["configurable"]["langgraph_auth_user"]
    user_id = auth_user["identity"]
    return {"user_id": user_id, "result": f"Data for {query}"}

ToolRuntime is a subclass of Runtime that includes config, state, and tool_call_id. It is only available inside tools (not graph nodes).

From graph nodes

Graph nodes can accept a config: RunnableConfig parameter. Aegra injects the authenticated user into config["configurable"]["langgraph_auth_user"] before the graph executes:

from langchain_core.runnables import RunnableConfig

async def my_node(state: State, config: RunnableConfig) -> dict:
    """Graph node that accesses the authenticated user."""
    auth_user = config["configurable"]["langgraph_auth_user"]
    user_id = auth_user["identity"]
    role = auth_user.get("role")
    # Use user_id and role for RBAC, personalization, etc.
    return {"response": f"Hello {user_id}"}

The Runtime object (from get_runtime()) does not include config. To access config from nodes, add a config: RunnableConfig parameter to your node function. In tools, use InjectedToolArg or ToolRuntime instead.

From factory graphs

Factory graphs receive a ServerRuntime object that includes runtime.user with the authenticated user’s data. This is available at factory time (when deciding graph structure) before the graph executes:

from langgraph_sdk.runtime import ServerRuntime

async def graph(runtime: ServerRuntime):
    """Factory graph that adapts structure based on the authenticated user."""
    user = runtime.user  # The authenticated user (or None)

    tools = [search_tool]

    # Grant admin-only tools based on auth
    if user and "admin" in user.permissions:
        tools.append(admin_delete_tool)

    builder = StateGraph(State)
    # ... build graph with user-appropriate tools
    return builder.compile()

The runtime.user object is the full User model with all fields from your @auth.authenticate handler. Standard fields (identity, display_name, permissions) and custom fields (role, team_id, etc.) are all accessible as attributes or via dict-style access:

user = runtime.user
user.identity          # "alice"
user.role              # "admin" (custom field, attribute access)
user["team_id"]        # "team42" (custom field, dict-style access)
user.to_dict()         # full dict including all custom fields

Factory graphs get user data in two places:

Factory time (runtime.user on ServerRuntime): for structural decisions like which tools to include, which nodes to add, or which model to use.
Execution time (config["configurable"]["langgraph_auth_user"]): available inside nodes and tools during the actual graph run (same as static graphs).

Available fields

The langgraph_auth_user dict contains everything your @auth.authenticate handler returns, including any custom fields:

auth_user = config["configurable"]["langgraph_auth_user"]
# Standard fields
auth_user["identity"]        # str - unique user ID
auth_user["display_name"]    # str - display name
auth_user["permissions"]     # list[str] - permission strings
auth_user["is_authenticated"]  # bool

# Custom fields from your auth handler
auth_user["role"]            # any custom field you returned
auth_user["team_id"]         # any custom field you returned

Convenience shortcuts are also available directly on config["configurable"]:

config["configurable"]["user_id"] — the user’s identity
config["configurable"]["user_display_name"] — the user’s display name

Provider examples

JWT
OAuth
Firebase

import os
from langgraph_sdk import Auth
import jwt

auth = Auth()

JWT_SECRET = os.environ.get("JWT_SECRET")
if not JWT_SECRET:
    raise RuntimeError("JWT_SECRET environment variable is required")

@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        raise Exception("Missing Authorization header")

    payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
    return {
        "identity": payload["sub"],
        "display_name": payload.get("name", ""),
        "permissions": payload.get("permissions", []),
        "role": payload.get("role", "user"),
    }

from langgraph_sdk import Auth
import httpx

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        raise Exception("Missing Authorization header")

    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://oauth-provider.com/userinfo",
            headers={"Authorization": f"Bearer {token}"},
        )
        response.raise_for_status()
        user_info = response.json()

    return {
        "identity": user_info["sub"],
        "display_name": user_info["name"],
        "permissions": user_info.get("permissions", []),
        "email": user_info["email"],
    }

from langgraph_sdk import Auth
from firebase_admin import auth as firebase_auth

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> dict:
    token = headers.get("Authorization", "").replace("Bearer ", "")
    if not token:
        raise Exception("Missing Authorization header")

    decoded_token = firebase_auth.verify_id_token(token)
    return {
        "identity": decoded_token["uid"],
        "display_name": decoded_token.get("name", ""),
        "permissions": decoded_token.get("permissions", []),
        "email": decoded_token.get("email", ""),
    }

Auth on custom routes

Custom routes can use authentication via the require_auth dependency:

from fastapi import Depends
from aegra_api.core.auth_deps import require_auth
from aegra_api.models.auth import User

@app.get("/custom/whoami")
async def whoami(user: User = Depends(require_auth)):
    return {
        "identity": user.identity,
        "display_name": user.display_name,
        "permissions": user.permissions,
    }

To require auth on all custom routes by default, set enable_custom_route_auth in your config:

{
  "http": {
    "app": "./custom_routes.py:app",
    "enable_custom_route_auth": true
  }
}

See the custom routes guide for more.

No-auth mode

If no auth is configured, Aegra runs in no-auth mode:

All requests are allowed
User is set to anonymous, with is_authenticated set to False
Authorization handlers are not called

This is the default for local development and testing. Every caller shares the anonymous user and its data, so don’t expose a server without auth to the internet.

Configuration options

Option	Type	Default	Description
`auth.path`	`string`	—	Import path to your auth handler (`./file.py:variable`)
`auth.disable_studio_auth`	`bool`	`false`	Disable auth for LangGraph Studio connections

Non-interruptive design

Authorization handlers are additive by default:

If no auth is configured, requests are allowed
If no handlers are defined, requests are allowed
Handlers only restrict access when they explicitly deny

This ensures Aegra works out of the box without requiring any auth setup.

Getting started

Core concepts

Data & storage

Configuration

Contributing

Reference

Authentication

Quick setup

The authenticate handler

Required fields

Optional fields

Who can use your server

Signed-in users only

Signed-in users and guests

Authorization handlers

Handler types

Resolution priority

Examples

Handler context

Return values

Which routes a handler covers

Accessing user data in your graph

From tools

From graph nodes

From factory graphs

Available fields

Provider examples

Auth on custom routes

No-auth mode

Configuration options

Non-interruptive design

​Quick setup

​The authenticate handler

​Required fields

​Optional fields

​Who can use your server

​Signed-in users only

​Signed-in users and guests

​Authorization handlers

​Handler types

​Resolution priority

​Examples

​Handler context

​Return values

​Which routes a handler covers

​Accessing user data in your graph

​From tools

​From graph nodes

​From factory graphs

​Available fields

​Provider examples

​Auth on custom routes

​No-auth mode

​Configuration options

​Non-interruptive design

Quick setup

The authenticate handler

Required fields

Optional fields

Who can use your server

Signed-in users only

Signed-in users and guests

Authorization handlers

Handler types

Resolution priority

Examples

Handler context

Return values

Which routes a handler covers

Accessing user data in your graph

From tools

From graph nodes

From factory graphs

Available fields

Provider examples

Auth on custom routes

No-auth mode

Configuration options

Non-interruptive design