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
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 | Authentication status |
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.
Denying access
Raise any exception to deny authentication:
@auth.authenticate
async def authenticate(headers: dict) -> dict:
token = headers.get("Authorization", "").replace("Bearer ", "")
if not token:
raise Exception("Authentication required")
if not is_valid_token(token):
raise Exception("Invalid token")
return user_data
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 |
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 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
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
- Authorization handlers are not called
This is the default for local development and testing.
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.