wl-apdp 0.2.0

Python SDK for WL-APDP (Watchlight Agentic Policy Decision Point) - secure your agentic AI systems

WL-APDP Python SDK

Python SDK for integrating WL-APDP (Watchlight Agentic Policy Decision Point) authorization into agentic AI systems. Framework-agnostic core with optional adapters for CrewAI, LangGraph, and AutoGen.

Features

• Framework-Agnostic: Core SDK works with any Python code
• Async/Sync Support: Both async and sync HTTP clients
• Context Propagation: Automatic context handling across async boundaries
• Tool Authorization: Decorators and wrappers for authorized tools
• Audit Logging: Built-in audit trail for all authorization decisions
• Multi-Tenant: First-class support for tenant isolation
• Framework Adapters: Optional integrations for popular agent frameworks

Installation

# Core SDK
pip install wl-apdp

# With framework adapters
pip install wl-apdp[crewai]      # CrewAI support
pip install wl-apdp[langgraph]   # LangGraph/LangChain support
pip install wl-apdp[autogen]     # AutoGen support
pip install wl-apdp[all]         # All frameworks

Quick Start

Basic Authorization Check

from wl_apdp import WlApdpClient

async with WlApdpClient("http://localhost:8081/api") as client:
    result = await client.authorize(
        principal='Agent::"my-agent"',
        action='Action::"execute"',
        resource='Tool::"web_search"'
    )

    if result.is_allowed:
        # Proceed with action
        pass

Authorized Tools

from wl_apdp import authorized_tool, AuthorizationContext, authorization_context

@authorized_tool(resource='Tool::"calculator"', action="execute")
def calculate(expression: str) -> float:
    return eval(expression)

# Set up authorization context
ctx = AuthorizationContext(
    principal="agent-123",
    principal_type="Agent",
    tenant_id="tenant-abc"
)

with authorization_context(ctx):
    result = calculate("2 + 2")  # Checks authorization first

Executor Pattern

from wl_apdp import AuthorizedExecutor

executor = AuthorizedExecutor(
    principal_id="agent-123",
    principal_type="Agent",
    tenant_id="tenant-abc"
)

# Check before executing
if executor.can_execute("web_search"):
    result = my_tool.run(query)

# Or execute with automatic check
result = executor.execute(
    action="execute",
    resource='Tool::"web_search"',
    func=lambda: my_tool.run(query)
)

Constraint Manifest (Agent-Native Interface)

The constraint manifest gives agents a complete picture of their operational boundaries at session start. This enables proactive self-governance rather than reactive discovery through failures.

Get Constraint Manifest

from wl_apdp import WlApdpSyncClient, ConstraintManifest

client = WlApdpSyncClient("http://localhost:8081/api")

# Fetch manifest at session start
manifest = client.get_constraint_manifest(
    agent_id="researcher",
    goal_id="goal-123",  # Optional
    include_guidance=True
)

# Manifest contains:
# - capabilities: What actions on what resources
# - constraints: What's forbidden, what needs approval
# - escalation_rules: When and how to escalate
# - guidance: Natural language role summary and best practices
print(manifest.guidance.role_summary)

Self-Governance with Cached Manifest

# Quick local check against cached constraints
if manifest.can_perform("read", "Document"):
    # Action appears to be allowed, safe to proceed
    pass
else:
    # Action is forbidden, don't even try
    handle_forbidden_action()

# Check if pre-flight is recommended
if manifest.should_preflight("execute"):
    # Use preflight before actual authorization
    pass

# Get required intents for an action
required_intents = manifest.get_required_intent("execute")

Pre-flight Checks

Pre-flight checks validate actions without consuming quota:

# Check if action would be allowed
result = client.preflight(
    principal='Agent::"researcher"',
    action='Action::"execute"',
    resource='Tool::"web_search"',
    intent_category="Research"
)

if result.allowed:
    # Safe to proceed with actual authorization
    auth_result = client.authorize(...)
else:
    print(f"Would be denied: {result.explanation}")
    for suggestion in result.suggestions:
        print(f"Suggestion: {suggestion.description}")
    if result.escalation_required:
        # Handle escalation
        print(f"Escalation needed: {result.escalation_required.reason}")

Query Capabilities

Discover what actions are available:

# What can I do with Documents?
result = client.query_capabilities(
    agent_id="researcher",
    resource_type="Document"
)

for cap in result.capabilities:
    print(f"Can {cap.action} with risk level {cap.risk_level}")
    if cap.requires_intent:
        print(f"  Requires intent: {cap.requires_intent}")

Integration Pattern

# At session start
manifest = client.get_constraint_manifest(agent_id)

# Before each action
def execute_action(action, resource, intent):
    # 1. Quick local check
    if not manifest.can_perform(action, resource):
        return {"denied": True, "reason": "Forbidden by manifest"}

    # 2. Pre-flight for uncertain actions
    if manifest.should_preflight(action):
        preflight = client.preflight(principal, action, resource, intent)
        if not preflight.allowed:
            return {"denied": True, "suggestions": preflight.suggestions}

    # 3. Actual authorization
    result = client.authorize(principal, action, resource, context)
    if result.is_allowed:
        return do_action()
    return {"denied": True}

Framework Integrations

CrewAI

from wl_apdp.contrib.crewai import AuthorizedCrewAITool, AuthorizedCrew
from crewai_tools import SerperDevTool

# Wrap individual tools
search_tool = AuthorizedCrewAITool(
    wrapped_tool=SerperDevTool(),
    resource='Tool::"web_search"'
)

# Or wrap entire crew
crew = AuthorizedCrew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    principal_id="crew-123",
    tenant_id="tenant-abc"
)
result = crew.kickoff()

LangGraph/LangChain

from wl_apdp.contrib.langgraph import authorized_langchain_tool, LangGraphAuthContext
from langchain_core.tools import tool

@authorized_langchain_tool(resource='Tool::"calculator"')
@tool
def calculator(expression: str) -> float:
    """Calculate a mathematical expression."""
    return eval(expression)

# Run graph with authorization
with LangGraphAuthContext(principal_id="agent-1", tenant_id="tenant-abc"):
    result = graph.invoke({"input": "calculate 2+2"})

AutoGen

from wl_apdp.contrib.autogen import AuthorizedFunction, AutoGenAuthContext

@authorized_function(resource='Tool::"weather"')
def get_weather(city: str) -> str:
    return f"Weather in {city}: Sunny"

# Run conversation with authorization
with AutoGenAuthContext(principal_id="user-123", tenant_id="org-abc"):
    user_proxy.initiate_chat(assistant, message="What's the weather?")

Authorization Context

The SDK uses Python's contextvars for propagating authorization context:

from wl_apdp import AuthorizationContext, authorization_context, get_current_context

ctx = AuthorizationContext(
    principal="agent-123",
    principal_type="Agent",      # User, Agent, Service, or Bot
    tenant_id="tenant-abc",
    session_id="session-001",
    attributes={"trust_level": 5}
)

# Sync context
with authorization_context(ctx):
    current = get_current_context()
    print(current.to_cedar_principal())  # 'Agent::"agent-123"'

# Async context
async with authorization_context(ctx):
    # Context available in all async code
    pass

Cedar Policy Examples

// Allow agents to execute tools
permit(
    principal is Agent,
    action == Action::"execute",
    resource is Tool
) when {
    principal.trust_level >= 3
};

// Tenant isolation
forbid(
    principal,
    action,
    resource
) when {
    context has tenant_id &&
    resource has tenant_id &&
    context.tenant_id != resource.tenant_id
};

// Rate limiting for bots
permit(
    principal is Bot,
    action,
    resource
) when {
    context.request_count < principal.rate_limit
};

API Reference

Clients

• WlApdpClient - Async HTTP client
• WlApdpSyncClient - Sync HTTP client

Context

• AuthorizationContext - Holds principal and context info
• authorization_context - Context manager for setting context
• get_current_context() - Get current context
• require_context() - Get context or raise

Decorators

• @authorized_tool - Add authorization to any function
• @authorize - Pre-authorization decorator
• @audit_action - Post-execution audit logging

Executors

• AuthorizedExecutor - Sync executor with authorization
• AsyncAuthorizedExecutor - Async executor

Models

• PrincipalType - Enum: User, Agent, Service, Bot
• AuthorizationRequest/Response - API models
• cedar_principal(), cedar_action(), cedar_resource() - Helpers

Exceptions

• AuthorizationDenied - Access was denied
• ServiceUnavailable - Cannot reach WL-APDP server
• ConfigurationError - Invalid configuration

Requirements

• Python 3.10+
• WL-APDP server running

License

Proprietary. See LICENSE for details.

Support

• Partner Portal: https://www.watchlight.ai/partner
• Email: [e-mail]