0
# Authentication & Authorization
1
2
Comprehensive authentication and authorization system supporting custom authentication handlers, fine-grained authorization rules, and flexible security policies for all resources and actions.
3
4
## Capabilities
5
6
### Authentication System
7
8
Custom authentication handlers for verifying user credentials and extracting user information from requests.
9
10
```python { .api }
11
from typing import Callable, TypeVar
12
from collections.abc import Sequence
13
from langgraph_sdk.auth import types, exceptions
14
15
TH = TypeVar("TH", bound=types.Handler)
16
AH = TypeVar("AH", bound=types.Authenticator)
17
18
class Auth:
19
"""Add custom authentication and authorization management to your LangGraph application.
20
21
The Auth class provides a unified system for handling authentication and
22
authorization in LangGraph applications. It supports custom user authentication
23
protocols and fine-grained authorization rules for different resources and
24
actions.
25
"""
26
27
types = types
28
"""Reference to auth type definitions.
29
30
Provides access to all type definitions used in the auth system,
31
like ThreadsCreate, AssistantsRead, etc."""
32
33
exceptions = exceptions
34
"""Reference to auth exception definitions.
35
36
Provides access to all exception definitions used in the auth system,
37
like HTTPException, etc.
38
"""
39
40
def __init__(self) -> None:
41
self.on: _On = ... # Authorization handlers
42
43
def authenticate(self, fn: AH) -> AH:
44
"""Register an authentication handler function.
45
46
The authentication handler is responsible for verifying credentials
47
and returning user scopes. It can accept any of the following parameters
48
by name:
49
50
- request (Request): The raw ASGI request object
51
- body (dict): The parsed request body
52
- path (str): The request path
53
- method (str): The HTTP method
54
- path_params (dict[str, str]): URL path parameters
55
- query_params (dict[str, str]): URL query parameters
56
- headers (dict[bytes, bytes]): Request headers
57
- authorization (str | None): The Authorization header value
58
59
Args:
60
fn: The authentication handler function to register.
61
Must return a representation of the user. This could be a:
62
- string (the user id)
63
- dict containing {"identity": str, "permissions": list[str]}
64
- or an object with identity and permissions properties
65
66
Returns:
67
The registered handler function.
68
"""
69
```
70
71
### Authorization Handlers
72
73
Fine-grained authorization control with resource-specific and action-specific handlers.
74
75
```python { .api }
76
class _On:
77
"""Authorization handler registration system."""
78
79
assistants: _AssistantsOn
80
threads: _ThreadsOn
81
crons: _CronsOn
82
store: _StoreOn
83
84
def __call__(
85
self,
86
fn: Callable = None,
87
*,
88
resources: str | list[str] = None,
89
actions: str | list[str] = None
90
) -> Callable:
91
"""
92
Register global or filtered authorization handler.
93
94
Parameters:
95
- fn: Handler function (for direct decoration)
96
- resources: Resource names to handle
97
- actions: Action names to handle
98
99
Returns:
100
Handler function or decorator
101
"""
102
103
# Resource-specific authorization handlers
104
class _AssistantsOn:
105
"""Authorization handlers for assistant operations."""
106
107
create: Callable # @auth.on.assistants.create
108
read: Callable # @auth.on.assistants.read
109
update: Callable # @auth.on.assistants.update
110
delete: Callable # @auth.on.assistants.delete
111
search: Callable # @auth.on.assistants.search
112
113
def __call__(self, fn: Callable) -> Callable:
114
"""Handle all assistant operations: @auth.on.assistants"""
115
116
class _ThreadsOn:
117
"""Authorization handlers for thread operations."""
118
119
create: Callable # @auth.on.threads.create
120
read: Callable # @auth.on.threads.read
121
update: Callable # @auth.on.threads.update
122
delete: Callable # @auth.on.threads.delete
123
search: Callable # @auth.on.threads.search
124
create_run: Callable # @auth.on.threads.create_run
125
126
def __call__(self, fn: Callable) -> Callable:
127
"""Handle all thread operations: @auth.on.threads"""
128
129
class _CronsOn:
130
"""Authorization handlers for cron operations."""
131
132
create: Callable # @auth.on.crons.create
133
read: Callable # @auth.on.crons.read
134
update: Callable # @auth.on.crons.update
135
delete: Callable # @auth.on.crons.delete
136
search: Callable # @auth.on.crons.search
137
138
def __call__(self, fn: Callable) -> Callable:
139
"""Handle all cron operations: @auth.on.crons"""
140
141
class _StoreOn:
142
"""Authorization handlers for store operations."""
143
144
def __call__(
145
self,
146
fn: Callable = None,
147
*,
148
actions: str | list[str] = None
149
) -> Callable:
150
"""
151
Handle store operations.
152
153
Parameters:
154
- fn: Handler function
155
- actions: Specific store actions ("put", "get", "search", "list_namespaces", "delete")
156
"""
157
```
158
159
### Authentication Types
160
161
Type definitions for user representation and authentication context.
162
163
```python { .api }
164
from typing import Protocol, TypedDict, Callable, Union, Literal, Any
165
from collections.abc import Sequence, Awaitable, Mapping
166
import typing_extensions
167
168
class MinimalUser(Protocol):
169
"""User objects must at least expose the identity property."""
170
171
@property
172
def identity(self) -> str:
173
"""The unique identifier for the user."""
174
...
175
176
class BaseUser(Protocol):
177
"""The base ASGI user protocol."""
178
179
@property
180
def is_authenticated(self) -> bool:
181
"""Whether the user is authenticated."""
182
...
183
184
@property
185
def display_name(self) -> str:
186
"""The display name of the user."""
187
...
188
189
@property
190
def identity(self) -> str:
191
"""The unique identifier for the user."""
192
...
193
194
@property
195
def permissions(self) -> Sequence[str]:
196
"""The permissions associated with the user."""
197
...
198
199
class StudioUser:
200
"""A user object that's populated from authenticated requests from the LangGraph studio."""
201
identity: str
202
display_name: str
203
is_authenticated: bool = True
204
kind: Literal["StudioUser"] = "StudioUser"
205
206
class BaseAuthContext:
207
"""Base class for authentication context."""
208
permissions: Sequence[str]
209
user: BaseUser
210
211
class AuthContext(BaseAuthContext):
212
"""Complete authentication context with resource and action information."""
213
resource: Literal["runs", "threads", "crons", "assistants", "store"]
214
action: Literal["create", "read", "update", "delete", "search", "create_run", "put", "get", "list_namespaces"]
215
216
class MinimalUserDict(TypedDict, total=False):
217
"""The dictionary representation of a user."""
218
identity: typing_extensions.Required[str]
219
display_name: str
220
is_authenticated: bool
221
permissions: Sequence[str]
222
223
# Filter types for authorization responses
224
FilterType = Union[
225
dict[str, Union[str, dict[Literal["$eq", "$contains"], str]]],
226
dict[str, str],
227
]
228
229
HandlerResult = Union[None, bool, FilterType]
230
"""The result of a handler can be:
231
* None | True: accept the request.
232
* False: reject the request with a 403 error
233
* FilterType: filter to apply
234
"""
235
236
Handler = Callable[..., Awaitable[HandlerResult]]
237
238
Authenticator = Callable[
239
...,
240
Awaitable[
241
Union[MinimalUser, str, BaseUser, MinimalUserDict, Mapping[str, Any]],
242
],
243
]
244
"""Type for authentication functions."""
245
```
246
247
### Authorization Data Types
248
249
Type definitions for operation-specific authorization data.
250
251
```python { .api }
252
# Thread operation types
253
class ThreadsCreate(TypedDict):
254
metadata: dict
255
thread_id: str
256
thread_ttl: int
257
258
class ThreadsRead(TypedDict):
259
thread_id: str
260
261
class ThreadsUpdate(TypedDict):
262
thread_id: str
263
metadata: dict
264
265
class ThreadsDelete(TypedDict):
266
thread_id: str
267
268
class ThreadsSearch(TypedDict):
269
metadata: dict
270
values: dict
271
status: str
272
273
# Assistant operation types
274
class AssistantsCreate(TypedDict):
275
graph_id: str
276
config: dict
277
metadata: dict
278
279
class AssistantsRead(TypedDict):
280
assistant_id: str
281
282
class AssistantsUpdate(TypedDict):
283
assistant_id: str
284
config: dict
285
metadata: dict
286
287
class AssistantsDelete(TypedDict):
288
assistant_id: str
289
290
class AssistantsSearch(TypedDict):
291
metadata: dict
292
graph_id: str
293
294
# Run operation types
295
class RunsCreate(TypedDict):
296
thread_id: str
297
assistant_id: str
298
input: dict
299
config: dict
300
301
# Cron operation types
302
class CronsCreate(TypedDict):
303
assistant_id: str
304
schedule: str
305
thread_id: str
306
config: dict
307
308
class CronsRead(TypedDict):
309
cron_id: str
310
311
class CronsDelete(TypedDict):
312
cron_id: str
313
314
class CronsSearch(TypedDict):
315
assistant_id: str
316
thread_id: str
317
318
# Store operation types
319
class StoreGet(TypedDict):
320
namespace: list[str]
321
key: str
322
323
class StorePut(TypedDict):
324
namespace: list[str]
325
key: str
326
value: dict
327
328
class StoreDelete(TypedDict):
329
namespace: list[str]
330
key: str
331
332
class StoreSearch(TypedDict):
333
namespace_prefix: list[str]
334
query: str
335
filter: dict
336
337
class StoreListNamespaces(TypedDict):
338
prefix: list[str]
339
suffix: list[str]
340
```
341
342
### Exception Types
343
344
```python { .api }
345
class HTTPException(Exception):
346
"""HTTP exception for authentication/authorization errors."""
347
348
def __init__(self, status_code: int, detail: str):
349
self.status_code = status_code
350
self.detail = detail
351
super().__init__(detail)
352
```
353
354
## Usage Examples
355
356
### Basic Authentication Setup
357
358
```python
359
from langgraph_sdk import Auth
360
361
# Create auth instance
362
auth = Auth()
363
364
@auth.authenticate
365
async def authenticate(authorization: str) -> str:
366
"""
367
Simple token-based authentication.
368
369
Returns user ID if token is valid.
370
"""
371
if not authorization or not authorization.startswith("Bearer "):
372
raise Auth.exceptions.HTTPException(
373
status_code=401,
374
detail="Missing or invalid authorization header"
375
)
376
377
token = authorization[7:] # Remove "Bearer "
378
user_id = await verify_token(token) # Your token verification logic
379
380
if not user_id:
381
raise Auth.exceptions.HTTPException(
382
status_code=401,
383
detail="Invalid token"
384
)
385
386
return user_id
387
388
async def verify_token(token: str) -> str:
389
"""Your token verification implementation."""
390
# Call your auth service, check database, etc.
391
if token == "valid-token-123":
392
return "user-123"
393
return None
394
```
395
396
### Advanced Authentication with Permissions
397
398
```python
399
@auth.authenticate
400
async def authenticate(
401
authorization: str,
402
path: str,
403
method: str
404
) -> Auth.types.MinimalUserDict:
405
"""
406
Authentication with user permissions.
407
"""
408
if not authorization:
409
raise Auth.exceptions.HTTPException(401, "Authorization required")
410
411
# Verify token and get user info
412
user_data = await verify_jwt_token(authorization)
413
414
return {
415
"identity": user_data["user_id"],
416
"permissions": user_data.get("permissions", []),
417
"display_name": user_data.get("name", "Unknown User")
418
}
419
420
async def verify_jwt_token(authorization: str) -> dict:
421
"""Verify JWT and return user data."""
422
# JWT verification logic
423
import jwt
424
425
try:
426
token = authorization.replace("Bearer ", "")
427
payload = jwt.decode(token, "secret", algorithms=["HS256"])
428
return payload
429
except jwt.InvalidTokenError:
430
raise Auth.exceptions.HTTPException(401, "Invalid token")
431
```
432
433
### Authorization Handlers
434
435
```python
436
# Global authorization handler
437
@auth.on
438
async def global_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
439
"""
440
Global handler for all requests.
441
Applied when no specific handler matches.
442
"""
443
# Log all requests
444
print(f"Request: {ctx.method} {ctx.path} by {ctx.user.identity}")
445
446
# Allow all requests (specific handlers can override)
447
return True
448
449
# Resource-specific handlers
450
@auth.on.threads
451
async def thread_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
452
"""
453
Handle all thread operations.
454
More specific than global handler.
455
"""
456
# Users can only access their own threads
457
thread_metadata = value.get("metadata", {})
458
owner = thread_metadata.get("owner")
459
460
return owner == ctx.user.identity
461
462
# Action-specific handlers
463
@auth.on.threads.create
464
async def thread_create_auth(
465
ctx: Auth.types.AuthContext,
466
value: Auth.types.ThreadsCreate
467
) -> bool:
468
"""
469
Handle thread creation specifically.
470
Most specific handler.
471
"""
472
# Check user has permission to create threads
473
return "threads:create" in getattr(ctx.user, "permissions", [])
474
475
@auth.on.threads.delete
476
async def thread_delete_auth(
477
ctx: Auth.types.AuthContext,
478
value: Auth.types.ThreadsDelete
479
) -> bool:
480
"""
481
Restrict thread deletion to admins.
482
"""
483
return "admin" in getattr(ctx.user, "permissions", [])
484
```
485
486
### Store Authorization
487
488
```python
489
@auth.on.store
490
async def store_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
491
"""
492
Authorize store operations.
493
Enforce user isolation in namespaces.
494
"""
495
namespace = value.get("namespace", [])
496
497
# Ensure user can only access their own namespace
498
if len(namespace) >= 2 and namespace[0] == "users":
499
return namespace[1] == ctx.user.identity
500
501
# Allow access to shared namespaces for certain users
502
if namespace[0] in ["shared", "public"]:
503
return True
504
505
# Admins can access everything
506
return "admin" in getattr(ctx.user, "permissions", [])
507
508
# Action-specific store handlers
509
@auth.on.store(actions=["put", "delete"])
510
async def store_write_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
511
"""
512
Restrict write operations.
513
"""
514
# Only users with write permission can modify data
515
return "store:write" in getattr(ctx.user, "permissions", [])
516
517
@auth.on.store(actions=["search", "list_namespaces"])
518
async def store_search_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
519
"""
520
Allow search operations for authenticated users.
521
"""
522
return not getattr(ctx.user, "is_anonymous", False)
523
```
524
525
### Advanced Authorization Patterns
526
527
```python
528
# Multi-resource authorization
529
@auth.on(resources=["threads", "runs"], actions=["create", "update"])
530
async def rate_limit_writes(ctx: Auth.types.AuthContext, value: dict) -> bool:
531
"""
532
Rate limit write operations across resources.
533
"""
534
user_id = ctx.user.identity
535
536
# Check rate limit
537
current_minute = datetime.now().strftime("%Y-%m-%d %H:%M")
538
rate_key = f"rate_limit:{user_id}:{current_minute}"
539
540
current_count = await redis_client.get(rate_key) or 0
541
if int(current_count) >= 100: # 100 writes per minute
542
raise Auth.exceptions.HTTPException(
543
status_code=429,
544
detail="Rate limit exceeded"
545
)
546
547
await redis_client.incr(rate_key)
548
await redis_client.expire(rate_key, 60)
549
550
return True
551
552
# Conditional authorization
553
@auth.on.assistants.update
554
async def assistant_update_auth(
555
ctx: Auth.types.AuthContext,
556
value: Auth.types.AssistantsUpdate
557
) -> bool:
558
"""
559
Allow assistant updates based on ownership or admin role.
560
"""
561
assistant_id = value["assistant_id"]
562
563
# Get assistant metadata
564
assistant = await get_assistant_metadata(assistant_id)
565
566
# Owner can always update
567
if assistant.get("owner") == ctx.user.identity:
568
return True
569
570
# Admins can update any assistant
571
if "admin" in getattr(ctx.user, "permissions", []):
572
return True
573
574
# Collaborators can update if they have permission
575
collaborators = assistant.get("collaborators", [])
576
return ctx.user.identity in collaborators
577
578
# Data filtering authorization
579
@auth.on.threads.search
580
async def thread_search_filter(
581
ctx: Auth.types.AuthContext,
582
value: Auth.types.ThreadsSearch
583
) -> dict:
584
"""
585
Filter search results based on user permissions.
586
Returns filter dict instead of boolean.
587
"""
588
# Regular users can only see their own threads
589
if "admin" not in getattr(ctx.user, "permissions", []):
590
return {
591
"metadata.owner": ctx.user.identity
592
}
593
594
# Admins see everything
595
return {}
596
```
597
598
### Configuration Integration
599
600
```python
601
# Example langgraph.json configuration
602
"""
603
{
604
"dependencies": ["."],
605
"graphs": {
606
"my_assistant": "./assistant.py:graph"
607
},
608
"env": ".env",
609
"auth": {
610
"path": "./auth.py:auth"
611
}
612
}
613
"""
614
615
# auth.py file
616
auth = Auth()
617
618
@auth.authenticate
619
async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:
620
# Your authentication logic
621
return await verify_user(authorization)
622
623
@auth.on
624
async def default_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
625
# Default authorization logic
626
return True
627
628
# Export the auth instance
629
__all__ = ["auth"]
630
```