Pydantic AI chat history in a database or Redis: store result.conversation per user and thread

Disclosure: I work on MemorySync (I'm Mohammed, the founder). I drafted this article with AI assistance and checked it against the GitHub issue, the Pydantic AI docs, the published PyPI packages and the MemorySync docs listed under Sources on Oct 31, 2026. Code status: I ran Steps 1 to 5 offline with outbound connections blocked and no API keys, on Python 3.13.5 with pydantic-ai-slim 2.64.0 (the latest release when I ran it on Oct 31), using Pydantic AI's own FunctionModel in place of a real model. The SQLite store ran against a local SQLite file, and the Redis version ran against fakeredis (an in-process stand-in), not a Redis server. I also ran the two UserErrors, the ImportError on pydantic-ai-slim 2.54.0, and the message_history fallback on 2.54.0 and 2.64.0. I didn't run a real model, PostgreSQL, a real Redis server, deferred tools, or the MemorySync snippet against the API (I only constructed the MemorySync capability offline).

Your Pydantic AI agent remembers the conversation while the process is up. Restart it, scale to two workers, or move it behind a web API, and every chat starts from zero, because message_history lives in memory unless you put it somewhere.

That's the ask in Pydantic AI issue #530, "Ability to Persist Messages in External Stores": keep message history in Redis or a database instead of in-process memory, so a conversation can be picked up later. It ran for almost two years and 30 comments. It was closed on Oct 5, 2026 with an answer: store the run's Conversation.

Short version: on pydantic-ai 2.55 or later, store result.conversation with ConversationTypeAdapter, keyed by user and thread, and pass it back as conversation= on the next run. It carries the messages plus the running usage, the conversation_id and any paused deferred tool requests. The details and what I checked are below.

MemorySync doesn't do this part for you. Storing and reloading a conversation is Pydantic AI plus your database. MemorySync comes up once, near the end, as a separate layer for facts about a user that should outlast any one thread.

Skip the long-term memory plumbing. The MemorySync Pydantic AI guide shows a MemorySyncCapability you add to Agent(capabilities=[...]) for per-user memory across conversations. It doesn't store your chat history for you. When you're ready, see MemorySync. Founding offer: use code FOUNDING30 for 30% off Starter for your first 3 months (Settings → Billing).

Status (Nov 15, 2026)

• #530 was opened Dec 23, 2024 (09:20 IST) and closed as completed on Oct 5, 2026 (03:06 IST). It has 30 comments and the labels feature, community snippets and message-history. The closing comment points to Conversation, ConversationTypeAdapter, the Persistence docs page and repair_messages, and says a store protocol in core is tracked in #9786 (checked Nov 15).
• Conversation and ConversationTypeAdapter first ship in pydantic-ai-slim 2.55.0 (Oct 10, 2026, 01:10 IST). I checked the published wheels: 2.54.0 doesn't have pydantic_ai/conversation.py, 2.55.0 does. The latest release on PyPI is pydantic-ai-slim 2.68.0 (and pydantic-ai 2.68.0; published Nov 12, 2026, 08:00 IST). I ran Steps 1 to 5 on 2.64.0 only. I read the 2.68.0 wheel (a source read, not a run): Conversation and ConversationTypeAdapter are still exported.
• The Pydantic AI Persistence page says the thing to store is a Conversation and that it fits "a jsonb column or equivalent, keyed by result.conversation_id". It also says Pydantic AI "gives you a full-fidelity serialization boundary and leaves the schema to you". The table design below is mine.
• A core ConversationStore (append-only log, conflict detection via revision, InMemoryConversationStore included) first shipped in 2.65.0 and is in 2.68.0. The Persistence page now lists it next to storing a Conversation yourself and says it does the append-only recipe for you. #9786 is still open (checked Nov 15). I have not run ConversationStore.

Step 1: Store the Conversation, not just the messages

from pydantic_ai import Agent, ConversationTypeAdapter

agent = Agent("openai:your-model", instructions="Be a helpful assistant.")

first = agent.run_sync("My name is Alice.", conversation_id="thread-42")
stored: bytes = ConversationTypeAdapter.dump_json(first.conversation)

# Later, in another request or process:
conversation = ConversationTypeAdapter.validate_json(stored)
second = agent.run_sync("What's my name?", conversation=conversation)
print(second.conversation_id, second.usage.requests)  # thread-42 2

In my run on 2.64.0, ConversationTypeAdapter.dump_json() returned JSON bytes with four keys: messages, usage, conversation_id and deferred_tool_requests. I read the 2.68.0 wheel (a source read, not a run): a Conversation also has revision and new_message_index. In my run the second call printed thread-42 2, and the fake model received both user prompts.

Why not just ModelMessagesTypeAdapter and message_history=? That still works, but per the message history docs the messages don't carry the conversation's usage or the deferred tool requests of a paused run. In my run, continuing with message_history= reported usage.requests == 1 for the new run, while conversation= on the same history reported 3. If you set a UsageLimits budget for a whole conversation, message_history alone resets it every turn.

Passing conversation_id="thread-42" on the first run makes the conversation ID your own thread ID. Without it, Pydantic AI generates one, and you read it back from result.conversation_id.

Step 2: Key it by user and thread

One row per (user_id, conversation_id). The user_id comes from your auth, never from the request body. This is the store I ran, against a SQLite file:

import sqlite3
from pydantic_ai import Conversation, ConversationTypeAdapter

class ConversationStore:
    """One row per (user_id, conversation_id). Stores the whole Conversation as JSON."""
    def __init__(self, path: str = "chats.db"):
        self.db = sqlite3.connect(path)
        self.db.execute(
            "CREATE TABLE IF NOT EXISTS conversations ("
            " user_id TEXT NOT NULL, conversation_id TEXT NOT NULL, data TEXT NOT NULL,"
            " PRIMARY KEY (user_id, conversation_id))"
        )

    def load(self, user_id: str, conversation_id: str) -> Conversation | None:
        row = self.db.execute(
            "SELECT data FROM conversations WHERE user_id = ? AND conversation_id = ?",
            (user_id, conversation_id),
        ).fetchone()
        return ConversationTypeAdapter.validate_json(row[0]) if row else None

    def save(self, user_id: str, conversation: Conversation) -> None:
        data = ConversationTypeAdapter.dump_json(conversation).decode()
        self.db.execute(
            "INSERT INTO conversations (user_id, conversation_id, data) VALUES (?, ?, ?)"
            " ON CONFLICT (user_id, conversation_id) DO UPDATE SET data = excluded.data",
            (user_id, conversation.conversation_id, data),
        )
        self.db.commit()

For Postgres, the same table works with a jsonb column for data, %s placeholders and the same ON CONFLICT ... DO UPDATE upsert. I didn't run it on Postgres.

Step 3: Load, run, save

store = ConversationStore()

async def chat(user_id: str, conversation_id: str, text: str) -> str:
    # user_id comes from your auth; conversation_id is your chat thread ID.
    conversation = store.load(user_id, conversation_id)
    if conversation is None:
        result = await agent.run(text, conversation_id=conversation_id)
    else:
        result = await agent.run(text, conversation=conversation)
    store.save(user_id, result.conversation)
    return result.output

I ran this with a fake model for Alice (thread t1, two messages), Bob (thread t1, one message), then Alice's thread t2. On her second message in t1 the model received both of Alice's prompts. Bob's t1 held only his message even though the thread ID was the same, because the user is part of the key. Alice's t2 started empty.

Two things to handle in production:

1. Concurrent messages on one thread. Two requests load the same conversation and the second save overwrites the first. Use a per-thread lock, or a version column you check on save.
2. Client-sent history. If your frontend sends the history (Vercel AI or AG-UI adapters), the docs say to treat it as untrusted and run it through sanitize_messages. Loading it from your own store, as above, avoids that.

Step 4: The same thing in Redis

import redis.asyncio as redis
from pydantic_ai import ConversationTypeAdapter

r = redis.Redis()  # your Redis URL here

async def chat_redis(user_id: str, conversation_id: str, text: str) -> str:
    key = f"chat:{user_id}:{conversation_id}"
    raw = await r.get(key)
    if raw is None:
        result = await agent.run(text, conversation_id=conversation_id)
    else:
        result = await agent.run(text, conversation=ConversationTypeAdapter.validate_json(raw))
    await r.set(key, ConversationTypeAdapter.dump_json(result.conversation), ex=30 * 24 * 3600)
    return result.output

The key includes the user, and the 30-day ex expiry is my choice; drop it to keep threads forever. I ran this against fakeredis in-process, not a Redis server: the second call saw both prompts and the key had a TTL.

Step 5: Long threads, append instead of rewrite

Rewriting the whole conversation each turn makes each write as large as the thread. The Persistence page suggests appending each run's new_messages() to a table of their own and keeping the rest of the conversation beside them, then reassembling a Conversation to continue. This is my version of that:

from pydantic import TypeAdapter
from pydantic_ai import Conversation, ModelMessagesTypeAdapter
from pydantic_ai.usage import RunUsage

UsageAdapter = TypeAdapter(RunUsage)

def save_run(db, user_id: str, result) -> None:
    # Append only this run's messages, so each write is the size of one turn.
    db.execute(
        "INSERT INTO chat_messages (user_id, conversation_id, data) VALUES (?, ?, ?)",
        (user_id, result.conversation_id,
         ModelMessagesTypeAdapter.dump_json(result.new_messages()).decode()),
    )
    # Keep the rest of the Conversation beside the messages.
    db.execute(
        "INSERT INTO chat_meta (user_id, conversation_id, usage) VALUES (?, ?, ?)"
        " ON CONFLICT (user_id, conversation_id) DO UPDATE SET usage = excluded.usage",
        (user_id, result.conversation_id, UsageAdapter.dump_json(result.usage).decode()),
    )
    db.commit()

def load_conversation(db, user_id: str, conversation_id: str) -> Conversation | None:
    meta = db.execute(
        "SELECT usage FROM chat_meta WHERE user_id = ? AND conversation_id = ?",
        (user_id, conversation_id),
    ).fetchone()
    if meta is None:
        return None
    rows = db.execute(
        "SELECT data FROM chat_messages WHERE user_id = ? AND conversation_id = ? ORDER BY rowid",
        (user_id, conversation_id),
    ).fetchall()
    messages = [m for (data,) in rows for m in ModelMessagesTypeAdapter.validate_json(data)]
    return Conversation(
        messages=messages,
        usage=UsageAdapter.validate_json(meta[0]),
        conversation_id=conversation_id,
    )

In my run each appended row held 2 messages (one request, one response), and the third turn saw all three prompts with usage.requests == 3. Loading the same thread for another user returned None.

This version doesn't store deferred_tool_requests. If your agent pauses for approvals or external tool results, store result.conversation.deferred_tool_requests beside the usage too; the docs say a paused conversation can't be resumed without it. I didn't run deferred tools.

A core ConversationStore (2.65.0 and later; listed on the Persistence page) does this append-only recipe for you. I have not run ConversationStore.

Check that it works

1. Send two messages on one thread, restart the server, then ask about the first message. The answer should use it.
2. Same thread ID, different user: the history must not cross over.
3. Print result.usage.requests across turns. With conversation= it should keep counting up; if it's back to 1 every turn, something is passing message_history instead.
4. ConversationTypeAdapter.validate_json(stored) should succeed on a row written by an older version. Per the docs, no migration is needed when Pydantic AI adds a message part.

Skip the plumbing with MemorySync

Disclosure: I build MemorySync.

A stored conversation brings back one thread. It doesn't tell a new thread what the user said last week. The Persistence page draws the same line: remembering a user across conversations is a different mechanism with a different key. Pydantic AI Harness has a Memory capability for that, and so does MemorySync.

The MemorySync Pydantic AI guide adds it as one capability. You keep the Conversation store from Steps 1 to 5 for the thread.

From the guide:

Unverified (not run):

pip install pydantic-ai-memorysync

from pydantic_ai import Agent
from pydantic_ai_memorysync import MemorySyncCapability

agent = Agent(
    "openai:your-model",
    instructions="You are a helpful assistant.",
    capabilities=[MemorySyncCapability(user_id="customer-7")],
)
result = await agent.run("What should I cook tonight?")

Per the guide, recall goes through the instructions channel rather than into the message list, so replaying a stored conversation doesn't stack old memory blocks into it. After each successful run, the user's prompt is sent for fact extraction; the model's reply isn't stored. It needs MEMORYSYNC_API_KEY, pydantic-ai (or -slim) 2.0+ and Python 3.10+. Use the same user_id you key your conversations with. Snippet from the MemorySync Pydantic AI guide. I only installed pydantic-ai-memorysync 1.1.0 and constructed the agent with MemorySyncCapability offline; I haven't run it against the API for this page.

Try it: MemorySync is optional here: the steps above work without it. If you want the memory layer from the snippet, see MemorySync, create an API key in the dashboard and drop the snippet into your agent.

Troubleshooting

• ImportError: cannot import name 'ConversationTypeAdapter' from 'pydantic_ai' (or run_sync() got an unexpected keyword argument 'conversation'). You're on pydantic-ai 2.54 or earlier; I saw both on 2.54.0. Upgrade to 2.55+, or use the older route below.
• UserError: `conversation` already carries `message_history`, so passing both is ambiguous. Pass conversation= on its own. The same error names conversation_id if you pass both of those.
• Usage limits never trigger across a chat. You're continuing with message_history=, which starts usage from zero each run (Step 1).
• A stored history was cut off mid tool call. The docs say Pydantic AI repairs history before each model request, and for the one case it raises on, repair_messages() closes out the dangling calls. I ran the docs example; it added a ToolReturnPart saying "The tool call was interrupted before a result was produced."
• Users see each other's history. The store key doesn't include the user, or the thread ID comes from the client without an ownership check (Step 2).

If you can't upgrade yet, this is the pre-2.55 route. I ran it on 2.54.0 and 2.64.0:

from pydantic_ai import ModelMessagesTypeAdapter

stored = ModelMessagesTypeAdapter.dump_json(result.all_messages())
history = ModelMessagesTypeAdapter.validate_json(stored)
result = agent.run_sync("What's my name?", message_history=history)

Common questions

Should I store the Conversation or an AgentRunResult? The Conversation, if you want to continue the chat. The docs also show storing an AgentRunResult as a field on a Pydantic model when you want the run's output kept with it.

Do I need Harness StepPersistence for this? Not for a chat app. The Persistence page says "a jsonb column is enough" to store a chat thread. StepPersistence is for continuing or forking runs from snapshots, and per the docs its snapshots hold messages only, not the usage and deferred tool requests.

Can the provider hold the history instead? On the OpenAI Responses API, yes (openai_conversation_id). The docs list the trade-offs: it's one provider's feature, earlier input tokens are still billed, and it isn't available with Zero Data Retention.

Will the stored conversation grow forever? Yes, unless you cap it. Step 5 keeps writes small; what reaches the model is a separate choice (Pydantic AI has compaction capabilities). I didn't test compaction for this page.

Related pages

• Add long-term memory to a Pydantic AI agent
• OpenAI Agents SDK RedisSession on Redis Cluster: fixing 'ClusterPipeline' object has no attribute 'raise_first_error'
• Add long-term memory to a Vercel AI SDK chatbot

Sources

• github.com/pydantic/pydantic-ai/issues/530 (opened 2024-12-23 09:20 IST, closed as completed 2026-10-05 03:06 IST, 30 comments, labels feature / community snippets / message-history; the original ask and the closing comment; read via the GitHub API 2026-10-31, re-checked 2026-11-15)
• github.com/pydantic/pydantic-ai/issues/9786 (still open, 5 comments, last update 2026-10-27 04:01 IST; re-checked 2026-11-15)
• Pydantic AI docs: Persistence and Messages and Chat History pages (fetched 2026-10-31; Persistence re-fetched 2026-11-15)
• PyPI: pydantic-ai-slim 2.64.0 (the run / then-latest, 2026-10-31 05:52 IST); pydantic-ai-slim 2.68.0 (latest as of 2026-11-15, published 2026-11-12 08:00 IST, source read / PyPI); 2.54.0 and 2.55.0 wheels inspected for pydantic_ai/conversation.py; 2.64.0 / 2.65.0 / 2.68.0 wheels inspected 2026-11-15 for conversation_store.py and Conversation fields (source read, not a run); pydantic-ai-memorysync 1.1.0 (installed 2026-10-31)
• Offline runs (2026-10-31): Steps 1 to 5 with FunctionModel, SQLite file store, fakeredis, append-only store, both UserErrors, the 2.54.0 ImportError, the message_history fallback, the docs' repair_messages example; no network, no model, no keys
• MemorySync Pydantic AI guide: docs.memorysync.io/guides/pydantic-ai (fetched 2026-10-31)

Try it

For an overview of MemorySync, see the MemorySync homepage.

Next step: read the MemorySync Pydantic AI guide, then see MemorySync. Founding offer: use code FOUNDING30 for 30% off Starter for your first 3 months (Settings → Billing).