Implementation:Langchain ai Langgraph SDK Schema
| Knowledge Sources | |
|---|---|
| Domains | SDK, Types |
| Last Updated | 2026-02-11 16:00 GMT |
Overview
The SDK schema module defines the TypedDict data models used to represent LangGraph Server API entities -- assistants, threads, runs, crons, store items, and graph schemas -- in the Python SDK client.
Description
`schema.py` is the canonical type definition layer for the LangGraph Python SDK. Every response from the LangGraph Server API is deserialized into one of the TypedDicts defined in this module, and many request payloads reference these types or their associated literal unions.
Core entity types include `Assistant` (with fields for `assistant_id`, `graph_id`, `config`, `context`, `metadata`, `version`, `name`, `description`, `created_at`, and `updated_at`), `Thread` (with `thread_id`, `status`, `values`, `interrupts`, `metadata`, and timestamps), `Run` (with `run_id`, `thread_id`, `assistant_id`, `status`, `metadata`, and `multitask_strategy`), and `Cron` (with scheduling fields like `schedule`, `end_time`, `next_run_date`, `payload`, and `enabled`). `AssistantVersion` extends `AssistantBase` for version-specific snapshots without the `updated_at` field.
State types include `ThreadState` (representing a full graph state snapshot with `values`, `next` nodes, `checkpoint`, `metadata`, `tasks`, and `interrupts`), `ThreadTask` (a task within a thread state, with `id`, `name`, `error`, `interrupts`, `checkpoint`, `state`, and `result`), and `Checkpoint` (with `thread_id`, `checkpoint_ns`, `checkpoint_id`, and `checkpoint_map`).
Store types include `Item` (a document in the graph's cross-thread store with `namespace`, `key`, `value`, and timestamps), `SearchItem` (extending `Item` with an optional relevance `score`), `ListNamespaceResponse`, and `SearchItemsResponse`.
Schema and config types include `GraphSchema` (describing a graph's input, output, state, config, and context JSON schemas), `Config` (with `tags`, `recursion_limit`, and `configurable`), `StreamPart` (a named tuple for stream responses), `Command` (for graph control flow with `goto`, `update`, and `resume`), and `Send` (for directing messages to specific nodes).
The module also defines numerous `Literal` type aliases for status enums (`RunStatus`, `ThreadStatus`), strategy enums (`MultitaskStrategy`, `OnConflictBehavior`, `DisconnectMode`), streaming modes (`StreamMode`, `ThreadStreamMode`), sort fields (`AssistantSortBy`, `ThreadSortBy`, `CronSortBy`), and select fields (`AssistantSelectField`, `ThreadSelectField`, `RunSelectField`, `CronSelectField`).
Usage
Import these types when working with the LangGraph Python SDK client to get full type safety for API interactions. They are used as return types from SDK client methods and as type annotations for request parameters.
Code Reference
Source Location
- Repository: Langchain_ai_Langgraph
- File: libs/sdk-py/langgraph_sdk/schema.py
Signature
class Assistant(TypedDict):
assistant_id: str
graph_id: str
config: Config
context: Context
created_at: datetime
updated_at: datetime
metadata: Json
version: int
name: str
description: str | None
class Thread(TypedDict):
thread_id: str
created_at: datetime
updated_at: datetime
metadata: Json
status: ThreadStatus
values: Json
interrupts: dict[str, list[Interrupt]]
class Run(TypedDict):
run_id: str
thread_id: str
assistant_id: str
created_at: datetime
updated_at: datetime
status: RunStatus
metadata: Json
multitask_strategy: MultitaskStrategy
class Cron(TypedDict):
cron_id: str
assistant_id: str
thread_id: str | None
schedule: str
created_at: datetime
updated_at: datetime
payload: dict
metadata: dict
enabled: bool
next_run_date: datetime | None
end_time: datetime | None
class Item(TypedDict):
namespace: list[str]
key: str
value: dict[str, Any]
created_at: datetime
updated_at: datetime
class GraphSchema(TypedDict):
graph_id: str
input_schema: dict | None
output_schema: dict | None
state_schema: dict | None
config_schema: dict | None
context_schema: dict | None
class ThreadState(TypedDict):
values: list[dict] | dict[str, Any]
next: Sequence[str]
checkpoint: Checkpoint
metadata: Json
created_at: str | None
parent_checkpoint: Checkpoint | None
tasks: Sequence[ThreadTask]
interrupts: list[Interrupt]
Import
from langgraph_sdk.schema import (
Assistant,
AssistantVersion,
Thread,
ThreadState,
ThreadTask,
Run,
Cron,
Item,
SearchItem,
ListNamespaceResponse,
GraphSchema,
Config,
Checkpoint,
StreamPart,
Command,
Send,
RunStatus,
ThreadStatus,
StreamMode,
MultitaskStrategy,
)
I/O Contract
Core Entity Types
| Type | Primary Key | Status Field | Timestamps | Description |
|---|---|---|---|---|
| `Assistant` | `assistant_id` | -- | `created_at`, `updated_at` | A configured graph instance with metadata and versioning |
| `Thread` | `thread_id` | `status: ThreadStatus` | `created_at`, `updated_at` | A conversation thread with state, interrupts, and metadata |
| `Run` | `run_id` | `status: RunStatus` | `created_at`, `updated_at` | A single graph execution within a thread |
| `Cron` | `cron_id` | `enabled: bool` | `created_at`, `updated_at` | A scheduled recurring graph execution |
Store Types
| Type | Key Fields | Description |
|---|---|---|
| `Item` | `namespace: list[str]`, `key: str` | A document in the cross-thread store |
| `SearchItem` | None` | An item with optional relevance score from search |
| `ListNamespaceResponse` | `namespaces: list[list[str]]` | Response listing namespace paths |
Status Enums
| Type | Values | Description |
|---|---|---|
| `RunStatus` | `"pending"`, `"running"`, `"error"`, `"success"`, `"timeout"`, `"interrupted"` | Lifecycle state of a run |
| `ThreadStatus` | `"idle"`, `"busy"`, `"interrupted"`, `"error"` | Current state of a thread |
| `StreamMode` | `"values"`, `"messages"`, `"updates"`, `"events"`, `"tasks"`, `"checkpoints"`, `"debug"`, `"custom"`, `"messages-tuple"` | Streaming output mode |
| `MultitaskStrategy` | `"reject"`, `"interrupt"`, `"rollback"`, `"enqueue"` | Concurrent run handling strategy |
Usage Examples
from langgraph_sdk import get_client
from langgraph_sdk.schema import Thread, Run, Assistant, Item
client = get_client(url="http://localhost:8123")
# Create an assistant
assistant: Assistant = await client.assistants.create(
graph_id="my_graph",
config={"configurable": {"model": "gpt-4"}},
metadata={"team": "engineering"},
)
print(f"Created assistant {assistant['assistant_id']} v{assistant['version']}")
# Create a thread
thread: Thread = await client.threads.create(
metadata={"owner": "user-123"},
)
print(f"Thread {thread['thread_id']} status: {thread['status']}")
# Start a run
run: Run = await client.runs.create(
thread_id=thread["thread_id"],
assistant_id=assistant["assistant_id"],
input={"messages": [{"role": "user", "content": "Hello"}]},
)
print(f"Run {run['run_id']} status: {run['status']}")
# Store a cross-thread memory
item: Item = await client.store.put_item(
namespace=["user-123", "preferences"],
key="theme",
value={"color": "dark", "language": "en"},
)
print(f"Stored item in {item['namespace']} with key {item['key']}")