codekingpro/portable-devtools

1from __future__ import annotations2 3import inspect4import typing5from collections.abc import Callable, Sequence6 7from langgraph_sdk.auth import exceptions, types8 9TH = typing.TypeVar("TH", bound=types.Handler)10AH = typing.TypeVar("AH", bound=types.Authenticator)11 12 13class Auth:14    """Add custom authentication and authorization management to your LangGraph application.15 16    The Auth class provides a unified system for handling authentication and17    authorization in LangGraph applications. It supports custom user authentication18    protocols and fine-grained authorization rules for different resources and19    actions.20 21    To use, create a separate python file and add the path to the file to your22    LangGraph API configuration file (`langgraph.json`). Within that file, create23    an instance of the Auth class and register authentication and authorization24    handlers as needed.25 26    Example `langgraph.json` file:27 28    ```json29    {30      "dependencies": ["."],31      "graphs": {32        "agent": "./my_agent/agent.py:graph"33      },34      "env": ".env",35      "auth": {36        "path": "./auth.py:my_auth"37      }38    ```39 40    Then the LangGraph server will load your auth file and run it server-side whenever a request comes in.41 42    ???+ example "Basic Usage"43 44        ```python45        from langgraph_sdk import Auth46 47        my_auth = Auth()48 49        @my_auth.authenticate50        async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:51            user = await verify_token(authorization)  # Your token verification logic52            if not user:53                raise Auth.exceptions.HTTPException(54                    status_code=401, detail="Unauthorized"55                )56            return {57                "identity": user["id"],58                "permissions": user.get("permissions", []),59            }60 61        # Default deny: reject all requests that don't have a specific handler62        @my_auth.on63        async def deny_all(ctx: Auth.types.AuthContext, value: Any) -> False:64            return False65 66        # Allow users to create threads with their own identity as owner67        @my_auth.on.threads.create68        async def allow_thread_create(69            ctx: Auth.types.AuthContext, value: Auth.types.on.threads.create.value70        ):71            metadata = value.setdefault("metadata", {})72            metadata["owner"] = ctx.user.identity73 74        # Allow users to read and search their own threads75        @my_auth.on.threads.read76        async def allow_thread_read(77            ctx: Auth.types.AuthContext, value: Auth.types.on.threads.read.value78        ) -> Auth.types.FilterType:79            return {"owner": ctx.user.identity}80 81        @my_auth.on.threads.search82        async def allow_thread_search(83            ctx: Auth.types.AuthContext, value: Auth.types.on.threads.search.value84        ) -> Auth.types.FilterType:85            return {"owner": ctx.user.identity}86 87        # Scope all store operations to the user's namespace88        @my_auth.on.store89        async def scope_store(ctx: Auth.types.AuthContext, value: Auth.types.on.store.value):90            namespace = tuple(value["namespace"]) if value.get("namespace") else ()91            if not namespace or namespace[0] != ctx.user.identity:92                namespace = (ctx.user.identity, *namespace)93            value["namespace"] = namespace94        ```95 96    ???+ note "Request Processing Flow"97 98        1. Authentication (your `@auth.authenticate` handler) is performed first on **every request**99        2. For authorization, the most specific matching handler is called:100            * If a handler exists for the exact resource and action, it is used (e.g., `@auth.on.threads.create`)101            * Otherwise, if a handler exists for the resource with any action, it is used (e.g., `@auth.on.threads`)102            * Finally, if no specific handlers match, the global handler is used (e.g., `@auth.on`)103            * If no global handler is set, the request is accepted104 105        This allows you to set default behavior with a global handler while106        overriding specific routes as needed.107    """108 109    __slots__ = (110        "_authenticate_handler",111        "_global_handlers",112        "_handler_cache",113        "_handlers",114        "on",115    )116    types = types117    """Reference to auth type definitions.118    119    Provides access to all type definitions used in the auth system,120    like ThreadsCreate, AssistantsRead, etc."""121 122    exceptions = exceptions123    """Reference to auth exception definitions.124    125    Provides access to all exception definitions used in the auth system,126    like HTTPException, etc.    127    """128 129    def __init__(self) -> None:130        self.on = _On(self)131        """Entry point for authorization handlers that control access to specific resources.132 133        The on class provides a flexible way to define authorization rules for different134        resources and actions in your application. It supports three main usage patterns:135 136        1. Global handlers that run for all resources and actions137        2. Resource-specific handlers that run for all actions on a resource138        3. Resource and action specific handlers for fine-grained control139 140        Each handler must be an async function that accepts two parameters:141            - ctx (AuthContext): Contains request context and authenticated user info142            - value: The data being authorized (type varies by endpoint)143 144        The handler should return one of:145 146            - None or True: Accept the request147            - False: Reject with 403 error148            - FilterType: Apply filtering rules to the response149        150        ???+ example "Examples"151 152            Start by denying all requests by default, then add specific handlers153            to allow access:154 155            ```python156            # Default deny: reject all unhandled requests157            @auth.on158            async def deny_all(ctx: AuthContext, value: Any) -> False:159                return False160            ```161 162            Resource-specific handler. This takes precedence over the global handler163            for all actions on the `threads` resource:164 165            ```python166            @auth.on.threads167            async def allow_thread_access(ctx: AuthContext, value: Any) -> Auth.types.FilterType:168                # Only allow access to threads owned by the user169                return {"owner": ctx.user.identity}170            ```171 172            Resource and action specific handler:173 174            ```python175            @auth.on.threads.delete176            async def allow_admin_thread_deletion(ctx: AuthContext, value: Any) -> bool:177                # Only admins can delete threads178                return "admin" in ctx.user.permissions179            ```180 181            Multiple resources or actions:182 183            ```python184            @auth.on(resources=["threads", "assistants"], actions=["read", "search"])185            async def allow_reads(ctx: AuthContext, value: Any) -> Auth.types.FilterType:186                # Allow read/search access to resources owned by the user187                return {"owner": ctx.user.identity}188            ```189 190            Auth for the `store` resource is a bit different since its structure is developer defined.191            You typically want to scope store operations by rewriting the namespace to include the user's identity.192            The `value` dict is mutable — changes to `value["namespace"]` are used by the server for the actual operation.193 194            ```python195            @auth.on.store196            async def scope_store(ctx: AuthContext, value: Auth.types.on.store.value):197                # Allow store access but scope to user's namespace198                namespace = tuple(value["namespace"]) if value.get("namespace") else ()199                if not namespace or namespace[0] != ctx.user.identity:200                    namespace = (ctx.user.identity, *namespace)201                value["namespace"] = namespace202            ```203 204            You can also register handlers for specific store actions:205 206            ```python207            @auth.on.store.put208            async def allow_put(ctx: AuthContext, value: Auth.types.on.store.put.value):209                # Allow puts, scoped to user's namespace210                value["namespace"] = (ctx.user.identity, *value["namespace"])211 212            @auth.on.store.get213            async def allow_get(ctx: AuthContext, value: Auth.types.on.store.get.value):214                # Allow gets, scoped to user's namespace215                value["namespace"] = (ctx.user.identity, *value["namespace"])216            ```217        """218        # These are accessed by the API. Changes to their names or types is219        # will be considered a breaking change.220        self._handlers: dict[tuple[str, str], list[types.Handler]] = {}221        self._global_handlers: list[types.Handler] = []222        self._authenticate_handler: types.Authenticator | None = None223        self._handler_cache: dict[tuple[str, str], types.Handler] = {}224 225    def authenticate(self, fn: AH) -> AH:226        """Register an authentication handler function.227 228        The authentication handler is responsible for verifying credentials229        and returning user scopes. It can accept any of the following parameters230        by name:231 232            - request (Request): The raw ASGI request object233            - path (str): The request path, e.g., "/threads/abcd-1234-abcd-1234/runs/abcd-1234-abcd-1234/stream"234            - method (str): The HTTP method, e.g., "GET"235            - path_params (dict[str, str]): URL path parameters, e.g., {"thread_id": "abcd-1234-abcd-1234", "run_id": "abcd-1234-abcd-1234"}236            - query_params (dict[str, str]): URL query parameters, e.g., {"stream": "true"}237            - headers (dict[bytes, bytes]): Request headers238            - authorization (str | None): The Authorization header value (e.g., "Bearer <token>")239 240        Args:241            fn: The authentication handler function to register.242                Must return a representation of the user. This could be a:243                    - string (the user id)244                    - dict containing {"identity": str, "permissions": list[str]}245                    - or an object with identity and permissions properties246                Permissions can be optionally used by your handlers downstream.247 248        Returns:249            The registered handler function.250 251        Raises:252            ValueError: If an authentication handler is already registered.253 254        ???+ example "Examples"255 256            Basic token authentication:257 258            ```python259            @auth.authenticate260            async def authenticate(authorization: str) -> str:261                user_id = verify_token(authorization)262                return user_id263            ```264 265            Accept the full request context:266 267            ```python268            @auth.authenticate269            async def authenticate(270                method: str,271                path: str,272                headers: dict[str, bytes]273            ) -> str:274                user = await verify_request(method, path, headers)275                return user276            ```277 278            Return user name and permissions:279 280            ```python281            @auth.authenticate282            async def authenticate(283                method: str,284                path: str,285                headers: dict[str, bytes]286            ) -> Auth.types.MinimalUserDict:287                permissions, user = await verify_request(method, path, headers)288                # Permissions could be things like ["runs:read", "runs:write", "threads:read", "threads:write"]289                return {290                    "identity": user["id"],291                    "permissions": permissions,292                    "display_name": user["name"],293                }294            ```295        """296        if self._authenticate_handler is not None:297            raise ValueError(298                f"Authentication handler already set as {self._authenticate_handler}."299            )300        self._authenticate_handler = fn301        return fn302 303 304## Helper types & utilities305 306V = typing.TypeVar("V", contravariant=True)307 308 309class _ActionHandler(typing.Protocol[V]):310    async def __call__(311        self, *, ctx: types.AuthContext, value: V312    ) -> types.HandlerResult: ...313 314 315T = typing.TypeVar("T", covariant=True)316 317 318class _ResourceActionOn(typing.Generic[T]):319    def __init__(320        self,321        auth: Auth,322        resource: typing.Literal["threads", "crons", "assistants"],323        action: typing.Literal[324            "create", "read", "update", "delete", "search", "create_run"325        ],326        value: type[T],327    ) -> None:328        self.auth = auth329        self.resource = resource330        self.action = action331        self.value = value332 333    def __call__(self, fn: _ActionHandler[T]) -> _ActionHandler[T]:334        _validate_handler(fn)335        _register_handler(self.auth, self.resource, self.action, fn)336        return fn337 338 339VCreate = typing.TypeVar("VCreate", covariant=True)340VUpdate = typing.TypeVar("VUpdate", covariant=True)341VRead = typing.TypeVar("VRead", covariant=True)342VDelete = typing.TypeVar("VDelete", covariant=True)343VSearch = typing.TypeVar("VSearch", covariant=True)344 345 346class _ResourceOn(typing.Generic[VCreate, VRead, VUpdate, VDelete, VSearch]):347    """348    Generic base class for resource-specific handlers.349    """350 351    value: type[VCreate | VUpdate | VRead | VDelete | VSearch]352 353    Create: type[VCreate]354    Read: type[VRead]355    Update: type[VUpdate]356    Delete: type[VDelete]357    Search: type[VSearch]358 359    def __init__(360        self,361        auth: Auth,362        resource: typing.Literal["threads", "crons", "assistants"],363    ) -> None:364        self.auth = auth365        self.resource = resource366        self.create: _ResourceActionOn[VCreate] = _ResourceActionOn(367            auth, resource, "create", self.Create368        )369        self.read: _ResourceActionOn[VRead] = _ResourceActionOn(370            auth, resource, "read", self.Read371        )372        self.update: _ResourceActionOn[VUpdate] = _ResourceActionOn(373            auth, resource, "update", self.Update374        )375        self.delete: _ResourceActionOn[VDelete] = _ResourceActionOn(376            auth, resource, "delete", self.Delete377        )378        self.search: _ResourceActionOn[VSearch] = _ResourceActionOn(379            auth, resource, "search", self.Search380        )381 382    @typing.overload383    def __call__(384        self,385        fn: (386            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]387            | _ActionHandler[dict[str, typing.Any]]388        ),389    ) -> _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]: ...390 391    @typing.overload392    def __call__(393        self,394        *,395        resources: str | Sequence[str],396        actions: str | Sequence[str] | None = None,397    ) -> Callable[398        [_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]],399        _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],400    ]: ...401 402    def __call__(403        self,404        fn: (405            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]406            | _ActionHandler[dict[str, typing.Any]]407            | None408        ) = None,409        *,410        resources: str | Sequence[str] | None = None,411        actions: str | Sequence[str] | None = None,412    ) -> (413        _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]414        | Callable[415            [_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]],416            _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],417        ]418    ):419        if fn is not None:420            _validate_handler(fn)421            return typing.cast(422                "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",423                _register_handler(self.auth, self.resource, "*", fn),424            )425 426        def decorator(427            handler: _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],428        ) -> _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]:429            _validate_handler(handler)430            return typing.cast(431                "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",432                _register_handler(self.auth, self.resource, "*", handler),433            )434 435        # Accept keyword-only parameters for future filtering behavior; referenced to satisfy linters.436        _ = resources, actions437        return decorator438 439 440class _AssistantsOn(441    _ResourceOn[442        types.AssistantsCreate,443        types.AssistantsRead,444        types.AssistantsUpdate,445        types.AssistantsDelete,446        types.AssistantsSearch,447    ]448):449    value = (450        types.AssistantsCreate451        | types.AssistantsRead452        | types.AssistantsUpdate453        | types.AssistantsDelete454        | types.AssistantsSearch455    )456    Create = types.AssistantsCreate457    Read = types.AssistantsRead458    Update = types.AssistantsUpdate459    Delete = types.AssistantsDelete460    Search = types.AssistantsSearch461 462 463class _ThreadsOn(464    _ResourceOn[465        types.ThreadsCreate,466        types.ThreadsRead,467        types.ThreadsUpdate,468        types.ThreadsDelete,469        types.ThreadsSearch,470    ]471):472    value = (473        types.ThreadsCreate474        | types.ThreadsRead475        | types.ThreadsUpdate476        | types.ThreadsDelete477        | types.ThreadsSearch478        | types.RunsCreate479    )480    Create = types.ThreadsCreate481    Read = types.ThreadsRead482    Update = types.ThreadsUpdate483    Delete = types.ThreadsDelete484    Search = types.ThreadsSearch485    CreateRun = types.RunsCreate486 487    def __init__(488        self,489        auth: Auth,490        resource: typing.Literal["threads", "crons", "assistants"],491    ) -> None:492        super().__init__(auth, resource)493        self.create_run: _ResourceActionOn[types.RunsCreate] = _ResourceActionOn(494            auth, resource, "create_run", self.CreateRun495        )496 497 498class _CronsOn(499    _ResourceOn[500        types.CronsCreate,501        types.CronsRead,502        types.CronsUpdate,503        types.CronsDelete,504        types.CronsSearch,505    ]506):507    value = type[508        types.CronsCreate509        | types.CronsRead510        | types.CronsUpdate511        | types.CronsDelete512        | types.CronsSearch513    ]514 515    Create = types.CronsCreate516    Read = types.CronsRead517    Update = types.CronsUpdate518    Delete = types.CronsDelete519    Search = types.CronsSearch520 521 522class _StoreActionOn(typing.Generic[T]):523    """Decorator for registering a handler for a specific store action."""524 525    def __init__(526        self,527        auth: Auth,528        action: typing.Literal["put", "get", "search", "delete", "list_namespaces"],529        value: type[T],530    ) -> None:531        self.auth = auth532        self.action = action533        self.value = value534 535    def __call__(self, fn: _ActionHandler[T]) -> _ActionHandler[T]:536        _validate_handler(fn)537        _register_handler(self.auth, "store", self.action, fn)538        return fn539 540 541class _StoreOn:542    def __init__(self, auth: Auth) -> None:543        self._auth = auth544        self.put = _StoreActionOn(auth, "put", types.StorePut)545        """Register a handler for store put operations.546 547        ???+ example "Example"548            If using `@auth.on` to deny by default, register this handler to allow549            put operations (scoped to the user's namespace):550 551            ```python552            @auth.on.store.put553            async def allow_store_put(ctx: Auth.types.AuthContext, value: Auth.types.on.store.put.value):554                # Allow puts, scoped to user's namespace555                value["namespace"] = (ctx.user.identity, *value["namespace"])556            ```557        """558        self.get = _StoreActionOn(auth, "get", types.StoreGet)559        """Register a handler for store get operations.560 561        ???+ example "Example"562            If using `@auth.on` to deny by default, register this handler to allow563            get operations (scoped to the user's namespace):564 565            ```python566            @auth.on.store.get567            async def allow_store_get(ctx: Auth.types.AuthContext, value: Auth.types.on.store.get.value):568                # Allow gets, scoped to user's namespace569                value["namespace"] = (ctx.user.identity, *value["namespace"])570            ```571        """572        self.search = _StoreActionOn(auth, "search", types.StoreSearch)573        """Register a handler for store search operations.574 575        ???+ example "Example"576            If using `@auth.on` to deny by default, register this handler to allow577            search operations (scoped to the user's namespace):578 579            ```python580            @auth.on.store.search581            async def allow_store_search(ctx: Auth.types.AuthContext, value: Auth.types.on.store.search.value):582                # Allow searches, scoped to user's namespace583                value["namespace"] = (ctx.user.identity, *value["namespace"])584            ```585        """586        self.delete = _StoreActionOn(auth, "delete", types.StoreDelete)587        """Register a handler for store delete operations.588 589        ???+ example "Example"590            If using `@auth.on` to deny by default, register this handler to allow591            delete operations (scoped to the user's namespace):592 593            ```python594            @auth.on.store.delete595            async def allow_store_delete(ctx: Auth.types.AuthContext, value: Auth.types.on.store.delete.value):596                # Allow deletes, scoped to user's namespace597                value["namespace"] = (ctx.user.identity, *value["namespace"])598            ```599        """600        self.list_namespaces = _StoreActionOn(601            auth, "list_namespaces", types.StoreListNamespaces602        )603        """Register a handler for store list_namespaces operations.604 605        ???+ example "Example"606            If using `@auth.on` to deny by default, register this handler to allow607            namespace listing (scoped to the user's prefix):608 609            ```python610            @auth.on.store.list_namespaces611            async def allow_list_ns(ctx: Auth.types.AuthContext, value: Auth.types.on.store.list_namespaces.value):612                # Allow listing, scoped to user's namespace prefix613                value["namespace"] = (ctx.user.identity,)614            ```615        """616 617    @typing.overload618    def __call__(619        self,620        *,621        actions: (622            typing.Literal["put", "get", "search", "list_namespaces", "delete"]623            | Sequence[624                typing.Literal["put", "get", "search", "list_namespaces", "delete"]625            ]626            | None627        ) = None,628    ) -> Callable[[AHO], AHO]: ...629 630    @typing.overload631    def __call__(self, fn: AHO) -> AHO: ...632 633    def __call__(634        self,635        fn: AHO | None = None,636        *,637        actions: (638            typing.Literal["put", "get", "search", "list_namespaces", "delete"]639            | Sequence[640                typing.Literal["put", "get", "search", "list_namespaces", "delete"]641            ]642            | None643        ) = None,644    ) -> AHO | Callable[[AHO], AHO]:645        """Register a handler for specific resources and actions.646 647        Can be used as a decorator or with explicit resource/action parameters:648 649        @auth.on.store650        async def handler(): ... # Handle all store ops651 652        @auth.on.store(actions=("put", "get", "search", "delete"))653        async def handler(): ... # Handle specific store ops654 655        @auth.on.store.put656        async def handler(): ... # Handle store.put ops657        """658        if fn is not None:659            # Used as a plain decorator660            _register_handler(self._auth, "store", None, fn)661            return fn662 663        # Used with parameters, return a decorator664        def decorator(665            handler: AHO,666        ) -> AHO:667            if isinstance(actions, str):668                action_list = [actions]669            else:670                action_list = list(actions) if actions is not None else ["*"]671            for action in action_list:672                _register_handler(self._auth, "store", action, handler)673            return handler674 675        return decorator676 677 678AHO = typing.TypeVar("AHO", bound=_ActionHandler[dict[str, typing.Any]])679 680 681class _On:682    """Entry point for authorization handlers that control access to specific resources.683 684    The _On class provides a flexible way to define authorization rules for different resources685    and actions in your application. It supports three main usage patterns:686 687    1. Global handlers that run for all resources and actions688    2. Resource-specific handlers that run for all actions on a resource689    3. Resource and action specific handlers for fine-grained control690 691    Each handler must be an async function that accepts two parameters:692    - ctx (AuthContext): Contains request context and authenticated user info693    - value: The data being authorized (type varies by endpoint)694 695    The handler should return one of:696        - None or True: Accept the request697        - False: Reject with 403 error698        - FilterType: Apply filtering rules to the response699 700    ???+ example "Examples"701 702        Start by denying all requests by default with a global handler,703        then add specific handlers to allow access:704 705        ```python706        # Default deny: reject all requests without a specific handler707        @auth.on708        async def deny_all(ctx: AuthContext, value: Any) -> False:709            return False710        ```711 712        Resource-specific handler to allow access (takes precedence713        over the global deny handler):714 715        ```python716        @auth.on.threads717        async def allow_thread_access(ctx: AuthContext, value: Any) -> Auth.types.FilterType:718            # Allow access only to threads owned by the user719            return {"owner": ctx.user.identity}720        ```721 722        Resource and action specific handler:723 724        ```python725        @auth.on.threads.create726        async def allow_thread_create(ctx: AuthContext, value: Any) -> None:727            # Allow thread creation, stamping the owner728            value.setdefault("metadata", {})["owner"] = ctx.user.identity729        ```730 731        Multiple resources or actions:732 733        ```python734        @auth.on(resources=["threads", "assistants"], actions=["read", "search"])735        async def allow_reads(ctx: AuthContext, value: Any) -> Auth.types.FilterType:736            # Allow read/search, scoped to user's resources737            return {"owner": ctx.user.identity}738        ```739    """740 741    __slots__ = (742        "_auth",743        "assistants",744        "crons",745        "runs",746        "store",747        "threads",748        "value",749    )750 751    def __init__(self, auth: Auth) -> None:752        self._auth = auth753        self.assistants = _AssistantsOn(auth, "assistants")754        self.threads = _ThreadsOn(auth, "threads")755        self.crons = _CronsOn(auth, "crons")756        self.store = _StoreOn(auth)757        self.value = dict[str, typing.Any]758 759    @typing.overload760    def __call__(761        self,762        *,763        resources: str | Sequence[str],764        actions: str | Sequence[str] | None = None,765    ) -> Callable[[AHO], AHO]: ...766 767    @typing.overload768    def __call__(self, fn: AHO) -> AHO: ...769 770    def __call__(771        self,772        fn: AHO | None = None,773        *,774        resources: str | Sequence[str] | None = None,775        actions: str | Sequence[str] | None = None,776    ) -> AHO | Callable[[AHO], AHO]:777        """Register a handler for specific resources and actions.778 779        Can be used as a decorator or with explicit resource/action parameters:780 781        @auth.on782        async def handler(): ...  # Global handler783 784        @auth.on(resources="threads")785        async def handler(): ...  # types.Handler for all thread actions786 787        @auth.on(resources="threads", actions="create")788        async def handler(): ...  # types.Handler for thread creation789        """790        if fn is not None:791            # Used as a plain decorator792            _register_handler(self._auth, None, None, fn)793            return fn794 795        # Used with parameters, return a decorator796        def decorator(797            handler: AHO,798        ) -> AHO:799            if isinstance(resources, str):800                resource_list = [resources]801            else:802                resource_list = list(resources) if resources is not None else ["*"]803 804            if isinstance(actions, str):805                action_list = [actions]806            else:807                action_list = list(actions) if actions is not None else ["*"]808            for resource in resource_list:809                for action in action_list:810                    _register_handler(self._auth, resource, action, handler)811            return handler812 813        return decorator814 815 816def _register_handler(817    auth: Auth,818    resource: str | None,819    action: str | None,820    fn: types.Handler,821) -> types.Handler:822    _validate_handler(fn)823    resource = resource or "*"824    action = action or "*"825    if resource == "*" and action == "*":826        if auth._global_handlers:827            raise ValueError("Global handler already set.")828        auth._global_handlers.append(fn)829    else:830        r = resource if resource is not None else "*"831        a = action if action is not None else "*"832        if (r, a) in auth._handlers:833            raise ValueError(f"types.Handler already set for {r}, {a}.")834        auth._handlers[(r, a)] = [fn]835    return fn836 837 838def _validate_handler(fn: Callable[..., typing.Any]) -> None:839    """Validates that an auth handler function meets the required signature.840 841    Auth handlers must:842    1. Be async functions843    2. Accept a ctx parameter of type AuthContext844    3. Accept a value parameter for the data being authorized845    """846    if not inspect.iscoroutinefunction(fn):847        raise ValueError(848            f"Auth handler '{getattr(fn, '__name__', fn)}' must be an async function. "849            "Add 'async' before 'def' to make it asynchronous and ensure"850            " any IO operations are non-blocking."851        )852 853    sig = inspect.signature(fn)854    if "ctx" not in sig.parameters:855        raise ValueError(856            f"Auth handler '{getattr(fn, '__name__', fn)}' must have a 'ctx: AuthContext' parameter. "857            "Update the function signature to include this required parameter."858        )859    if "value" not in sig.parameters:860        raise ValueError(861            f"Auth handler '{getattr(fn, '__name__', fn)}' must have a 'value' parameter. "862            " The value contains the mutable data being sent to the endpoint."863            "Update the function signature to include this required parameter."864        )865 866 867def is_studio_user(868    user: types.MinimalUser | types.BaseUser | types.MinimalUserDict,869) -> bool:870    return (871        isinstance(user, types.StudioUser)872        or (isinstance(user, dict) and user.get("kind") == "StudioUser")  # ty: ignore[invalid-argument-type]873    )874 875 876__all__ = ["Auth", "exceptions", "types"]877