Pydantic AI (`edri.integrations.pydantic_ai`)¶

Install with pip install edri[pydantic-ai].

A bridge between Pydantic AI agents and EDRI's Worker model. It lets an agent's tools call other EDRI services as plain Events and can publish a bounded, privacy-controlled action stream for operational inspection. See Pydantic AI: a support agent with tools that dispatch to a different Manager for a full, runnable example — cross-Manager dispatch, parallel tool calls, the strict gotcha below in context.

Core pieces:

Class	Role
`AgentDeps[ResourcesT]`	Runtime deps handed to tools: `dispatch` (the door back into the router) + arbitrary `resources`.
`AgentBase[DepsT, OutputT]`	Wraps a `pydantic_ai.Agent`; domain methods call `self._run(...)` so deps never leak into their signatures.
`BaseAgentWorker[EventT]`	A `WorkerThread` that builds the declared agents, bakes their deps in, and implements `dispatch` — concurrent dispatches from the same worker run in parallel, each correlated to its own reply via `event._id`.
`ToolResponse`	Base for the responses your tool-target events reply with: a `tool_result` property says what value a dispatching tool receives.
`AgentAction`	Local EDRI event describing one START or terminal lifecycle point. Any ordinary Manager can consume it.
`AgentInspectorManager`	Optional bounded in-memory projection with query events and subscribed updates.

Your model/provider construction lives in each agent's build_agent().

How it fits together¶

An agent declares its model and tools and exposes domain methods.
A worker lists the agents it uses (agents), builds them in setup(), and runs the orchestration in do(); agents are reachable via self.agent[SomeAgent]. A worker isn't limited to one agent — one hosting several agents on different models gives each its own entry in agent_configs = {FastAgent: {"model_name": "..."}, DeepAgent: {"model_name": "..."}}; override agent_kwargs(agent_cls) instead when kwargs must be computed at runtime (env lookups, secrets, ...) — agent_configs is just its default implementation's lookup table.
A tool that needs another service calls deps.dispatch(SomeRequestEvent()) and gets back the reply's tool_result. Any Event works — no special base class — as long as its reply is a ToolResponse; a reply that isn't one, or an event with no declared response at all, raises a clear error instead of hanging. dispatch takes a keyword-only timeout (default 30s) per call — a tool known to hit a slow downstream Worker can afford a longer wait, one that should fail fast can ask for less. This also works if the tool-target event is handled by a different Manager's Worker: ManagerBase tracks the pending reply generically, keyed on Event._id (universal on every Event), and forwards it back to the dispatching worker's pipe once it arrives — no per-type registration or manual forwarding handler needed on either Manager.
dispatch does not serialize concurrent calls: pydantic-ai can run a turn's tool calls from multiple threads (e.g. with model_settings={"parallel_tool_calls": True}, letting the model call several different tools in one turn), so more than one dispatch can be in flight on a worker's pipe at once — each is correlated to its own reply via event._id, so unrelated dispatches never block each other.
Agents are built in setup(), which the base Worker runs before do() inside its own error-handling scope — so don't override run(); put one-time preparation in setup() (call super().setup()) and the orchestration in do().

See Pydantic AI: a support agent with tools that dispatch to a different Manager for all of this worked out in a full, runnable, cross-Manager example.

Observability and audit actions¶

Set EDRI_PYDANTIC_AI_AUDIT before starting workers:

Value	Published data
`off`	No `AgentAction` events and no observability hook overhead. This is the default.
`metadata`	Identity, timing, outcome, model labels, usage counters and bounded structural error fields.
`content`	Metadata plus only the content categories allowed by `AgentActionPolicy`.

The environment value is a deployment ceiling. Application policy can remove content but cannot enable content when the environment permits metadata only. For example:

from edri.integrations.pydantic_ai import (
    AgentActionPolicy,
    BaseAgentWorker,
    NormalizationLimits,
)


class SupportWorker(BaseAgentWorker):
    def agent_action_policy(self, agent_cls):
        return AgentActionPolicy(
            include_tool_arguments=True,
            include_tool_results=True,
            include_error_messages=True,
            redactor=redact_secrets,
            limits=NormalizationLimits(max_total_bytes=32 * 1024),
        )

Redaction receives an already detached and bounded value. Its result is bounded again. A failing redactor drops the affected content; it never changes the agent result. Content capture and OpenTelemetry instrumentation are independent: enabling either one does not enable or configure the other.

Every logical operation has an action_id; attempts are one-based and pair a START with END or ERROR. root_action_id groups the complete run and parent_action_id forms its tree. Agent runs, model requests, tool validation, tool execution and EDRI dispatch are observed. Usage is copied from supported Pydantic AI counters at each terminal point; consumers decide whether to show per-operation values or aggregate them, and must avoid double-counting nested usage.

Cancellation, timeout and deferral are terminal outcomes rather than failures. A process lost outside an observable seam can leave a START without a terminal; the inspector reports that run as partial instead of inventing an outcome.

Custom consumers and migration¶

AgentBase.build_audit_json() was removed. It produced one ad-hoc dictionary after a run and could not describe retries, concurrent tools, partial runs or dispatches. Consume AgentAction through a normal Manager instead:

from edri.abstract import ManagerBase, request
from edri.integrations.pydantic_ai import AgentAction, AgentActionPhase


class AuditSink(ManagerBase):
    @request
    def record(self, event: AgentAction) -> None:
        durable_store.append({
            "action_id": str(event.action_id),
            "root_action_id": str(event.root_action_id),
            "attempt": event.attempt,
            "phase": event.phase.value,
            "outcome": event.outcome.value if event.outcome else None,
            "usage": event.usage,
        })

Register AuditSink() with EDRI.add_component() like any other manager. Actions are local and non-shareable: they do not cross Switch automatically. A durable consumer can persist what reaches it, but cannot recover an action lost before delivery. The stream is therefore operational evidence, not a durable audit log or an acknowledged transport.

Inspector¶

Add AgentInspectorManager() when the application needs bounded recent history:

from edri.integrations.pydantic_ai import AgentInspectorManager, StoreLimits

app.add_component(AgentInspectorManager(
    limits=StoreLimits(max_runs=500, max_actions=10_000),
))

The manager exposes agent-inspector-runs, agent-inspector-run, agent-inspector-action and agent-inspector-stats through the application's existing Listener, Broker and middleware. Its defaults are finite. Eviction removes only the inspector's projection; it cannot delete another consumer's copy. The stats response reports retained counts, incomplete runs, trimming and eviction.

Live clients register for agent-inspector-update. No registration means no updates. An empty param_set subscribes to every change in the store; a root_action_id predicate narrows the feed to one run. The safe handshake is:

Start a bounded client-side update buffer.
Send register and wait for its acknowledgement.
Request agent-inspector-run as the snapshot.
Apply buffered updates from the same generation_id with revisions greater than the snapshot revision, in order.
Fetch a new snapshot after a revision gap, generation change, invalidation or buffer overflow.

Query and registration events pass through configured middleware, so protect both. Extra subscription predicates are useful for a selective feed, but they intentionally create revision gaps and therefore cannot maintain a complete run view.

Dashboard¶

GET /edri/agents answers JSON to a REST client and an HTML page to a browser. The page is served by agent-inspector-runs itself, so it is reachable exactly where the listing is: an application that authorizes the query authorizes the page, one whose middleware refuses it refuses both, and one that does not run AgentInspectorManager has no page.

The page is a shell. It carries no run data, opens a WebSocket back to the application, registers for agent-inspector-update with an empty param_set and then answers itself out of the four query events -- the listing on connection and after a filter change, the run tree when a row is opened, the retained content when an action is expanded, and the store statistics on a timer. Updates patch the open run while revisions stay consecutive within a generation, and a gap, a generation change or an invalidation makes it re-read the snapshot. A dropped connection is retried with a backoff, and each reconnection repeats the handshake from the start.

Everything it displays comes from a model run, so it is written to the DOM as text and never as markup.

MCP coverage¶

Install the pydantic-ai extra, which includes Pydantic AI's MCP support, and wire the public callback into the application's own toolset:

from pydantic_ai.mcp import MCPToolset
from edri.integrations.pydantic_ai import audited_tool_calls

toolset = MCPToolset(
    client,
    id="inventory",
    process_tool_call=audited_tool_calls("inventory"),
)

On the supported Pydantic AI range (>=2.41,<3), this observes tools/call made inside an observed agent run. Direct MCPToolset.direct_call_tool() calls and protocol operations initialize, tools/list, resources/* and prompts/* have no public observation seam and are not reported. EDRI does not replace an existing callback implicitly. The resolved MCP SDK 2 server replaces server exception messages with a generic tool error before the callback, so those original details cannot appear in the action stream.

MCP is unavailable on Python 3.15. pydantic_ai.mcp raises an ImportError asking for the fastmcp client there even when fastmcp is installed; the cause is beartype, a transitive dependency of the mcp extra, which up to and including 0.22.9 does not import on 3.15. Nothing else in the integration is affected.

Pydantic AI capabilities¶

Pydantic AI capabilities are compatible with this integration. A capability tool receives the same RunContext[AgentDeps[ResourcesT]] as a tool registered directly on an agent, so it can dispatch ordinary EDRI events through ctx.deps.dispatch(...).

Capabilities are constructed and registered by the application's build_agent() method. EDRI does not automatically forward an AgentBase.capabilities attribute to pydantic_ai.Agent; pass the capabilities explicitly when constructing the agent:

from typing import Any

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import Capability

from edri.integrations.pydantic_ai import AgentBase, AgentDeps


orders = Capability[AgentDeps[Any]](
    id="orders",
    description="Look up and list customer orders.",
)


@orders.tool(strict=False)
def lookup_order_status(
    ctx: RunContext[AgentDeps[Any]],
    order_id: str,
) -> Any:
    return ctx.deps.dispatch(OrderLookup(order_id=order_id))


class SupportAgent(AgentBase):
    capabilities = [orders]

    def build_agent(self) -> Agent:
        return Agent(
            self.model,
            deps_type=AgentDeps,
            capabilities=self.capabilities,
        )

Declaring capabilities = [...] on AgentBase is only an application-level convention. The capability becomes active only after it is passed to Agent(capabilities=...).

Capability tools can use all values exposed by AgentDeps:

@orders.tool
def lookup_order_status(
    ctx: RunContext[AgentDeps[AppResources]],
    order_id: str,
) -> Any:
    audit = ctx.deps.resources.audit_logger
    audit.info("Looking up order %s", order_id)

    return ctx.deps.dispatch(
        OrderLookup(order_id=order_id),
        timeout=5.0,
    )

The dispatched event has the same requirements as a dispatch from a regular agent tool:

the event must declare a response;
the response must subclass ToolResponse;
concurrent capability tool calls are correlated independently by Event._id;
dispatch can cross Manager boundaries without capability-specific wiring.

Deferred capabilities¶

Capabilities declared with defer_loading=True also work with EDRI:

orders = Capability[AgentDeps[Any]](
    id="orders",
    description="Use for questions about customer orders.",
    defer_loading=True,
)

Pydantic AI initially exposes the capability through its capability-loading mechanism and registers its instructions and tools only after the model loads it. Once loaded, its tools receive the same AgentDeps and can call dispatch(...) normally.

Be aware that deferred capabilities introduce framework-managed tools such as load_capability and, depending on the model/provider, search_tools. These tools may themselves be emitted with strict: true when their JSON schemas qualify for strict mode.

On inference backends affected by the strict-schema decoding issue described below, setting strict=False only on the capability's domain tools might not be sufficient: decoding may hang while calling load_capability, before the domain tool or EDRI dispatch is reached.

For such backends, either avoid deferred loading or configure the Pydantic AI model profile so strict tool definitions are not advertised, where supported by the selected model/provider.

Per-run state¶

Avoid storing mutable per-run state directly on a shared module-level Capability or in capability tool closures. For values such as audit information, request metadata, or application services, use AgentDeps.resources.

Similarly, flags such as tool_used are better derived from the completed Pydantic AI run or stored in request-scoped resources than mutated on a shared agent or capability object. This prevents state from leaking between concurrent runs.

Tool schemas and `strict`¶

pydantic-ai auto-promotes a tool's JSON schema to strict: true when its shape qualifies (roughly: simple required parameters, no unions/optionals). This constrains the model's output via the backend's guided/grammar decoding. Some inference backends (observed with vLLM + xgrammar, particularly with speculative decoding and reasoning models) can hang indefinitely — no exception, no timeout, no log line — decoding a strict schema for a tool call the model is not confident about (e.g. no valid argument value is available yet). This happens inside Agent.run_sync(), before dispatch is ever called, so nothing in this module can catch or time it out.

If a tool's calls hang under load with a particular backend, pass strict=False explicitly when registering it:

@agent.tool(strict=False)
def lookup_order_status(ctx: RunContext[AgentDeps], order_id: str) -> Any:
    ...

Pydantic AI (edri.integrations.pydantic_ai)¶