# Pydantic AI + ACP — Policy & Audit Install Guide

Add per-user policy, audit logging, and PII detection to Pydantic AI agents. One pip install, one ACPHooks() registration, every tool call audited and policy-checked.

# Pydantic AI + ACP — Policy & Audit Install Guide

[Pydantic AI](https://ai.pydantic.dev) is a Python framework for building agents with type-safe tools, structured outputs, and provider-agnostic model strings. Out of the box, a production deployment shares one backend API key across every end user's request — no per-user policy enforcement, no per-user audit trail, no way to tell downstream systems which human triggered which action.

`acp-pydantic-ai` closes that gap with one line. Pydantic AI 2 ships a first-class `Hooks` capability, and `ACPHooks()` registers on that surface: every tool the agent has — and every tool you add later — is policy-checked before it runs and audited after, with zero per-function decorators. Bind the end user's identity per request with `set_context`. Same control model as Claude Code — same `/govern/tool-use` endpoint, same workspace policies.

> **Starter · 5-minute install.** `pip install acp-pydantic-ai`, add `capabilities=[ACPHooks()]` to your `Agent(...)`, bind the JWT per request. See [the runnable starter](https://github.com/agentic-control-plane/acp-governance-sdks/tree/main/examples/starters/pydantic-ai), [the control model](/docs/governance-model), or the [frameworks index](/frameworks).

## Install

```bash
pip install acp-pydantic-ai "pydantic-ai>=2"
```

## Minimal agent under policy

```python
from fastapi import FastAPI, Header
from pydantic_ai import Agent
from acp_pydantic_ai import ACPHooks, configure, set_context

configure(base_url="https://api.agenticcontrolplane.com")
app = FastAPI()

# One registration covers every tool on the agent — present and future.
agent = Agent(
    "anthropic:claude-sonnet-4-6",
    instructions="You are an ACP-governed agent. Use the tools available.",
    capabilities=[ACPHooks()],
)


@agent.tool_plain
def lookup_record(id: str) -> str:
    """Look up a record by ID."""
    return json.dumps(db.records.find_one({"id": id}))


@agent.tool_plain
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    return mailer.send(to=to, subject=subject, body=body)


@app.post("/run")
def run(prompt: str, authorization: str = Header(...)):
    set_context(
        user_token=authorization.removeprefix("Bearer ").strip(),
        agent_name="my-pydantic-agent",
        agent_tier="interactive",
    )
    result = agent.run_sync(prompt)
    return {"result": result.output}
```

No decorator on either tool. That's the point: the coverage is a property of the agent, not of each function — there is nothing to forget on the tool you add next month.

## What happens on every tool call

1. `ACPHooks()`'s wrap hook POSTs to ACP's `/govern/tool-use` with the tool name, validated args, and the user JWT bound by `set_context`.
2. **Deny** → the tool function is **never called**. The model receives `"tool_error: <reason>"` as the tool result, sees the denial, and adapts.
3. **Allow** → your function runs.
4. Post-audit: ACP scans the output for PII / secrets. `redact` → the redacted version replaces the original before the model sees it; `block` → the model sees `"[ACP] Blocked: <reason>"`. Audit row written, rooted in the end user's identity.

## How it hooks in

Pydantic AI 2's `Hooks` capability exposes ~20 hook points (`before_tool_execute`, `after_tool_execute`, wrap forms, error forms, and more). `ACPHooks()` returns a `Hooks` instance carrying a single `tool_execute` wrap hook — Pydantic AI invokes it around every tool execution, and the hook only calls through to your function on allow. It composes with your own capabilities and hooks; register it alongside them:

```python
agent = Agent(model, capabilities=[my_metrics_hooks, ACPHooks()])
```

Scope coverage when you need to:

```python
ACPHooks(tools=["send_email", "delete_record"])   # only these are checked
ACPHooks(exclude=["get_time"])                    # everything but these
```

## Per-tier policy

`set_context(agent_tier="...")` controls the policy tier:

- **`interactive`** — human at the keyboard, permissive default.
- **`subagent`** — invoked by another agent, no human in the immediate loop.
- **`background`** — autonomous, most restrictive.
- **`api`** — programmatic call from your backend.

## Compatibility: the decorator pattern (pre-v2)

Before Pydantic AI 2, this guide's pattern was stacking `@governed` under the tool decorator:

```python
from acp_governance import governed

@agent.tool_plain     # outer — registers with Pydantic AI
@governed("lookup_record")      # inner — wraps the call with the policy check
def lookup_record(id: str) -> str: ...
```

This still works on v2 — `functools.wraps` preserves `__wrapped__`, so Pydantic AI's introspection reads the original signature for the tool schema — and `acp-pydantic-ai` re-exports `governed` so existing code keeps importing from one place. Use it only if you're pinned below v2. Don't combine it with `ACPHooks()` on the same tool, or the call is checked (and audited) twice.

## Price and meter the model calls

`ACPHooks()` is the **interception plane** — what your agent *does*. The **proxy
plane** covers what it *spends*. Both run against the same gateway, and `init()`
wires them together:

```python
import acp_governance as acp

acp.init()            # call before you construct any model client
```

Model calls now land on the same trail as tool calls, carrying real cost,
tokens, and cache economics. Constructing clients explicitly instead? Pass the
config rather than calling `init()`:

```python
from acp_governance import model_client_kwargs
from anthropic import Anthropic

client = Anthropic(**model_client_kwargs("anthropic"))
```

Shapes are `"anthropic"`, `"openai"` (chat completions), and
`"openai-responses"`. The last two are **not** interchangeable — `/v1` serves
chat completions, `/openai/v1` serves responses.

`init()` is all-or-nothing per provider: it sets the base URL *and* the key, or
it leaves that provider completely alone and tells you which one it skipped. If
`OPENAI_BASE_URL` already points at your own gateway, ACP won't silently
reroute you.

**Your coverage:** interception ✓ (hooks) · proxy ✓ (model calls).

## Pydantic AI–specific notes

- **Multi-provider model strings.** `Agent("anthropic:claude-sonnet-4-6")`, `Agent("openai:gpt-4o-mini")`, `Agent("google-gla:gemini-...")` all resolve to provider-specific clients using their respective API keys. The policy layer is identical across providers.
- **Structured outputs.** Pydantic AI's `output_type=MyModel` works unchanged — the hook runs on tool calls, not on the agent's structured-output validation.
- **Async and sync tools both covered.** The wrap hook sits above Pydantic AI's tool dispatch, so `def` and `async def` tools flow through the same check.
- **Test without keys.** Pydantic AI's `TestModel` / `FunctionModel` run the full tool loop — including `ACPHooks()` — without an LLM key, which is how the starter's integration test scripts a denial and asserts the tool never ran.

## FAQ

**Do I still need a decorator on each tool?**
No. `capabilities=[ACPHooks()]` covers every tool registered on the agent, including ones added later. The `@governed` decorator remains available for pre-v2 codebases.

**What does the model see on a denial?**
The tool result is `"tool_error: <reason>"` — the function itself never executed. The model treats it like any tool result and routes around it; the run completes instead of crashing.

**Does this touch my LLM traffic?**
Not by default — LLM calls go direct to your provider with your own key while ACP checks and records tool calls. Opt model calls in with the proxy plane above (`acp.init()`), which puts cost, tokens, and cache economics on the same trail.

**What if the ACP gateway is unreachable?**
Fail-open: the tool proceeds, marked `"fail-open"` in the reason. Policy checks are never a single point of failure for the agent. (5s timeout, configurable via `configure(timeout_s=...)`.)

**Which Pydantic AI versions?**
`ACPHooks()` needs `pydantic-ai >= 2` (the `Hooks` capability). On 1.x, use the decorator pattern above with plain `acp-governance`.

## Related

- [Pydantic AI docs — agents](https://pydantic.dev/docs/ai/core-concepts/agents/)
- [Pydantic AI docs — hooks](https://pydantic.dev/docs/ai/core-concepts/hooks/)
- [`acp-pydantic-ai` on PyPI](https://pypi.org/project/acp-pydantic-ai)
- [`acp-governance` (core SDK)](https://pypi.org/project/acp-governance)
- [LangChain / LangGraph integration](/integrations/langgraph)
- [Deep Agents integration](/integrations/deepagents)
- [CrewAI integration](/integrations/crewai)

<script type="application/ld+json">
{
  "@context": "https://schema.org",
  "@type": "HowTo",
  "name": "Add policy and audit to Pydantic AI with Agentic Control Plane",
  "totalTime": "PT5M",
  "step": [
    {"@type": "HowToStep", "name": "Install acp-pydantic-ai", "text": "pip install acp-pydantic-ai \"pydantic-ai>=2\""},
    {"@type": "HowToStep", "name": "Register ACPHooks once", "text": "Add capabilities=[ACPHooks()] to Agent(...) — every tool on the agent is covered, no per-tool decorators."},
    {"@type": "HowToStep", "name": "Bind the user's JWT per request", "text": "Call set_context(user_token=...) at the start of each request."}
  ]
}
</script>