Persistence

Store contracts, their value types, and the shipped implementations. See Storage for owner scoping, thread keying, and the reference models.

Conversations

ConversationStore

ConversationStore

Bases: Protocol

Pluggable server-side persistence for AG-UI conversations.

Handed to a transport. The package ships NullConversationStore (the server stays stateless) and a session-backed implementation; projects supply their own. All methods are async so an implementation can use the async ORM or a network backend.

Threads key by (owner_id, thread_id), so two endpoints sharing a store share one user's thread list. Wrap with ScopedConversationStore to partition them.

list returns owner-scoped metadata only, no message bodies, capped at limit rows (None for the store's own default); a store that cannot enumerate returns an empty list. exists is a presence check that loads no message body, so a rename or probe does not deserialize a whole thread just to 404. rename sets a display title, and is a no-op in a store that cannot persist one.

Conversation

Conversation dataclass

A persisted conversation, keyed by thread_id.

messages are JSON-serialisable records whose shape the calling transport owns: this substrate persists and returns them verbatim and never interprets them, which is what keeps the storage contract neutral. The AG-UI transport stores its own wire Message shape, so client message ids survive a round trip untouched; another transport stores its own.

owner_id scopes the conversation to a user for authorization.

ConversationMeta

ConversationMeta dataclass

Lightweight metadata for one conversation: the thread-drawer row shape.

Returned by ConversationStore.list, and carrying no message bodies, which is what keeps a thread list cheap. title defaults to a truncation of the first user message unless a store records a rename, preview is a one-line excerpt of the latest message, and updated_at is None in a store that does not track it. owner_id scopes the conversation to a user and is not surfaced on the wire.

NullConversationStore

NullConversationStore

The default store: no-op, keeping the server stateless.

load returns None and save / delete do nothing, so the conversation lives entirely in the client's posted history. A transport treats this store as "persistence off".

DjangoSessionConversationStore

DjangoSessionConversationStore

Conversation persistence in the Django session, needing no migration.

Conversations are namespaced by thread_id inside the user's own session, so owner scoping is implicit and durability lasts as long as that browser session. For cross-device or audited persistence, use a model-backed store.

ModelConversationStore

ModelConversationStore

Bases: ABC

Abstract base for a model-backed (or any sync) ConversationStore.

Provides the async wrapping and per-request owner scoping; a subclass implements the three synchronous row operations against its own Django model. Model-agnostic on purpose — the package ships no concrete model, so it forces no migration and consumers define the fields and the owner relation.

allow_anonymous governs whether anonymous requests are served. False refuses them rather than collapsing every anonymous visitor into one shared owner bucket where they could read and delete each other's data. It is a store policy, so two endpoints sharing a store necessarily agree on it. Pass it explicitly: this substrate reads no Django settings.

Example:

class MyStore(ModelConversationStore):
    def _fetch(self, thread_id, owner_id):
        row = MyConversation.objects.filter(
            thread_id=thread_id, owner_id=owner_id,
        ).first()
        return None if row is None else Conversation(...)
    def _store(self, conversation, owner_id): ...
    def _remove(self, thread_id, owner_id): ...

ScopedConversationStore

ScopedConversationStore

Partition another ConversationStore by a scope name.

Stores key threads by (owner_id, thread_id). Two AG-UI endpoints sharing one store therefore share one user's thread list: a conversation started at /internal/agent appears in /public/agent's history drawer and can be resumed there — under the public agent's model, tools and guard policy.

Wrapping fixes that without a migration:

internal = AGUIServer(
    registry,
    conversation_store=ScopedConversationStore(store, scope="internal"),
)
public = AGUIServer(
    registry,
    conversation_store=ScopedConversationStore(store, scope="public"),
)

The partition is a thread-id prefix, so this composes with any implementation, third-party ones included, where a scope column would mean a migration and a breaking change to the ConversationStore protocol every custom store implements.

Opt in explicitly. A transport does not wrap by itself: doing so from its namespace would silently orphan the whole thread history of an existing single-endpoint project the moment it set one.

The scope is invisible on the wire. Thread ids are echoed back to the client unchanged; only the storage key carries the prefix.

: is reserved and a scope containing one is refused. The prefix is the whole partition, so an ambiguous prefix is an ambiguous partition: with scopes admin and admin:readonly, the readonly mount's thread keys to admin:readonly:t1, which the admin mount's own prefix filter matches. It would list that thread as readonly:t1, and load, rename and delete would all resolve there — silently, in both directions, since thread ids come from the client. Refused at construction rather than escaped at the key, because escaping would rewrite the storage key of every thread already saved. Scopes that merely share a prefix (admin / administrators) are unaffected: the separator ends the scope.

list async

list(*, request: HttpRequest, limit: int | None = None) -> ConversationMetaList

This scope's threads only, with storage keys translated back.

limit is applied by the inner store before this filter, so a busy sibling scope can crowd out rows. A store needing exact per-scope paging should partition at the query rather than by wrapping.

Attachments

AttachmentStore

AttachmentStore

Bases: Protocol

Pluggable server-side storage for files a user attaches to a conversation.

Handed to a transport. The package ships NullAttachmentStore (uploads off) and the abstract ModelAttachmentStore; the opt-in django_pydantic_agent.contrib.store app adds a ready DefaultAttachmentStore keeping bytes in Django Storage and metadata in a row.

Every method is async and owner-scoped: a store filters by the acting user so one user can never read or delete another's files, the security boundary for the whole feature. save validates nothing about size or type — the view does that from its own config — and just persists the bytes, returning a durable AttachmentRef. open returns None for a missing or cross-owner id rather than raising, so a caller maps both to a 404 and the two stay indistinguishable.

Attachments need no scoped wrapper of the kind conversations have: they are id-referenced with no enumeration and already owner-scoped, so two endpoints sharing a store expose nothing across the user boundary. Thread lists are the case that leaks.

AttachmentRef

AttachmentRef dataclass

A durable, lightweight reference to one uploaded file.

What an upload returns and what travels on the wire — never the bytes. The file is uploaded out of band, the client holds this ref on the message, and the agent reads the bytes server-side through the read_attachment tool.

id is the opaque, owner-scoped handle the store resolves back to bytes. mime is client-declared, so treat it as a hint. url is an optional direct fetch URL, such as an owner-checked download endpoint, and stays None unless a store fills it in.

OpenedAttachment

OpenedAttachment dataclass

An attachment's metadata paired with a readable byte stream.

Returned by AttachmentStore.open, so a download view and the read_attachment tool both get the content and the AttachmentRef in one owner-scoped call.

content is an open binary stream rather than the bytes, so a large attachment streams out instead of being buffered. The consumer owns it and must read it exactly once — hand it to FileResponse, which closes it, or read it under a with block.

NullAttachmentStore

NullAttachmentStore

The default attachment store: uploads disabled, server stays stateless.

A transport's attachments view detects this store and answers 410 Gone, so a misconfigured client gets a clear "uploads are off" signal rather than a silent success, and save is never reached. Called directly it raises, rather than fabricating a ref. open returns None so every fetch is a 404, and delete is a no-op: the endpoint is inert until a real store is configured.

ModelAttachmentStore

ModelAttachmentStore

Bases: ABC

Abstract base for a model-backed (or any sync) AttachmentStore.

The attachment twin of ModelConversationStore — same async wrapping, same per-request owner scoping, same allow_anonymous policy — over a subclass's own storage: a Django Storage for the bytes, a model row for the metadata. The opt-in django_pydantic_agent.contrib.store app supplies a concrete pair.

Each _save / _open / _remove receives the resolved owner_id (None for anonymous) and must filter by it, so files never cross users.

Example:

class MyStore(ModelAttachmentStore):
    def _save(self, upload, owner_id):
        row = MyAttachment.objects.create(owner_id=owner_id or "", ...)
        row.file.save(row.attachment_id, upload, save=True)
        return AttachmentRef(id=row.attachment_id, name=..., mime=..., size=...)
    def _open(self, attachment_id, owner_id): ...
    def _remove(self, attachment_id, owner_id): ...

Memory

memory_namespace

memory_namespace

memory_namespace(request: HttpRequest) -> str

The per-user namespace for pydantic_ai_harness.memory.Memory, from a request.

For a host that builds the capability per request and therefore holds one::

Memory(store, namespace=lambda ctx: memory_namespace(request))

Most transports do not: AGUIServer(capabilities=...) resolves its list once at mount time, where no request exists. Reach for memory_namespace_for_user there. The one thing this resolver can do that the other cannot is key an anonymous caller to their browser session, so anonymous visitors get separate namespaces instead of sharing one.

Not resolve_owner_id, and the difference is load-bearing. That helper returns anon:<session_key> for an anonymous request, and a colon is not in the alphabet the harness accepts for a path segment -- so Memory raises ValueError: invalid memory path for every anonymous visitor. That raise happens in the capability's for_run, which is outside the store read that injection_errors guards, so the harness's own "ignore" default does not catch it and the whole run aborts. Hence a separate resolver whose only contract is that its result is always a valid segment.

An identifier that is already segment-safe is used as-is behind a prefix; one that is not is replaced by a digest of it rather than sanitised, because stripping the offending characters maps tenant/42 and tenant-42 onto one namespace.

memory_namespace_for_user

memory_namespace_for_user

memory_namespace_for_user(user: Any) -> str

The namespace for pydantic_ai_harness.memory.Memory, from the acting user.

The resolver to use when the capability is constructed at mount time, which is where every transport takes it — AGUIServer(capabilities=...) resolves its list once, and no request exists yet::

Memory(store, namespace=lambda ctx: memory_namespace_for_user(ctx.deps.user))

Reading the user off ctx.deps rather than closing over a request is what lets one agent, built once, serve every caller: pydantic-ai clones each capability per run and Memory re-resolves its scope in the clone. AgentDeps.user is set by the transport from the authenticated request, so it is server-resolved and not something a client can choose.

Use memory_namespace instead when the store really is built per request and you hold one: it can key an anonymous caller to their browser session, which this cannot. With no request there is no session, so every unauthenticated caller shares one namespace — a real limitation rather than an oversight, and rarely reached, since a transport that serves anonymous callers at all has opted into it deliberately.

The result is always a valid harness path segment: a segment-safe primary key is carried through readably behind a u- prefix, and anything else — an email-address primary key, a natural key with a slash, a pathological 200-character pk that no longer fits once prefixed — is replaced by a digest.

Errors

AnonymousOperationError

AnonymousOperationError

Bases: Exception

Raised when a model-backed store is asked to act for an anonymous request.

The reference stores refuse anonymous operations unless constructed with allow_anonymous=True, since otherwise every anonymous visitor would share one owner bucket and could read or delete the others' data. A transport's persistence views catch this and return 403.

Reference implementations

The opt-in django_pydantic_agent.contrib.store app. Add it to INSTALLED_APPS and run migrate; the base package ships no model, so projects that don't opt in get no migration.

DefaultConversationStore

Bases: ModelConversationStore

A ready-to-use model-backed store over StoredConversation.

Cross-device, per-user history with a cheap thread list. Add "django_pydantic_agent.contrib.store" to INSTALLED_APPS, run migrate, and pass an instance to your transport's conversation_store=. For a bespoke schema, subclass ModelConversationStore instead.

Every query filters by the owner_id the base resolves. A title is derived from the first user message at first save and then left alone except by a rename; the preview re-derives on every save.

Saving also reconciles which attachments the thread refers to, and deleting it drops the ones nothing else refers to, so an attachment's lifetime is tied to the conversations quoting it. An upload that was never sent belongs to no conversation, and the agent_store_prune_attachments command collects those instead.

DefaultAttachmentStore

Bases: ModelAttachmentStore

A ready-to-use model-backed store over StoredAttachment.

Bytes live in Django Storage (filesystem by default, S3 or GCS through STORAGES), metadata in a row. Add "django_pydantic_agent.contrib.store" to INSTALLED_APPS, run migrate, and pass an instance to your transport's attachment_store=. For a bespoke schema, subclass ModelAttachmentStore instead.

Every query filters by the owner_id the base resolves, so one user's id never reaches another's file, and the public attachment_id is an opaque UUID kept separate from the storage filename.

Uploads are deduplicated by content hash, within one owner: the same file sent into five threads is written to storage once and pointed at five times, while still getting a row of its own each time so it keeps the name the composer showed. The blob goes when the last row pointing at it does.

DefaultStepStore

A durable, owner-scoped StepStore over the reference models.

The database equivalent of the harness's own SqliteStepStore / FileStepStore. It structurally satisfies pydantic-ai-harness's StepStore protocol — that protocol is upstream's, not this package's — while partitioning every row by the resolved owner, so one user can never read or resume another's runs even by guessing a run_id. The owner is entirely ours; no harness record carries one.

Built per request, unlike a ConversationStore singleton whose methods each take a request: the protocol's methods carry none, so the request is bound at construction. Owner resolution runs inside each sync_to_async hop, because it may create a session row for the anonymous bucket and must stay off the event loop.

An anonymous request degrades rather than crashing. With no owner and allow_anonymous off, every write no-ops and every read returns empty: the capability's hooks fire mid-run, so refusing by raising would abort the run, and an anonymous visitor has no durable identity to resume under anyway. Pair the store with an authenticated mount for it to persist.

Add "django_pydantic_agent.contrib.store" to INSTALLED_APPS and run migrate for the backing tables. Requires the django-pydantic-agent[harness] extra.

list_runs async

list_runs(
    *, parent_run_id: str | None = None, conversation_id: str | None = None
) -> list[RunRecord]

The owner's runs by started_at, oldest first — a contract, not a default.

StepStore.list_runs documents ascending order and tells callers they may take the most recent run with [-1], so answering newest-first would quietly hand upstream's own idiom the wrong run. Presenting recent runs first is a reading order, and belongs where the list is rendered.

DefaultMemoryStore

A durable, owner-scoped MemoryStore over the reference models.

The database equivalent of the harness's own SqliteMemoryStore / FileStore. It structurally satisfies pydantic-ai-harness's MemoryStore protocol — that protocol is upstream's, not this package's — while partitioning every row by the resolved owner, so one user's memory can never be read or overwritten by another even when the namespace handed to the capability is wrong. Attach it with::

Memory(DefaultMemoryStore(request), namespace=lambda ctx: memory_namespace(request))

Built per request, like DefaultStepStore and for the same reason: the protocol's methods carry no request, so it is bound at construction. Owner resolution runs inside each sync_to_async hop, because it may create a session row for the anonymous bucket and must stay off the event loop.

The path's namespace is not trusted as the boundary. The harness composes the key as <namespace>/<agent_name>/<file>.md from a resolver the host supplies, and a / inside a resolved namespace is accepted — it simply opens further path segments. So a resolver reading anything user-controlled could otherwise address another scope. Filtering every query on the server-resolved owner_id makes that harmless, and memory_namespace keeps the namespace valid in the first place.

Stored content cannot close the fence it is injected inside. The harness wraps injected memory in <memory> markers and is explicit in its own README that this "is not a hard prompt-injection boundary"; without help, a note containing the closing tag ends the block early and everything after it reads as the user's own turn, durably, on every future run. Every write here escapes the angle brackets of both tags, so the stored bytes are safe for every consumer — the injection, read_memory, search_memory and an app-side read alike. Doing it on write rather than read is also what keeps write_memory's old_text replacement working: the model edits against the same escaped text it was shown.

A per-owner ceiling, which the capability does not have. Its max_memory_size bounds one file and its injection budget bounds what is read; nothing upstream caps how many files a namespace accumulates or how large they grow in total. max_files and max_total_chars do, at the write.

An anonymous request degrades rather than crashing. With no owner and allow_anonymous off, every write no-ops and every read returns empty: the capability's hooks fire mid-run, so refusing by raising would abort the run, and an anonymous visitor has no durable identity worth remembering under. Note the asymmetry with the other stores — the memory tools are model-facing, so a no-op write still reports success to the model. Pair the store with an authenticated mount for memory to persist.

Add "django_pydantic_agent.contrib.store" to INSTALLED_APPS and run migrate for the backing tables. Requires the django-pydantic-agent[harness] extra.

purge staticmethod

purge(owner_id: str) -> int

Delete every memory row for one owner, returning how many files went.

Not part of MemoryStore, and it has to be here because composing the protocol cannot express it: list_paths then delete needs the current version of each path, so a purge becomes an unbounded read-then-delete loop that a concurrent write_memory can lose to. Memory is durable personal data written about a user, so an erasure request has to be able to reach it in one statement.

Deliberately not wired to a post_delete signal on the user model: whether deleting an account erases its memory is a product policy, and a library's job here is to make the operation possible.

Synchronous — call it from a management command, an admin action, or an async caller through sync_to_async.

Attachment lifecycle

Contrib-level, not part of either protocol. See Storage for the semantics and the management commands.

reconcile_conversation_attachments

reconcile_conversation_attachments(
    conversation: StoredConversation, messages: Iterable[Any]
) -> None

Point conversation at exactly the attachments its messages quote.

The resolver behind attachment lifecycle, needing no wire or client change: the web component already augments each user message with an attachments array of the refs the composer uploaded, and that undeclared field survives into the stored message list. Reading it here turns an id in some JSON into a relation the database can enforce a lifecycle on.

Reconciled rather than appended, so a message dropped on a re-save drops its reference too and the relation always describes the conversation as it stands.

Owner-scoped, which is the load-bearing part: an id resolves only against the attachments of the conversation's own owner, so a guessed or copied id links nothing — the rule that makes AttachmentStore.open return None across owners.

The parse is total. A message that is not a mapping, an attachments field of the wrong type, an entry with no usable id, an id resolving to an attachment since deleted: all degrade to "no reference" in silence. This runs inside a conversation save, where an exception would lose the user's message.

strip_inline_binary_parts

strip_inline_binary_parts(messages: Any) -> tuple[Any, bool]

Rebuild messages without any inline file bytes, and say if it changed.

A transport that inlines an attachment for the model serialises the file into the message list as base64, which persisted turns a 2.6 MB PDF into roughly 3.5 MB of text in one row, shipped to the browser on every load.

The edit is structural: the list is walked as plain JSON and only the offending parts are dropped. Nothing is round-tripped through a message type, because validating and re-dumping is what would silently discard the fields this must not lose — every message's id, and the non-standard attachments array that both renders the chips and drives attachment reconciliation.

A message whose parts were all bytes keeps its place with an empty content list, since dropping it would take its id too.

Returns:

Name	Type	Description
	Any	The rebuilt messages and whether anything was removed, so a caller can
	bool	leave untouched rows untouched. A non-list messages is handed back
unchanged	tuple[Any, bool]	the column is JSON, and a row written by something other than
	tuple[Any, bool]	this store is left alone.

AttachmentDeletion dataclass

What one attachment deletion pass removed.

Three counts rather than one, because deduplication pulls them apart: deleting one of two rows sharing a file removes a row and no blob, so a command reporting only rows would claim space it did not reclaim.

bytes_freed is summed from the declared size of the rows whose blob went, not from a fresh stat of the backend, so it estimates what a remote store gives back rather than measuring it.