Pydantic AI message history

Pydantic AI serializes a run’s message history to JSON and leaves storage to you. PerSQL makes that one table: store the serialized history per thread, load it back with ModelMessagesTypeAdapter, and pass it to the next run.

pip install pydantic-ai persql

import os
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessagesTypeAdapter
from persql import PerSQL

client = PerSQL(token=os.environ["PERSQL_TOKEN"])
db = client.database("acme/agent-state")

db.query(
    """CREATE TABLE IF NOT EXISTS chat_history (
         thread_id TEXT PRIMARY KEY,
         messages TEXT NOT NULL,
         updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)"""
)

agent = Agent("openai:gpt-5", instructions="Be concise.")


def load_history(thread_id: str):
    rows = db.query(
        "SELECT messages FROM chat_history WHERE thread_id = ?", [thread_id]
    )["data"]
    if not rows:
        return None
    return ModelMessagesTypeAdapter.validate_json(rows[0]["messages"])


def save_history(thread_id: str, result) -> None:
    db.query(
        """INSERT INTO chat_history (thread_id, messages) VALUES (?, ?)
           ON CONFLICT(thread_id) DO UPDATE SET
             messages = excluded.messages,
             updated_at = CURRENT_TIMESTAMP""",
        [thread_id, result.all_messages_json().decode()],
    )


result = agent.run_sync("hi", message_history=load_history("user-42"))
save_history("user-42", result)

The next run_sync with the same thread_id continues the conversation — history survives restarts and lives in a database you can inspect with SQL.

Tool calls and SQL belong next to the history

The same database that holds the history can be the agent’s workspace: hand the agent typed query tools over its own tables, or lease it a branch with a TTL for ephemeral runs. For durable graph state in LangGraph instead, use the LangGraph checkpointer.