pypi-langgraph-sdk

1
# Authentication & Authorization
2

3
Comprehensive authentication and authorization system supporting custom authentication handlers, fine-grained authorization rules, and flexible security policies for all resources and actions.
4

5
## Capabilities
6

7
### Authentication System
8

9
Custom authentication handlers for verifying user credentials and extracting user information from requests.
10

11
```python { .api }
12
from typing import Callable, TypeVar
13
from collections.abc import Sequence
14
from langgraph_sdk.auth import types, exceptions
15

16
TH = TypeVar("TH", bound=types.Handler)
17
AH = TypeVar("AH", bound=types.Authenticator)
18

19
class Auth:
20
    """Add custom authentication and authorization management to your LangGraph application.
21

22
    The Auth class provides a unified system for handling authentication and
23
    authorization in LangGraph applications. It supports custom user authentication
24
    protocols and fine-grained authorization rules for different resources and
25
    actions.
26
    """
27

28
    types = types
29
    """Reference to auth type definitions.
30

31
    Provides access to all type definitions used in the auth system,
32
    like ThreadsCreate, AssistantsRead, etc."""
33

34
    exceptions = exceptions
35
    """Reference to auth exception definitions.
36

37
    Provides access to all exception definitions used in the auth system,
38
    like HTTPException, etc.
39
    """
40

41
    def __init__(self) -> None:
42
        self.on: _On = ...  # Authorization handlers
43

44
    def authenticate(self, fn: AH) -> AH:
45
        """Register an authentication handler function.
46

47
        The authentication handler is responsible for verifying credentials
48
        and returning user scopes. It can accept any of the following parameters
49
        by name:
50

51
            - request (Request): The raw ASGI request object
52
            - body (dict): The parsed request body
53
            - path (str): The request path
54
            - method (str): The HTTP method
55
            - path_params (dict[str, str]): URL path parameters
56
            - query_params (dict[str, str]): URL query parameters
57
            - headers (dict[bytes, bytes]): Request headers
58
            - authorization (str | None): The Authorization header value
59

60
        Args:
61
            fn: The authentication handler function to register.
62
                Must return a representation of the user. This could be a:
63
                    - string (the user id)
64
                    - dict containing {"identity": str, "permissions": list[str]}
65
                    - or an object with identity and permissions properties
66

67
        Returns:
68
            The registered handler function.
69
        """
70
```
71

72
### Authorization Handlers
73

74
Fine-grained authorization control with resource-specific and action-specific handlers.
75

76
```python { .api }
77
class _On:
78
    """Authorization handler registration system."""
79

80
    assistants: _AssistantsOn
81
    threads: _ThreadsOn
82
    crons: _CronsOn
83
    store: _StoreOn
84

85
    def __call__(
86
        self,
87
        fn: Callable = None,
88
        *,
89
        resources: str | list[str] = None,
90
        actions: str | list[str] = None
91
    ) -> Callable:
92
        """
93
        Register global or filtered authorization handler.
94

95
        Parameters:
96
        - fn: Handler function (for direct decoration)
97
        - resources: Resource names to handle
98
        - actions: Action names to handle
99

100
        Returns:
101
        Handler function or decorator
102
        """
103

104
# Resource-specific authorization handlers
105
class _AssistantsOn:
106
    """Authorization handlers for assistant operations."""
107

108
    create: Callable  # @auth.on.assistants.create
109
    read: Callable    # @auth.on.assistants.read
110
    update: Callable  # @auth.on.assistants.update
111
    delete: Callable  # @auth.on.assistants.delete
112
    search: Callable  # @auth.on.assistants.search
113

114
    def __call__(self, fn: Callable) -> Callable:
115
        """Handle all assistant operations: @auth.on.assistants"""
116

117
class _ThreadsOn:
118
    """Authorization handlers for thread operations."""
119

120
    create: Callable     # @auth.on.threads.create
121
    read: Callable       # @auth.on.threads.read
122
    update: Callable     # @auth.on.threads.update
123
    delete: Callable     # @auth.on.threads.delete
124
    search: Callable     # @auth.on.threads.search
125
    create_run: Callable # @auth.on.threads.create_run
126

127
    def __call__(self, fn: Callable) -> Callable:
128
        """Handle all thread operations: @auth.on.threads"""
129

130
class _CronsOn:
131
    """Authorization handlers for cron operations."""
132

133
    create: Callable  # @auth.on.crons.create
134
    read: Callable    # @auth.on.crons.read
135
    update: Callable  # @auth.on.crons.update
136
    delete: Callable  # @auth.on.crons.delete
137
    search: Callable  # @auth.on.crons.search
138

139
    def __call__(self, fn: Callable) -> Callable:
140
        """Handle all cron operations: @auth.on.crons"""
141

142
class _StoreOn:
143
    """Authorization handlers for store operations."""
144

145
    def __call__(
146
        self,
147
        fn: Callable = None,
148
        *,
149
        actions: str | list[str] = None
150
    ) -> Callable:
151
        """
152
        Handle store operations.
153

154
        Parameters:
155
        - fn: Handler function
156
        - actions: Specific store actions ("put", "get", "search", "list_namespaces", "delete")
157
        """
158
```
159

160
### Authentication Types
161

162
Type definitions for user representation and authentication context.
163

164
```python { .api }
165
from typing import Protocol, TypedDict, Callable, Union, Literal, Any
166
from collections.abc import Sequence, Awaitable, Mapping
167
import typing_extensions
168

169
class MinimalUser(Protocol):
170
    """User objects must at least expose the identity property."""
171

172
    @property
173
    def identity(self) -> str:
174
        """The unique identifier for the user."""
175
        ...
176

177
class BaseUser(Protocol):
178
    """The base ASGI user protocol."""
179

180
    @property
181
    def is_authenticated(self) -> bool:
182
        """Whether the user is authenticated."""
183
        ...
184

185
    @property
186
    def display_name(self) -> str:
187
        """The display name of the user."""
188
        ...
189

190
    @property
191
    def identity(self) -> str:
192
        """The unique identifier for the user."""
193
        ...
194

195
    @property
196
    def permissions(self) -> Sequence[str]:
197
        """The permissions associated with the user."""
198
        ...
199

200
class StudioUser:
201
    """A user object that's populated from authenticated requests from the LangGraph studio."""
202
    identity: str
203
    display_name: str
204
    is_authenticated: bool = True
205
    kind: Literal["StudioUser"] = "StudioUser"
206

207
class BaseAuthContext:
208
    """Base class for authentication context."""
209
    permissions: Sequence[str]
210
    user: BaseUser
211

212
class AuthContext(BaseAuthContext):
213
    """Complete authentication context with resource and action information."""
214
    resource: Literal["runs", "threads", "crons", "assistants", "store"]
215
    action: Literal["create", "read", "update", "delete", "search", "create_run", "put", "get", "list_namespaces"]
216

217
class MinimalUserDict(TypedDict, total=False):
218
    """The dictionary representation of a user."""
219
    identity: typing_extensions.Required[str]
220
    display_name: str
221
    is_authenticated: bool
222
    permissions: Sequence[str]
223

224
# Filter types for authorization responses
225
FilterType = Union[
226
    dict[str, Union[str, dict[Literal["$eq", "$contains"], str]]],
227
    dict[str, str],
228
]
229

230
HandlerResult = Union[None, bool, FilterType]
231
"""The result of a handler can be:
232
    * None | True: accept the request.
233
    * False: reject the request with a 403 error
234
    * FilterType: filter to apply
235
"""
236

237
Handler = Callable[..., Awaitable[HandlerResult]]
238

239
Authenticator = Callable[
240
    ...,
241
    Awaitable[
242
        Union[MinimalUser, str, BaseUser, MinimalUserDict, Mapping[str, Any]],
243
    ],
244
]
245
"""Type for authentication functions."""
246
```
247

248
### Authorization Data Types
249

250
Type definitions for operation-specific authorization data.
251

252
```python { .api }
253
# Thread operation types
254
class ThreadsCreate(TypedDict):
255
    metadata: dict
256
    thread_id: str
257
    thread_ttl: int
258

259
class ThreadsRead(TypedDict):
260
    thread_id: str
261

262
class ThreadsUpdate(TypedDict):
263
    thread_id: str
264
    metadata: dict
265

266
class ThreadsDelete(TypedDict):
267
    thread_id: str
268

269
class ThreadsSearch(TypedDict):
270
    metadata: dict
271
    values: dict
272
    status: str
273

274
# Assistant operation types
275
class AssistantsCreate(TypedDict):
276
    graph_id: str
277
    config: dict
278
    metadata: dict
279

280
class AssistantsRead(TypedDict):
281
    assistant_id: str
282

283
class AssistantsUpdate(TypedDict):
284
    assistant_id: str
285
    config: dict
286
    metadata: dict
287

288
class AssistantsDelete(TypedDict):
289
    assistant_id: str
290

291
class AssistantsSearch(TypedDict):
292
    metadata: dict
293
    graph_id: str
294

295
# Run operation types
296
class RunsCreate(TypedDict):
297
    thread_id: str
298
    assistant_id: str
299
    input: dict
300
    config: dict
301

302
# Cron operation types
303
class CronsCreate(TypedDict):
304
    assistant_id: str
305
    schedule: str
306
    thread_id: str
307
    config: dict
308

309
class CronsRead(TypedDict):
310
    cron_id: str
311

312
class CronsDelete(TypedDict):
313
    cron_id: str
314

315
class CronsSearch(TypedDict):
316
    assistant_id: str
317
    thread_id: str
318

319
# Store operation types
320
class StoreGet(TypedDict):
321
    namespace: list[str]
322
    key: str
323

324
class StorePut(TypedDict):
325
    namespace: list[str]
326
    key: str
327
    value: dict
328

329
class StoreDelete(TypedDict):
330
    namespace: list[str]
331
    key: str
332

333
class StoreSearch(TypedDict):
334
    namespace_prefix: list[str]
335
    query: str
336
    filter: dict
337

338
class StoreListNamespaces(TypedDict):
339
    prefix: list[str]
340
    suffix: list[str]
341
```
342

343
### Exception Types
344

345
```python { .api }
346
class HTTPException(Exception):
347
    """HTTP exception for authentication/authorization errors."""
348

349
    def __init__(self, status_code: int, detail: str):
350
        self.status_code = status_code
351
        self.detail = detail
352
        super().__init__(detail)
353
```
354

355
## Usage Examples
356

357
### Basic Authentication Setup
358

359
```python
360
from langgraph_sdk import Auth
361

362
# Create auth instance
363
auth = Auth()
364

365
@auth.authenticate
366
async def authenticate(authorization: str) -> str:
367
    """
368
    Simple token-based authentication.
369

370
    Returns user ID if token is valid.
371
    """
372
    if not authorization or not authorization.startswith("Bearer "):
373
        raise Auth.exceptions.HTTPException(
374
            status_code=401,
375
            detail="Missing or invalid authorization header"
376
        )
377

378
    token = authorization[7:]  # Remove "Bearer "
379
    user_id = await verify_token(token)  # Your token verification logic
380

381
    if not user_id:
382
        raise Auth.exceptions.HTTPException(
383
            status_code=401,
384
            detail="Invalid token"
385
        )
386

387
    return user_id
388

389
async def verify_token(token: str) -> str:
390
    """Your token verification implementation."""
391
    # Call your auth service, check database, etc.
392
    if token == "valid-token-123":
393
        return "user-123"
394
    return None
395
```
396

397
### Advanced Authentication with Permissions
398

399
```python
400
@auth.authenticate
401
async def authenticate(
402
    authorization: str,
403
    path: str,
404
    method: str
405
) -> Auth.types.MinimalUserDict:
406
    """
407
    Authentication with user permissions.
408
    """
409
    if not authorization:
410
        raise Auth.exceptions.HTTPException(401, "Authorization required")
411

412
    # Verify token and get user info
413
    user_data = await verify_jwt_token(authorization)
414

415
    return {
416
        "identity": user_data["user_id"],
417
        "permissions": user_data.get("permissions", []),
418
        "display_name": user_data.get("name", "Unknown User")
419
    }
420

421
async def verify_jwt_token(authorization: str) -> dict:
422
    """Verify JWT and return user data."""
423
    # JWT verification logic
424
    import jwt
425

426
    try:
427
        token = authorization.replace("Bearer ", "")
428
        payload = jwt.decode(token, "secret", algorithms=["HS256"])
429
        return payload
430
    except jwt.InvalidTokenError:
431
        raise Auth.exceptions.HTTPException(401, "Invalid token")
432
```
433

434
### Authorization Handlers
435

436
```python
437
# Global authorization handler
438
@auth.on
439
async def global_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
440
    """
441
    Global handler for all requests.
442
    Applied when no specific handler matches.
443
    """
444
    # Log all requests
445
    print(f"Request: {ctx.method} {ctx.path} by {ctx.user.identity}")
446

447
    # Allow all requests (specific handlers can override)
448
    return True
449

450
# Resource-specific handlers
451
@auth.on.threads
452
async def thread_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
453
    """
454
    Handle all thread operations.
455
    More specific than global handler.
456
    """
457
    # Users can only access their own threads
458
    thread_metadata = value.get("metadata", {})
459
    owner = thread_metadata.get("owner")
460

461
    return owner == ctx.user.identity
462

463
# Action-specific handlers
464
@auth.on.threads.create
465
async def thread_create_auth(
466
    ctx: Auth.types.AuthContext,
467
    value: Auth.types.ThreadsCreate
468
) -> bool:
469
    """
470
    Handle thread creation specifically.
471
    Most specific handler.
472
    """
473
    # Check user has permission to create threads
474
    return "threads:create" in getattr(ctx.user, "permissions", [])
475

476
@auth.on.threads.delete
477
async def thread_delete_auth(
478
    ctx: Auth.types.AuthContext,
479
    value: Auth.types.ThreadsDelete
480
) -> bool:
481
    """
482
    Restrict thread deletion to admins.
483
    """
484
    return "admin" in getattr(ctx.user, "permissions", [])
485
```
486

487
### Store Authorization
488

489
```python
490
@auth.on.store
491
async def store_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
492
    """
493
    Authorize store operations.
494
    Enforce user isolation in namespaces.
495
    """
496
    namespace = value.get("namespace", [])
497

498
    # Ensure user can only access their own namespace
499
    if len(namespace) >= 2 and namespace[0] == "users":
500
        return namespace[1] == ctx.user.identity
501

502
    # Allow access to shared namespaces for certain users
503
    if namespace[0] in ["shared", "public"]:
504
        return True
505

506
    # Admins can access everything
507
    return "admin" in getattr(ctx.user, "permissions", [])
508

509
# Action-specific store handlers
510
@auth.on.store(actions=["put", "delete"])
511
async def store_write_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
512
    """
513
    Restrict write operations.
514
    """
515
    # Only users with write permission can modify data
516
    return "store:write" in getattr(ctx.user, "permissions", [])
517

518
@auth.on.store(actions=["search", "list_namespaces"])
519
async def store_search_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
520
    """
521
    Allow search operations for authenticated users.
522
    """
523
    return not getattr(ctx.user, "is_anonymous", False)
524
```
525

526
### Advanced Authorization Patterns
527

528
```python
529
# Multi-resource authorization
530
@auth.on(resources=["threads", "runs"], actions=["create", "update"])
531
async def rate_limit_writes(ctx: Auth.types.AuthContext, value: dict) -> bool:
532
    """
533
    Rate limit write operations across resources.
534
    """
535
    user_id = ctx.user.identity
536

537
    # Check rate limit
538
    current_minute = datetime.now().strftime("%Y-%m-%d %H:%M")
539
    rate_key = f"rate_limit:{user_id}:{current_minute}"
540

541
    current_count = await redis_client.get(rate_key) or 0
542
    if int(current_count) >= 100:  # 100 writes per minute
543
        raise Auth.exceptions.HTTPException(
544
            status_code=429,
545
            detail="Rate limit exceeded"
546
        )
547

548
    await redis_client.incr(rate_key)
549
    await redis_client.expire(rate_key, 60)
550

551
    return True
552

553
# Conditional authorization
554
@auth.on.assistants.update
555
async def assistant_update_auth(
556
    ctx: Auth.types.AuthContext,
557
    value: Auth.types.AssistantsUpdate
558
) -> bool:
559
    """
560
    Allow assistant updates based on ownership or admin role.
561
    """
562
    assistant_id = value["assistant_id"]
563

564
    # Get assistant metadata
565
    assistant = await get_assistant_metadata(assistant_id)
566

567
    # Owner can always update
568
    if assistant.get("owner") == ctx.user.identity:
569
        return True
570

571
    # Admins can update any assistant
572
    if "admin" in getattr(ctx.user, "permissions", []):
573
        return True
574

575
    # Collaborators can update if they have permission
576
    collaborators = assistant.get("collaborators", [])
577
    return ctx.user.identity in collaborators
578

579
# Data filtering authorization
580
@auth.on.threads.search
581
async def thread_search_filter(
582
    ctx: Auth.types.AuthContext,
583
    value: Auth.types.ThreadsSearch
584
) -> dict:
585
    """
586
    Filter search results based on user permissions.
587
    Returns filter dict instead of boolean.
588
    """
589
    # Regular users can only see their own threads
590
    if "admin" not in getattr(ctx.user, "permissions", []):
591
        return {
592
            "metadata.owner": ctx.user.identity
593
        }
594

595
    # Admins see everything
596
    return {}
597
```
598

599
### Configuration Integration
600

601
```python
602
# Example langgraph.json configuration
603
"""
604
{
605
  "dependencies": ["."],
606
  "graphs": {
607
    "my_assistant": "./assistant.py:graph"
608
  },
609
  "env": ".env",
610
  "auth": {
611
    "path": "./auth.py:auth"
612
  }
613
}
614
"""
615

616
# auth.py file
617
auth = Auth()
618

619
@auth.authenticate
620
async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:
621
    # Your authentication logic
622
    return await verify_user(authorization)
623

624
@auth.on
625
async def default_auth(ctx: Auth.types.AuthContext, value: dict) -> bool:
626
    # Default authorization logic
627
    return True
628

629
# Export the auth instance
630
__all__ = ["auth"]
631
```