watchlight-langgraph

Governance primitives for LangGraph agents with Watchlight: authorize the actions you gate, validate plans, and spawn attenuated sub-agents.

pip install watchlight-langgraph

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

What it does

watchlight-langgraph gives you explicit governance primitives that you call from your LangGraph nodes. LangGraph is extended by graph composition, so the plugin does not intercept your tools or model calls by itself: an action is governed only when your code routes it through one of the calls below. The policy decisions run on Watchlight's compiled engine, either in-process for local development or against the governed control plane in production.

What is enforced, and where:

• handle.authorize_action(...) and handle.authorize_action_detailed(...) ask for a decision before an action. A Deny returns False (or a Deny result). If the backend cannot be reached, the call raises, so the action does not run as long as you do not catch and ignore that exception.
• governed_tool_call(handle, tool_name, ...) (in watchlight_langgraph.governance) wraps a tool implementation: it runs the tool only on Allow, raises ToolDenied on Deny, and pauses the graph with LangGraph interrupt when a human approval is needed.
• handle.submit_plan(...) and handle.preflight_step(...) validate a plan or a step and return Allow, Modify or Deny. Your code must act on the result.
• handle.spawn_subagent(...) asks the backend for an attenuated child run. A refused spawn raises, and the child gets its own credential and scope.
• plugin.start_run(...) refuses to start (raises AgentNotRegistered or GovernanceUnavailable) when the agent is not registered, the backend does not accept the session, or a wl-proxy URL is configured but governed egress cannot be set up.

What is not enforced:

• The callback handlers (WatchlightCallbackHandler, WatchlightAsyncCallbackHandler, handle.callback_handler()) only observe. They record tool, model and chain events for lineage and never block a call.
• A tool or model call that your graph makes without going through authorize_action or governed_tool_call is observed by the callbacks but not authorized.

Quickstart

Point the plugin at a backend and wrap your run. Your agent code doesn't change — only the backend does.

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

pip install watchlight-langgraph watchlight-engine

from watchlight_langgraph import WatchlightLangGraphPlugin
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);'},
]

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

async def run_agent(question: str):
    async with await plugin.start_run(agent_slug="research-agent") as handle:
        # Validate a multi-step plan ahead of execution (Allow / Modify / Deny).
        result = await handle.submit_plan(["read dataset", "summarize"])
        if result.is_deny():
            raise PermissionError(f"Plan denied: {result.violations}")

        # Gate each action authoritatively. Returns True on Allow, False on Deny.
        if not await handle.authorize_action("read", "dataset"):
            raise PermissionError("Denied by policy")

        # ... your LangGraph agent runs, every action governed ...

authorize_action returns False on a denial and raises when the backend cannot be reached. Either way the guarded action does not run, as long as your code does not swallow the exception.

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 WatchlightLangGraphPlugin(...), or per run to start_run(...), or set WATCHLIGHT_APPLICATION_ID in the environment — an explicit argument wins over the env var. 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:

from watchlight_core import ApdpClient, Credential

plugin = WatchlightLangGraphPlugin()
# The plugin credential comes from your secret store or the environment
# (WATCHLIGHT_PLUGIN_TOKEN), never from source code.
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