auth

from langgraph_sdk import Auth

my_auth = Auth()

async def verify_token(token: str) -> str:
    # Verify token and return user_id
    # This would typically be a call to your auth server
    return "user_id"

@auth.authenticate
async def authenticate(authorization: str) -> str:
    # Verify token and return user_id
    result = await verify_token(authorization)
    if result != "user_id":
        raise Auth.exceptions.HTTPException(
            status_code=401, detail="Unauthorized"
        )
    return result

# Global fallback handler
@auth.on
async def authorize_default(params: Auth.on.value):
    return False # Reject all requests (default behavior)

@auth.on.threads.create
async def authorize_thread_create(params: Auth.on.threads.create.value):
    # Allow the allowed user to create a thread
    assert params.get("metadata", {}).get("owner") == "allowed_user"

@auth.on.store
async def authorize_store(ctx: Auth.types.AuthContext, value: Auth.types.on.store.value):
    # Automatically scope all store operations to the user's namespace.
    namespace = tuple(value["namespace"]) if value.get("namespace") else ()
    assert isinstance(namespace, tuple)
    if not namespace or namespace[0] != ctx.user.identity:
        namespace = (ctx.user.identity, *namespace)
    value["namespace"] = namespace

1. Authentication (your @auth.authenticate handler) is performed first on every request
2. For authorization, the most specific matching handler is called:
   • If a handler exists for the exact resource and action, it is used (e.g., @auth.on.threads.create)
   • Otherwise, if a handler exists for the resource with any action, it is used (e.g., @auth.on.threads)
   • Finally, if no specific handlers match, the global handler is used (e.g., @auth.on)
   • If no global handler is set, the request is accepted

This allows you to set default behavior with a global handler while overriding specific routes as needed.