watchlight-pydantic-ai

Watchlight governance for Pydantic AI agents: the tools you wrap with guarded_tool() are authorized before they run, and every Agent.run opens a governed run (nested runs as attenuated sub-agents), fail-closed. Undecorated tools and model turns are not authorized.

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 each tool call you wrap with guarded_tool() — the call is allowed or denied before it runs, never after — and opens every Agent.run as a governed run, with nested runs spawned as attenuated sub-agents. Tools you do not wrap, and the model turns themselves, are not authorized. 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, guarded_tool
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", name="research-agent")

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

async def run_agent(question: str):
    # Constructing the plugin instruments Agent.run: this call opens the
    # governed run, and the model and every tool call run under it.
    result = await agent.run(question)
    return result.output

The @guarded_tool() decorator runs authorize_action("execute", "web_search") before every invocation of the tool, on the run that is driving the agent at that moment: the root run for a top-level agent.run, or the sub-agent's own run when the agent is called from inside another agent's tool. A sub-agent's tools are therefore always authorized on the sub-agent's own credential and scope, never on its parent's. 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, an unreachable backend, or a call with no live governed run (raised as GovernanceUnavailable) stops the action.

Only tools you decorate are authorized; an undecorated tool runs without a Watchlight decision.

RunHandle.guarded_tool() (the decorator bound to a handle from start_run) is deprecated. It still authorizes each call on the run driving the agent at call time, and falls back to its own handle only when no instrumented run is active.

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.

If the plugin is configured to route egress through wl-proxy (connect_proxy_url= or WATCHLIGHT_PROXY_URL), a run is refused with GovernanceUnavailable when Watchlight issues no proxy token for it or the local CONNECT sidecar cannot be installed; it never falls back to direct egress. A run whose setup fails after its session was opened has that session terminated.

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, Credential

plugin = WatchlightPydanticAIPlugin()
# The credential is loaded from the environment or your credential provider
# (for example WATCHLIGHT_CREDENTIAL_FILE); never put a key in source.
plugin.apdp = ApdpClient("https://apdp.your-company.example", credential=Credential.from_default())

Links

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

License

Apache-2.0.