Metadata-Version: 2.4
Name: watchlight-pydantic-ai
Version: 0.3.3
Summary: Watchlight governance for Pydantic AI agents: authorize the tools you wrap with guarded_tool() and open every Agent.run as a governed run (nested runs as attenuated sub-agents), fail-closed. Undecorated tools and model turns are not authorized.
Author-email: Watchlight AI <team@watchlight.ai>
License: Apache-2.0
Project-URL: Homepage, https://watchlight.ai
Project-URL: Documentation, https://docs.watchlight.ai/de
Project-URL: Pydantic AI, https://github.com/pydantic/pydantic-ai
Keywords: watchlight,pydantic-ai,ai-agents,ai-governance,agent-runtime
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: watchlight-agent-sdk>=0.9.4
Requires-Dist: pydantic-ai<2,>=0.0.13
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: mypy>=1.8; extra == "dev"

# 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.

```bash
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):

```bash
pip install watchlight-pydantic-ai watchlight-engine
```

```python
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

|  | Backend | Runs |
|---|---|---|
| **Developer Edition** | `InProcessClient` | The compiled engine, **in-process** — no server, no network |
| **Enterprise** | `ApdpClient` | The governed control plane — signed lineage, drift detection, fleet-wide governance |

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

```python
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.
