pypi-langgraph-sdk

1
# LangGraph SDK
2

3
A comprehensive Python SDK for interacting with the LangGraph Platform REST API. The SDK enables developers to build and manage AI assistants and conversational workflows with async and sync client interfaces, automatic local server discovery, streaming support, and fine-grained authentication and authorization management.
4

5
## Package Information
6

7
- **Package Name**: langgraph-sdk
8
- **Language**: Python
9
- **Installation**: `pip install langgraph-sdk`
10

11
## Core Imports
12

13
```python
14
from langgraph_sdk import get_client, get_sync_client, Auth
15
```
16

17
## Basic Usage
18

19
```python
20
from langgraph_sdk import get_client
21

22
# Connect to LangGraph server (auto-detects local server at localhost:8123)
23
client = await get_client()
24

25
# List all assistants
26
assistants = await client.assistants.search()
27
agent = assistants[0]
28

29
# Create a new conversation thread
30
thread = await client.threads.create()
31

32
# Start a streaming run
33
input_data = {"messages": [{"role": "human", "content": "Hello!"}]}
34
async for chunk in client.runs.stream(
35
    thread['thread_id'],
36
    agent['assistant_id'],
37
    input=input_data
38
):
39
    print(chunk)
40

41
# Close the client
42
await client.aclose()
43
```
44

45
## Architecture
46

47
The LangGraph SDK follows a resource-oriented design with distinct client managers:
48

49
- **Client Factory**: `get_client()` and `get_sync_client()` create configured HTTP clients
50
- **Resource Clients**: Dedicated managers for assistants, threads, runs, crons, and store operations
51
- **Authentication System**: Pluggable auth handlers for custom security implementations
52
- **Streaming Support**: Server-sent events for real-time execution monitoring
53
- **Type System**: Comprehensive TypedDict schemas for all API interactions
54

55
Both async and sync versions provide identical APIs, enabling integration into any Python application architecture.
56

57
## Capabilities
58

59
### Client Management
60

61
Core client creation and HTTP operations for connecting to LangGraph servers with automatic discovery, custom authentication, and connection management.
62

63
```python { .api }
64
from collections.abc import Mapping
65
from typing import Union, Optional
66
import httpx
67

68
# Type aliases
69
TimeoutTypes = Union[
70
    None,
71
    float,
72
    tuple[Optional[float], Optional[float]],
73
    tuple[Optional[float], Optional[float], Optional[float], Optional[float]],
74
    httpx.Timeout,
75
]
76

77
def get_client(
78
    *,
79
    url: str | None = None,
80
    api_key: str | None = None,
81
    headers: Mapping[str, str] | None = None,
82
    timeout: TimeoutTypes | None = None,
83
) -> LangGraphClient: ...
84

85
def get_sync_client(
86
    *,
87
    url: str | None = None,
88
    api_key: str | None = None,
89
    headers: Mapping[str, str] | None = None,
90
    timeout: TimeoutTypes | None = None,
91
) -> SyncLangGraphClient: ...
92
```
93

94
[Client Management](./client-management.md)
95

96
### Assistant Management
97

98
Create, configure, and manage AI assistants based on registered graphs. Assistants serve as the execution engines for conversational workflows, with support for versioning, configuration, and metadata management.
99

100
```python { .api }
101
from collections.abc import Mapping
102
from langgraph_sdk.schema import (
103
    Assistant, AssistantSelectField, AssistantSortBy, SortOrder,
104
    Config, Context, Json, OnConflictBehavior, QueryParamTypes
105
)
106

107
# Via client.assistants
108
async def get(
109
    assistant_id: str,
110
    *,
111
    headers: Mapping[str, str] | None = None,
112
    params: QueryParamTypes | None = None,
113
) -> Assistant: ...
114

115
async def create(
116
    graph_id: str | None,
117
    config: Config | None = None,
118
    *,
119
    context: Context | None = None,
120
    metadata: Json = None,
121
    assistant_id: str | None = None,
122
    if_exists: OnConflictBehavior | None = None,
123
    name: str | None = None,
124
    headers: Mapping[str, str] | None = None,
125
    params: QueryParamTypes | None = None,
126
) -> Assistant: ...
127

128
async def update(
129
    assistant_id: str,
130
    *,
131
    graph_id: str | None = None,
132
    config: Config | None = None,
133
    context: Context | None = None,
134
    metadata: Json = None,
135
    name: str | None = None,
136
    headers: Mapping[str, str] | None = None,
137
    description: str | None = None,
138
    params: QueryParamTypes | None = None,
139
) -> Assistant: ...
140

141
async def search(
142
    *,
143
    metadata: Json = None,
144
    graph_id: str | None = None,
145
    limit: int = 10,
146
    offset: int = 0,
147
    sort_by: AssistantSortBy | None = None,
148
    sort_order: SortOrder | None = None,
149
    select: list[AssistantSelectField] | None = None,
150
    headers: Mapping[str, str] | None = None,
151
    params: QueryParamTypes | None = None,
152
) -> list[Assistant]: ...
153
```
154

155
[Assistant Management](./assistant-management.md)
156

157
### Thread Management
158

159
Manage conversation threads that maintain state across multiple interactions. Threads provide isolation for conversations and support state inspection, updates, and history tracking.
160

161
```python { .api }
162
from collections.abc import Mapping, Sequence
163
from typing import Any
164
from langgraph_sdk.schema import (
165
    Thread, ThreadSelectField, ThreadSortBy, ThreadState, ThreadStatus,
166
    ThreadUpdateStateResponse, Checkpoint, Json, OnConflictBehavior, QueryParamTypes
167
)
168

169
# Via client.threads
170
async def create(
171
    *,
172
    metadata: Json = None,
173
    thread_id: str | None = None,
174
    if_exists: OnConflictBehavior | None = None,
175
    supersteps: Sequence[dict[str, Sequence[dict[str, Any]]]] | None = None,
176
    graph_id: str | None = None,
177
    ttl: int | Mapping[str, Any] | None = None,
178
    headers: Mapping[str, str] | None = None,
179
    params: QueryParamTypes | None = None,
180
) -> Thread: ...
181

182
async def get(
183
    thread_id: str,
184
    *,
185
    select: list[ThreadSelectField] | None = None,
186
    headers: Mapping[str, str] | None = None,
187
    params: QueryParamTypes | None = None,
188
) -> Thread: ...
189

190
async def update(
191
    thread_id: str,
192
    *,
193
    metadata: Mapping[str, Any],
194
    ttl: int | Mapping[str, Any] | None = None,
195
    headers: Mapping[str, str] | None = None,
196
    params: QueryParamTypes | None = None,
197
) -> Thread: ...
198

199
async def search(
200
    *,
201
    metadata: Json = None,
202
    values: Json = None,
203
    ids: Sequence[str] | None = None,
204
    status: ThreadStatus | None = None,
205
    limit: int = 10,
206
    offset: int = 0,
207
    sort_by: ThreadSortBy | None = None,
208
    sort_order: SortOrder | None = None,
209
    select: list[ThreadSelectField] | None = None,
210
    headers: Mapping[str, str] | None = None,
211
    params: QueryParamTypes | None = None,
212
) -> list[Thread]: ...
213

214
async def get_state(
215
    thread_id: str,
216
    *,
217
    checkpoint: Checkpoint | None = None,
218
    checkpoint_id: str | None = None,  # deprecated
219
    subgraphs: bool = False,
220
    headers: Mapping[str, str] | None = None,
221
    params: QueryParamTypes | None = None,
222
) -> ThreadState: ...
223

224
async def update_state(
225
    thread_id: str,
226
    values: dict[str, Any] | Sequence[dict] | None,
227
    *,
228
    as_node: str | None = None,
229
    checkpoint: Checkpoint | None = None,
230
    checkpoint_id: str | None = None,  # deprecated
231
    headers: Mapping[str, str] | None = None,
232
    params: QueryParamTypes | None = None,
233
) -> ThreadUpdateStateResponse: ...
234
```
235

236
[Thread Management](./thread-management.md)
237

238
### Run Execution
239

240
Execute assistant workflows on threads with support for streaming, interrupts, configuration, and completion handling. Runs represent individual executions of an assistant on a thread.
241

242
```python { .api }
243
from collections.abc import AsyncIterator, Mapping, Sequence
244
from typing import Any
245
from langgraph_sdk.schema import (
246
    Run, StreamPart, StreamMode, Config, Context, Checkpoint,
247
    Command, CancelAction, QueryParamTypes
248
)
249

250
# Via client.runs
251
def stream(
252
    thread_id: str | None,
253
    assistant_id: str,
254
    *,
255
    input: Mapping[str, Any] | None = None,
256
    command: Command | None = None,
257
    stream_mode: StreamMode | Sequence[StreamMode] = "values",
258
    stream_subgraphs: bool = False,
259
    stream_resumable: bool = False,
260
    metadata: Mapping[str, Any] | None = None,
261
    config: Config | None = None,
262
    context: Context | None = None,
263
    checkpoint: Checkpoint | None = None,
264
    checkpoint_id: str | None = None,  # deprecated
265
    webhook: str | None = None,
266
    webhook_mode: str | None = None,
267
    headers: Mapping[str, str] | None = None,
268
    params: QueryParamTypes | None = None,
269
) -> AsyncIterator[StreamPart]: ...
270

271
async def create(
272
    thread_id: str | None,
273
    assistant_id: str,
274
    *,
275
    input: Mapping[str, Any] | None = None,
276
    command: Command | None = None,
277
    stream_mode: StreamMode | Sequence[StreamMode] = "values",
278
    stream_subgraphs: bool = False,
279
    stream_resumable: bool = False,
280
    metadata: Mapping[str, Any] | None = None,
281
    config: Config | None = None,
282
    context: Context | None = None,
283
    checkpoint: Checkpoint | None = None,
284
    checkpoint_id: str | None = None,  # deprecated
285
    webhook: str | None = None,
286
    webhook_mode: str | None = None,
287
    headers: Mapping[str, str] | None = None,
288
    params: QueryParamTypes | None = None,
289
) -> Run: ...
290

291
async def wait(
292
    thread_id: str | None,
293
    assistant_id: str,
294
    *,
295
    input: Mapping[str, Any] | None = None,
296
    command: Command | None = None,
297
    metadata: Mapping[str, Any] | None = None,
298
    config: Config | None = None,
299
    context: Context | None = None,
300
    checkpoint: Checkpoint | None = None,
301
    checkpoint_id: str | None = None,  # deprecated
302
    webhook: str | None = None,
303
    webhook_mode: str | None = None,
304
    checkpoint_during: bool | None = None,
305
    headers: Mapping[str, str] | None = None,
306
    params: QueryParamTypes | None = None,
307
) -> Run: ...
308

309
async def cancel(
310
    thread_id: str,
311
    run_id: str,
312
    *,
313
    wait: bool = False,
314
    action: CancelAction = "interrupt",
315
    headers: Mapping[str, str] | None = None,
316
    params: QueryParamTypes | None = None,
317
) -> None: ...
318
```
319

320
[Run Execution](./run-execution.md)
321

322
### Scheduled Tasks
323

324
Create and manage cron jobs for automated execution of assistants on threads or with dynamic thread creation. Supports timezone handling, webhook notifications, and flexible scheduling.
325

326
```python { .api }
327
from collections.abc import Mapping
328
from typing import Any
329
from langgraph_sdk.schema import (
330
    Cron, CronSelectField, CronSortBy, SortOrder,
331
    Config, Context, All, QueryParamTypes
332
)
333

334
# Via client.crons
335
async def create(
336
    assistant_id: str,
337
    *,
338
    schedule: str,
339
    input: Mapping[str, Any] | None = None,
340
    metadata: Mapping[str, Any] | None = None,
341
    config: Config | None = None,
342
    context: Context | None = None,
343
    checkpoint_during: bool | None = None,
344
    interrupt_before: All | list[str] | None = None,
345
    interrupt_after: All | list[str] | None = None,
346
    webhook: str | None = None,
347
    webhook_mode: str | None = None,
348
    headers: Mapping[str, str] | None = None,
349
    params: QueryParamTypes | None = None,
350
) -> Cron: ...
351

352
async def create_for_thread(
353
    thread_id: str,
354
    assistant_id: str,
355
    *,
356
    schedule: str,
357
    input: Mapping[str, Any] | None = None,
358
    metadata: Mapping[str, Any] | None = None,
359
    config: Config | None = None,
360
    context: Context | None = None,
361
    checkpoint_during: bool | None = None,
362
    interrupt_before: All | list[str] | None = None,
363
    interrupt_after: All | list[str] | None = None,
364
    webhook: str | None = None,
365
    webhook_mode: str | None = None,
366
    headers: Mapping[str, str] | None = None,
367
    params: QueryParamTypes | None = None,
368
) -> Cron: ...
369

370
async def search(
371
    *,
372
    assistant_id: str | None = None,
373
    thread_id: str | None = None,
374
    limit: int = 10,
375
    offset: int = 0,
376
    sort_by: CronSortBy | None = None,
377
    sort_order: SortOrder | None = None,
378
    select: list[CronSelectField] | None = None,
379
    headers: Mapping[str, str] | None = None,
380
    params: QueryParamTypes | None = None,
381
) -> list[Cron]: ...
382

383
async def delete(
384
    cron_id: str,
385
    *,
386
    headers: Mapping[str, str] | None = None,
387
    params: QueryParamTypes | None = None,
388
) -> None: ...
389
```
390

391
[Scheduled Tasks](./scheduled-tasks.md)
392

393
### Persistent Storage
394

395
Cross-thread persistent memory system for storing and retrieving documents, configuration, and application data with namespacing, search capabilities, and flexible data organization.
396

397
```python { .api }
398
from collections.abc import Mapping, Sequence
399
from typing import Any, Literal
400
from langgraph_sdk.schema import (
401
    Item, SearchItemsResponse, ListNamespaceResponse, QueryParamTypes
402
)
403

404
# Via client.store
405
async def put_item(
406
    namespace: Sequence[str],
407
    /,
408
    key: str,
409
    value: Mapping[str, Any],
410
    index: Literal[False] | list[str] | None = None,
411
    ttl: int | None = None,
412
    headers: Mapping[str, str] | None = None,
413
    params: QueryParamTypes | None = None,
414
) -> None: ...
415

416
async def get_item(
417
    namespace: Sequence[str],
418
    /,
419
    key: str,
420
    *,
421
    refresh_ttl: bool | None = None,
422
    headers: Mapping[str, str] | None = None,
423
    params: QueryParamTypes | None = None,
424
) -> Item: ...
425

426
async def search_items(
427
    namespace_prefix: Sequence[str],
428
    /,
429
    filter: Mapping[str, Any] | None = None,
430
    limit: int = 10,
431
    offset: int = 0,
432
    query: str | None = None,
433
    refresh_ttl: bool | None = None,
434
    headers: Mapping[str, str] | None = None,
435
    params: QueryParamTypes | None = None,
436
) -> SearchItemsResponse: ...
437

438
async def delete_item(
439
    namespace: Sequence[str],
440
    /,
441
    key: str,
442
    headers: Mapping[str, str] | None = None,
443
    params: QueryParamTypes | None = None,
444
) -> None: ...
445

446
async def list_namespaces(
447
    *,
448
    prefix: Sequence[str] | None = None,
449
    suffix: Sequence[str] | None = None,
450
    limit: int = 100,
451
    offset: int = 0,
452
    headers: Mapping[str, str] | None = None,
453
    params: QueryParamTypes | None = None,
454
) -> ListNamespaceResponse: ...
455
```
456

457
[Persistent Storage](./persistent-storage.md)
458

459
### Authentication & Authorization
460

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

463
```python { .api }
464
from typing import Callable, TypeVar
465
from collections.abc import Sequence
466
from langgraph_sdk.auth import types, exceptions
467

468
TH = TypeVar("TH", bound=types.Handler)
469
AH = TypeVar("AH", bound=types.Authenticator)
470

471
class Auth:
472
    types = types
473
    exceptions = exceptions
474

475
    def __init__(self) -> None:
476
        self.on: _On = ...  # Authorization handlers
477

478
    def authenticate(self, fn: AH) -> AH: ...
479

480
    # Authorization context classes
481
    class _On:
482
        assistants: _AssistantsOn
483
        threads: _ThreadsOn
484
        crons: _CronsOn
485
        store: _StoreOn
486
        value: type[dict[str, Any]]
487

488
        def __call__(
489
            self,
490
            fn: Callable | None = None,
491
            *,
492
            resources: str | Sequence[str] | None = None,
493
            actions: str | Sequence[str] | None = None,
494
        ) -> Callable: ...
495
```
496

497
[Authentication & Authorization](./authentication.md)
498

499
## Core Types
500

501
```python { .api }
502
# Client Types
503
class LangGraphClient:
504
    assistants: AssistantsClient
505
    threads: ThreadsClient
506
    runs: RunsClient
507
    crons: CronClient
508
    store: StoreClient
509
    async def aclose(self) -> None: ...
510

511
class SyncLangGraphClient:
512
    assistants: SyncAssistantsClient
513
    threads: SyncThreadsClient
514
    runs: SyncRunsClient
515
    crons: SyncCronClient
516
    store: SyncStoreClient
517
    def close(self) -> None: ...
518

519
# Core Data Models
520
class Assistant(TypedDict):
521
    assistant_id: str
522
    graph_id: str
523
    config: Config
524
    created_at: str
525
    updated_at: str
526
    metadata: dict
527

528
class Thread(TypedDict):
529
    thread_id: str
530
    created_at: str
531
    updated_at: str
532
    metadata: dict
533
    status: ThreadStatus
534

535
class Run(TypedDict):
536
    run_id: str
537
    thread_id: str
538
    assistant_id: str
539
    created_at: str
540
    updated_at: str
541
    status: RunStatus
542
    kwargs: dict
543

544
class Cron(TypedDict):
545
    cron_id: str
546
    thread_id: str
547
    assistant_id: str
548
    schedule: str
549
    timezone: str
550
    created_at: str
551

552
class Item(TypedDict):
553
    namespace: list[str]
554
    key: str
555
    value: dict
556
    created_at: str
557
    updated_at: str
558

559
# Status Types
560
RunStatus = Literal["pending", "running", "error", "success", "timeout", "interrupted"]
561
ThreadStatus = Literal["idle", "busy", "interrupted", "error"]
562
StreamMode = Literal["values", "messages", "updates", "events", "tasks", "checkpoints", "debug"]
563

564
# Configuration Types
565
class Config(TypedDict, total=False):
566
    tags: list[str]
567
    recursion_limit: int
568
    configurable: dict
569
```