on

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

# 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:

@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:

@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:

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

@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:

@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"])