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 == Agent::"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.

Runs that cannot be governed do not run

Governed entry points: Agent.run, run_sync, run_stream, iter and run_stream_sync are instrumented directly, and run_stream_events is governed through run, which pydantic-ai drives it with. Each call opens a governed run first: a root session for a top-level call, a sub-agent spawn for a call made inside another agent's tool. If that fails, the call raises before the model is called, and the agent's tools never run. A refusal from Watchlight (for example ScopeAttenuationDenied when the delegation depth is exceeded, or AgentNotRegistered) is raised as is. Any other failure, such as an unreachable backend, is raised as GovernanceUnavailable. A refused sub-agent never runs on its parent's run. A call made from a run that has ended (for example from a task a tool started that outlives the run) is refused with GovernanceUnavailable.

When a run returns, its session is completed and execution_completed is emitted; a run that raises is terminated. For run_stream_sync the run closes with its result: it completes when the stream is fully consumed or get_output() returns, and it is terminated when consuming it raises, when the caller stops iterating before the run finished, or, as a last resort, when the result object is dropped unconsumed (reason pydantic_ai_stream_abandoned). Once the run is terminated, reading the result further (get_output(), the stream_* methods, validate_response_output()) raises GovernanceUnavailable instead of driving the run. The result is still a StreamedRunResultSync.

Attributing runs to an AI application

Every governed run can be attributed to the AI application the agent belongs to. Pass application_id="<application UUID>" to WatchlightPydanticAIPlugin(...), or per run to start_run(...), or set WATCHLIGHT_APPLICATION_ID in the environment — an explicit argument wins over the env var. The zero-touch instrumentation opens runs with the plugin-level value, so a deployment attributes an agent without a code change. The id is sent on session create and bound to the run credential and to every sub-agent the run spawns; wl-apdp refuses the session if the agent is not a member of that application. An agent in exactly one application needs nothing here. The id is explicit configuration, never inferred from prompts or names, and it is an Enterprise feature: the in-process Developer Edition engine refuses it.

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.