Auth(
self,
)Add custom authentication and authorization management to your LangGraph application.
The Auth class provides a unified system for handling authentication and authorization in LangGraph applications. It supports custom user authentication protocols and fine-grained authorization rules for different resources and actions.
To use, create a separate python file and add the path to the file to your
LangGraph API configuration file (langgraph.json). Within that file, create
an instance of the Auth class and register authentication and authorization
handlers as needed.
Example langgraph.json file:
{
"dependencies": ["."],
"graphs": {
"agent": "./my_agent/agent.py:graph"
},
"env": ".env",
"auth": {
"path": "./auth.py:my_auth"
}Then the LangGraph server will load your auth file and run it server-side whenever a request comes in.
from langgraph_sdk import Auth
my_auth = Auth()
@my_auth.authenticate
async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:
user = await verify_token(authorization) # Your token verification logic
if not user:
raise Auth.exceptions.HTTPException(
status_code=401, detail="Unauthorized"
)
return {
"identity": user["id"],
"permissions": user.get("permissions", []),
}
# Default deny: reject all requests that don't have a specific handler
@my_auth.on
async def deny_all(ctx: Auth.types.AuthContext, value: Any) -> False:
return False
# Allow users to create threads with their own identity as owner
@my_auth.on.threads.create
async def allow_thread_create(
ctx: Auth.types.AuthContext, value: Auth.types.on.threads.create.value
):
metadata = value.setdefault("metadata", {})
metadata["owner"] = ctx.user.identity
# Allow users to read and search their own threads
@my_auth.on.threads.read
async def allow_thread_read(
ctx: Auth.types.AuthContext, value: Auth.types.on.threads.read.value
) -> Auth.types.FilterType:
return {"owner": ctx.user.identity}
@my_auth.on.threads.search
async def allow_thread_search(
ctx: Auth.types.AuthContext, value: Auth.types.on.threads.search.value
) -> Auth.types.FilterType:
return {"owner": ctx.user.identity}
# Scope all store operations to the user's namespace
@my_auth.on.store
async def scope_store(ctx: Auth.types.AuthContext, value: Auth.types.on.store.value):
namespace = tuple(value["namespace"]) if value.get("namespace") else ()
if not namespace or namespace[0] != ctx.user.identity:
namespace = (ctx.user.identity, *namespace)
value["namespace"] = namespace@auth.authenticate handler) is performed first on every request@auth.on.threads.create)@auth.on.threads)@auth.on)This allows you to set default behavior with a global handler while overriding specific routes as needed.
Reference to auth type definitions.
Provides access to all type definitions used in the auth system, like ThreadsCreate, AssistantsRead, etc.
Reference to auth exception definitions.
Provides access to all exception definitions used in the auth system, like HTTPException, etc.
Entry point for authorization handlers that control access to specific resources.
The on class provides a flexible way to define authorization rules for different resources and actions in your application. It supports three main usage patterns:
Register an authentication handler function.
The authentication handler is responsible for verifying credentials and returning user scopes. It can accept any of the following parameters by name:
- request (Request): The raw ASGI request object
- path (str): The request path, e.g., "/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream"
- method (str): The HTTP method, e.g., "GET"
- path_params (dict[str, str]): URL path parameters, e.g., {"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}
- query_params (dict[str, str]): URL query parameters, e.g., {"stream": "true"}
- headers (dict[bytes, bytes]): Request headers
- authorization (str | None): The Authorization header value (e.g., "Bearer <token>")