watchlight-pydantic-ai

Governed Pydantic AI agents with Watchlight — authorize every agent action, fail-closed, with zero infrastructure.

pip install watchlight-pydantic-ai

Independent third-party plugin. This is an independent integration built by Watchlight AI. It is not affiliated with, endorsed by, or sponsored by Pydantic. Pydantic AI and related names are trademarks of Pydantic, used here nominatively only to describe compatibility.

What it does

watchlight-pydantic-ai puts a Watchlight authorization decision in front of every action your Pydantic AI agent takes — so each tool call is allowed or denied before it runs, never after. It's open-source glue: a thin, framework-specific layer that threads Watchlight's governance primitives into Pydantic AI's @agent.tool decorators and nested Agent.run() composition. The actual policy decisions run on Watchlight's compiled engine — either in-process for local development or against the governed control plane in production.

Quickstart

Point the plugin at a backend and wrap your agent tools with one decorator. Your agent code stays vanilla Pydantic AI — only the backend changes.

For local development, the zero-infrastructure Developer Edition runs the compiled engine in-process (requires the watchlight-engine package):

pip install watchlight-pydantic-ai watchlight-engine

from pydantic_ai import Agent
from watchlight_pydantic_ai import WatchlightPydanticAIPlugin
from watchlight_core import InProcessClient

# A Cedar policy: the research agent may execute web_search, nothing else.
POLICIES = [
    {"name": "reader",
     "code": 'permit(principal == User::"research-agent", action == Action::"execute", resource == Resource::"web_search");'},
]

plugin = WatchlightPydanticAIPlugin()
plugin.apdp = InProcessClient(POLICIES)   # decisions run in-process, no server, no network

agent = Agent("openai:gpt-4o-mini")

async def run_agent(question: str):
    async with await plugin.start_run("research-agent") as handle:

        @agent.tool
        @handle.guarded_tool()             # ← the governance line
        async def web_search(ctx, query: str) -> list[dict]:
            return await fetch(query)

        result = await agent.run(question)
        return result.output

The @handle.guarded_tool() decorator runs authorize_action("execute", "web_search") against the backend before every invocation of the tool. On a policy denial the decorator raises PermissionError, which propagates up through Pydantic AI's normal tool-error path — the tool body never runs. guarded_tool fails closed: a denial, or an unreachable backend, stops the action.

functools.wraps preserves the wrapped function's signature and docstring, so Pydantic AI's tool-schema introspection works unchanged and the decorator composes cleanly with @agent.tool.

Two backends, same code

Moving from local to production is a one-line change — swap the backend, keep your agent code:

from watchlight_core import ApdpClient

plugin = WatchlightPydanticAIPlugin()
plugin.apdp = ApdpClient("https://apdp.your-company.example", api_key="...")

Links

• Documentation: https://docs.watchlight.ai/de
• Website: https://watchlight.ai

License

Apache-2.0.