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

Install with `pip install edri[pydantic-ai]`.

A bridge between [Pydantic AI](https://ai.pydantic.dev/) agents and EDRI's
{doc}`/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 {doc}`/examples/pydantic_ai/support_agent`
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)=
## 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
  {doc}`/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 {doc}`/examples/pydantic_ai/support_agent`
for all of this worked out in a full, runnable, cross-Manager example.

(agent-observability)=
## 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:

```python
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:

```python
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:

```python
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:

1. Start a bounded client-side update buffer.
2. Send `register` and wait for its acknowledgement.
3. Request `agent-inspector-run` as the snapshot.
4. Apply buffered updates from the same `generation_id` with revisions greater
   than the snapshot revision, in order.
5. 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:

```python
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](https://ai.pydantic.dev/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:

```python
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`:

```python
@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:

```python
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)=
## 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:

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