edri.integrations.pydantic_ai package¶

Submodules¶

edri.integrations.pydantic_ai.agent module¶

Agent-facing half of the Pydantic AI bridge.

AgentDeps and ToolResponse are the contract between an agent's tools and the worker hosting them; AgentBase is what an application subclasses to own domain behaviour without touching runtime concerns. The worker side lives in edri.integrations.pydantic_ai.worker.

class edri.integrations.pydantic_ai.agent.AgentBase(*, model_name: str, name: str | None = None)[source]¶

Bases: ABC, Generic

Base class for EDRI-hosted Pydantic AI agents.

An agent owns only its domain behavior: model/provider construction (build_agent()), prompt, output configuration and domain methods. Runtime concerns (the EDRI pipe, shared resources) arrive as AgentDeps, which the worker bakes in via bind_deps() after construction. Tools reach them through RunContext[AgentDeps], so domain methods stay clean and never expose deps in their signatures.

bind_deps(deps: AgentDeps[Any]) → AgentBase[DepsT, OutputT][source]¶

Bake runtime dependencies into the agent (called once by the worker).

Returns self to allow fluent chaining.

abstractmethod build_agent() → Agent[DepsT, OutputT][source]¶

Build the underlying Pydantic AI agent.

Implementations construct the model/provider and return a configured Agent. Pass deps_type=AgentDeps (and your capabilities) so tools can resolve runtime dependencies.

capabilities: ClassVar[list[Any]] = []¶: Stateless, shareable capability/tool bundles. Tools read runtime values from RunContext[AgentDeps]; capabilities themselves hold no per-run state.

classmethod default_name() → str[source]¶

Derive a stable snake_case agent name from the class name.

A trailing Agent is stripped: ClaimDebateJudgeAgent -> claim_debate_judge.

framework: str = 'pydantic_ai'¶

model_version: str | None = None¶

output_mode: str = 'unknown'¶

prompt_version: str | None = None¶

provider: str = 'unknown'¶

version: str | None = None¶

class edri.integrations.pydantic_ai.agent.AgentDeps(dispatch: DispatchFn, resources: ResourcesT | None = None)[source]¶

Bases: Generic

Runtime dependencies injected into an agent by its worker.

Pydantic AI tools receive this as RunContext[AgentDeps].deps. Agents never construct or inspect it — the worker builds one and bakes it in via AgentBase.bind_deps().

Attributes:

dispatch -- Sends a request Event to the EDRI router and returns the tool result extracted from the response. Implemented by the worker (BaseAgentWorker._dispatch_event()); blocking for the caller, but concurrent dispatches from the same worker don't block each other. This is the only path from a tool back into the rest of the system. Pass timeout= to override the 30s default for a given call, e.g. ctx.deps.dispatch(SlowLookup(...), timeout=5.0).
resources -- Optional worker-level shared state (in-memory stores, DB handles, caches). Parametrize AgentDeps[MyResources] for typing.

resources: ResourcesT | None = None¶

class edri.integrations.pydantic_ai.agent.DispatchFn(*args, **kwargs)[source]¶

Bases: Protocol

Callable shape of AgentDeps.dispatch.

timeout is keyword-only so each call site (i.e. each tool) can override the default wait for its own dispatched event — a tool known to hit a slow downstream Worker can afford a longer wait; one that should fail fast can ask for less than the 30s default.

class edri.integrations.pydantic_ai.agent.ToolResponse[source]¶

Bases: Response, ABC

Base for responses that an agent tool consumes via dispatch().

A tool calls deps.dispatch(SomeRequest()) and gets back whatever the reply's response exposes as tool_result. Subclass this on the responses your tool-target events use: the response keeps its own typed fields for normal callers and additionally declares which value a tool receives — so the worker needs no extraction hook.

tool_result is a property, not a field, so it never appears in the serialized payload; implement it to return an existing field (or a derived, JSON-friendly view of several).

abstract property tool_result: Any¶: The value handed to the dispatching agent tool.

edri.integrations.pydantic_ai.inspector module¶

The inspector manager: a standard EDRI manager that watches agent runs.

It consumes AgentAction through ordinary routing and answers queries about what it has seen. An application includes it like any other component; the environment switch decides whether workers emit actions, this manager decides whether EDRI also offers the standard browsing experience over them.

What it deliberately does not do is own anything on an agent's path. It holds no agent workers, no Pydantic AI objects and no queue a run waits on, so an inspector that is slow, wedged or entirely absent cannot slow a run down or fail one. The worst it can do to an agent is forget what the agent did.

class edri.integrations.pydantic_ai.inspector.AgentInspectorManager(router_queue: Any | None = None, logger: Logger | None = None, from_time: datetime | None = None, max_workers: int = -1, *, limits: StoreLimits | None = None, log_actions: bool = False, publish_updates: bool = True, clock: Callable[[], datetime] | None = None)[source]¶

Bases: ManagerBase

Bounded, in-memory history of observed agent activity.

limits is the application's retention choice; log_actions turns on a concise human-readable line per action, off by default because an application that wanted the query API did not necessarily ask for its log to double in size.

publish_updates is on because live inspection is what the manager is for, and the broker sends an update only to clients whose subscription filter matches it -- no browser is fanned the whole stream. What it does cost is one router message per observed action. An application that wants the query API and nothing else turns it off and pays nothing.

resolve(event: Event) → None[source]¶

Resolve one event, then publish what the store noticed doing it.

Retention by age has no arrival to hang off: nothing reaches an idle store, so a run that timed out is discovered by whichever read ran next -- a query, or even a refused one. Draining here rather than in each handler means no read can quietly swallow an invalidation a subscriber is waiting for.

solve_req_get_action(event: AgentInspectorAction) → None¶: One action in the detail the store still holds.

solve_req_get_run(event: AgentInspectorRun) → None¶: One run and its action tree.

solve_req_get_stats(event: AgentInspectorStats) → None¶: What the store holds and what retention has already removed.

solve_req_list_runs(event: AgentInspectorRuns) → None¶: A page of run summaries, newest first.

solve_req_observe(event: AgentAction) → None¶

Take one observed action into the store.

The action itself is never sent on -- publishing the observed event back would feed the stream into its own input. What does go back is a subscribed update per state change, which is a different event.

edri.integrations.pydantic_ai.inspector_events module¶

The inspector's query surface, as ordinary EDRI API events.

These are plain @api events: they reach the inspector through the same broker, the same registration and the same middleware as every other event an application exposes. There is no inspector-specific web server and no bypass endpoint -- an application that authenticates its API authenticates these too.

Request fields stay scalar on purpose. The global @api validator admits scalars, UUIDs, timestamps and lists of them, and that is exactly what a filterable query needs; the nested trees the inspector returns live in the response, which the serializer already carries.

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorAction(action_id: UUID | None = None)[source]¶

Bases: Event

One action, with the attempts and captured content the store kept.

action_id: UUID | None = None¶

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorActionResponse[source]¶

Bases: Response

One action in full retained detail.

action: dict[str, Any] | None = None¶

error: str | None = None¶

found: bool = None¶

generation_id: str | None = None¶

revision: int | None = None¶

root_action_id: str | None = None¶

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorRun(root_action_id: UUID | None = None, run_id: str | None = None)[source]¶

Bases: Event

One run, addressed by the root action that identifies it.

root_action_id: UUID | None = None¶

run_id: str | None = None¶: An alternative for a caller that only knows the upstream id. Ambiguous by nature, and answered as ambiguous rather than by picking one.

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorRunResponse[source]¶

Bases: Response

One reconstructed run with its action tree.

error: str | None = None¶

found: bool = None¶: False means the store does not have this run -- never seen, or evicted. It is not the same answer as a run that is present and empty.

generation_id: str | None = None¶

revision: int | None = None¶

run: dict[str, Any] | None = None¶

Bases: Event

Recent runs, filtered and paged.

A browser asking for HTML gets the inspector dashboard instead of the listing: one page that opens a WebSocket, subscribes to AgentInspectorUpdate and then answers itself out of the four query events. It is the same event either way, so an application that authorizes the listing authorizes the page, and one that never routes this event has no page to reach.

agent: str | None = None¶

errors_only: bool | None = None¶

limit: int | None = None¶

offset: int | None = None¶

run_id: str | None = None¶: The upstream identifier. It may match several runs -- a caller can hand one run_id to two invocations -- so it filters rather than selects.

since: datetime | None = None¶

until: datetime | None = None¶

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorRunsResponse[source]¶

Bases: Response

A page of run summaries, newest first.

error: str | None = None¶

limit: int = 50¶

offset: int = None¶

total: int = None¶

a client paging needs to know what it has not seen.

Type:: Matching runs, not returned runs

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorStats[source]¶

Bases: Event

Retention and store statistics.

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorStatsResponse[source]¶

Bases: Response

What the store holds, and what it has already dropped.

actions: int = None¶

attempts: int = None¶

bytes: int = None¶

dedupe_entries: int = None¶

evicted_runs: int = None¶

incomplete_runs: int = None¶

runs: int = None¶

trimmed_runs: int = None¶

Bases: Event

One store change, delivered to whoever registered for it.

Every field here exists to be filtered on. The broker compares a subscription's param_set against top-level event fields, so the keys a client is expected to narrow by -- the run above all -- are carried flat and typed, and the change itself travels in the response.

Only root_action_id, generation_id, revision and kind are filled on every update. The rest describe the action that caused the change and are absent on an invalidation, which no action caused. A client that filters on one of them therefore receives a view of the run rather than its complete history -- which is a legitimate feed, but not the stream the snapshot handshake detects loss on.

action_id: UUID | None = None¶: The action the change is about, absent on an invalidation.

agent: str | None = None¶

edri_event_id: UUID | None = None¶

an application watching what its own request caused filters on this.

Type:: The business event that started the run, when one did

generation_id: UUID | None = None¶: A revision is only comparable within its generation. A run evicted and later rebuilt starts counting again under a new generation_id.

kind: ChangeKind | None = None¶

namespace: str | None = None¶

operation: str | None = None¶

phase: AgentActionPhase | None = None¶

revision: int | None = None¶

root_action_id: UUID | None = None¶

the one filter a browsing client is always expected to set.

Type:: The run this change belongs to

run_id: str | None = None¶

class edri.integrations.pydantic_ai.inspector_events.AgentInspectorUpdateResponse[source]¶

Bases: Response

What changed, in the detail a live view needs to redraw.

The rich body lives here rather than on the event because @api request fields are restricted to API-safe scalars. That restriction is the reason the filterable keys above are flat: a subscription predicate can only ever compare a top-level field, so anything a client filters on has to be one.

action: dict[str, Any] | None = None¶: The one action this change touched, without captured content. A live view redraws a node from it; the detail query is where content lives.

kind: str = 'updated'¶: updated replaces; invalidated says the run is gone from the store and a client holding it should stop trusting what it has.

run: dict[str, Any] | None = None¶: The run summary as the listing would render it, or None for an invalidation -- there is no longer a run to summarize.

edri.integrations.pydantic_ai.inspector_store module¶

Bounded reconstruction of agent history from the action stream.

The store turns AgentAction events back into the runs, actions and attempts they describe. It is development history, not an audit log: it is bounded, it is in memory, and it is allowed to forget. What it is not allowed to do is pretend.

Three properties follow from that, and most of this module exists to keep them:

Nothing is invented. An attempt with no terminal event stays unfinished; an action whose parent was never seen keeps parent_missing; an action that names no run is not filed under one. Absence is reported as absence.
Duplicates are free. EDRI delivers an event once, but a store that advanced its revision on a replay would hand a subscriber a change that never happened. Re-applying one Event._id is a no-op.
Forgetting is visible. Evicting attempts marks the surviving action truncated; evicting a run invalidates it, so a client cannot read the next generation's low revision as an old update.

Run identity is root_action_id, never the upstream run_id: one run_id may be handed to two concurrent invocations, and an action's parent may be missing exactly when it is needed.

class edri.integrations.pydantic_ai.inspector_store.ActionState(action_id: UUID, root_action_id: UUID, parent_action_id: UUID | None = None, namespace: str = '', operation: str = '', name: str | None = None, agent: str | None = None, model: str | None = None, provider: str | None = None, metadata: dict[str, ~typing.Any]=<factory>, attempts: dict[int, ~edri.integrations.pydantic_ai.inspector_store.AttemptState]=<factory>, truncated: bool = False, bytes: int = 0)[source]¶

Bases: object

One logical operation, with every attempt of it the store still holds.

measure() → int[source]¶

property start_missing: bool¶

property terminal_missing: bool¶

class edri.integrations.pydantic_ai.inspector_store.AttemptState(attempt: int, started_at: datetime | None = None, ended_at: datetime | None = None, phase: AgentActionPhase | None = None, outcome: AgentActionOutcome | None = None, duration: float | None = None, usage: dict[str, Any] | None = None, request: Any = None, response_data: Any = None, error: dict[str, Any] | None = None, bytes: int = 0)[source]¶

Bases: object

One execution attempt of one logical action.

START and the terminal event are separate routed events and either may be absent, so both halves are optional and their absence is a reported flag rather than a reason to drop the attempt.

measure() → int[source]¶

property start_missing: bool¶

property terminal_missing: bool¶

class edri.integrations.pydantic_ai.inspector_store.ChangeKind(*values)[source]¶

Bases: Enum

What a consumer must do with a change record.

UPDATED carries a revision a subscriber can compare against what it holds. INVALIDATED says the run's retained state is gone -- evicted, expired or cleared -- so comparing revisions is meaningless and the only correct response is to stop trusting what is held.

INVALIDATED = 'invalidated'¶

UPDATED = 'updated'¶

class edri.integrations.pydantic_ai.inspector_store.InspectorStore(*, limits: StoreLimits | None = None, clock: Callable[[], ~datetime.datetime]=<function _utc_now>)[source]¶

Bases: object

Bounded in-memory history of observed agent runs.

Mutation is synchronous on the caller's thread: the manager owns the store and ingests on its own event loop, so there is no background thread and no lock. clock is injected because retention is a function of time and tests must control it without sleeping.

action(action_id: UUID) → tuple[RunState, ActionState] | None[source]¶: One action and the run it belongs to, both detached.

clear() → tuple[StoreChange, ...][source]¶: Forget everything, invalidating every run a client may be holding.

drain_changes() → tuple[StoreChange, ...][source]¶

Take the invalidations the store has accumulated.

A query can expire a run, and a client holding that run has to be told. The manager drains after serving a query for exactly that reason.

ingest(action: AgentAction) → tuple[StoreChange, ...][source]¶

Apply one observed action and report what changed.

Returns one UPDATED record when the run changed, nothing when the event was already applied, and additionally an INVALIDATED record for anything retention removed while making room.

run(root_action_id: UUID) → RunState | None[source]¶: A detached snapshot of one run, or None if it is not retained.

Matching runs, newest first, and how many matched in total.

The count is of matches, not of the page: a client paging through needs to know what it has not seen yet. Every returned run is a detached copy.

stats() → StoreStats[source]¶: What is retained now, and what retention has already removed.

class edri.integrations.pydantic_ai.inspector_store.RunState(root_action_id: UUID, generation_id: UUID, revision: int = 0, run_id: str | None = None, conversation_id: str | None = None, agent: str | None = None, created_at: datetime | None = None, updated_at: datetime | None = None, actions: dict[~uuid.UUID, ~edri.integrations.pydantic_ai.inspector_store.ActionState]=<factory>, truncated: bool = False, event_ids: set[UUID] = <factory>, bytes: int = 0)[source]¶

Bases: object

One agent invocation: its identity, its actions and its revision.

property completed: bool¶

property failed: bool¶

Something in this run ended badly.

A timed-out or errored action counts even when the run itself finished: a retry that eventually succeeded still describes a run an operator looking for failures wants to find. Cancellation and deferral are not failures -- they are how a run ends when something asked it to.

measure() → int[source]¶

property outcome: AgentActionOutcome | None¶

property partial: bool¶

Something this run should contain is not here.

Either the run itself was never fully observed, or one of its actions was not -- a missing START, a missing terminal, a parent that never arrived, or history retention removed.

property root: ActionState | None¶

class edri.integrations.pydantic_ai.inspector_store.StoreChange(kind: ChangeKind, root_action_id: UUID, generation_id: UUID, revision: int, action_id: UUID | None = None, run_id: str | None = None, agent: str | None = None)[source]¶

Bases: object

One thing that happened to the store, for the manager to publish.

generation_id travels with every record because revision means nothing without it: a run re-materialized after eviction restarts at 1, and a subscriber holding revision 40 of the previous generation must not read that as stale.

class edri.integrations.pydantic_ai.inspector_store.StoreLimits(max_runs: int = 256, max_actions: int = 8192, max_attempts_per_action: int = 32, max_bytes: int = 16777216, ttl: timedelta | None = None)[source]¶

Bases: object

What the store is allowed to keep.

Every limit applies to the whole retained state, auxiliary indexes included: a bookkeeping dict that grows while max_actions is still satisfied is the same leak, one level down.

class edri.integrations.pydantic_ai.inspector_store.StoreStats(runs: int, actions: int, attempts: int, bytes: int, incomplete_runs: int, dedupe_entries: int, evicted_runs: int, trimmed_runs: int, limits: StoreLimits)[source]¶

Bases: object

What the store currently holds, and what it has already lost.

evicted_runs is retention's own count. It is not a guess about actions the producer never published: the store cannot see those and does not pretend to.

edri.integrations.pydantic_ai.lifecycle module¶

Observing one run's model and tool seams.

The capability built here is what an observed run installs for itself: it holds the run's correlation state, so nothing is stored on the agent and two runs on one built agent cannot read each other's identity.

Every seam is a wrapper hook. A wrapper owns both ends of the operation it observes -- START, duration and exactly one terminal -- so pairing is a local variable rather than a table keyed on an upstream id, and there is no window in which a crash leaves a half-written pairing entry behind. That choice is forced for tools: wrap_tool_execute is the only hook that sees every terminal state, because a ModelRetry surfaces inside it as ToolRetryError and never reaches on_tool_execute_error.

What the seams deliberately do not do:

They never refuse, delay or alter an operation. An observation failure is contained (see _emit_safely()); an observed failure is re-raised unchanged, including the original exception instance.
They never merge two executions into one logical action. Upstream gives each retry of a tool call a fresh tool_call_id, so there is no stable identity to merge on, and inferring one from tool name, arguments or timing would silently fuse two concurrent calls of the same tool.
They never read an application-defined conversion. Content is projected from an explicit list of attributes and anything else is skipped, not stringified.

Limits of what these seams can see, measured against the supported floor:

A capability that refuses a tool from before_tool_execute runs outside the wrap_tool_execute frame, so such a tool produces no execution action at all -- not a START left without a terminal. The failure is still visible on the run.
A tool that exceeds its own upstream timeout arrives as a ToolRetryError whose message mentions a timeout. The type says nothing, so the attempt is reported as an ordinary error; TIMEOUT is reserved for an expiry that is visible as a type.
Losing the worker outside any seam leaves the attempt with no terminal action. On the supported Pydantic AI floor, a BaseException escaping a sync tool cancels the enclosing run task: the tool stays partial while the run records that observed cancellation.

edri.integrations.pydantic_ai.lifecycle.audit_hooks(state: _RunState, emitter: _ActionEmitter, policy: _EffectivePolicy) → pydantic_ai.capabilities.Hooks[source]¶

The audit capability for one run.

Registered per run through run_sync(capabilities=[...]), so it is not installed on the agent and does not persist beyond the invocation it observes.

edri.integrations.pydantic_ai.mcp module¶

Auditing the one MCP operation Pydantic AI exposes a public seam for.

Of the MCP protocol operations an agent performs, exactly one can be observed without reaching into private internals on the supported dependency floor: tools/call, through MCPToolset(process_tool_call=...). initialize, tools/list, resources/* and prompts/* round-trip correctly but offer no public observation point, so this module makes no claim about them -- and MCPToolset.direct_call_tool() bypasses the seam entirely, so an application calling a toolset directly is outside the observed set.

Installation is explicit, in the agent's own build_agent():

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

class WeatherAgent(AgentBase[AgentDeps, str]):
    def build_agent(self):
        toolset = MCPToolset(
            client, id="weather",
            process_tool_call=audited_tool_calls("weather"),
        )
        return Agent(self.model_name, deps_type=AgentDeps, toolsets=[toolset])

Explicit because the toolset is the application's object: EDRI does not reach into an agent's toolsets to instrument them, and an integration that silently replaced a user's own process_tool_call would be worse than one that asks for a line of wiring.

edri.integrations.pydantic_ai.mcp.audited_tool_calls(server: str | None = None) → ProcessToolCallback[source]¶

A process_tool_call callback that publishes one action per call.

server labels the server in the action stream. The seam is handed only the run context, the transport callable, a tool name and its arguments -- never the connection -- so the label cannot be discovered here; pass the same name the toolset carries as its id.

The returned callback is transparent. It awaits the real call, returns its result unchanged and re-raises its exceptions unchanged, including the ModelRetry that a server-side failure arrives as. A call made while no observed run is in progress -- a shared connection's own traffic -- is passed straight through: its attribution is unknown, and EDRI publishes nothing rather than guessing a run for it.

The return type is the real ProcessToolCallback, imported only under TYPE_CHECKING: pydantic_ai.mcp raises ImportError unless the mcp extra is installed, and this package must import without it.

edri.integrations.pydantic_ai.normalization module¶

Detach, bound and sanitize captured content before it is published.

This is the security boundary of the audit feature. Content is sanitized here, before the action reaches the router queue -- the inspector is not the boundary, and there is no "queue it raw and redact later" path.

The order is snapshot, then redact, then normalize again:

raw value -> bounded detached snapshot -> redactor -> bounded normalization

The application redactor therefore never receives a live business object, so destructive code such as dict.pop() can only damage the observability copy. The second pass exists because a redactor can return a new oversized, cyclic or otherwise unsupported structure even when its input was safe.

What makes the snapshot an isolation boundary rather than a shallow copy is that it accepts only exact built-in scalars and containers (plus UUID and datetime, converted by the stdlib). An unknown object is a fixed sentinel: the normalizer never calls repr, str, __iter__, keys, asdict, model_dump, JSON hooks or any other application-defined conversion on it, because those run arbitrary code -- on the agent's hot path, on data the operator may have chosen not to capture at all.

edri.integrations.pydantic_ai.normalization.BINARY: Final = '<binary>'¶

a size is still information about content.

Type:: Raw binary. Fixed, with no length

edri.integrations.pydantic_ai.normalization.CYCLE: Final = '<cycle>'¶: The value is its own ancestor.

class edri.integrations.pydantic_ai.normalization.NormalizationBudget(limits: NormalizationLimits | None = None)[source]¶

Bases: object

Mutable spend against one action's limits.

One budget is shared by every field of one action -- request, response, metadata, usage and error -- rather than each field getting the full allowance, so an action cannot multiply the limit by its field count.

fit_text(text: str) → str | None[source]¶

Return as much of text as fits, or None if nothing does.

Truncation is by characters, never by bytes: slicing an encoded string at a byte offset can split a codepoint, and the result has to stay valid text. The byte budget is then applied by shrinking the character count until the encoded form fits.

spend(byte_count: int = 8) → bool[source]¶: Charge one value; False means it does not fit and nothing is spent.

edri.integrations.pydantic_ai.normalization.REDACTION_FAILED: Final = '<redaction-failed>'¶: The application redactor raised. The original is not published instead.

edri.integrations.pydantic_ai.normalization.REQUEST_USAGE_FIELDS: Final[tuple[str, ...]] = ('input_tokens', 'output_tokens', 'cache_write_tokens', 'cache_read_tokens', 'input_audio_tokens', 'cache_audio_read_tokens', 'output_audio_tokens')¶: Counters of one model request, which is a different measurement from the run totals above: requests and tool_calls count whole-run activity and a single request has no opinion about them. Listed separately rather than filtered at the call site, so a future upstream release that adds a run-scoped counter to the per-request object cannot leak it into a snapshot that a consumer would then be tempted to sum with the run's own.

edri.integrations.pydantic_ai.normalization.TOO_DEEP: Final = '<max-depth>'¶: Nesting passed max_depth.

edri.integrations.pydantic_ai.normalization.TRUNCATED: Final = '<truncated>'¶: A budget ran out -- value count, string length or total bytes.

edri.integrations.pydantic_ai.normalization.UNSUPPORTED: Final = '<unsupported>'¶

an application object, a container subclass, a set, or a float that JSON cannot represent.

Type:: A value the normalizer will not expand

edri.integrations.pydantic_ai.normalization.USAGE_FIELDS: Final[tuple[str, ...]] = ('requests', 'tool_calls', 'input_tokens', 'output_tokens', 'cache_write_tokens', 'cache_read_tokens', 'input_audio_tokens', 'cache_audio_read_tokens', 'output_audio_tokens')¶: Usage counters EDRI publishes, read explicitly so an upstream release adding a field cannot silently change the audit shape. Measured against pydantic_ai_slim 2.18.0; extending it is an intentional EDRI change.

edri.integrations.pydantic_ai.normalization.capture_content(value: Any, *, budget: NormalizationBudget, redactor: Callable[[Any], Any] | None = None) → Any[source]¶

Snapshot value, hand the snapshot to redactor, bound the result.

The redactor runs on the lifecycle hot path and is application code, so its failure is contained here: the result degrades to REDACTION_FAILED and never falls back to the unredacted original, which is the one outcome that would defeat the point of having a redactor.

edri.integrations.pydantic_ai.normalization.describe_error(exception: BaseException, *, include_message: bool, budget: NormalizationBudget, redactor: Callable[[Any], Any] | None = None) → dict[str, Any][source]¶

Describe exception under the effective content policy.

Exception text is not metadata-safe: provider messages carry request bodies, model output, URLs with credentials and application rows, and truncation is not sanitization. So metadata mode publishes only allowlisted structural fields, and a message appears solely when the application enabled include_error_messages under content.

The message is built from args, never from str(exception): __str__ is an application-defined conversion, and calling it to produce a nicer message is exactly the arbitrary evaluation this module refuses. Neither the live exception nor its traceback is retained -- a traceback holds frames, locals and open resources, and an action outlives all of them.

edri.integrations.pydantic_ai.normalization.normalize(value: Any, budget: NormalizationBudget) → Any[source]¶

Return a detached, bounded, JSON-like clone of value.

Total: no input raises, recurses without limit, or allocates without limit. The emitter drops observations it cannot build, so a normalizer that raised would silently delete audit records rather than produce a degraded one.

Iterative rather than recursive, because max_depth is caller-configurable and a generous one would otherwise hit the interpreter's stack limit instead of the documented bound.

edri.integrations.pydantic_ai.normalization.usage_snapshot(usage: Any, budget: NormalizationBudget, counters: tuple[str, ...] = ('requests', 'tool_calls', 'input_tokens', 'output_tokens', 'cache_write_tokens', 'cache_read_tokens', 'input_audio_tokens', 'cache_audio_read_tokens', 'output_audio_tokens')) → dict[str, Any] | None[source]¶

Build EDRI's stable usage shape from an explicit field list.

Counters that are absent are omitted rather than reported as zero: a missing counter is a measurement EDRI did not make, and zero is a claim. Unknown upstream fields are ignored so a dependency upgrade cannot change the published schema.

counters names the scope being snapshotted: USAGE_FIELDS for a whole run, REQUEST_USAGE_FIELDS for one model request. The caller picks it from what it is observing, because the two objects overlap and a per-request object read through the run-scoped list would publish what looks like a run total with counters missing.

Returns None when nothing supported is available -- "no usage" and "an empty usage mapping" are different claims.

edri.integrations.pydantic_ai.observability module¶

Public observability contract for the Pydantic AI bridge.

This module holds the parts of the audit feature that are not Pydantic AI specific: the action event that enters ordinary EDRI routing, the environment switch that arms the feature, and the application policy that narrows what its payloads may contain.

Nothing here imports pydantic_ai. The action event is what application managers and the inspector consume, and that side of the system should not need the optional dependency to understand it.

class edri.integrations.pydantic_ai.observability.AgentAction(action_id: UUID, namespace: str, operation: str, phase: AgentActionPhase, timestamp: datetime = <factory>, attempt: int = 1, parent_action_id: UUID | None = None, root_action_id: UUID | None = None, duration: float | None = None, outcome: AgentActionOutcome | None = None, run_id: str | None = None, conversation_id: str | None = None, edri_event_id: UUID | None = None, agent: str | None = None, provider: str | None = None, framework: str = 'pydantic_ai', model: str | None = None, model_version: str | None = None, prompt_version: str | None = None, output_mode: str | None = None, name: str | None = None, usage: dict[str, ~typing.Any] | None=None, metadata: dict[str, ~typing.Any]=<factory>, request: Any = None, response_data: Any = None, error: dict[str, ~typing.Any] | None=None)[source]¶

Bases: Event

One observed lifecycle point of one agent operation.

Identity comes in three layers. Event._id identifies this concrete routed event; (action_id, attempt) identifies the logical operation attempt, so a START and its terminal event pair up while retries stay distinguishable; root_action_id names the run the action belongs to. The run's root action is the key because upstream run_id is not unique -- a caller may pass one run_id to two concurrent invocations -- and because a consumer must be able to bucket an action whose parent it never saw or has already evicted.

The event is deliberately not frozen -- Event carries mutable framework transport state -- but its payloads are detached and normalized before publication, and the emitter never mutates a published instance. The router queue serializes on a feeder thread, so a payload still aliased to a live object could otherwise change mid-serialization.

request/response_data are typed Any because they hold normalized JSON-like values (built-in scalars and containers only). response_data rather than response because Event.response is framework state.

agent: str | None = None¶

attempt: int = 1¶

conversation_id: str | None = None¶

duration: float | None = None¶

edri_event_id: UUID | None = None¶

error: dict[str, Any] | None = None¶

framework: str = 'pydantic_ai'¶

model: str | None = None¶

model_version: str | None = None¶

name: str | None = None¶

outcome: AgentActionOutcome | None = None¶

output_mode: str | None = None¶

parent_action_id: UUID | None = None¶

prompt_version: str | None = None¶

provider: str | None = None¶

request: Any = None¶

response_data: Any = None¶

root_action_id: UUID | None = None¶

run_id: str | None = None¶

usage: dict[str, Any] | None = None¶

class edri.integrations.pydantic_ai.observability.AgentActionNamespace[source]¶

Bases: object

Broad subsystems this integration owns.

Plain strings, deliberately: AgentAction.namespace is extensible, and adding a future subsystem such as memory must not require changing the event model. These constants only spare callers a literal.

AGENT: Final = 'agent'¶

EDRI: Final = 'edri'¶

MCP: Final = 'mcp'¶

MODEL: Final = 'model'¶

TOOL: Final = 'tool'¶

class edri.integrations.pydantic_ai.observability.AgentActionOperation[source]¶

Bases: object

Operations this integration owns, by namespace.

The MCP operations are reserved by the event model from day one; which of them the implementation can actually observe on the supported dependency floor is a separate, measured question (see the spec's MCP coverage matrix).

DISPATCH: Final = 'dispatch'¶

EXECUTE: Final = 'execute'¶

MCP_INITIALIZE: Final = 'initialize'¶

MCP_PROMPTS_GET: Final = 'prompts.get'¶

MCP_PROMPTS_LIST: Final = 'prompts.list'¶

MCP_RESOURCES_LIST: Final = 'resources.list'¶

MCP_RESOURCES_READ: Final = 'resources.read'¶

MCP_TOOLS_CALL: Final = 'tools.call'¶

MCP_TOOLS_LIST: Final = 'tools.list'¶

REQUEST: Final = 'request'¶

RUN: Final = 'run'¶

VALIDATE: Final = 'validate'¶

class edri.integrations.pydantic_ai.observability.AgentActionOutcome(*values)[source]¶

Bases: Enum

Terminal result of one execution attempt.

Cancellation, timeout and deferral are not failures merely because an upstream library uses an exception to unwind control flow, so they terminate with AgentActionPhase.END and their own outcome.

CANCELLED = 'cancelled'¶

DEFERRED = 'deferred'¶

ERROR = 'error'¶

SUCCESS = 'success'¶

TIMEOUT = 'timeout'¶

class edri.integrations.pydantic_ai.observability.AgentActionPhase(*values)[source]¶

Bases: Enum

Lifecycle point of one execution attempt.

Closed, unlike namespace/operation: START/END/ERROR has stable transport semantics that a new subsystem does not extend.

END = 'end'¶

ERROR = 'error'¶

START = 'start'¶

class edri.integrations.pydantic_ai.observability.AgentActionPolicy(include_model_requests: bool = False, include_model_responses: bool = False, include_tool_arguments: bool = False, include_tool_results: bool = False, include_mcp_requests: bool = False, include_mcp_responses: bool = False, include_dispatch_requests: bool = False, include_dispatch_responses: bool = False, include_error_messages: bool = False, redactor: ~typing.Callable[[~typing.Any], ~typing.Any] | None = None, limits: ~edri.integrations.pydantic_ai.observability.NormalizationLimits = <factory>)[source]¶

Bases: object

What an application would like captured, per content category.

This is the application half of the content decision. The deployment half is AGENT_AUDIT_ENV_VAR, and the effective policy is their intersection: a policy can always remove content, never add content the operator disallowed. Everything therefore defaults to off, error messages included.

class edri.integrations.pydantic_ai.observability.AgentAuditMode(*values)[source]¶

Bases: Enum

Deployment-level ceiling on what observability may capture.

CONTENT = 'content'¶

METADATA = 'metadata'¶

OFF = 'off'¶

property enabled: bool¶

class edri.integrations.pydantic_ai.observability.NormalizationLimits(max_depth: int = 8, max_values: int = 1024, max_string_characters: int = 4096, max_total_bytes: int = 65536)[source]¶

Bases: object

Bounds applied to every captured payload, before and after redaction.

max_total_bytes is one shared budget across an action's request, response, metadata, usage and error fields rather than a per-field allowance, so a single action cannot multiply the limit by its field count.

edri.integrations.pydantic_ai.observability.resolve_audit_mode(environ: Mapping[str, str] | None = None) → AgentAuditMode[source]¶

Read AGENT_AUDIT_ENV_VAR; missing means AgentAuditMode.OFF.

Read here rather than as an import-time constant in edri.config.setting because the mode is resolved once per worker during BaseAgentWorker.setup(). An import-time constant would fix the value for the whole process and could not be resolved per worker at all.

Unknown values are configuration errors, not silently reinterpreted: an operator who typed EDRI_PYDANTIC_AI_AUDIT=metdata asked for less exposure than a fallback to content would give them, and more than a fallback to off.

edri.integrations.pydantic_ai.worker module¶

Worker half of the Pydantic AI bridge.

BaseAgentWorker hosts one or more AgentBase agents on an EDRI WorkerThread, bakes runtime deps into each, and exposes dispatch so agent tools can call other services as plain Events.

class edri.integrations.pydantic_ai.worker.BaseAgentWorker(pipe: Connection, event: T, name: str, *args: Any, **kwargs: Any)[source]¶

Bases: WorkerThread[_EventT], ABC

A Worker that hosts one or more AgentBase agents.

The worker is an orchestrator, not an agent. It declares which agents it uses (agents), constructs them, bakes runtime deps into each, then runs domain logic in do(). Agents are reachable via self.agent[SomeAgentClass] after setup. The EDRI dispatch and shared resources are injected via deps, so agent methods never expose them.

Subclasses must set agents, supply each agent's constructor arguments — at least model_name — via agent_configs or agent_kwargs(), and implement do().

The agents are built in setup(), which the base Worker runs before do() inside its own error-handling scope — so this class does not override run().

agent_action_policy(agent_cls: type[AgentBase[Any, Any]]) → AgentActionPolicy[source]¶

Content policy for agent_cls, resolved once during setup().

Defaults to capturing no payload bodies at all. Override to describe what would be useful to observe; the deployment environment decides separately what is permitted, and the effective policy is their intersection, so an override here can never widen what EDRI_PYDANTIC_AI_AUDIT allows.

agent_configs: ClassVar[dict[type[AgentBase[Any, Any]], dict[str, Any]]] = {}¶: Per-agent constructor kwargs for the common case — each agent class in agents maps to the dict passed to its constructor (e.g. {"model_name": "gpt-4o"}), so a worker hosting several agents on different models is a flat declaration instead of a dispatch you write yourself. See agent_kwargs() for kwargs that must be computed at runtime instead (env lookups, secrets, ...).

agent_deps(agent_cls: type[AgentBase[Any, Any]]) → AgentDeps[Any][source]¶

Runtime dependencies for agent_cls.

Defaults to sharing the worker's dispatch and exposing the worker itself as resources. Override to scope resources per agent.

agent_kwargs(agent_cls: type[AgentBase[Any, Any]]) → dict[str, Any][source]¶

Constructor keyword arguments for agent_cls.

Defaults to looking agent_cls up in agent_configs — set that class-level mapping for the common case (each agent just needs a different model_name or similar static config). Override this method instead when kwargs must be computed at runtime (env lookups, secrets, ...); an agent needs at least model_name, and only the worker knows the application's model/provider configuration.

Raises:: KeyError -- If agent_cls isn't in agent_configs and this wasn't overridden.

agents: ClassVar[list[type[AgentBase[Any, Any]]]] = []¶: Agent classes used by this worker. Identity is the class itself; the logical name comes from AgentBase.default_name().

setup() → None[source]¶

Construct all declared agents and bake runtime deps into each.

Overrides the setup() hook, so it runs once before do(). Override further for worker-specific setup, but call super().setup() so agents get built.

The audit mode is resolved once here, not per run: a running worker is not a runtime toggle, and a worker must be restarted for a new mode to take effect.

Raises:

ValueError -- If two agent classes resolve to the same logical name, or if EDRI_PYDANTIC_AI_AUDIT holds an unsupported value.
RuntimeError -- If observability is enabled but no router queue is bound, which would leave the actions with nowhere to go.

Module contents¶

Pydantic AI agent integration for EDRI — optional.

Requires pydantic-ai, which is not a core EDRI dependency. Install it with:

pip install edri[pydantic-ai]

Importing this module without pydantic-ai installed raises a clear ImportError naming the extra, rather than a bare ModuleNotFoundError.

What this module provides¶

A thin bridge between Pydantic AI agents and EDRI's Worker model:

AgentDeps — runtime dependencies handed to an agent's tools: the only door back into the EDRI router (dispatch) plus arbitrary shared resources.
AgentBase — wraps a pydantic_ai.Agent and keeps runtime concerns (the EDRI pipe, shared state) out of an agent's domain methods.
BaseAgentWorker — a WorkerThread that hosts one or more agents, builds them, bakes their deps in, and exposes the EDRI dispatch so agent tools can call other services as plain Events.
AgentInspectorManager — an optional standard manager that consumes the AgentAction stream into bounded in-memory history, answers the agent-inspector-* query events over the application's existing API, and publishes AgentInspectorUpdate to whoever registered for it.

A tool can dispatch any Event — including one handled by a Worker owned by a different Manager — and get the reply back on this worker's pipe without any per-type wiring: ManagerBase tracks the reply generically (see ManagerBase._track_worker_response) keyed on Event._id, which every event carries unconditionally.

From pydantic-ai this package uses Agent, Agent.run_sync(deps=...) and, when observability is enabled, the public Hooks capability installed for the duration of one run. Model/provider construction lives in your concrete agents' build_agent().