Restore SQLite history

Reopen a real temporary database, restore history, paginate messages, update the title, and delete the session.

Run it

From the repository root:

uv run python docs/_examples/sessions.py

The output includes:

Raw roles: system, user, assistant, system, user, assistant

This program uses deterministic offline models and temporary storage where needed. It requires no API key and is executed by the documentation test suite.

What to take into your application

Keep one stable session_id for a conversation and open a persistence extension over the same database when the process restarts. The example closes the first connection before creating a second client, so the restored history comes from SQLite, not an in-memory cache.

async def main() -> None:
    # No demo touches the default database under the user's home directory.
    with TemporaryDirectory(prefix="zett-docs-") as directory:
        path = Path(directory) / "sessions.sqlite"
        session_id = "project-notes"
        config = AgentRunConfig(session_id=session_id)
        first_storage = SQLiteSessionExtension(path)
        try:
            first = await create_agent(TurnModel(), config=config, extensions=[first_storage])
            print((await first.run("Remember Python", metadata={"origin": "tutorial"})).content)
        finally:
            await first_storage.close()

        restored_storage = SQLiteSessionExtension(path)
        try:
            restored = await create_agent(TurnModel(), config=config, extensions=[restored_storage])
            print((await restored.run("Continue")).content)
            rows = await restored_storage.list_raw_messages(session_id)
            assert [row.message.role.value for row in rows] == [
                "system",
                "user",
                "assistant",
                "system",
                "user",
                "assistant",
            ]
            assert rows[0].metadata == {"origin": "tutorial"}
            assert len(await restored_storage.list_raw_messages(session_id, limit=2, offset=2)) == 2
            view = await restored_storage.storage.load(session_id)
            assert view.snapshot is None and len(view.raw_tail) == 6
            await restored_storage.update_session(session_id, title="Project notes")
            renamed = await restored_storage.get_session(session_id)
            assert renamed is not None and renamed.title == "Project notes"
            assert await restored_storage.delete_session(session_id)
            assert await restored_storage.list_raw_messages(session_id) == []
            print("Raw roles: system, user, assistant, system, user, assistant")
            print("Session restored, renamed, and deleted in a temporary database")
        finally:
            await restored_storage.close()

The example also paginates messages, renames the conversation, and deletes it. It only deletes a session in a temporary database. In your application, session deletion removes history; put it behind an explicit user action.

Complete source

Download sessions.py.

"""Restore a conversation through a new SQLite connection and a new Agent."""

import asyncio
from collections.abc import AsyncIterator
from pathlib import Path
from tempfile import TemporaryDirectory

from zett_agent.agent import AgentRunConfig
from zett_agent.client import create_agent
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.messages import (
    AssistantMessage,
    UserMessage,
)
from zett_agent.model import (
    ModelEvent,
    ModelRequest,
    ModelResponse,
    RetryOptions,
)


class TurnModel:
    retry = RetryOptions(max_retries=0)

    async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
        turn = sum(isinstance(message, UserMessage) for message in request.messages)
        yield ModelEvent.completed(ModelResponse(AssistantMessage(content=f"Stored turn {turn}")))


async def main() -> None:
    # No demo touches the default database under the user's home directory.
    with TemporaryDirectory(prefix="zett-docs-") as directory:
        path = Path(directory) / "sessions.sqlite"
        session_id = "project-notes"
        config = AgentRunConfig(session_id=session_id)
        first_storage = SQLiteSessionExtension(path)
        try:
            first = await create_agent(TurnModel(), config=config, extensions=[first_storage])
            print((await first.run("Remember Python", metadata={"origin": "tutorial"})).content)
        finally:
            await first_storage.close()

        restored_storage = SQLiteSessionExtension(path)
        try:
            restored = await create_agent(TurnModel(), config=config, extensions=[restored_storage])
            print((await restored.run("Continue")).content)
            rows = await restored_storage.list_raw_messages(session_id)
            assert [row.message.role.value for row in rows] == [
                "system",
                "user",
                "assistant",
                "system",
                "user",
                "assistant",
            ]
            assert rows[0].metadata == {"origin": "tutorial"}
            assert len(await restored_storage.list_raw_messages(session_id, limit=2, offset=2)) == 2
            view = await restored_storage.storage.load(session_id)
            assert view.snapshot is None and len(view.raw_tail) == 6
            await restored_storage.update_session(session_id, title="Project notes")
            renamed = await restored_storage.get_session(session_id)
            assert renamed is not None and renamed.title == "Project notes"
            assert await restored_storage.delete_session(session_id)
            assert await restored_storage.list_raw_messages(session_id) == []
            print("Raw roles: system, user, assistant, system, user, assistant")
            print("Session restored, renamed, and deleted in a temporary database")
        finally:
            await restored_storage.close()


if __name__ == "__main__":
    asyncio.run(main())

Next: Keep and restore conversations for storage choices and resource ownership, or Compact without losing history for long conversations.