langgraph/libs/sdk-py/langgraph_sdk/auth/__init__.py

from __future__ import annotations

import inspect
import typing
from collections.abc import Callable, Sequence

from langgraph_sdk.auth import exceptions, types

TH = typing.TypeVar("TH", bound=types.Handler)
AH = typing.TypeVar("AH", bound=types.Authenticator)


class Auth:
    """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:

    ```json
    {
      "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.

    ???+ example "Basic Usage"

        ```python
        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
        ```

    ???+ note "Request Processing Flow"

        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.
    """

    __slots__ = (
        "_authenticate_handler",
        "_global_handlers",
        "_handler_cache",
        "_handlers",
        "on",
    )
    types = types
    """Reference to auth type definitions.
    
    Provides access to all type definitions used in the auth system,
    like ThreadsCreate, AssistantsRead, etc."""

    exceptions = exceptions
    """Reference to auth exception definitions.
    
    Provides access to all exception definitions used in the auth system,
    like HTTPException, etc.    
    """

    def __init__(self) -> None:
        self.on = _On(self)
        """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:

        1. Global handlers that run for all resources and actions
        2. Resource-specific handlers that run for all actions on a resource
        3. Resource and action specific handlers for fine-grained control

        Each handler must be an async function that accepts two parameters:
            - ctx (AuthContext): Contains request context and authenticated user info
            - value: The data being authorized (type varies by endpoint)

        The handler should return one of:

            - None or True: Accept the request
            - False: Reject with 403 error
            - FilterType: Apply filtering rules to the response
        
        ???+ example "Examples"

            Start by denying all requests by default, then add specific handlers
            to allow access:

            ```python
            # Default deny: reject all unhandled requests
            @auth.on
            async def deny_all(ctx: AuthContext, value: Any) -> False:
                return False
            ```

            Resource-specific handler. This takes precedence over the global handler
            for all actions on the `threads` resource:

            ```python
            @auth.on.threads
            async def allow_thread_access(ctx: AuthContext, value: Any) -> Auth.types.FilterType:
                # Only allow access to threads owned by the user
                return {"owner": ctx.user.identity}
            ```

            Resource and action specific handler:

            ```python
            @auth.on.threads.delete
            async def allow_admin_thread_deletion(ctx: AuthContext, value: Any) -> bool:
                # Only admins can delete threads
                return "admin" in ctx.user.permissions
            ```

            Multiple resources or actions:

            ```python
            @auth.on(resources=["threads", "assistants"], actions=["read", "search"])
            async def allow_reads(ctx: AuthContext, value: Any) -> Auth.types.FilterType:
                # Allow read/search access to resources owned by the user
                return {"owner": ctx.user.identity}
            ```

            Auth for the `store` resource is a bit different since its structure is developer defined.
            You typically want to scope store operations by rewriting the namespace to include the user's identity.
            The `value` dict is mutable — changes to `value["namespace"]` are used by the server for the actual operation.

            ```python
            @auth.on.store
            async def scope_store(ctx: AuthContext, value: Auth.types.on.store.value):
                # Allow store access but scope to user's namespace
                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
            ```

            You can also register handlers for specific store actions:

            ```python
            @auth.on.store.put
            async def allow_put(ctx: AuthContext, value: Auth.types.on.store.put.value):
                # Allow puts, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])

            @auth.on.store.get
            async def allow_get(ctx: AuthContext, value: Auth.types.on.store.get.value):
                # Allow gets, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])
            ```
        """
        # These are accessed by the API. Changes to their names or types is
        # will be considered a breaking change.
        self._handlers: dict[tuple[str, str], list[types.Handler]] = {}
        self._global_handlers: list[types.Handler] = []
        self._authenticate_handler: types.Authenticator | None = None
        self._handler_cache: dict[tuple[str, str], types.Handler] = {}

    def authenticate(self, fn: AH) -> AH:
        """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>")

        Args:
            fn: The authentication handler function to register.
                Must return a representation of the user. This could be a:
                    - string (the user id)
                    - dict containing {"identity": str, "permissions": list[str]}
                    - or an object with identity and permissions properties
                Permissions can be optionally used by your handlers downstream.

        Returns:
            The registered handler function.

        Raises:
            ValueError: If an authentication handler is already registered.

        ???+ example "Examples"

            Basic token authentication:

            ```python
            @auth.authenticate
            async def authenticate(authorization: str) -> str:
                user_id = verify_token(authorization)
                return user_id
            ```

            Accept the full request context:

            ```python
            @auth.authenticate
            async def authenticate(
                method: str,
                path: str,
                headers: dict[str, bytes]
            ) -> str:
                user = await verify_request(method, path, headers)
                return user
            ```

            Return user name and permissions:

            ```python
            @auth.authenticate
            async def authenticate(
                method: str,
                path: str,
                headers: dict[str, bytes]
            ) -> Auth.types.MinimalUserDict:
                permissions, user = await verify_request(method, path, headers)
                # Permissions could be things like ["runs:read", "runs:write", "threads:read", "threads:write"]
                return {
                    "identity": user["id"],
                    "permissions": permissions,
                    "display_name": user["name"],
                }
            ```
        """
        if self._authenticate_handler is not None:
            raise ValueError(
                f"Authentication handler already set as {self._authenticate_handler}."
            )
        self._authenticate_handler = fn
        return fn


## Helper types & utilities

V = typing.TypeVar("V", contravariant=True)


class _ActionHandler(typing.Protocol[V]):
    async def __call__(
        self, *, ctx: types.AuthContext, value: V
    ) -> types.HandlerResult: ...


T = typing.TypeVar("T")


class _ResourceActionOn(typing.Generic[T]):
    def __init__(
        self,
        auth: Auth,
        resource: typing.Literal["threads", "crons", "assistants"],
        action: typing.Literal[
            "create", "read", "update", "delete", "search", "create_run"
        ],
        value: type[T],
    ) -> None:
        self.auth = auth
        self.resource = resource
        self.action = action
        self.value = value

    def __call__(self, fn: _ActionHandler[T]) -> _ActionHandler[T]:
        _validate_handler(fn)
        _register_handler(self.auth, self.resource, self.action, fn)
        return fn


VCreate = typing.TypeVar("VCreate", covariant=True)
VUpdate = typing.TypeVar("VUpdate", covariant=True)
VRead = typing.TypeVar("VRead", covariant=True)
VDelete = typing.TypeVar("VDelete", covariant=True)
VSearch = typing.TypeVar("VSearch", covariant=True)
ResourceActionT = typing.TypeVar("ResourceActionT", bound=str)

_ResourceAction = typing.Literal["create", "read", "update", "delete", "search"]
_ThreadAction = _ResourceAction | typing.Literal["create_run"]


class _ResourceOn(
    typing.Generic[VCreate, VRead, VUpdate, VDelete, VSearch, ResourceActionT]
):
    """
    Generic base class for resource-specific handlers.
    """

    value: type[VCreate | VUpdate | VRead | VDelete | VSearch]

    Create: type[VCreate]
    Read: type[VRead]
    Update: type[VUpdate]
    Delete: type[VDelete]
    Search: type[VSearch]

    def __init__(
        self,
        auth: Auth,
        resource: typing.Literal["threads", "crons", "assistants"],
    ) -> None:
        self.auth = auth
        self.resource = resource
        self.create: _ResourceActionOn[VCreate] = _ResourceActionOn(
            auth, resource, "create", self.Create
        )
        self.read: _ResourceActionOn[VRead] = _ResourceActionOn(
            auth, resource, "read", self.Read
        )
        self.update: _ResourceActionOn[VUpdate] = _ResourceActionOn(
            auth, resource, "update", self.Update
        )
        self.delete: _ResourceActionOn[VDelete] = _ResourceActionOn(
            auth, resource, "delete", self.Delete
        )
        self.search: _ResourceActionOn[VSearch] = _ResourceActionOn(
            auth, resource, "search", self.Search
        )

    @typing.overload
    def __call__(
        self,
        fn: (
            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]
            | _ActionHandler[dict[str, typing.Any]]
        ),
    ) -> _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]: ...

    @typing.overload
    def __call__(
        self,
        *,
        resources: str | Sequence[str] | None = None,
        actions: ResourceActionT | Sequence[ResourceActionT] | None = None,
    ) -> Callable[
        [_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]],
        _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],
    ]: ...

    def __call__(
        self,
        fn: (
            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]
            | _ActionHandler[dict[str, typing.Any]]
            | None
        ) = None,
        *,
        resources: str | Sequence[str] | None = None,
        actions: ResourceActionT | Sequence[ResourceActionT] | None = None,
    ) -> (
        _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]
        | Callable[
            [_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]],
            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],
        ]
    ):
        def decorator(
            handler: _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],
        ) -> _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]:
            _validate_handler(handler)
            if resources is None:
                resource_list = [self.resource]
            elif isinstance(resources, str):
                resource_list = [resources]
            elif isinstance(resources, Sequence):
                resource_list = list(resources)
            else:
                raise TypeError("resources must be a string or sequence of strings")
            if resource_list != [self.resource]:
                raise ValueError(
                    f"Resource-specific decorator for {self.resource!r} cannot "
                    f"register handlers for {resource_list!r}. Use @auth.on(...) "
                    "for other or multiple resources."
                )
            if actions is None:
                action_list = ["*"]
            elif isinstance(actions, str):
                action_list = [actions]
            elif isinstance(actions, Sequence):
                action_list = list(actions)
            else:
                raise TypeError("actions must be a string or sequence of strings")
            if not action_list:
                raise ValueError("actions must not be empty")
            if not all(isinstance(action, str) for action in action_list):
                raise TypeError("actions must be a string or sequence of strings")
            valid_actions = {
                value.action
                for value in vars(self).values()
                if isinstance(value, _ResourceActionOn)
            }
            invalid_actions = (
                sorted(set(action_list) - valid_actions) if actions is not None else []
            )
            if invalid_actions:
                raise ValueError(
                    f"Invalid action(s) for {self.resource}: {', '.join(invalid_actions)}"
                )
            if len(action_list) != len(set(action_list)):
                raise ValueError("actions must not contain duplicates")
            for action in action_list:
                if (self.resource, action) in self.auth._handlers:
                    raise ValueError(
                        f"types.Handler already set for {self.resource}, {action}."
                    )
            for action in action_list:
                _register_handler(self.auth, self.resource, action, handler)
            return handler

        if fn is not None:
            return decorator(
                typing.cast(
                    "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",
                    fn,
                )
            )
        return decorator


class _AssistantsOn(
    _ResourceOn[
        types.AssistantsCreate,
        types.AssistantsRead,
        types.AssistantsUpdate,
        types.AssistantsDelete,
        types.AssistantsSearch,
        _ResourceAction,
    ]
):
    value = (
        types.AssistantsCreate
        | types.AssistantsRead
        | types.AssistantsUpdate
        | types.AssistantsDelete
        | types.AssistantsSearch
    )
    Create = types.AssistantsCreate
    Read = types.AssistantsRead
    Update = types.AssistantsUpdate
    Delete = types.AssistantsDelete
    Search = types.AssistantsSearch


class _ThreadsOn(
    _ResourceOn[
        types.ThreadsCreate,
        types.ThreadsRead,
        types.ThreadsUpdate,
        types.ThreadsDelete,
        types.ThreadsSearch,
        _ThreadAction,
    ]
):
    value = (
        types.ThreadsCreate
        | types.ThreadsRead
        | types.ThreadsUpdate
        | types.ThreadsDelete
        | types.ThreadsSearch
        | types.RunsCreate
    )
    Create = types.ThreadsCreate
    Read = types.ThreadsRead
    Update = types.ThreadsUpdate
    Delete = types.ThreadsDelete
    Search = types.ThreadsSearch
    CreateRun = types.RunsCreate

    def __init__(
        self,
        auth: Auth,
        resource: typing.Literal["threads", "crons", "assistants"],
    ) -> None:
        super().__init__(auth, resource)
        self.create_run: _ResourceActionOn[types.RunsCreate] = _ResourceActionOn(
            auth, resource, "create_run", self.CreateRun
        )


class _CronsOn(
    _ResourceOn[
        types.CronsCreate,
        types.CronsRead,
        types.CronsUpdate,
        types.CronsDelete,
        types.CronsSearch,
        _ResourceAction,
    ]
):
    value = type[
        types.CronsCreate
        | types.CronsRead
        | types.CronsUpdate
        | types.CronsDelete
        | types.CronsSearch
    ]

    Create = types.CronsCreate
    Read = types.CronsRead
    Update = types.CronsUpdate
    Delete = types.CronsDelete
    Search = types.CronsSearch


class _StoreActionOn(typing.Generic[T]):
    """Decorator for registering a handler for a specific store action."""

    def __init__(
        self,
        auth: Auth,
        action: typing.Literal["put", "get", "search", "delete", "list_namespaces"],
        value: type[T],
    ) -> None:
        self.auth = auth
        self.action = action
        self.value = value

    def __call__(self, fn: _ActionHandler[T]) -> _ActionHandler[T]:
        _validate_handler(fn)
        _register_handler(self.auth, "store", self.action, fn)
        return fn


class _StoreOn:
    def __init__(self, auth: Auth) -> None:
        self._auth = auth
        self.put = _StoreActionOn(auth, "put", types.StorePut)
        """Register a handler for store put operations.

        ???+ example "Example"
            If using `@auth.on` to deny by default, register this handler to allow
            put operations (scoped to the user's namespace):

            ```python
            @auth.on.store.put
            async def allow_store_put(ctx: Auth.types.AuthContext, value: Auth.types.on.store.put.value):
                # Allow puts, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])
            ```
        """
        self.get = _StoreActionOn(auth, "get", types.StoreGet)
        """Register a handler for store get operations.

        ???+ example "Example"
            If using `@auth.on` to deny by default, register this handler to allow
            get operations (scoped to the user's namespace):

            ```python
            @auth.on.store.get
            async def allow_store_get(ctx: Auth.types.AuthContext, value: Auth.types.on.store.get.value):
                # Allow gets, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])
            ```
        """
        self.search = _StoreActionOn(auth, "search", types.StoreSearch)
        """Register a handler for store search operations.

        ???+ example "Example"
            If using `@auth.on` to deny by default, register this handler to allow
            search operations (scoped to the user's namespace):

            ```python
            @auth.on.store.search
            async def allow_store_search(ctx: Auth.types.AuthContext, value: Auth.types.on.store.search.value):
                # Allow searches, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])
            ```
        """
        self.delete = _StoreActionOn(auth, "delete", types.StoreDelete)
        """Register a handler for store delete operations.

        ???+ example "Example"
            If using `@auth.on` to deny by default, register this handler to allow
            delete operations (scoped to the user's namespace):

            ```python
            @auth.on.store.delete
            async def allow_store_delete(ctx: Auth.types.AuthContext, value: Auth.types.on.store.delete.value):
                # Allow deletes, scoped to user's namespace
                value["namespace"] = (ctx.user.identity, *value["namespace"])
            ```
        """
        self.list_namespaces = _StoreActionOn(
            auth, "list_namespaces", types.StoreListNamespaces
        )
        """Register a handler for store list_namespaces operations.

        ???+ example "Example"
            If using `@auth.on` to deny by default, register this handler to allow
            namespace listing (scoped to the user's prefix):

            ```python
            @auth.on.store.list_namespaces
            async def allow_list_ns(ctx: Auth.types.AuthContext, value: Auth.types.on.store.list_namespaces.value):
                # Allow listing, scoped to user's namespace prefix
                value["namespace"] = (ctx.user.identity,)
            ```
        """

    @typing.overload
    def __call__(
        self,
        *,
        actions: (
            typing.Literal["put", "get", "search", "list_namespaces", "delete"]
            | Sequence[
                typing.Literal["put", "get", "search", "list_namespaces", "delete"]
            ]
            | None
        ) = None,
    ) -> Callable[[AHO], AHO]: ...

    @typing.overload
    def __call__(self, fn: AHO) -> AHO: ...

    def __call__(
        self,
        fn: AHO | None = None,
        *,
        actions: (
            typing.Literal["put", "get", "search", "list_namespaces", "delete"]
            | Sequence[
                typing.Literal["put", "get", "search", "list_namespaces", "delete"]
            ]
            | None
        ) = None,
    ) -> AHO | Callable[[AHO], AHO]:
        """Register a handler for specific resources and actions.

        Can be used as a decorator or with explicit resource/action parameters:

        @auth.on.store
        async def handler(): ... # Handle all store ops

        @auth.on.store(actions=("put", "get", "search", "delete"))
        async def handler(): ... # Handle specific store ops

        @auth.on.store.put
        async def handler(): ... # Handle store.put ops
        """
        if fn is not None:
            # Used as a plain decorator
            _register_handler(self._auth, "store", None, fn)
            return fn

        # Used with parameters, return a decorator
        def decorator(
            handler: AHO,
        ) -> AHO:
            if isinstance(actions, str):
                action_list = [actions]
            else:
                action_list = list(actions) if actions is not None else ["*"]
            for action in action_list:
                _register_handler(self._auth, "store", action, handler)
            return handler

        return decorator


AHO = typing.TypeVar("AHO", bound=_ActionHandler[dict[str, typing.Any]])


class _On:
    """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:

    1. Global handlers that run for all resources and actions
    2. Resource-specific handlers that run for all actions on a resource
    3. Resource and action specific handlers for fine-grained control

    Each handler must be an async function that accepts two parameters:
    - ctx (AuthContext): Contains request context and authenticated user info
    - value: The data being authorized (type varies by endpoint)

    The handler should return one of:
        - None or True: Accept the request
        - False: Reject with 403 error
        - FilterType: Apply filtering rules to the response

    ???+ example "Examples"

        Start by denying all requests by default with a global handler,
        then add specific handlers to allow access:

        ```python
        # Default deny: reject all requests without a specific handler
        @auth.on
        async def deny_all(ctx: AuthContext, value: Any) -> False:
            return False
        ```

        Resource-specific handler to allow access (takes precedence
        over the global deny handler):

        ```python
        @auth.on.threads
        async def allow_thread_access(ctx: AuthContext, value: Any) -> Auth.types.FilterType:
            # Allow access only to threads owned by the user
            return {"owner": ctx.user.identity}
        ```

        Resource and action specific handler:

        ```python
        @auth.on.threads.create
        async def allow_thread_create(ctx: AuthContext, value: Any) -> None:
            # Allow thread creation, stamping the owner
            value.setdefault("metadata", {})["owner"] = ctx.user.identity
        ```

        Multiple resources or actions:

        ```python
        @auth.on(resources=["threads", "assistants"], actions=["read", "search"])
        async def allow_reads(ctx: AuthContext, value: Any) -> Auth.types.FilterType:
            # Allow read/search, scoped to user's resources
            return {"owner": ctx.user.identity}
        ```
    """

    __slots__ = (
        "_auth",
        "assistants",
        "crons",
        "runs",
        "store",
        "threads",
        "value",
    )

    def __init__(self, auth: Auth) -> None:
        self._auth = auth
        self.assistants = _AssistantsOn(auth, "assistants")
        self.threads = _ThreadsOn(auth, "threads")
        self.crons = _CronsOn(auth, "crons")
        self.store = _StoreOn(auth)
        self.value = dict[str, typing.Any]

    @typing.overload
    def __call__(
        self,
        *,
        resources: str | Sequence[str],
        actions: str | Sequence[str] | None = None,
    ) -> Callable[[AHO], AHO]: ...

    @typing.overload
    def __call__(self, fn: AHO) -> AHO: ...

    def __call__(
        self,
        fn: AHO | None = None,
        *,
        resources: str | Sequence[str] | None = None,
        actions: str | Sequence[str] | None = None,
    ) -> AHO | Callable[[AHO], AHO]:
        """Register a handler for specific resources and actions.

        Can be used as a decorator or with explicit resource/action parameters:

        @auth.on
        async def handler(): ...  # Global handler

        @auth.on(resources="threads")
        async def handler(): ...  # types.Handler for all thread actions

        @auth.on(resources="threads", actions="create")
        async def handler(): ...  # types.Handler for thread creation
        """
        if fn is not None:
            # Used as a plain decorator
            _register_handler(self._auth, None, None, fn)
            return fn

        # Used with parameters, return a decorator
        def decorator(
            handler: AHO,
        ) -> AHO:
            if isinstance(resources, str):
                resource_list = [resources]
            else:
                resource_list = list(resources) if resources is not None else ["*"]

            if isinstance(actions, str):
                action_list = [actions]
            else:
                action_list = list(actions) if actions is not None else ["*"]
            for resource in resource_list:
                for action in action_list:
                    _register_handler(self._auth, resource, action, handler)
            return handler

        return decorator


def _register_handler(
    auth: Auth,
    resource: str | None,
    action: str | None,
    fn: types.Handler,
) -> types.Handler:
    _validate_handler(fn)
    resource = resource or "*"
    action = action or "*"
    if resource == "*" and action == "*":
        if auth._global_handlers:
            raise ValueError("Global handler already set.")
        auth._global_handlers.append(fn)
    else:
        r = resource if resource is not None else "*"
        a = action if action is not None else "*"
        if (r, a) in auth._handlers:
            raise ValueError(f"types.Handler already set for {r}, {a}.")
        auth._handlers[(r, a)] = [fn]
    return fn


def _validate_handler(fn: Callable[..., typing.Any]) -> None:
    """Validates that an auth handler function meets the required signature.

    Auth handlers must:
    1. Be async functions
    2. Accept a ctx parameter of type AuthContext
    3. Accept a value parameter for the data being authorized
    """
    if not inspect.iscoroutinefunction(fn):
        raise ValueError(
            f"Auth handler '{getattr(fn, '__name__', fn)}' must be an async function. "
            "Add 'async' before 'def' to make it asynchronous and ensure"
            " any IO operations are non-blocking."
        )

    sig = inspect.signature(fn)
    if "ctx" not in sig.parameters:
        raise ValueError(
            f"Auth handler '{getattr(fn, '__name__', fn)}' must have a 'ctx: AuthContext' parameter. "
            "Update the function signature to include this required parameter."
        )
    if "value" not in sig.parameters:
        raise ValueError(
            f"Auth handler '{getattr(fn, '__name__', fn)}' must have a 'value' parameter. "
            " The value contains the mutable data being sent to the endpoint."
            "Update the function signature to include this required parameter."
        )


def is_studio_user(
    user: types.MinimalUser | types.BaseUser | types.MinimalUserDict,
) -> bool:
    return isinstance(user, types.StudioUser) or (
        isinstance(user, dict) and user.get("kind") == "StudioUser"
    )


__all__ = ["Auth", "exceptions", "types"]