Skip to content
Agentic Control Plane

Governed Pydantic AI in 3 minutes

David Crowe · 5 min read
pydantic-ai python governance three-minute-governance

Pydantic AI is what you get when the team behind Pydantic builds an agent framework. Type-safe tools, structured outputs, provider-prefixed model strings (anthropic:claude-sonnet-4-6, openai:gpt-4o-mini, google-gla:gemini-...). It’s the most opinionated Python agent framework in 2026, and the opinions are good.

What it doesn’t ship: per-user identity attribution on tool calls, cross-tenant audit, output redaction, or pluggable policy. The framework’s emerging Hooks API is heading there, but the production-ready integration today is decorator-based.

This post is the 3-minute path with Agentic Control Plane. Stack @governed under @agent.tool_plain, bind the user JWT, ship.

The pattern

Tool-layer governance via decorator stacking. @agent.tool_plain (or @agent.tool if you need run context) registers the tool with Pydantic AI. @governed wraps the function with ACP’s pre/post hooks. The decorator order matters — @governed sits closer to the function so the governance check runs inside Pydantic AI’s tool dispatch. functools.wraps preserves __wrapped__, so Pydantic AI’s signature introspection reads the original function’s type hints and docstring for tool schema generation.

Three minutes from blank slate

1. Install

pip install acp-governance pydantic-ai

2. Wrap your tools

import os, json
from pydantic_ai import Agent
from acp_governance import configure, governed, set_context

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

agent = Agent(
    "anthropic:claude-sonnet-4-6",
    instructions="You are an ACP-governed agent. Use the tools available.",
)


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


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


def run(prompt: str, user_jwt: str):
    set_context(
        user_token=user_jwt,
        agent_name="my-pydantic-agent",
        agent_tier="interactive",
    )
    return agent.run_sync(prompt).output

3. Run it

export ACP_USER_TOKEN=gsk_...
export ANTHROPIC_API_KEY=...
python my_agent.py

Open cloud.agenticcontrolplane.com/activity. One row per tool call with actor, tool, decision, session, input/output preview. Three minutes, two integration calls, full audit.

What @governed does

Every call to a wrapped tool:

  1. POSTs to /govern/tool-use with the tool name, input arguments, and the user token bound by set_context.
  2. Deny → returns "tool_error: <reason>". Pydantic AI delivers it to the agent as tool output; the model adapts.
  3. Allow → runs your function.
  4. Post-audit: ACP scans the output for PII / secrets, optionally redacts.

Pydantic AI’s introspection sees the original function via __wrapped__, so tool schema generation works correctly with the decorator in place.

Decorator order

@agent.tool_plain     # outer — registers with Pydantic AI
@governed("...")      # inner — wraps the call with governance
def my_tool(...): ...

Reverse the order and @agent.tool_plain registers the un-governed function with Pydantic AI, and @governed becomes a no-op wrapper that’s never called. Always @governed closer to the function.

If you need access to the agent’s run context inside the tool, use @agent.tool (not @agent.tool_plain) and keep RunContext[Deps] as the first parameter — @governed composes with both.

Per-tier policy

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

  • interactive — human at the keyboard.
  • subagent — invoked by another agent.
  • background — autonomous, most restrictive.
  • api — programmatic call from your backend.

A destructive verb that’s allowed when a human is supervising gets denied when the same agent runs on a schedule. Match the tier to actual deployment reality.

Async tools

@agent.tool_plain
@governed("fetch")
async def fetch(url: str) -> str:
    """Fetch a URL."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
    return resp.text

@governed detects coroutine functions and dispatches accordingly.

Future migration: Hooks API

Pydantic AI ships a first-class Hooks capability (before_tool_execute, after_tool_execute, wrap_tool_execute) that’s a cleaner integration point than per-function decorator stacking — it governs every tool registered with the agent without requiring users to decorate each one.

A future acp-pydantic-ai v0.2 package will expose an ACPHooks() helper using this surface. The decorator pattern documented here will keep working — the migration is an ergonomic upgrade, not a correctness requirement. If you want one-line governance setup at the agent level, watch for the v0.2 release; if you’re shipping today, the decorator pattern is the production-ready path.

What this unlocks

Pydantic AI is the most type-safe Python agent framework in 2026. ACP adds the governance, audit, and identity attribution layer it doesn’t ship. Three lines of integration — configure, @governed, set_context — and your typed agent becomes a typed governed agent.

Pydantic AI integration guide → · Three-minute integrations → · Get started →

Get the next post
Agentic governance, AgentGovBench updates, the occasional incident post-mortem. One email per post. No marketing fluff.
Share: Twitter LinkedIn
Related posts

← back to blog