Threads and state

Threads are the core persistence unit in Aegra. Each thread represents a conversation with its own state, message history, and checkpoints. Runs execute within threads, and state is automatically persisted after each node execution.

Creating threads

import asyncio
from langgraph_sdk import get_client


async def main():
    client = get_client(url="http://localhost:2026")

    # Create a thread
    thread = await client.threads.create()

    # Create with metadata
    thread = await client.threads.create(
        metadata={"user_name": "Alice", "session": "onboarding"},
    )

    # Create with a specific ID (idempotent)
    thread = await client.threads.create(
        thread_id="my-thread-123",
        if_exists="do_nothing",  # Don't error if it already exists
    )


asyncio.run(main())

A client-provided thread_id must be 1–255 non-blank characters. Omit the field (or send null) to let the server generate a UUID. Longer or blank values are rejected with 422 before they hit PostgreSQL’s btree key limit (~2704 bytes).

Thread status

Threads have a status that reflects their current state:

Status	Meaning
`idle`	No active run, ready for new input
`busy`	A run is currently executing
`interrupted`	A run paused for human input
`error`	The last run ended with an error

thread = await client.threads.get(thread_id)
print(thread["status"])  # "idle", "busy", "interrupted", "error"

The snippets below assume you are inside an async def function with an initialized client — see the example above.

Getting thread state

The state contains the current values of all state fields, information about what nodes will execute next, and any pending interrupts.

state = await client.threads.get_state(thread_id)

# Current state values (your State schema fields)
print(state["values"])

# Next nodes to execute (empty if completed)
print(state["next"])

# Pending tasks
print(state["tasks"])

# Active interrupts (if thread is interrupted)
print(state["interrupts"])

# Checkpoint info
print(state["checkpoint"]["checkpoint_id"])

Updating thread state

You can modify thread state directly, for example to inject data or correct values:

await client.threads.update_state(
    thread_id,
    values={"messages": [{"type": "human", "content": "Injected message"}]},
    as_node="agent",  # Apply update as if it came from this node
)

Checkpoint history

Every state change creates a checkpoint. You can browse the full history:

# Get recent checkpoints
history = await client.threads.get_history(thread_id)

for entry in history:
    print(f"Checkpoint: {entry['checkpoint']['checkpoint_id']}")
    print(f"  Values: {entry['values']}")
    print(f"  Next: {entry['next']}")

Get state at a specific checkpoint

# Go back in time to a specific checkpoint
state = await client.threads.get_state(
    thread_id,
    checkpoint={"checkpoint_id": "your-checkpoint-id"},
)

This is useful for debugging, replaying, or branching conversations from a previous point.

Run from a specific checkpoint

Pass checkpoint_id (or the full checkpoint object) to start a run from an earlier checkpoint. input is optional: without it the run replays from that point, with it the run branches with new input.

run = await client.runs.create(
    thread_id,
    "agent",
    checkpoint_id="your-checkpoint-id",
)

If both are set, keys in checkpoint win over checkpoint_id. A checkpoint_id that is not a UUID returns 422.

Checkpoint durability

By default LangGraph writes a checkpoint after every step of a run, in the background. Set durability on a run to change when checkpoints are written:

run = await client.runs.create(
    thread_id,
    "agent",
    input={"messages": [{"role": "user", "content": "Hi"}]},
    durability="exit",
)

Mode	When checkpoints are written	After a crash or requeue
`sync`	After every step, before the next step starts	Resumes from the last completed step
`async` (default)	After every step, in the background while the next step runs	Resumes from the last checkpoint that finished writing
`exit`	Once, when the run finishes, fails, or stops at an interrupt	Starts over from the thread’s state before the run

exit writes one checkpoint per run instead of one per step, which suits high-volume runs that never use time travel or mid-run recovery. It has two costs:

With Redis workers, a run requeued after a worker crash or a shutdown drain starts over from the thread’s last checkpoint, so its steps run a second time. See worker architecture.
The thread history holds one checkpoint per run, so you cannot inspect or fork from the steps inside it. Interrupts still write a checkpoint, so human-in-the-loop resumes work.

checkpoint_during is the deprecated form of the same setting: true means async and false means exit. If both are sent, durability wins. Creating a cron accepts both fields and updating one accepts durability; the cron applies them to every run it fires. An unknown value returns 422. To change the default for runs that do not set a mode, set checkpointer.durability in aegra.json or the AEGRA_CHECKPOINT_DURABILITY environment variable. See the configuration reference.

Searching threads

Find threads by status or metadata:

# Search by metadata
threads = await client.threads.search(
    metadata={"user_name": "Alice"},
)

# Search by status
threads = await client.threads.search(
    status="interrupted",
)

# With pagination
threads = await client.threads.search(
    limit=20,
    offset=0,
)

limit accepts 1 through MAX_SEARCH_LIMIT (default 1000, matching LangGraph Platform). Omitted or null limit uses min(20, MAX_SEARCH_LIMIT). Values above the cap return 422.

Listing threads

threads = await client.threads.search()
for thread in threads:
    print(f"{thread['thread_id']}: {thread['status']}")

Deleting threads

Deleting a thread cancels any active runs and removes all state, including the checkpoint history stored in the graph backend:

await client.threads.delete(thread_id)

Thread TTL

Threads can expire automatically. Configure a server-wide default under checkpointer.ttl in aegra.json (or the AEGRA_THREAD_TTL environment variable) — see the configuration reference. A background sweep then applies each thread’s policy:

delete — the thread, its runs and crons, and all checkpoint data are removed.
keep_latest — checkpoint history is compacted down to the latest state; the thread stays usable and the TTL re-arms, so history is pruned periodically.

Individual threads can override the default at creation:

# Delete after 24 hours (minutes; the SDK sends strategy "delete")
thread = await client.threads.create(ttl=1440)

# Keep the thread, prune its history every 7 days
thread = await client.threads.create(ttl={"ttl": 10080, "strategy": "keep_latest"})

Expired threads can also be cleaned up immediately for the calling user:

import httpx

resp = httpx.post("http://localhost:2026/threads/prune")
print(resp.json())  # {"deleted": 3, "pruned": 1}

Threads with pending or running runs are never expired mid-run — they are picked up on a later sweep once their runs settle. TTL applies to threads created after it is enabled; pre-existing threads are not backfilled.

Thread metadata

Threads automatically track metadata about their usage:

owner: User identity at creation time (mirrors user_id on the Thread object, which is used for access control)
assistant_id: Last assistant used
graph_id: Last graph executed

You can add your own metadata at creation or update time:

# Update metadata
await client.threads.update(
    thread_id,
    metadata={"priority": "high"},
)

User isolation

When authentication is enabled, threads are automatically scoped to the authenticated user. Users can only see and interact with their own threads.

Getting started

Core concepts

Data & storage

Configuration

Contributing

Reference

Threads and state

Creating threads

Thread status

Getting thread state

Updating thread state

Checkpoint history

Get state at a specific checkpoint

Run from a specific checkpoint

Checkpoint durability

Searching threads

Listing threads

Deleting threads

Thread TTL

Thread metadata

User isolation

​Creating threads

​Thread status

​Getting thread state

​Updating thread state

​Checkpoint history

​Get state at a specific checkpoint

​Run from a specific checkpoint

​Checkpoint durability

​Searching threads

​Listing threads

​Deleting threads

​Thread TTL

​Thread metadata

​User isolation

Creating threads

Thread status

Getting thread state

Updating thread state

Checkpoint history

Get state at a specific checkpoint

Run from a specific checkpoint

Checkpoint durability

Searching threads

Listing threads

Deleting threads

Thread TTL

Thread metadata

User isolation