watchlight-agent-sdk

Build governed AI agents. The Watchlight Agent SDK lets you put a real, fail-closed authorization decision in front of every action your agent takes — and record what happened — with the same code whether you run it on your laptop or across a production fleet.

pip install watchlight-agent-sdk

Import it as watchlight_core.

What it does

Your agent asks "may I do this?" before it acts; Watchlight answers Allow or Deny by evaluating your policies, and the action only runs if it's permitted. The SDK gives you:

• Authorization — authorize_action(action, resource) gates every step, fail-closed.
• Preflight — a cheap read-only "would this be allowed?" check.
• Lifecycle — a RunHandle per agent run: submit a plan, authorize steps, complete or terminate.
• Lineage — canonical, value-free execution events for audit and debugging.
• Framework plugins — ready-made integrations for LangGraph, Google ADK, AWS Bedrock, Microsoft Agent Framework, Pydantic AI, and more.

The SDK is glue: it shapes a request, hands it to a Watchlight backend, and projects the answer. Every decision is made by the backend — never in this package — which is why it's open source (Apache-2.0).

Two backends, same code

The SDK talks to a backend through one small interface, so the exact same agent and plugin code runs against either:

Moving from one to the other is a one-line change, never a rewrite.

Quickstart — a governed decision, zero infrastructure

Install the in-process engine alongside the SDK:

pip install watchlight-agent-sdk watchlight-engine

import asyncio
from watchlight_core import InProcessClient

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

async def main():
    gov = InProcessClient(POLICIES)          # real engine decisions, in-process — no network
    print((await gov.authorize("research-agent", "read", "dataset"))["decision"])    # Allow
    print((await gov.authorize("research-agent", "delete", "dataset"))["decision"])  # Deny

asyncio.run(main())

Allow
Deny

That Deny is the point: the action is refused before it runs — by the same engine that powers the Watchlight platform, with no database and no network.

Govern a framework agent

Install the plugin for your framework and point it at a backend — your agent code doesn't change:

from watchlight_langgraph import WatchlightLangGraphPlugin   # or -adk, -bedrock, -pydantic-ai, …
from watchlight_core import InProcessClient

plugin = WatchlightLangGraphPlugin()
plugin.apdp = InProcessClient(POLICIES)      # the one line that differs from production

async with await plugin.start_run("research-agent") as handle:
    if not await handle.authorize_action("read", "dataset"):
        raise PermissionError("denied by policy")
    # ... your agent runs, every action governed ...

Custom agents (any framework)

For an agent not built on a supported framework, the @watchlight decorator governs any function:

from watchlight_core import InProcessClient, watchlight

gov = InProcessClient(POLICIES)

@watchlight(agent_id="custom-research", client=gov)
async def run_research(query: str) -> str:
    ...

Optional: OpenTelemetry export

Install the [otlp] extra to export a subset of the lineage stream to your OTel collector (off by default):

pip install 'watchlight-agent-sdk[otlp]'

Lineage delivery and retries

Lineage events and the run's session close are retried when the server answers 429, 502, 503 (it cannot yet know the run's state) or 504, or the connection fails. The SDK waits with exponential backoff and jitter and honours Retry-After. A retried event is deduplicated by its event_id. A 500 and every other 4xx are final and never retried.

A run's events are sent in call order. Authorization is never retried or delayed by lineage. The execution_started sent before authorize is one direct attempt that never waits behind a retry; if it fails, authorize fails closed (an action whose execution was never started would be dropped from lineage), and a transient failure is also queued for background delivery. Other events emitted while a decision is applied (execution_quarantined) get one attempt if nothing else is being sent for the run, and are otherwise queued; such an event can land after a batch that was being retried when it was emitted, and the lineage view orders events by timestamp. The quarantine kill signal fires before any lineage is sent.

When a batch runs out of budget, the SDK logs one WARNING with reason=lineage_retry_exhausted (event types and counts, no payloads), and the run's later events get one attempt each and are then queued, until one is delivered. When the run ends, the SDK waits up to one budget for its queued events. ApdpClient.lineage_delivery_counters counts retries, give-ups, queued batches and cancelled publishes.

Learn more

• Documentation: https://docs.watchlight.ai/de
• Developer Edition (zero-infra quickstart) and Enterprise (the governed control plane): https://watchlight.ai

License

Apache-2.0.