langgraph_agent_toolkit.core.memory.migration module

Plan and apply checkpoint maintenance while service workers are stopped.

class langgraph_agent_toolkit.core.memory.migration.MigrationEntry(*, old_thread_id, user_id, agent_id, thread_id=None)

Bases: BaseModel

Map one legacy thread to an explicit owner and agent.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

• old_thread_id (Annotated[str, MinLen(min_length=1)])

• user_id (Annotated[str, MinLen(min_length=1), MaxLen(max_length=256)])

• agent_id (Annotated[str, MinLen(min_length=1), MaxLen(max_length=256)])

• thread_id (Annotated[str | None, MinLen(min_length=1), MaxLen(max_length=256)])

model_config = {'extra': 'forbid', 'strict': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

old_thread_id: str

user_id: str

agent_id: str

thread_id: str | None

property public_thread_id: str

property destination: str

langgraph_agent_toolkit.core.memory.migration.load_manifest(path)

Read explicit mappings from a JSON array.

Parameters:

path (str | Path)

Return type:

list[MigrationEntry]

langgraph_agent_toolkit.core.memory.migration.migrate_sqlite(path, entries, *, apply=False)

Plan or atomically apply a SQLite thread migration.

Parameters:

• path (str | Path)

• entries (Any)

• apply (bool)

Return type:

dict[str, Any]

langgraph_agent_toolkit.core.memory.migration.migrate_postgres(conninfo, entries, *, schema='public', apply=False)

Plan or atomically apply a PostgreSQL thread migration.

Parameters:

• conninfo (str)

• entries (Any)

• schema (str)

• apply (bool)

Return type:

dict[str, Any]

langgraph_agent_toolkit.core.memory.migration.retain_sqlite(path, before, *, apply=False)

Plan or delete complete inactive SQLite threads before a timestamp.

Parameters:

• path (str | Path)

• before (str)

• apply (bool)

Return type:

dict[str, Any]

langgraph_agent_toolkit.core.memory.migration.retain_postgres(conninfo, before, *, schema='public', apply=False)

Plan or delete complete inactive PostgreSQL threads before a timestamp.

Parameters:

• conninfo (str)

• before (str)

• schema (str)

• apply (bool)

Return type:

dict[str, Any]

langgraph_agent_toolkit.core.memory.migration.main(backend, manifest=None, before=None, sqlite_path=None, postgres_env='DATABASE_URL', schema='public', apply=False)

Preview maintenance. Set –apply=true to commit after review.

Parameters:

• backend (str)

• manifest (str | None)

• before (str | None)

• sqlite_path (str | None)

• postgres_env (str)

• schema (str)

• apply (bool)

Return type:

None