API reference¶

Autodoc of the public surface re-exported from django_ag_ui. Everything below is importable directly, e.g. from django_ag_ui import ToolRegistry.

Registry¶

ToolRegistry ¶

An ordered, named collection of server-side tools.

State lives on the instance — a transport holds one, tests build a fresh one per scenario. Each tool's JSON Schema is derived once at registration, and either sync or async callables can be dispatched.

register ¶

register(spec: ToolSpec) -> ToolBinding

Register spec and return its binding.

Raises:

Type	Description
`ValueError`	when `spec.name` is already registered.

get ¶

get(name: str) -> ToolBinding

Return the binding for name or raise KeyError.

call ¶

call(name: str, arguments: dict[str, Any], *, ctx: Any = None) -> Any

Dispatch a sync call to the registered tool.

arguments is the model's payload and nothing else. A tool declaring a leading ctx: RunContext[...] gets it from ctx here, the way pydantic-ai would supply it from the run; passing one to a tool that takes none is harmless.

Refuses coroutine functions to avoid silently returning an un-awaited coroutine. Use acall for async tools.

Raises:

Type	Description
`TypeError`	when the tool is async, or declares a `RunContext` parameter and no `ctx` was given.

acall `async` ¶

acall(name: str, arguments: dict[str, Any], *, ctx: Any = None) -> Any

Dispatch an async call; transparently awaits sync callables.

ctx is bound exactly as in call.

tool ¶

tool(
    registry: ToolRegistry,
    *,
    name: str | None = None,
    description: str | None = None,
    destructive: bool = False,
    category: ToolCategory = ToolCategory.OTHER,
    confirm: str | None = None,
    summary: str | None = None,
) -> Callable[[F], F]

Register the decorated callable on registry as a tool.

name defaults to the function's name and description to the first paragraph of its docstring. confirm is the confirmation prompt for a destructive tool and summary a short display label; both reach the client as the matching x-* schema key.

build_input_schema ¶

build_input_schema(
    fn: Callable[..., Any],
    *,
    destructive: bool = False,
    category: ToolCategory = ToolCategory.OTHER,
    confirm: str | None = None,
    summary: str | None = None,
) -> dict[str, Any]

Derive a JSON Schema object from fn's parameters.

Covers str, int, float, bool, list[T], dict[str, Any] and X | None unions. Anything richer falls back to an empty fragment, imposing no type constraint but staying wire-valid.

destructive / category / confirm / summary are stamped at the schema root as the matching x-* extension keys, which AG-UI passes through verbatim to the client.

A leading ctx: RunContext[...] parameter is not an argument and is left out. Pydantic-AI fills it from the run — it is how a tool reaches the acting user — so advertising it would ask the model to invent a value for something it cannot supply.

ToolSpec `dataclass` ¶

Canonical declaration of a server-side tool.

Bundles the callable with the metadata the registry needs to expose it to a Pydantic-AI agent and to a frontend.

name `instance-attribute` ¶

name: str

Stable identifier exposed to the agent, unique within a ToolRegistry.

fn `instance-attribute` ¶

fn: Callable[..., Any]

The callable implementing the tool. Its parameters must be typed; the registry derives a JSON Schema from the signature.

description `instance-attribute` ¶

description: str

Summary shown to the agent. Most clients display the first line.

destructive `class-attribute` `instance-attribute` ¶

destructive: bool = False

Whether calling this tool may mutate state. Stamped as x-destructive so a frontend can gate it behind a confirmation step.

category `class-attribute` `instance-attribute` ¶

category: ToolCategory = OTHER

Coarse capability grouping, stamped as x-category.

confirm `class-attribute` `instance-attribute` ¶

confirm: str | None = None

A confirmation prompt for a destructive tool, stamped as x-confirm, shown instead of a generic "Run ?".

summary `class-attribute` `instance-attribute` ¶

summary: str | None = None

A short label, stamped as x-summary, shown on the tool-call card instead of the raw tool name.

ToolBinding `dataclass` ¶

A registered tool plus the JSON Schema derived from its signature.

The schema is computed once at registration and carried alongside the spec, so a tool listing does not re-introspect on every request.

ToolCategory ¶

Bases: str, Enum

Coarse grouping for a tool, surfaced to the agent and the UI.

Advisory metadata: it lets a frontend group tools, a system prompt reason about capability classes, and a project apply category-wide policy. It does not gate execution; that is the destructive flag's job.

X_DESTRUCTIVE_KEY `module-attribute` ¶

X_DESTRUCTIVE_KEY = 'x-destructive'

X_CATEGORY_KEY `module-attribute` ¶

X_CATEGORY_KEY = 'x-category'

X_CONFIRM_KEY `module-attribute` ¶

X_CONFIRM_KEY = 'x-confirm'

X_SUMMARY_KEY `module-attribute` ¶

X_SUMMARY_KEY = 'x-summary'

Skills¶

SkillRegistry ¶

An ordered, named collection of SkillSpecs.

State lives on the instance (like ToolRegistry). payload produces the JSON-serialisable catalog the frontend consumes (camelCase keys, optional fields omitted when default).

register ¶

register(spec: SkillSpec) -> SkillSpec

Register spec; raise ValueError if the name is taken.

add ¶

add(
    name: str,
    title: str,
    prompt: str | None = None,
    *,
    description: str | None = None,
    send_immediately: bool = False,
    chip: bool = False,
) -> SkillSpec

Construct a SkillSpec and register it (convenience).

payload ¶

payload() -> list[dict[str, Any]]

The client catalog: a list of skill dicts with camelCase keys.

SkillSpec `dataclass` ¶

A pre-defined action offered to the user (a "skill").

Serialised into the client catalog the frontend surfaces as chips and/or the /-command palette.

A skill need not ship its prompt to the browser. With prompt unset the catalog advertises only the name and label, and picking the skill sends the bare /name token for the agent to resolve — from the harness Skills capability or its own instructions — so the wording of an internal workflow never leaves the server. Setting prompt hands the text to the client, which fills any {placeholder}s from its skill context before sending: right for a prompt that is genuinely a user-facing convenience, and for placeholders only the page can fill.

name `instance-attribute` ¶

name: str

Stable id; the /token in the palette. Unique within a registry.

title `instance-attribute` ¶

title: str

Label shown in chips and the palette.

prompt `class-attribute` `instance-attribute` ¶

prompt: str | None = None

Prompt text handed to the client, or None to keep it server-side and have the client send /name instead. May contain {placeholder}s the client fills before sending.

description `class-attribute` `instance-attribute` ¶

description: str | None = None

Optional secondary line shown in the palette.

send_immediately `class-attribute` `instance-attribute` ¶

send_immediately: bool = False

Send on pick instead of pre-filling the input. Surfaced as sendImmediately.

chip `class-attribute` `instance-attribute` ¶

chip: bool = False

Also surface as a chip (the palette shows all skills regardless).

SkillsView ¶

A read-only endpoint returning a SkillRegistry's client catalog.

A callable instance (like DjangoAGUIView) so it can hold the registry and a project can mount several. GET returns the JSON skill list the frontend fetches via data-skills-url.

Skill prompts can encode internal workflows — worth gating — so the view carries the same authentication seam as DjangoAGUIView (require_authenticated / get_user, sync or async) and the same closed default: an anonymous GET is a 401 until you pass require_authenticated=False.

Agent and view¶

AGUIServer ¶

One config object that mounts an AG-UI endpoint and its sub-views.

The Django-idiomatic front door for the package — the admin.site idiom and the mirror of drf-mcp's MCPServer. Construct it once with the tool registry (plus optional stores / auth), then mount its namespaced urls with path():

from django_ag_ui import AGUIServer

agent = AGUIServer(registry, csrf_exempt=False)

urlpatterns = [
    path("agent/", agent.urls),
]
# reverse("ag_ui:endpoint") · "ag_ui:tools" · "ag_ui:threads" · ...

The registry is passed once: the object builds the agent view (DjangoAGUIView) and the read-only tool catalog (ToolsView) from it. The mount point is the consumer's to choose the Django way, so there is no prefix=.

What gets mounted. The agent endpoint (endpoint) and its tool catalog (tools) always mount. The rest mount only when their backend is active:

skills — a SkillRegistry was passed (skills/, GET JSON for data-skills-url).
threads / thread — the conversation store is not a NullConversationStore (threads/ + threads/<id>/, the history drawer's data-threads-url). A thread_activity_source= puts pushed activities back into what that route serves; without one, a restored thread is the model's message history and nothing else, which is where a pushed chart is missing from. See ThreadActivitySource.
attachments / attachment — the attachment store is not a NullAttachmentStore (attachments/ + attachments/<id>/, the composer's data-attachments-url).
transcribe — the transcription backend is not a NullTranscriptionBackend (the mic's data-transcribe-url).
resume / fork / runs — a step_store is configured.

Collaborators are passed here or absent: there is no settings fallback, and the keys that once held a dotted path are refused at startup by check_removed_settings rather than ignored. Unpassed, each falls back to its Null* backend, so a bare AGUIServer(registry) serves the agent endpoint and its tool catalog and nothing else.

A collaborator handed a dotted path is refused here too (check_no_dotted_paths): there is no import_string to resolve it, and a string otherwise constructs, mounts that collaborator's endpoints, and fails on the first request instead.

Request policy, closed by default. require_authenticated / get_user / authorize / csrf_exempt are forwarded to every view this object builds, so one policy governs the whole mount — including the write endpoints (attachment upload / delete, thread rename / delete, transcribe), which csrf_exempt=True exempts alongside the run endpoint. require_authenticated defaults to True, so a bare AGUIServer(registry) serves nobody who is not logged in; pass require_authenticated=False to serve anonymous runs deliberately. The agent view's model and instructions fall back to the DJANGO_AG_UI settings when not passed.

Anonymous scoping caveat. With require_authenticated=False and a model-backed store, an anonymous request has no owner id. The reference contrib stores refuse anonymous thread / attachment operations unless built with allow_anonymous=True (which buckets per browser session), so leave the default in place — or pass a get_user hook — whenever the store persists, rather than relying on owner scoping to isolate anonymous visitors from one another.

Spec tools. service_specs takes a name -> spec mapping, a spec registry (drf-services' SpecRegistry, the single declaration site for a project exposing the same specs over several transports), or an already-built SpecToolset / SpecCapability:

AGUIServer(registry, service_specs=spec_registry.by_tag("public"))
AGUIServer(registry, service_specs=SpecToolset(SPECS, max_page_size=50))

Prefer the registry over registry.specs(). An entry carries more than its spec -- its tags, and the OfflineContract saying what a caller with no HTTP request has to be told (the URL kwargs, query params and field-audience overrides an HTTP caller gets from the URLconf and query string for free). The flattened mapping has none of it, and losing it is silent: the tools are all there and merely missing declarations nobody asked for.

A filtered registry view (by_tag / subset) is itself a registry, so two endpoints can be given different projections with no shared state. The pre-built form is the only way to reach a toolset knob (max_page_size, an exception_map, a build_context override, require_permissions=False while migrating) without abandoning service_specs= for capabilities=, which the tool catalog never sees; it is attached as itself and its specs are read for the catalog, so the powerful form keeps the tool-call card labels. The parameter's union covers all four shapes, the last two matched structurally because drf-pydantic-ai's own types cannot be named from a package that only optionally depends on it. Requires the django-ag-ui[spec-tools] extra.

One agent, many runs. The endpoint builds its agent once and reuses it rather than re-deriving every tool's JSON Schema per request. model_for_request(request) and instructions_for_request(request) are the two hooks that vary it — the per-tenant model and the per-tenant system prompt — and they ride the run, through pydantic-ai's own per-run model / instructions:

AGUIServer(registry, model_for_request=lambda r: r.tenant.model)

Rate limiting. throttle takes a Throttle — one consume(request) returning the suggested Retry-After in seconds, or None to allow the run — and applies to the agent endpoint only, the one that costs a model call per request. It runs after authentication, so a limiter can key on the acting user rather than only an IP:

AGUIServer(registry, throttle=FixedWindowThrottle(max_runs=20, per_seconds=60))

transcribe_throttle is the same seam on transcribe/, the other route that spends provider money per request — authentication bounds who may call it, not how often. It is a separate argument rather than a second use of throttle because one limiter instance is one counter: sharing would let voice clips eat the run budget, and the two want different numbers anyway.

AGUIServer(
    registry,
    throttle=FixedWindowThrottle(max_runs=20, per_seconds=60),
    transcribe_throttle=FixedWindowThrottle(
        max_runs=60, per_seconds=60, namespace="transcribe"
    ),
)

Per-run dependencies. deps_factory is a request -> AgentDeps callable replacing the default, which binds only the acting user and their IP. Use it to carry project-specific per-run context on an AgentDeps subclass, or to seed AgentDeps.state with a Pydantic model — the only way to have AG-UI's inbound shared state validated, since pydantic-ai validates it against type(deps.state). Whatever it returns reaches every tool, toolset and capability as ctx.deps.

Durable step persistence. step_store is a factory — a request -> StepStore callable rather than a shared store, because the pydantic-ai-harness step-store protocol carries no request. When set, every run attaches a StepPersistence capability recording an owner-scoped run / event / snapshot / tool-effect ledger, and three owner-scoped endpoints mount: resume/<run_id>/ and fork/<run_id>/ seed a new run with a prior run's last continuable snapshot, and runs/ indexes the user's runs so a client can discover what it may resume rather than only continuing a run whose id it still holds. Pass DefaultStepStore (its constructor is the factory) for the reference model-backed store, or any such callable. Requires the django-ag-ui[harness] extra.

The ledger is scoped by owner, not by endpoint, so two mounts handed the same factory share one user's run list: a run recorded at /internal/agent is listed at /public/agent/runs/ and can be resumed there, under the public agent's model, tools and guard policy. Wrap the factory in a ScopedStepStore to keep them apart, exactly as ScopedConversationStore does for thread history.

Namespacing. urls returns the (patterns, app_name, namespace) triple path() mounts directly (like admin.site.urls — no include()), so endpoint names are namespaced (namespace, default "ag_ui") and multiple mounts don't collide — reverse("ag_ui:endpoint").

urls `property` ¶

urls: tuple[list[URLPattern], str, str]

The namespaced (patterns, app_name, namespace) triple path() mounts.

Mounts directly at any prefix — path("agent/", server.urls), no include() — exactly like admin.site.urls. Every route name listed under "What gets mounted" reverses within the namespace, as reverse("<namespace>:endpoint") and so on.

DjangoAGUIView ¶

An async Django view that serves an AG-UI endpoint.

Bridges a Django HttpRequest to Pydantic-AI's AGUIAdapter without Starlette: it parses the posted RunAgentInput, builds a Pydantic-AI Agent from the server-side tool registry, and returns a StreamingHttpResponse of AG-UI events (Server-Sent Events). Frontend tools declared in the request are merged by the adapter automatically.

The view is a callable instance, so configuration lives on self and a project can mount several with independent registries. model, instructions, and audit_logger fall back to the DJANGO_AG_UI settings when not passed explicitly.

The agent is built once and reused by every run; model_for_request and instructions_for_request are the only per-request hooks on it, and everything else varies by riding the run instead. agent_factory takes over construction wholesale — see AGUIServer for what that turns off.

Authentication is the host's responsibility, and the view fails closed. Tools (and the drf-mcp bridge) act as request.user; if your middleware hasn't authenticated the request, that is AnonymousUser — a data-exposure footgun. require_authenticated therefore defaults to True: an anonymous request gets a 401 before any agent runs. Pass require_authenticated=False to serve anonymous runs deliberately, and/or a get_user(request) hook to establish the user (e.g. from a token) before tools run. get_user may be sync or async; a sync hook runs off the event loop, so a plain ORM token lookup is fully supported. A hook that raises propagates as an unhandled error (500) — return AnonymousUser (or None) for a clean 401 instead.

CSRF: the view is CSRF-exempt unless told otherwise, because AG-UI clients typically authenticate via headers (Bearer / API key), where CSRF does not apply. If your deployment authenticates with session cookies, pass csrf_exempt=False and send the CSRF token from the client — tools act as request.user, so a cookie-auth endpoint without CSRF protection lets any third-party page drive the agent as the logged-in user (mitigated, not eliminated, by Django's default SameSite=Lax cookie). Leaving csrf_exempt unset while supplying no get_user warns at construction: that pairing states nothing about how requests authenticate, and the likeliest reading is the dangerous one. Any of the three answers silences it.

AgentSession ¶

Per-run orchestration between the HTTP transport and the agent.

Owns everything one AG-UI run needs after the transport has authenticated the request, parsed the RunAgentInput, and built the agent — and before the response object exists: the AGUIAdapter (an OutcomeAGUIAdapter, the stock one plus a tool call's outcome), the composed event stream (native → transformed → reasoning-filtered → encoded → disconnect-guarded), and persistence on all three of a run's exits — completed, failed and cancelled — the last two audited as well.

Splitting it from DjangoAGUIView makes the streaming pipeline testable without a StreamingHttpResponse (drive stream directly) and keeps the SSE transport swappable — a future WebSocket transport reuses the session unchanged.

stream ¶

stream() -> AsyncIterator[str]

The encoded AG-UI event stream for this run, disconnect-guarded.

Composed by hand (rather than adapter.run_stream) so the session keeps a reference to the native event stream — the innermost generator, whose context manager owns the provider's streaming request. On client disconnect the guard closes it explicitly, then persists the partial exchange and audits the cancellation.

ToolsView ¶

A read-only endpoint returning the agent's server-tool catalog (GET, JSON).

A callable instance (like DjangoAGUIView) holding the same ToolRegistry the view uses. GET returns the build_tool_catalog list the web component fetches via data-tools-url to label tool-call cards for server-side tools (whose schema never reaches the browser).

The catalog names every server tool the agent can wield — an inventory worth gating — so the view carries the same authentication seam as DjangoAGUIView (require_authenticated / get_user, sync or async) and the same closed default: an anonymous GET is a 401 until you pass require_authenticated=False.

build_agent ¶

build_agent(registry: ToolRegistry, config: AgentConfig) -> Agent[AgentDeps, Any]

Build a Pydantic-AI Agent from a registry and an AgentConfig.

Each registry tool is registered as a plain Pydantic-AI tool, and config supplies the model, the toolsets and capabilities composed alongside them, and the policies below. Frontend tools declared in the AG-UI RunAgentInput are merged by the adapter, not registered here.

Three capabilities are added from config on request: an AuditCapability that times and records every tool the agent runs — registry tools and composed toolsets alike — a ToolGuard that flips destructive tools to require approval, and a ToolFailurePolicy, on unless turned off, so a raising tool fails its own call rather than the whole run. Each declares its position through get_ordering and pydantic-ai sorts them, so the list needs no pre-ordering.

A destructive tool is not confirmed unless a config asks for it. The approval interrupt exists only when config.tool_guard is enabled, and even then it reaches a tool only if that tool's source declares the mutation — a registry @tool(destructive=True), an MCP readOnlyHint of False, an x-destructive schema stamp, or an explicit require_approval name. With no tool_guard, every server-side tool here runs the moment the model calls it. The browser's own confirmation card is no substitute: it is driven by a client-registered tool's schema and never sees a tool that executes server-side. A transport describing this agent to the model should say which of the two it configured.

The agent is typed Agent[AgentDeps, ...], so every run must be given AgentDeps through deps=.

AgentConfig `dataclass` ¶

Resolved construction parameters for a Pydantic-AI Agent.

Bundles everything build_agent needs so the call site passes one record instead of a long keyword list. A transport resolves these from its own configuration and hands the record down; toolsets and capabilities arrive already resolved to instances (never dotted paths — this substrate resolves nothing from settings).

model `instance-attribute` ¶

model: Any

The Pydantic-AI model (a model string or Model instance).

instructions `class-attribute` `instance-attribute` ¶

instructions: str | None = None

System/instructions prompt for the agent.

audit_logger `class-attribute` `instance-attribute` ¶

audit_logger: AuditLogger | None = None

Wraps every server-side tool call for timing and success/failure records. None means no auditing.

audit_ip_address `class-attribute` `instance-attribute` ¶

audit_ip_address: str | None = None

Client IP stamped onto every audit event this agent records (the view fills it from the driving request). None leaves the field unset.

model_settings `class-attribute` `instance-attribute` ¶

model_settings: dict[str, Any] | None = None

Pydantic-AI ModelSettings (temperature, max_tokens, …).

retries `class-attribute` `instance-attribute` ¶

retries: int | None = None

Default tool/output retry budget.

toolsets `class-attribute` `instance-attribute` ¶

toolsets: Sequence[Any] | None = None

Extra Pydantic-AI toolsets composed alongside the registry tools.

capabilities `class-attribute` `instance-attribute` ¶

capabilities: Sequence[Any] | None = None

Pydantic-AI capabilities passed to the Agent.

tool_guard `class-attribute` `instance-attribute` ¶

tool_guard: ToolGuardConfig | None = None

Server-side destructive-tool approval policy. When set and enabled, build_agent composes a ToolGuard built from the registry's destructive tools; None or disabled leaves the agent ungated.

tool_failure `class-attribute` `instance-attribute` ¶

tool_failure: ToolFailureConfig = field(default_factory=ToolFailureConfig)

What an unhandled tool exception costs. On by default, so a raising tool fails its own call and the run carries on, with the exception still reaching the audit logger and the Python logger. A plain record rather than ... | None because there is no third state, and None would have to mean "on" to keep that default — backwards next to tool_guard.

AgentFactoryFn ¶

Bases: Protocol

The escape-hatch signature for a transport's agent_factory= argument.

A callable of this shape fully replaces build_agent, giving a project complete control over Agent construction. It receives the server-side tool registry and the calling transport's own resolved config record, which is untyped here because this substrate owns no settings namespace; each transport documents the concrete type its users receive.

The returned agent must be built with deps_type=AgentDeps. Transports hand every run an AgentDeps, and that is how the acting user reaches spec tools and how AG-UI state reaches deps.state. A factory that omits it produces an agent whose tools see no user.

DEFAULT_SYSTEM_PROMPT `module-attribute` ¶

DEFAULT_SYSTEM_PROMPT = "You are an assistant embedded in a web application. You can call tools to read data and to drive the user interface on the user's behalf. Prefer the most specific tool available. The application, not you, decides which actions need an explicit confirmation from the user: where one is needed it interrupts the call and comes back to you with the answer, so treat a tool call as a request the application may still refuse rather than as something you have already done. Do not re-ask in text for an action the user has clearly asked for. When a destructive or irreversible action is NOT clearly covered by what the user asked for, say what you are about to do and wait for them to agree before calling the tool. Briefly state what you are doing. When the user refers to something by name, use a listing or search tool's arguments to find it and then act on the result — don't stop after the lookup. Treat 'open', 'go to', or 'show me' as a request to navigate. Always finish your turn with a short reply or a completed action — never an empty turn. Keep replies concise."

Configuration¶

AGUIConfig `dataclass` ¶

An endpoint's resolved scalar configuration.

Every field is already resolved — there is no "unset" state and no settings lookup left to do. AGUIServer builds one in __init__ (via build_ag_ui_config, which reads DJANGO_AG_UI) and threads it to the agent view and every sub-view.

Resolving once is what lets two endpoints in one project differ: read at request time these values could only ever be global, forcing an /internal/agent and a /public/agent to share one tool-guard policy, one retry budget, one upload cap. Collaborators are not here at all — they are constructor arguments taking real objects.

Do not construct this directly to override a field: a partially-specified config would silently discard the project's own DJANGO_AG_UI values. Use build_ag_ui_config(retries=3), which layers overrides over the settings.

model `instance-attribute` ¶

model: Any

The Pydantic-AI model: a "provider:name" string (e.g. "anthropic:claude-sonnet-4.6") or a pre-built Model instance. May be None here; the agent factory raises a clear error if it is still unset when an agent is actually built.

api_key `class-attribute` `instance-attribute` ¶

api_key: str | None = field(repr=False)

API key handed to the provider when model is a "provider:name" string, so the key comes from settings rather than the environment. Ignored when a provider is passed or model is already a Model.

repr=False because this record is bound to a plainly-named local on every path that builds an agent, so the generated repr would print the provider secret into the frame locals of a technical-500 page or an error-reporting event. Name-based scrubbing does not catch it there: the key is nested inside another object's repr rather than sitting in a field called api_key. Read the attribute to use it -- only the rendering is suppressed.

system_prompt `instance-attribute` ¶

system_prompt: str | None

Override for the agent's default system prompt. None uses DEFAULT_SYSTEM_PROMPT.

model_settings `instance-attribute` ¶

model_settings: dict[str, Any] | None

Pydantic-AI ModelSettings (e.g. {"temperature": 0.2}) passed straight to the Agent. None leaves the model defaults untouched.

retries `instance-attribute` ¶

retries: int | None

Default tool/output retry budget passed to the Agent. None uses Pydantic-AI's default.

attachment_max_bytes `instance-attribute` ¶

attachment_max_bytes: int

Maximum accepted upload size in bytes (server-authoritative). 0 disables the cap.

attachment_allowed_types `instance-attribute` ¶

attachment_allowed_types: tuple[str, ...]

Allowed (client-declared) content types for uploads. Empty accepts any.

attachment_inline `instance-attribute` ¶

attachment_inline: AttachmentInlineConfig | None

How much of an attachment read_attachment hands the model, or None for the substrate's defaults.

A separate budget from the two above, and deliberately a smaller one: those bound what may be stored, this bounds what rides in every model request for the rest of the run. But the two must be set together, because a file above this limit and below attachment_max_bytes uploads, shows a chip, and can never be read -- indistinguishable from success on screen. Raising the upload cap without raising this one widens that band.

manage_system_prompt `instance-attribute` ¶

manage_system_prompt: str

Who owns the system prompt on the wire: "server" (the agent's prompt is authoritative and a client-posted system message is ignored) or "client". instructions are always server-side regardless.

allow_uploaded_files `instance-attribute` ¶

allow_uploaded_files: bool

Whether UploadedFile references in client-submitted messages are honoured. False drops them with a warning before they reach the agent.

forward_reasoning `instance-attribute` ¶

forward_reasoning: bool

Whether to forward a reasoning model's chain-of-thought to the client as AG-UI reasoning events — a pure adapter pass-through.

Reasoning events are emphatically not gated on an operator opting in, and the nearest counterexample is this package. A failed tool call emits a REASONING_ENCRYPTED_VALUE — that is where Pydantic-AI carries a non-success outcome — on any model whatsoever, including one that cannot think. It is forwarded rather than filtered because the event belongs to upstream and its general form carries provider continuity data.

The models supply the other half. Pydantic-AI's OpenAI-compatible chat path builds a ThinkingPart out of whatever the provider returned in reasoning / reasoning_content, consulting no setting at all, and its DeepSeek profile marks deepseek-reasoner with thinking_always_enabled because that model cannot be told to stop. On such a provider the default here — True — streams a chain-of-thought to every browser with nothing configured and nothing asked for. Set False to keep it server-side.

transcription_max_bytes `instance-attribute` ¶

transcription_max_bytes: int

Maximum accepted audio-clip size in bytes (server-authoritative). 0 disables the cap.

transcription_allowed_types `instance-attribute` ¶

transcription_allowed_types: tuple[str, ...]

Allowed (client-declared) content types for voice clips. Empty accepts any.

thread_list_limit `instance-attribute` ¶

thread_list_limit: int

Maximum threads the index endpoint returns in one call. A larger ?limit is clamped to this ceiling.

run_list_limit `instance-attribute` ¶

run_list_limit: int

Maximum runs the run index returns in one call, newest first.

A much tighter ceiling than thread_list_limit because the rows cost far more: the thread index answers from metadata alone, while every run row loads that run's last snapshot and holds its whole message list resident while the response is built. 0 disables the cap and restores the unbounded behaviour.

tool_guard `instance-attribute` ¶

tool_guard: ToolGuardConfig

Server-side destructive-tool approval policy. When enabled, a ToolGuard capability flips destructive tools to require the AG-UI approval interrupt.

approval_prompts `instance-attribute` ¶

approval_prompts: Mapping[str, str]

Human-readable questions for gated tools, by tool name, stamped onto the approval interrupt as x-confirm.

The question a client would otherwise ask is the call spelled out, which is accurate and unreadable. A registry tool's own @tool(confirm=...) is folded in here by AGUIServer, so this only needs entries for tools whose schema carries none — a spec tool reaching the agent in-process, or a bridged MCP tool. A tool with no entry keeps the generated question.

tool_failure `instance-attribute` ¶

tool_failure: ToolFailureConfig

What an unhandled tool exception costs. On by default, so a raising tool fails its own call and the turn carries on rather than ending in RUN_ERROR with the answer so far discarded.

run_context `instance-attribute` ¶

run_context: RunContextConfig

What client-supplied context reaches the model: the host page's own RunAgentInput.context entries and the attachment refs riding the posted messages, fenced and labelled as data, capped by a character ceiling.

RunContextConfig `dataclass` ¶

What client-supplied context reaches the model, and how much of it.

A RunAgentInput carries a context list the host page fills in, and pydantic-ai's adapter deliberately does not read it. This record is where a project says whether it wants that text delivered, by which channel, whether the attachment refs the web component posts become a manifest the model can act on, and what either is allowed to cost.

client_context and attachment_manifest choose what is delivered; delivery chooses how, for whatever survives them.

Every field is already resolved, matching AGUIConfig's contract — there is no unset state here. Build it through build_ag_ui_config, which layers the RUN_CONTEXT settings dict under any override.

client_context `instance-attribute` ¶

client_context: bool

Whether RunAgentInput.context entries are delivered to the model. False restores the behaviour of every release before this feature: the client can announce whatever it likes and the model never sees it.

attachment_manifest `instance-attribute` ¶

attachment_manifest: bool

Whether attachment refs carried on the posted user messages are derived into a manifest of files the model can read with read_attachment.

max_chars `instance-attribute` ¶

max_chars: int

Ceiling on the combined length of the delivered values. Client context is unbounded text that reaches the model on every request of a run, so this is a ceiling rather than a budget — content over it is truncated visibly, not dropped in silence.

delivery `class-attribute` `instance-attribute` ¶

delivery: ContextDelivery = 'instructions'

Which channel carries the block — ContextDelivery. "instructions" survives compaction and is never echoed back; "tool" follows pydantic-ai's documented preference and keeps client text out of the slot that carries operator authority. Read that type's docstring before changing it: the two are a genuine trade, not a default and a fallback.

Last and defaulted, unlike its siblings, so that adding a channel did not break every existing construction of this record. The default is what the package has always done.

ContextDelivery `module-attribute` ¶

ContextDelivery = Literal['instructions', 'tool']

Which channel carries the fenced client-context block.

There are two defensible answers and this package holds neither as the only one, because the right one depends on who writes the client. The block is identical either way — same fence, same sentinel neutralisation, same budget — and only the delivery differs.

"instructions" (the default) delivers it as additional run instructions. They are re-rendered on every model request, so the block survives compaction: the attachment manifest is still there at step 20 when the model decides to call read_attachment. They land on ModelRequest.instructions, which the AG-UI adapter does not emit, so the text is neither persisted into the thread nor echoed back to the browser.

The cost is the one pydantic-ai names in its own documentation: instructions carry operator authority, so building them out of text a client sent lets a prompt injection inherit it. Upstream left an AGUIAdapter.context accessor out deliberately for that reason and points consumers at tool output instead. The fence here is labelling, not sanitisation, and does not change that.

"tool" follows upstream: the block becomes the return value of a get_client_context tool, so the text arrives as data the model fetched rather than as instructions it was given. Three costs, all real. A tool result can be compacted away and is not re-supplied, so an attachment handle can stop being referenceable partway through a long run. The model has to decide to call it — ambient facts like which page the user is on are only considered if the model thinks to ask. And a tool result is an ordinary part of the exchange, so the block is streamed back to the browser and persisted into the thread, where the instructions channel is neither. That is auditability for some projects and an unwanted copy of a page map for others; it is a property of the channel, not a bug in it.

Pick "tool" where the page that fills RunAgentInput.context is not fully under your control, and "instructions" where it is and the manifest matters more than the authority boundary.

build_ag_ui_config ¶

build_ag_ui_config(
    *,
    model: Any = None,
    api_key: str | None = None,
    system_prompt: str | None = None,
    model_settings: dict[str, Any] | None = None,
    retries: int | None = None,
    attachment_max_bytes: int | None = None,
    attachment_allowed_types: tuple[str, ...] | list[str] | None = None,
    attachment_inline: AttachmentInlineConfig | None = None,
    manage_system_prompt: str | None = None,
    allow_uploaded_files: bool | None = None,
    forward_reasoning: bool | None = None,
    transcription_max_bytes: int | None = None,
    transcription_allowed_types: tuple[str, ...] | list[str] | None = None,
    thread_list_limit: int | None = None,
    run_list_limit: int | None = None,
    approval_prompts: Mapping[str, str] | None = None,
    tool_guard: ToolGuardConfig | None = None,
    tool_failure: ToolFailureConfig | None = None,
    run_context: RunContextConfig | None = None,
) -> AGUIConfig

Resolve an AGUIConfig from DJANGO_AG_UI, applying overrides.

The single place the scalar settings are read. AGUIServer calls this once in __init__; nothing reads these settings per request, which is what lets two endpoints in one project hold different values.

Every argument is None by default, meaning "take it from settings". Pass one to override just that field for this endpoint:

AGUIServer(registry, config=build_ag_ui_config(retries=3))

Use this rather than constructing AGUIConfig directly — it is what layers your overrides over the project's settings instead of discarding them.

Policy and audit¶

AuditLogger ¶

Bases: Protocol

Sink for tool-invocation records.

An implementation may drop, sample or forward events. The package ships NullAuditLogger and LoggingAuditLogger; a project passes its own to its transport's audit_logger=.

AuditCapability ¶

Bases: AbstractCapability[Any]

Records every tool execution to an AuditLogger sink.

A Pydantic-AI capability on the wrap_tool_execute lifecycle hook, so it times and records every tool the agent runs: registry tools, the drf-mcp and spec bridges, attachment and skill tools alike.

Recording is non-raising. A sink that throws is caught and logged to the django_pydantic_agent.audit Python logger, so a broken audit backend costs audit records rather than the run.

Parameters:

Name	Type	Description	Default
`logger`	`AuditLogger`	The sink each `AuditEvent` is recorded to.	required
`ip_address`	`str \| None`	Fallback client IP, used only when the run's deps carry no `ip_address`. Per-run deps come first because a constructor argument is per-agent: taking the IP from it alone forces a fresh agent per request, and building once anyway fails silently, with every record carrying the IP of whoever arrived first.	`None`
`organization_id`	`str \| None`	Org scope stamped onto every event, for a multi-tenant host.	`None`

get_ordering ¶

get_ordering() -> CapabilityOrdering

Pin audit as the outermost capability in the chain.

Its wrap_tool_execute has to surround every other capability's execution hooks so the tool is recorded whatever else composes the run. Declaring it here rather than relying on list order at the build_agent call site keeps that true however the capabilities are inserted, since pydantic-ai sorts by these constraints.

AuditEvent `dataclass` ¶

A single tool invocation as seen by the audit logger.

Arguments are stored as a string, typically JSON, to keep records cheap to serialize and to discourage retaining sensitive raw values.

One run-level record rides this shape: a client disconnecting mid-run is recorded as tool_name="agent.run", success=False and an error starting "cancelled:", so a sink can tell cancelled runs apart without widening the AuditLogger protocol.

organization_id `class-attribute` `instance-attribute` ¶

organization_id: str | None = None

Multi-tenant scope of the acting user. None at this layer; a custom AuditLogger fills it from its own tenancy model.

target_type `class-attribute` `instance-attribute` ¶

target_type: str | None = None

Kind of domain object the call acted on. None at this layer, where tool arguments are domain-opaque; a sink that knows its tools can classify them.

target_id `class-attribute` `instance-attribute` ¶

target_id: str | None = None

Identifier of the acted-on object, paired with target_type.

ip_address `class-attribute` `instance-attribute` ¶

ip_address: str | None = None

Client IP of the request that drove the run, when the view knows it.

NullAuditLogger ¶

Discards every event. The default when no audit logger is configured.

LoggingAuditLogger ¶

Writes audit events to the Python logging framework.

Successful invocations log at INFO; failures log at WARNING with the error message included.

Rate limiting¶

The seam both spending routes take — the agent endpoint via throttle= and transcribe/ via transcribe_throttle=. See throttle= for the contract and its ordering.

Throttle ¶

Bases: Protocol

Rate limiter for the agent run endpoint, evaluated after authentication.

A single consume call is the gate and the bookkeeping update — there is deliberately no separate "check, then commit", a shape that races under exactly the concurrency an agent endpoint sees. Implementations decrement quotas atomically in shared storage and return the suggested Retry-After in seconds, or None to allow the run. 0 is allowed (denied, but the window resets immediately); most implementations return a positive integer.

Running after authentication is what lets a limiter key on the acting user rather than only on an IP: request.user is resolved by the time this is called, including through a get_user hook.

consume is synchronous and is run off the event loop, so it may safely touch the Django cache or the ORM. An async def consume is refused at construction rather than awaited, because a coroutine returned into an is not None check would throttle every request on a bogus Retry-After.

State that crosses requests must live in shared storage (the Django cache, Redis), never on the instance — an endpoint is constructed once per process, so instance counters would enforce a per-worker limit while reading like a global one.

Mirrors djangorestframework-mcp-server's MCPRateLimit, so a project protecting both transports writes one kind of limiter.

FixedWindowThrottle ¶

A fixed-window run limiter backed by django.core.cache.

The window is bucketed by absolute time: every integer multiple of per_seconds since the epoch starts a fresh counter. Simpler than a sliding window and sufficient for the thing this protects against — one client starting runs faster than a human could read them, on an endpoint where each run costs a model call.

namespace separates counters so two throttles on one endpoint (a burst limit and a steady-state limit) do not share a bucket. key chooses the bucket dimension; the default is per-user, falling back to per-IP.

The cache must be a shared backend in a multi-process deployment. Django's locmem cache is fine in tests but enforces a per-worker limit that reads like a global one.

Mirrors djangorestframework-mcp-server's FixedWindowRateLimit, so the two transports behave the same way under the same configuration.

Conversation persistence¶

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 `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 `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 ¶

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".

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.

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 ¶

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): ...

ThreadsView ¶

Owner-scoped thread index endpoint for the chat-history drawer (async, JSON).

Mounted by AGUIServer whenever conversation_store= is a live store, over the same ConversationStore the agent view uses:

GET <prefix>threads/ → the user's threads, metadata only ({"threads": [...]});
GET <prefix>threads/<id>/ → that thread's messages ({"thread_id", "messages"});
PATCH <prefix>threads/<id>/ → rename (body {"title": "..."});
DELETE <prefix>threads/<id>/ → delete the thread (204).

Every operation is scoped to the acting user: the store filters by owner, so a thread owned by another user simply isn't found (404) — never another user's history. The view carries the same authentication seam as DjangoAGUIView (require_authenticated / get_user, sync or async), and the closed default is load-bearing here: every route is owner-scoped, so an anonymous caller has no history to reach.

A thread_activity_source= merges pushed activities back into the read thread — see ThreadActivitySource for why they are not in the stored history to begin with.

ThreadActivitySource ¶

Bases: Protocol

Where a restored thread's pushed activities come back from.

Passed as AGUIServer(thread_activity_source=...); consulted by ThreadsView on GET <prefix>threads/<id>/, whose messages then carry the returned activities alongside the stored turns. Off unless supplied.

Why this is a hook and not a setting. A pushed activity deliberately never enters the model's message history -- that is the entire reason to push one rather than let the agent call a tool -- and the stored thread is that history. So the server has nothing to redraw from, and cannot get it without keeping a second record beside the conversation, with its own ordering, its own identity rules and its own answer for what a resumed run does with a snapshot. The project already holds the data (it charted it), so the smaller, more honest seam is to ask.

Materialise, do not replay. event is an ActivitySnapshotEvent and deliberately only that: a chart that was moved with chart_points_delta comes back as a fresh snapshot built from the current numbers, not as the patches that produced them. The project holds those numbers already -- it computed them, which is where the deltas came from -- so materialising is a constructor call. Widening this to accept deltas would instead ask every implementation for an ordered event log, a replay on every thread load, and an answer for what a resumed run does with a half-applied patch, all to arrive at a value that was already in a variable.

The stored messages are handed over so an implementation can work out where each activity belongs -- the tool result it accompanied is in there, and its id is what ThreadActivity.after_message_id wants. They are the thread as stored; editing them changes nothing, since only the returned activities are merged in.

Async because it runs on the event loop, next to the store's own reads: a Django ORM lookup here needs the a-prefixed queryset methods or sync_to_async, the same as any other view code in this package.

ThreadActivity `dataclass` ¶

One pushed activity to put back into a restored thread, and where.

Returned by a ThreadActivitySource. The event is the same object the run pushed -- chart_activity builds one -- so a project re-pushes what it already knows how to build rather than learning a second vocabulary for the restore path.

after_message_id is the stored message this activity followed, and it is the caller's answer to the ordering question the library cannot answer for itself: the activity was never in the model's history, so nothing in the stored thread records where it belonged. Name the message it came after and it lands there; leave it None -- or name a message this thread does not have -- and it lands at the end, which is right for a chart pushed after the last turn and wrong for one pushed three turns ago.

Reference store (opt-in)¶

The django_pydantic_agent.contrib.store app ships a ready-to-use durable store. Add it to INSTALLED_APPS, run migrate, then set conversation_store= to django_pydantic_agent.contrib.store.default_conversation_store.DefaultConversationStore. Projects that don't opt in get no model and 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.

StoredConversation ¶

Bases: Model

The reference durable conversation row, one per (owner_id, thread_id).

owner_id is the resolved owner (the user's pk, or an anon:<session_key> bucket) and every query filters by it: it is the security boundary. title / preview are denormalised so the thread drawer's list query never loads message bodies. attachments is derived from messages on every save and is what attachment lifecycle runs on.

Used by DefaultConversationStore.

Step persistence¶

The durable run / event / snapshot ledger behind resume/, fork/ and runs/. The store itself is a request -> StepStore factory from pydantic-ai-harness (the reference one is django_pydantic_agent.contrib.store.default_step_store.DefaultStepStore); this is the wrapper that keeps two endpoints' ledgers apart, the way ScopedConversationStore does for thread history.

ScopedStepStore ¶

Partition a step-store factory by a scope name.

A step ledger is keyed by (owner_id, run_id). Two AG-UI endpoints handed the same step_store therefore share one user's runs: a run recorded at /internal/agent is listed by /public/agent/runs/ and, because resume/<run_id>/ addresses a run by id, can be continued there — under the public agent's model, tools and guard policy. Owner scoping does not catch it: it is the same user on both mounts.

Wrapping fixes that without a migration:

internal = AGUIServer(
    registry,
    step_store=ScopedStepStore(DefaultStepStore, scope="internal"),
    config=build_ag_ui_config(tool_guard=ToolGuardConfig(enabled=True)),
)
public = AGUIServer(
    registry,
    step_store=ScopedStepStore(DefaultStepStore, scope="public"),
)

A factory in, a factory out, unlike ScopedConversationStore, which wraps a store: the harness protocol's methods carry no request, so step_store= takes a request -> StepStore callable and the store is built per call. This is that callable, and calling it wraps whatever the inner factory returned.

The partition is a run-id prefix, so it composes with any implementation, third-party ones included, where a scope column would mean a migration and a breaking change to a protocol that is upstream's, not ours. A run belonging to another scope is not refused on resume/ — it is simply not found, so a probe cannot confirm the id exists either.

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

Opt in explicitly. A transport does not wrap by itself: doing so from its namespace would orphan every run an existing single-endpoint project had already recorded, the moment it set one. For the same reason, adding a scope to a mount that has been running hides that mount's earlier runs from runs/ rather than migrating them.

File uploads¶

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 `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 `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 ¶

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 ¶

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): ...

AttachmentsView ¶

Owner-scoped file-upload + download endpoint (async, multipart/JSON).

Mounted by AGUIServer whenever attachment_store= is a live AttachmentStore:

POST <prefix>attachments/ → multipart upload under the file field; validates size/type from DJANGO_AG_UI settings, persists the bytes, and returns 201 with the AttachmentRef JSON ({"id", "name", "mime", "size", "url"?}) — a durable ref, not bytes.
GET <prefix>attachments/<id>/ → stream the bytes back (owner-checked) for preview/download; missing or cross-owner → 404.
DELETE <prefix>attachments/<id>/ → drop the attachment (204).

Every operation is scoped to the acting user: the store filters by owner, so one user's id can never resolve another's file. Downloads are served as an attachment with X-Content-Type-Options: nosniff so an uploaded text/html can't execute as a same-origin page. The view carries the same authentication seam as DjangoAGUIView (require_authenticated / get_user), and the closed default is load-bearing here: every route is owner-scoped, so an anonymous caller has no files to reach.

With the default NullAttachmentStore an upload returns 410 (off): mount the view with a real store to enable it.

Reference attachment store (opt-in)¶

The same django_pydantic_agent.contrib.store app ships a ready-to-use durable file store. With the app installed and migrated, set attachment_store= to django_pydantic_agent.contrib.store.default_attachment_store.DefaultAttachmentStore. The bytes go to Django Storage (S3/GCS via STORAGES / DEFAULT_FILE_STORAGE); projects that don't opt in get no model and no migration.

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.

StoredAttachment ¶

Bases: Model

The reference durable attachment row for a model-backed store.

attachment_id is the opaque handle the wire ref carries; file holds the bytes via Django Storage; name / mime / size are denormalised so metadata is returned without reading the file back. owner_id is the resolved owner and every query filters by it: it is the security boundary. sha256 is the chunked digest that lets a re-upload of the same bytes reuse the blob already in storage, blank on rows predating the column until agent_store_backfill_hashes fills them in.

thread_id is a loose label a project may set; the reference store leaves it blank. Lifecycle runs on the ConversationAttachment relation, not on this column, which is kept only because projects read it.

Used by DefaultAttachmentStore.

Voice input¶

TranscriptionBackend ¶

Bases: Protocol

Pluggable server-side speech-to-text for the composer's voice input.

Passed as AGUIServer(transcription_backend=...). The package ships a no-op default (NullTranscriptionBackend — voice off) and an opt-in reference implementation over an OpenAI-compatible /audio/transcriptions endpoint (OpenAITranscriptionBackend).

The single method is async and receives the acting request so a backend can scope by user, rate-limit, or bill per principal. Unlike an AttachmentStore, transcription keeps no durable artifact — audio in, text out, nothing to open or delete. transcribe validates nothing about size or type; the view does that from settings.

NullTranscriptionBackend ¶

The default transcription backend: voice input disabled.

TranscribeView detects this backend and answers 410 Gone, so transcribe is never reached through the endpoint and a misconfigured client gets a clear "voice is off" signal. Called directly it raises, rather than fabricating a transcript. Pass AGUIServer(transcription_backend=...) to enable voice.

TranscribeView ¶

Owner-scoped speech-to-text endpoint (async, multipart in / JSON out).

Mounted by AGUIServer whenever transcription_backend= is a live TranscriptionBackend:

POST <prefix>transcribe/ → multipart audio under the audio field; validates size/type from DJANGO_AG_UI settings, runs the backend, and returns 200 with {"text": "<transcript>"}.

The audio is transcribed and discarded — nothing is stored — so unlike AttachmentsView there is no download/delete route. The view carries the same authentication seam as DjangoAGUIView (require_authenticated / get_user), closed by default — which matters here beyond consistency: the backend spends money per request, so an open route is a bill as well as a leak.

Authentication is not a spend limit, which is why throttle is here too: an authenticated caller looping small valid clips reaches the provider on every request. It takes the same Throttle the agent endpoint takes and runs at the same point — after authentication, so a limiter can key on the acting user, and before the body is parsed. Give it its own limiter rather than sharing the agent's: one instance is one counter, so a shared one would let voice input consume the run budget.

The size cap aborts the upload rather than measuring it afterwards. A CappedUploadHandler is inserted before the multipart body is parsed, so a clip over TRANSCRIPTION_MAX_BYTES is refused mid-stream instead of being spooled to a temp file in full and answered 413 once it is already on disk.

With the default NullTranscriptionBackend a request returns 410 (off): mount the view with a real backend to enable it.

Reference transcription backend (opt-in)¶

A ready-to-use backend over any OpenAI-compatible /audio/transcriptions endpoint. Install the [openai] extra and set transcription_backend= to django_ag_ui.contrib.transcription.openai_transcription_backend.OpenAITranscriptionBackend; subclass it to change the model or point at another OpenAI-compatible server.

OpenAITranscriptionBackend ¶

A ready-to-use transcription backend over an OpenAI-compatible API.

The batteries-included voice backend: it forwards the recorded clip to an OpenAI /audio/transcriptions endpoint and returns the text. Enable it by installing the [openai] extra and passing an instance to the server, which is what mounts transcribe/:

AGUIServer(registry, transcription_backend=OpenAITranscriptionBackend())

Constructible with no arguments: the API key comes from the OPENAI_API_KEY environment variable (the SDK default). Override the model or point at an OpenAI-compatible server (Azure OpenAI, a local Whisper server, Groq, …) by subclassing and setting the class attributes:

class GroqTranscription(OpenAITranscriptionBackend):
    model = "whisper-large-v3"
    base_url = "https://api.groq.com/openai/v1"
    api_key = os.environ["GROQ_API_KEY"]

base_url and api_key travel together. The SDK sends whatever key it holds as the bearer token to whatever host base_url names, and with api_key left unset that key is OPENAI_API_KEY from the environment — so pointing at a third-party endpoint without setting one hands that vendor the OpenAI credential, on every clip, and shows only a 401 for it. Set both, or neither.

The openai SDK is imported lazily inside transcribe so the base package keeps it an optional dependency (the [openai] extra).

model `class-attribute` `instance-attribute` ¶

model: str = 'whisper-1'

The transcription model name passed to the API.

base_url `class-attribute` `instance-attribute` ¶

base_url: str | None = None

Optional base URL for an OpenAI-compatible endpoint (None → OpenAI).

api_key `class-attribute` `instance-attribute` ¶

api_key: str | None = None

Key sent to base_url as the bearer token. None falls back to the SDK's own default, the OPENAI_API_KEY environment variable — correct for OpenAI itself and wrong for anywhere else, so set this whenever base_url is set.

timeout `class-attribute` `instance-attribute` ¶

timeout: float | None = 60.0

Per-request timeout (seconds) for the transcription call. The SDK default is 10 minutes; a bounded default keeps a stalled upstream from pinning a worker. Override on a subclass for slower endpoints.

Charts¶

ChartSpec `dataclass` ¶

A chart, as data rather than as markup.

Sent as the content of an ACTIVITY_SNAPSHOT, which the client draws itself. Nothing here is HTML and nothing here is interpreted as HTML: the server chooses the numbers and the browser chooses the DOM, which is what keeps a pushed visual off the sanitiser's surface entirely.

Frozen at the top level: labels, series and the points inside them are copied to tuples and metadata to a read-only mapping, so a list handed in cannot be appended to after it has been checked. A nested structure inside metadata is still shared with the caller, and nothing reads it here anyway.

Frozen, and not usefully hashable -- __hash__ exists but raises, because metadata is part of a spec's value, so excluding it to buy a hash would make two specs carrying different payloads compare equal. That is the ordinary situation for a record with a mapping field; nothing here needs to be a dict key.

The client is the authority on shape. It refuses a spec whose series disagree in length with the labels, because a chart that is subtly misaligned still reads as authoritative -- validate catches that here instead, on the side that can name the offending series.

title `class-attribute` `instance-attribute` ¶

title: str | None = None

Shown above the chart. Anything but a string is refused rather than sent: the client treats a non-string title as absent, so it would vanish quietly.

metadata `class-attribute` `instance-attribute` ¶

metadata: Mapping[str, Any] = field(default_factory=dict)

Extra keys merged into the payload, for a client that reads more than this package knows about.

Copied and made read-only at construction, so a spec cannot be edited into an invalid one after it has been checked. Its values are not inspected: anything that Pydantic cannot serialise raises at encode time, which is mid-stream, after the response has already begun.

validate ¶

validate() -> None

Raise ValueError for a spec the client would refuse to draw.

Checked at construction rather than at send time so a mistake surfaces where the data is assembled. The client refuses the same shapes and silently draws nothing -- it has no channel to complain on -- so a spec that gets past here reaches a user as a chart that simply is not there. Failing at construction names which series is wrong, on the side that can fix it.

as_content ¶

as_content() -> dict[str, Any]

The activity payload the client reads.

ChartSeries `dataclass` ¶

One named series, carrying exactly one point per chart label.

The length agreement is checked by ChartSpec, which is the only thing that knows how many labels there are.

ChartKind `module-attribute` ¶

ChartKind = Literal['bar', 'line', 'pie', 'scatter', 'stacked']

How the client draws a spec.

An unrecognised value is drawn as a bar rather than refused: the numbers are still worth showing, and a chart that appears in the wrong shape is easier to notice than one that never appears at all.

chart_activity ¶

chart_activity(
    spec: ChartSpec, *, chart_id: str | None = None
) -> ActivitySnapshotEvent

An event drawing spec, or redrawing a chart already on screen.

Emitted by the project from its own code, where it holds the data. There is no setting that turns this on: pushing a chart is an act rather than a mode, and a flag would suggest the framework emits one on your behalf, which it cannot -- it has no idea what you want charted.

The data never enters the model's context. That is the whole reason to push rather than let the agent call a tool: a large or sensitive dataset is drawn for the user without being sent to the provider, and without costing a model round. The trade is that the model cannot then discuss what it never saw.

chart_id is the identity of this chart, not of this event. Send the same one again to replace what is on screen -- a chart that redraws as a computation advances is one chart moving, and the client swaps it in place rather than stacking copies. Omit it and every call draws a new chart.

It is an AG-UI message_id and shares that namespace, so choose something unlikely to collide: a chart sent under an id already used by an assistant turn replaces that message in the client's transcript. Prefixing is enough, and the generated ids do it (chart-<uuid>).

chart_points_delta ¶

chart_points_delta(
    chart_id: str,
    *,
    series: int = 0,
    points: tuple[float, ...] | list[float],
    spec: ChartSpec | None = None,
) -> ActivityDeltaEvent

Replace one series' points on the chart already under chart_id.

The cheap half of live updating. A snapshot re-sends the whole spec, which is right when the shape changes; this sends a JSON Patch touching one array, which is right when a long-running computation is only moving the numbers.

chart_id must name a chart the client has already drawn. A delta for an id it does not hold is dropped -- there is nothing to patch -- so send the snapshot first and keep the id.

series is the index in the spec's series list, and points must be the same length as the series it replaces. Neither can be checked from chart_id alone -- an id is a name, not a shape -- and both fail the same quiet way: a patch is applied positionally, so it cannot tell that series 2 is now something else, and a wrong-length array applies cleanly and leaves a chart the client then refuses to redraw -- stale numbers on screen, and the chart gone entirely on the next reload.

spec= is how you get told. Pass the ChartSpec the chart on screen was drawn from and both of those become construction-time errors, the way a spec's own shape already is. It stays optional and the default is unchanged: a caller that does not declare a shape gets exactly the behaviour above, because a delta sent from somewhere the spec is no longer in hand is a legitimate call, not a mistake.

Why the spec and not an expected point count. A count was the cheaper argument and is the wrong one twice over. It is derived -- len(spec.labels) -- and deriving it by hand is the same step that goes wrong in the first place, so a caller who miscounts the points miscounts the count too and the guard cheerfully agrees with the mistake. And it can only see half the problem: series is an index into spec.series, so only the spec itself can say that index 2 is past the end. The spec also costs nothing the caller does not already have, since a delta requires a snapshot to have gone first and the spec is what that snapshot was built from -- keeping it is the same act as keeping the chart_id.

It is read, never sent. This helper still emits a patch touching one array; spec is the caller's declaration of what that array is expected to replace, and nothing about it reaches the wire.

CHART_ACTIVITY_TYPE `module-attribute` ¶

CHART_ACTIVITY_TYPE = 'chart'

activity_type the client matches on to draw a chart.

A convention inside an extension point the protocol already provides, not a protocol extension: AG-UI defines the envelope and leaves activity_type an open string. An ACTIVITY_SNAPSHOT rather than a CUSTOM event, for the reason inject_compaction_events gives for the same choice -- the wire stays vanilla AG-UI and ours is not a privileged client. A client that does not know this name ignores the event, which is the graceful outcome.

Follow-up suggestions¶

suggestions_activity ¶

suggestions_activity(
    prompts: Sequence[str], *, suggestions_id: str | None = None
) -> ActivitySnapshotEvent

Offer up to MAX_SUGGESTIONS follow-up prompts as clickable chips.

Emitted by the project from its own code, after a tool ran or a turn finished -- wherever it knows what the user is likely to want next. Registered skill chips cannot do this: they are static and host-configured, so they can offer "summarize this" but never "want me to update the shipping address too?".

Clicking a chip sends that text as the user's message. So a prompt is written as the user would say it, first person and complete, not as a label: "Update the shipping address too", not "Shipping address".

suggestions_id is the identity of this set, not of this event -- the same contract as chart_id. Omit it and every push draws its own row, under the answer it follows, which is what suits a set offered per turn. Send the same one again to replace a row already on screen, in the place it was drawn.

It is an AG-UI message_id and shares that namespace, so a set sent under an id an assistant turn already used replaces that message. Prefixing is enough, and the generated ids do it (suggestions-<uuid>).

Chips are content, so they persist and a reload puts them back. Rows from earlier turns stay above, reading as what was offered then; nothing expires them, and a project that wants exactly one row keeps one id.

Raises:

Type	Description
`ValueError`	If `prompts` is empty, carries more than `MAX_SUGGESTIONS`, or holds a prompt that is blank or longer than `MAX_SUGGESTION_CHARS`. Raised here rather than trimmed, because the client cannot report what it dropped and the failure would be a suggestion that silently never appears.

SUGGESTIONS_ACTIVITY_TYPE `module-attribute` ¶

SUGGESTIONS_ACTIVITY_TYPE = 'suggestions'

activity_type the client matches on to draw follow-up chips.

A convention inside an extension point the protocol already provides, exactly as chart is: AG-UI defines the envelope and leaves activity_type open. A client that does not know this name ignores the event, which is the graceful outcome.

MAX_SUGGESTIONS `module-attribute` ¶

MAX_SUGGESTIONS = 4

Most prompts one push may carry.

Slack's assistant.threads.setSuggestedPrompts caps at four and it is the right number for the same reason here: the chips wrap onto a second row beyond it, and a wall of suggestions is a menu rather than a nudge.

Keep in step with MAX_SUGGESTIONS in the component's src/ui/suggestion_chips.ts. The client silently draws no more than its own limit and has no channel to report the difference, so a producer that does not know the same number ships suggestions that never appear -- the exact hole chart_limits exists to close, and it was found there by shipping it.

MAX_SUGGESTION_CHARS `module-attribute` ¶

MAX_SUGGESTION_CHARS = 120

Longest one prompt may be.

A suggestion is a question the user might send, not an answer: past a sentence it stops being scannable, and a chip that wraps to three lines is a paragraph with a border. Mirrored on the client for the same reason as the count.

Sub-agent progress¶

SubAgentObserver ¶

Bases: WrapperCapability[Any]

Wraps a SubAgents capability and reports each delegation as it happens.

A delegated child runs to completion inside one delegate_task tool call, and a tool call emits nothing between its arguments and its result. So a parent that hands a long task to a sub-agent shows a tool card that simply sits there -- for a minute, for five -- with no way to tell a working run from a wedged one. Wrapping the capability is the seam for saying so:

capabilities=[SubAgentObserver(SubAgents(agents=[...], agent_folders=None))]

Opt-in by construction: passing SubAgents unwrapped emits nothing, and costs nothing.

What reaches the client rides two carriers, on purpose. The delegation's own lifetime goes on the protocol's SUBAGENT_STARTED / _FINISHED / _ERROR events, built by subagent_lifecycle; each tool call the child makes goes on a CUSTOM event, whose wire contract is SUBAGENT_EVENT_NAME. Both key on the parent's own delegate_task tool call id -- as parentToolCallId and as delegationId respectively -- so a client augments the card it already drew rather than opening a second row beside it.

The split is not a transitional state. Moving the steps to the protocol's own vocabulary would mean ordinary TOOL_CALL_* events tagged with subagentRunId, and those are materialised into the persisted message list and replayed on every thread restore -- which would redraw a finished run's progress as though it were live. subagent_progress carries the full reasoning.

The observer installs itself onto the capability you hand it. Reporting a child's tool calls needs SubAgents.event_stream_handler, which only the capability that starts the child run can pass on, so construction sets that field on the wrapped instance. Two consequences worth stating rather than discovering: wrapping a SubAgents that already carries a handler is refused (silently replacing it would lose whatever it was for), and the instance you passed is the one that changed. The second is harmless in a way worth knowing -- the installed handler announces onto SUBAGENT_SINK, which no run outside this transport's stream ever binds, so the same instance reused at an unwrapped endpoint behaves exactly as it did before.

Subclassing WrapperCapability (pydantic-ai's supported wrapper, analogous to WrapperToolset) rather than hand-rolling a proxy keeps the rest of the capability protocol intact -- ordering, the has_* hook-introspection flags and every other lifecycle method delegate untouched.

wrap_tool_execute `async` ¶

wrap_tool_execute(
    ctx: Any, *, call: Any, tool_def: Any, args: dict[str, Any], handler: Any
) -> Any

Announce the delegation this call is, then run it.

This hook fires for every tool in the parent run, not only the ones the wrapped capability contributed -- pydantic-ai composes one root capability and asks it about each call -- so the delegate tool is picked out by name and everything else is delegated untouched.

Every exit path closes the delegation it opened, cancellation included -- which reverses an earlier choice here, and the reason is worth keeping. Not announcing on asyncio.CancelledError was defended as "a cancelled run is a client that has gone away, and there is nobody left to tell", and that holds only when the whole run was cancelled. A single tool call can be cancelled while the run continues, and under the protocol's own lifecycle an unclosed delegation is not merely a missing line: @ag-ui/client refuses RUN_FINISHED while any delegation is still open, so the omission would take down the run it was trying not to disturb. Announcing into a sink nobody is draining costs nothing, which makes the safe direction the cheap one.

SUBAGENT_EVENT_NAME `module-attribute` ¶

SUBAGENT_EVENT_NAME = 'ag_ui.subagent'

name a client matches on to render one step a delegated sub-agent took.

A convention inside an extension point the protocol already provides, not a protocol extension: AG-UI defines the envelope and leaves name an open string. A client that does not know this name ignores the event, which is the graceful outcome and the whole reason the field is open -- a run that delegates streams exactly as it did before, plus events nobody has to read.

Half a contract, and the half it is¶

A delegation's lifetime is no longer described here. AG-UI 0.1.21 gave a sub-agent a first-class lifecycle, so SUBAGENT_STARTED opens a delegation and SUBAGENT_FINISHED / SUBAGENT_ERROR closes it, built by subagent_lifecycle. What stayed on this carrier is what the child does in between: one event per tool call it makes and one per result it gets back.

The value¶

Every event carries the same five keys::

{
  "delegationId": "call_abc123",
  "agent": "researcher",
  "phase": "tool_call",
  "status": "researcher: calling search_docs",
  "tool": {"toolCallId": "call_def456", "name": "search_docs", "ok": null}
}

delegationId is the parent's delegate_task tool call id -- the toolCallId the client already received on TOOL_CALL_START and already drew a tool card for. That is what makes this an augmentation of a card on screen rather than a second row beside it, and it is why the id is the parent's call rather than the child's run id: the child's run id names something the client has never heard of. It is also the value the lifecycle events carry as parentToolCallId, which is what joins the two carriers.
agent is the sub-agent's name, as the parent model asked for it.
phase is tool_call or tool_result. Any number of the pair sit between the SUBAGENT_STARTED that opens the delegation and the SUBAGENT_FINISHED / SUBAGENT_ERROR that closes it.
status is a rendered one-line summary. A client that never expands a row needs nothing but this, which is the point of sending it: the collapsed line costs no client-side assembly and no phase-to-wording table. A client that wants its own wording (a localised UI) builds it from the structured keys and ignores this one.
tool always has all three keys. ok is null on tool_call (the call has not returned yet), true on a result the child accepted, and false on one it did not -- which pydantic-ai surfaces to the child as a retry, and which the status line calls failed, because that is what a reader watching the row sees happen. A client rendering its own wording should follow status; ok is the machine-readable half of the same fact. Sending the key as null rather than omitting it keeps the shape fixed, so a client can create the row on tool_call and update it in place on tool_result.

Note that the null survives on the wire, and it is worth knowing why it is safe to rely on. From 0.1.21 the protocol omits an unset optional field rather than serialising it as null -- but that applies to declared model fields, and ok is a value inside a plain dict, which it leaves alone.

The child's text output is deliberately absent. Progress is a status line, not a second transcript.

Why `CUSTOM`, still, for these two¶

The lifecycle moved to the protocol's own events and these did not, and the reason is the one that chose CUSTOM in the first place: lifetime.

The protocol does have a way to say "the child called a tool" -- an ordinary TOOL_CALL_START tagged with subagentRunId. But @ag-ui/client materialises those into agent.messages exactly as it does the parent's own, the transport persists the message list wholesale, and the replay path re-fires it on every thread restore. Replayed progress is a lie: a run that finished last week would redraw "working, step 4 of 8" on every reload of the thread. The three lifecycle events have no such problem -- the client dispatches them to callbacks and never pushes them into messages -- which is precisely why they were adoptable and these are not.

ACTIVITY_SNAPSHOT is for content; CUSTOM is for an imperative. Progress is the second -- it has no meaning once acted on. That is also the choice chart_activity makes in the other direction, because a chart is content and should come back.

STATE_SNAPSHOT / STATE_DELTA are out for a third reason: shared state round-trips into the next RunAgentInput, so progress placed there would be echoed back to the model as though it were something the model had said.

Host data invalidation¶

publish_invalidation ¶

publish_invalidation(*keys: str, reason: str | None = None) -> bool

Queue an invalidation for the run currently streaming, without returning it.

The sibling of :func:resource_invalidation, and the one to reach for inside a transaction. Returning an event as tool metadata is the simpler route and is right when nothing is being written transactionally, but it cannot answer the ordering question at all::

@agent.tool_plain
def place_order(sku: str) -> str:
    with transaction.atomic():
        order = Order.objects.create(sku=sku)
        # Fires when the transaction commits -- not before, and not at
        # all if it rolls back.
        transaction.on_commit(
            lambda: publish_invalidation("orders", f"orders/{order.pk}",
                                         reason="place_order")
        )
    return f"Ordered {sku}."

Announcing before the commit is wrong by default, not in an edge case. A ServiceSpec is atomic=True unless told otherwise, so the naive version tells the page to refetch data that has not committed -- and may never commit. The page then re-reads the old row and caches it as fresh, which is worse than not having been told.

Register the callback with transaction.on_commit, from the thread that owns the transaction. Django connections are thread-local, so code running on the event loop sees no transaction at all and any callback it registers runs immediately -- announcing early while looking correct. In this package's async topology that means registering from inside whatever sync_to_async(..., thread_sensitive=True) wrapper is already doing the write, which is where the ORM call is anyway.

Returns whether a run was listening. False means there was no stream to queue onto -- a call from off-HTTP code -- and nothing happened.

resource_invalidation ¶

resource_invalidation(*keys: str, reason: str | None = None) -> CustomEvent

An event naming the resources a write has just moved.

Emitted by the project from its own code, where it knows what it wrote. There is no setting that turns this on and nothing derives it for you: the framework has no idea which of your pages read which of your tables.

keys are opaque host-defined strings. This package never interprets them, compares them to anything, or requires a scheme -- it carries them. That is the load-bearing choice: every alternative (model labels, spec names, URLs, MCP-style URIs) encodes one side's vocabulary into a contract both sides have to read, and the two sides here are a Django app and a frontend that may have no notion of Django models at all.

Name every key you want invalidated, including the collection. Matching on the wire is exact and never by prefix, because a prefix rule guesses at a scheme this package does not own and fails both ways -- orders/1 would match orders/11. So a write to one row that should also refresh the list says both::

resource_invalidation("orders", "orders/42", reason="place_order")

The host may then match hierarchically in its own vocabulary, where the scheme is known and hierarchy is the point (TanStack query keys are built this way). That is the host's to interpret, not this package's to guess.

reason is the tool or action that caused the write, carried for the host's logging and filtering. It is never interpreted here either.

The page it reaches is the one that started the run, during the run. There is one StreamingHttpResponse per run and no channel to anybody else, so this cannot tell another user's open page that something moved -- see the invalidation guide for why that is a different subsystem rather than a missing argument.

INVALIDATE_EVENT_NAME `module-attribute` ¶

INVALIDATE_EVENT_NAME = 'ag_ui.invalidate'

name the client matches on to route an invalidation to the host page.

A convention inside an extension point the protocol already provides, not a protocol extension: AG-UI defines the envelope and leaves name an open string. A client that does not know this name ignores the event, which is the graceful outcome and the whole reason the field is open.

A CUSTOM event rather than an ACTIVITY_SNAPSHOT, which is the opposite of the choice chart_activity makes -- and the difference is lifetime, not taste. @ag-ui/client materialises an activity into a role: "activity" message; the transport persists the message list wholesale, and the client's replay path re-fires it on every thread restore. That is right for a chart, which is content and should come back. An invalidation is an imperative: it has no place in the conversation and no meaning once acted on, so replaying it on every thread load would be a refetch storm.

ACTIVITY_SNAPSHOT is for content; CUSTOM is for an imperative.

STATE_SNAPSHOT is out for a third reason: shared state round-trips into the next RunAgentInput, so an invalidation placed there would be echoed back to the model as though it were something the model had said.

Internal helpers¶

These are not part of the public re-export surface but are referenced from the guides.

build_model ¶

build_model(model: str, *, api_key: str | None = None, provider: Any = None) -> Any

Build a Pydantic-AI model from a "provider:name" string and explicit key.

Prefix resolution is delegated to Pydantic-AI's infer_model, with a provider_factory that injects credentials rather than letting it read the environment. A provider instance takes precedence and is used as-is, so it may carry a custom base_url or client; otherwise api_key goes to the prefix's default Provider class. Every provider Pydantic-AI knows therefore works with no table to maintain here, and a bare model name it can map to a provider is accepted too.

Raises:

Type	Description
`ImproperlyConfigured`	The provider could not be resolved — an unknown or uninferable prefix, or its extra is not installed. Pass a `Provider` instance for anything Pydantic-AI cannot infer.

build_tool_catalog ¶

build_tool_catalog(
    registry: ToolRegistry,
    *,
    drf_mcp_server: Any = None,
    service_specs: dict[str, Any] | None = None,
) -> list[dict[str, Any]]

The agent's server-tool catalog for the frontend to label tool-call cards.

Server-side tools execute server-side, so their JSON Schema never reaches the browser and a web component cannot read an x-summary off it. This catalog is that channel: the component fetches it and maps tool name to label.

Each entry is {"name", "summary", "description"?}. summary is always present, resolved from @tool(summary=...) for registry tools and from display_name then title for drf-mcp ones, falling back to a prettified name. description is carried through when the source has one.

The catalog covers these sources and no others. A tool attached through a transport's capabilities= / toolsets= is not listed and its card falls back to a prettified name — a degraded label, not a broken call. Enumerating those is not possible here: pydantic-ai exposes tool names only through AbstractToolset.get_tools, which is async and needs a RunContext, while this runs at configuration time with no run in sight. Route spec tools through a transport's service_specs= to keep labels.

Parameters:

Name	Type	Description	Default
`registry`	`ToolRegistry`	The `@tool` registry, listed first; it wins name collisions.	required
`drf_mcp_server`	`Any`	A drf-mcp server whose registered tools are appended.	`None`
`service_specs`	`dict[str, Any] \| None`	A `name -> spec` mapping whose tools are appended, each described by its service or selector callable's docstring.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	One entry per tool, in that source order.

DRFMCPToolset ¶

Bases: AbstractToolset[Any]

Exposes a drf-mcp MCPServer's tools as a Pydantic-AI toolset.

Built per request, so the agent acts as the request's logged-in user. Both schemas and execution route through drf-mcp's public in-process surface (MCPServer.list_tools / acall_tool, drf-mcp 0.9+), so the advertised parameters, serializer validation and permissions match the HTTP transport exactly — without the network hop. Tool definitions carry the default kind="function", the in-process kind the run loop calls itself; an external tool would instead be deferred to the client and never run.

Failures split three ways, along MCP's protocol-vs-tool boundary:

JSON-RPC -32602 and tool-level validation_error results raise pydantic_ai.ModelRetry, so the model retries with the field errors instead of the run dying;
other tool-level failures (service_error / not_found) are returned as the tool's content, for the model to read;
protocol faults (auth, rate limits, an internal error) raise RuntimeError and abort the run.

Parameters:

Name	Type	Description	Default
`server`	`Any`	The drf-mcp `MCPServer` whose registry is bridged.	required
`request`	`HttpRequest`	The request carried into every call; its `user` is the acting user.	required
`exclude_names`	`frozenset[str]`	Names the `@tool` registry has already claimed. A colliding drf-mcp tool is skipped, so the registry wins — the rule `build_tool_catalog` applies — because pydantic-ai raises `UserError` for a duplicate name at run time.	`frozenset()`
`max_retries`	`int`	Per-tool retry budget: how many times a `ModelRetry` is fed back to the model before the run aborts. The default matches pydantic-ai's own function-tool default.	`1`

get_tools `async` ¶

get_tools(ctx: Any) -> dict[str, ToolsetTool[Any]]

Load tool defs from drf-mcp's tools/list once, then wrap them.

build_attachment_toolset ¶

build_attachment_toolset(
    store: AttachmentStore,
    request: HttpRequest,
    *,
    inline: AttachmentInlineConfig | None = None,
) -> FunctionToolset[None]

Build a per-request toolset exposing read_attachment over store.

The request is captured in a closure so store.open stays owner-scoped to the acting user: the model can only read files that user uploaded, never another's by id. Attachments therefore travel the wire as lightweight refs and the bytes are reached here, server-side, only when the model asks.

inline decides which binary types come back as file content and how large a file may be before it is described rather than attached; None takes the AttachmentInlineConfig defaults. Read its docstring before widening either — inlined bytes go to the provider on every model request left in the run.

API reference¶

Registry¶

ToolRegistry ¶

register ¶

get ¶

call ¶

acall async ¶

tool ¶

build_input_schema ¶

ToolSpec dataclass ¶

name instance-attribute ¶

fn instance-attribute ¶

description instance-attribute ¶

destructive class-attribute instance-attribute ¶

category class-attribute instance-attribute ¶

confirm class-attribute instance-attribute ¶

summary class-attribute instance-attribute ¶

ToolBinding dataclass ¶

ToolCategory ¶

X_DESTRUCTIVE_KEY module-attribute ¶

X_CATEGORY_KEY module-attribute ¶

X_CONFIRM_KEY module-attribute ¶

X_SUMMARY_KEY module-attribute ¶

Skills¶

SkillRegistry ¶

register ¶

add ¶

payload ¶

SkillSpec dataclass ¶

name instance-attribute ¶

title instance-attribute ¶

prompt class-attribute instance-attribute ¶

description class-attribute instance-attribute ¶

send_immediately class-attribute instance-attribute ¶

chip class-attribute instance-attribute ¶

SkillsView ¶

Agent and view¶

AGUIServer ¶

urls property ¶

DjangoAGUIView ¶

AgentSession ¶

stream ¶

ToolsView ¶

build_agent ¶

AgentConfig dataclass ¶

model instance-attribute ¶

instructions class-attribute instance-attribute ¶

audit_logger class-attribute instance-attribute ¶

audit_ip_address class-attribute instance-attribute ¶

model_settings class-attribute instance-attribute ¶

retries class-attribute instance-attribute ¶

toolsets class-attribute instance-attribute ¶

capabilities class-attribute instance-attribute ¶

tool_guard class-attribute instance-attribute ¶

tool_failure class-attribute instance-attribute ¶

AgentFactoryFn ¶

DEFAULT_SYSTEM_PROMPT module-attribute ¶

Configuration¶

AGUIConfig dataclass ¶

model instance-attribute ¶

api_key class-attribute instance-attribute ¶

system_prompt instance-attribute ¶

model_settings instance-attribute ¶

retries instance-attribute ¶

attachment_max_bytes instance-attribute ¶

attachment_allowed_types instance-attribute ¶

attachment_inline instance-attribute ¶

manage_system_prompt instance-attribute ¶

allow_uploaded_files instance-attribute ¶

forward_reasoning instance-attribute ¶

transcription_max_bytes instance-attribute ¶

transcription_allowed_types instance-attribute ¶

thread_list_limit instance-attribute ¶

run_list_limit instance-attribute ¶

tool_guard instance-attribute ¶

approval_prompts instance-attribute ¶

tool_failure instance-attribute ¶

run_context instance-attribute ¶

RunContextConfig dataclass ¶

client_context instance-attribute ¶

acall `async` ¶

ToolSpec `dataclass` ¶

name `instance-attribute` ¶

fn `instance-attribute` ¶

description `instance-attribute` ¶

destructive `class-attribute` `instance-attribute` ¶

category `class-attribute` `instance-attribute` ¶

confirm `class-attribute` `instance-attribute` ¶

summary `class-attribute` `instance-attribute` ¶

ToolBinding `dataclass` ¶

X_DESTRUCTIVE_KEY `module-attribute` ¶

X_CATEGORY_KEY `module-attribute` ¶

X_CONFIRM_KEY `module-attribute` ¶

X_SUMMARY_KEY `module-attribute` ¶

SkillSpec `dataclass` ¶

name `instance-attribute` ¶

title `instance-attribute` ¶

prompt `class-attribute` `instance-attribute` ¶

description `class-attribute` `instance-attribute` ¶

send_immediately `class-attribute` `instance-attribute` ¶

chip `class-attribute` `instance-attribute` ¶

urls `property` ¶

AgentConfig `dataclass` ¶

model `instance-attribute` ¶

instructions `class-attribute` `instance-attribute` ¶

audit_logger `class-attribute` `instance-attribute` ¶

audit_ip_address `class-attribute` `instance-attribute` ¶

model_settings `class-attribute` `instance-attribute` ¶

retries `class-attribute` `instance-attribute` ¶

toolsets `class-attribute` `instance-attribute` ¶

capabilities `class-attribute` `instance-attribute` ¶

tool_guard `class-attribute` `instance-attribute` ¶

tool_failure `class-attribute` `instance-attribute` ¶

DEFAULT_SYSTEM_PROMPT `module-attribute` ¶

AGUIConfig `dataclass` ¶

model `instance-attribute` ¶

api_key `class-attribute` `instance-attribute` ¶

system_prompt `instance-attribute` ¶

model_settings `instance-attribute` ¶

retries `instance-attribute` ¶

attachment_max_bytes `instance-attribute` ¶

attachment_allowed_types `instance-attribute` ¶

attachment_inline `instance-attribute` ¶

manage_system_prompt `instance-attribute` ¶

allow_uploaded_files `instance-attribute` ¶

forward_reasoning `instance-attribute` ¶

transcription_max_bytes `instance-attribute` ¶

transcription_allowed_types `instance-attribute` ¶

thread_list_limit `instance-attribute` ¶

run_list_limit `instance-attribute` ¶

tool_guard `instance-attribute` ¶

approval_prompts `instance-attribute` ¶

tool_failure `instance-attribute` ¶

run_context `instance-attribute` ¶

RunContextConfig `dataclass` ¶

client_context `instance-attribute` ¶

attachment_manifest `instance-attribute` ¶

max_chars `instance-attribute` ¶

delivery `class-attribute` `instance-attribute` ¶

ContextDelivery `module-attribute` ¶

AuditEvent `dataclass` ¶

organization_id `class-attribute` `instance-attribute` ¶

target_type `class-attribute` `instance-attribute` ¶

target_id `class-attribute` `instance-attribute` ¶

ip_address `class-attribute` `instance-attribute` ¶

Conversation `dataclass` ¶

ConversationMeta `dataclass` ¶

list `async` ¶

ThreadActivity `dataclass` ¶

AttachmentRef `dataclass` ¶

OpenedAttachment `dataclass` ¶

model `class-attribute` `instance-attribute` ¶

base_url `class-attribute` `instance-attribute` ¶

api_key `class-attribute` `instance-attribute` ¶

timeout `class-attribute` `instance-attribute` ¶

ChartSpec `dataclass` ¶

title `class-attribute` `instance-attribute` ¶

metadata `class-attribute` `instance-attribute` ¶

ChartSeries `dataclass` ¶

ChartKind `module-attribute` ¶