docs

Persistence

Checkpointers save agent state (conversation history, tool results, token usage) so you can resume sessions and recover from crashes. Without a checkpointer, each .run() starts fresh.

PostgresCheckpointer

The recommended checkpointer for production. Stores state in PostgreSQL with async I/O.

Python
from ninetrix import Agent, PostgresCheckpointer

checkpointer = PostgresCheckpointer("postgresql://localhost/agents")

agent = Agent(
    name="assistant",
    provider="anthropic",
    model="claude-sonnet-4-6",
    checkpointer=checkpointer,
)

# First run — creates checkpoint
result = agent.run("What is MCP?", thread_id="session-1")

# Later (even after process restart) — resumes from checkpoint
result = agent.run("Tell me more about tool servers", thread_id="session-1")
# ^ agent remembers the previous conversation

InMemoryCheckpointer

Volatile storage for development and testing. State is lost when the process exits.

Python
from ninetrix import Agent, InMemoryCheckpointer

agent = Agent(
    name="assistant",
    provider="anthropic",
    model="claude-sonnet-4-6",
    checkpointer=InMemoryCheckpointer(),
)

Shared checkpointer via Ninetrix factory

Python
from ninetrix import Ninetrix, PostgresCheckpointer

# All agents share the same checkpointer
ntx = Ninetrix(
    provider="anthropic",
    model="claude-sonnet-4-6",
    checkpointer=PostgresCheckpointer("postgresql://localhost/agents"),
)

agent_a = ntx.agent(name="agent-a", instructions="...")
agent_b = ntx.agent(name="agent-b", instructions="...")
# Both persist to the same database, keyed by (agent_name, thread_id)

Custom checkpointer

Implement the Checkpointer abstract class to use any backend (Redis, S3, SQLite, etc.):

Python
from ninetrix import Checkpointer

class RedisCheckpointer(Checkpointer):
    async def save(self, thread_id: str, agent_id: str, data: dict) -> None:
        ...

    async def load(self, thread_id: str, agent_id: str) -> dict | None:
        ...
On this page