Pydantic AI + ACP — Policy & Audit Install Guide

Pydantic AI 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, the control model, or the frameworks index.

Install

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

Minimal agent under policy

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

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.
Deny → the tool function is never called. The model receives "tool_error: <reason>" as the tool result, sees the denial, and adapts.
Allow → your function runs.
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:

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

Scope coverage when you need to:

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:

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:

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

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.