WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-11 04:07:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

docs(sdk-py): update auth docstrings to default-deny pattern (#6933)

Browse Source 
## Summary
- Updated all auth docstring examples in `libs/sdk-py` to follow a
**default-closed** pattern
- Every example now first registers a global `@auth.on` handler that
**denies** all requests, then adds resource/action-specific handlers to
selectively **allow** access
- Fixed inconsistencies (e.g., `@auth.on` vs `@my_auth.on`, sync vs
async handlers) and made examples more realistic

## Test plan
- [x] `make format` passes
- [x] `make lint` passes
- [x] Changes are docstring-only — no runtime behavior affected

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
William FH
2026-02-24 20:36:59 -08:00
committed by GitHub
co-authored by Claude Opus 4.6
parent 8fbdb14487
commit 8087e6a42c
2 changed files with 126 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/sdk-py/langgraph_sdk/auth/__init__.py
+105 -76
View File
@@ -46,36 +46,48 @@ class 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":
@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 result
return {
"identity": user["id"],
"permissions": user.get("permissions", []),
}
# Global fallback handler
@auth.on
async def authorize_default(params: Auth.on.value):
return False # Reject all requests (default behavior)
# 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
@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"
# 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
@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.
# 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 ()
assert isinstance(namespace, tuple)
if not namespace or namespace[0] != ctx.user.identity:
namespace = (ctx.user.identity, *namespace)
value["namespace"] = namespace
@@ -137,30 +149,31 @@ class Auth:
???+ example "Examples"
Global handler for all requests:
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 reject_unhandled_requests(ctx: AuthContext, value: Any) -> None:
print(f"Request to {ctx.path} by {ctx.user.identity}")
async def deny_all(ctx: AuthContext, value: Any) -> False:
return False
```
Resource-specific handler. This would take precedence over the global handler
Resource-specific handler. This takes precedence over the global handler
for all actions on the `threads` resource:
```python
@auth.on.threads
async def check_thread_access(ctx: AuthContext, value: Any) -> bool:
# Allow access only to threads created by the user
return value.get("created_by") == ctx.user.identity
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 prevent_thread_deletion(ctx: AuthContext, value: Any) -> bool:
async def allow_admin_thread_deletion(ctx: AuthContext, value: Any) -> bool:
# Only admins can delete threads
return "admin" in ctx.user.permissions
```
@@ -168,10 +181,10 @@ class Auth:
Multiple resources or actions:
```python
@auth.on(resources=["threads", "runs"], actions=["create", "update"])
async def rate_limit_writes(ctx: AuthContext, value: Any) -> bool:
# Implement rate limiting for write operations
return await check_rate_limit(ctx.user.identity)
@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.
@@ -180,10 +193,9 @@ class Auth:
```python
@auth.on.store
async def authorize_store(ctx: AuthContext, value: Auth.types.on.store.value):
# Automatically scope all store operations to the user's namespace.
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 ()
assert isinstance(namespace, tuple)
if not namespace or namespace[0] != ctx.user.identity:
namespace = (ctx.user.identity, *namespace)
value["namespace"] = namespace
@@ -193,14 +205,14 @@ class Auth:
```python
@auth.on.store.put
async def on_put(ctx: AuthContext, value: Auth.types.on.store.put.value):
# value has typed fields: namespace, key, value, index
...
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 on_get(ctx: AuthContext, value: Auth.types.on.store.get.value):
# value has typed fields: namespace, key
...
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
@@ -533,44 +545,56 @@ class _StoreOn:
"""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 on_store_put(ctx: Auth.types.AuthContext, value: Auth.types.on.store.put.value):
# Scope puts to user's namespace
...
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 on_store_get(ctx: Auth.types.AuthContext, value: Auth.types.on.store.get.value):
# Scope gets to user's namespace
...
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 on_store_search(ctx: Auth.types.AuthContext, value: Auth.types.on.store.search.value):
# Scope searches to user's namespace
...
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 on_store_delete(ctx: Auth.types.AuthContext, value: Auth.types.on.store.delete.value):
# Scope deletes to user's namespace
...
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(
@@ -579,11 +603,14 @@ class _StoreOn:
"""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 on_list_ns(ctx: Auth.types.AuthContext, value: Auth.types.on.store.list_namespaces.value):
# Scope namespace listing to user's prefix
...
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,)
```
"""
@@ -672,40 +699,42 @@ class _On:
???+ example "Examples"
Global handler for all requests:
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 log_all_requests(ctx: AuthContext, value: Any) -> None:
print(f"Request to {ctx.path} by {ctx.user.identity}")
return True
async def deny_all(ctx: AuthContext, value: Any) -> False:
return False
```
Resource-specific handler:
Resource-specific handler to allow access (takes precedence
over the global deny handler):
```python
@auth.on.threads
async def check_thread_access(ctx: AuthContext, value: Any) -> bool:
# Allow access only to threads created by the user
return value.get("created_by") == ctx.user.identity
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.delete
async def prevent_thread_deletion(ctx: AuthContext, value: Any) -> bool:
# Only admins can delete threads
return "admin" in ctx.user.permissions
@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", "runs"], actions=["create", "update"])
async def rate_limit_writes(ctx: AuthContext, value: Any) -> bool:
# Implement rate limiting for write operations
return await check_rate_limit(ctx.user.identity)
@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}
```
"""
libs/sdk-py/langgraph_sdk/auth/types.py
+21 -11
View File
@@ -233,13 +233,20 @@ class StudioUser:
???+ example "Examples"
Use `@auth.on` to deny by default, but allow Studio users through:
```python
@auth.on
async def allow_developers(ctx: Auth.types.AuthContext, value: Any) -> None:
async def deny_all_except_studio(ctx: Auth.types.AuthContext, value: Any) -> bool:
# Allow Studio users, deny everyone else by default
if isinstance(ctx.user, Auth.types.StudioUser):
return None
...
return True
return False
# Then add specific handlers to allow access for non-Studio users
@auth.on.threads
async def allow_thread_access(ctx: Auth.types.AuthContext, value: Any) -> Auth.types.FilterType:
return {"owner": ctx.user.identity}
```
"""
@@ -973,24 +980,27 @@ class on:
and search operations across different resources (threads, assistants, crons).
???+ note "Usage"
Start by denying all requests by default, then add handlers to allow access:
```python
from langgraph_sdk import Auth
auth = Auth()
# Default deny: reject all requests without a specific handler
@auth.on
def handle_all(params: Auth.on.value):
raise Exception("Not authorized")
async def deny_all(ctx: Auth.types.AuthContext, value: Auth.on.value):
return False
# Allow thread creation, stamping the owner
@auth.on.threads.create
def handle_thread_create(params: Auth.on.threads.create.value):
# Handle thread creation
pass
async def allow_thread_create(ctx: Auth.types.AuthContext, value: Auth.on.threads.create.value):
value.setdefault("metadata", {})["owner"] = ctx.user.identity
# Allow assistant search, scoped to user's resources
@auth.on.assistants.search
def handle_assistant_search(params: Auth.on.assistants.search.value):
# Handle assistant search
pass
async def allow_assistant_search(ctx: Auth.types.AuthContext, value: Auth.on.assistants.search.value):
return {"owner": ctx.user.identity}
```
"""