pydantic-ai-persistence

Backend-agnostic persistence for PydanticAI: chat history and durable step persistence, over one small async KV interface. Backends (DynamoDB, CosmosDB, Firestore) implement AsyncKV and get both stores.

PydanticAI has two persistence layers:

• History — you serialize result.all_messages() and store it yourself. KVHistoryStore does exactly that.
• Step persistence — PydanticAI's StepStore (append-only events, continuable snapshots, tool-effect ledger) for durable/resumable runs. KVStepStore implements the async StepStore protocol.

Install

pip install pydantic-ai-persistence            # core (in-memory backend)
pip install "pydantic-ai-persistence[dynamodb]"  # + DynamoDB backend

Core pieces

History

from pydantic_ai import Agent
from pydantic_ai_persistence import KVHistoryStore, InMemoryAsyncKV

store = KVHistoryStore(InMemoryAsyncKV())

result = agent.run_sync("Hi, I'm Kamal")
await store.save("conv-1", result.all_messages())

prior = await store.load("conv-1")
result = agent.run_sync("What's my name?", message_history=prior)

Step persistence

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import StepPersistence
from pydantic_ai_persistence import KVStepStore, InMemoryAsyncKV

step_store = KVStepStore(InMemoryAsyncKV(), max_snapshots_per_run=10)
agent = Agent("openai:gpt-4o", capabilities=[StepPersistence(store=step_store)])

KVStepStore records runs, append-only events, continuable snapshots (with pruning), and the tool-effect ledger — everything the StepStore protocol requires.

Build your own backend

Implement AsyncKV (four async methods) and hand it to KVStepStore / KVHistoryStore. See pydantic-ai-dynamodb-persistence for an example.

Note: StepStore is a beta/experimental PydanticAI harness feature; its API may still change.

License

MIT