Skip to main content

Python SDK Quickstart

Get started with the Watchlight Python SDK for AI agent authorization.

Installation​

pip install wl-apdp

Or with optional framework integrations:

pip install wl-apdp[crewai]      # CrewAI integration
pip install wl-apdp[langgraph]   # LangGraph integration
pip install wl-apdp[all]         # All integrations

Basic Usage​

Initialize the Client​

from wl_apdp import WatchlightClient

# Connect to WL-APDP server
client = WatchlightClient(
    base_url="http://localhost:8081",
    # Optional: API key for authentication
    api_key="your-api-key"
)

Check Authorization​

# Simple authorization check
result = client.authorize(
    principal="User::\"alice\"",
    action="read",
    resource="Document::\"quarterly-report\""
)

if result.allowed:
    print("Access granted!")
else:
    print(f"Access denied: {result.reasons}")

Authorization with Context​

# Include intent, goal, and delegation chain
result = client.authorize(
    principal="Agent::\"research-bot\"",
    action="execute",
    resource="Tool::\"web-search\"",
    context={
        "intent": "search for market data",
        "goal": "quarterly_report_preparation",
        "delegation_chain": [
            {
                "from": "User::\"analyst\"",
                "to": "Agent::\"research-bot\"",
                "scope": ["read", "execute"]
            }
        ]
    }
)

Async Support​

import asyncio
from wl_apdp import AsyncWatchlightClient

async def main():
    async with AsyncWatchlightClient("http://localhost:8081") as client:
        result = await client.authorize(
            principal="User::\"alice\"",
            action="read",
            resource="Document::\"report\""
        )
        print(result)

asyncio.run(main())

Policy Management​

Create a Policy​

policy = client.create_policy(
    id="agent-tool-access",
    name="Agent Tool Access Policy",
    code='''
        permit(
            principal is Agent,
            action == Action::"execute",
            resource is Tool
        ) when {
            context.intent != "" &&
            context.delegation_chain.length > 0
        };
    ''',
    description="Allow agents to execute tools with valid intent and delegation"
)

List Policies​

policies = client.list_policies()
for policy in policies:
    print(f"{policy.id}: {policy.name}")

Validate a Policy​

validation = client.validate_policy('''
    permit(
        principal == User::"admin",
        action,
        resource
    );
''')

if validation.valid:
    print("Policy is valid")
else:
    print(f"Validation errors: {validation.errors}")

Delegation Chains​

Create a Delegation​

from wl_apdp import Delegation

# Create delegation from user to agent
delegation = Delegation(
    from_principal="User::\"alice\"",
    to_principal="Agent::\"assistant\"",
    scope=["read", "execute"],
    resources=["Document::*", "Tool::*"],
    expires_at="2024-12-31T23:59:59Z"
)

# Use in authorization request
result = client.authorize(
    principal="Agent::\"assistant\"",
    action="read",
    resource="Document::\"report\"",
    delegation_chain=[delegation]
)

Build Delegation Chains​

from wl_apdp import DelegationChain

chain = DelegationChain()

# User delegates to coordinator
chain.add(
    from_principal="User::\"manager\"",
    to_principal="Agent::\"coordinator\"",
    scope=["read", "write", "delegate"]
)

# Coordinator delegates to worker
chain.add(
    from_principal="Agent::\"coordinator\"",
    to_principal="Agent::\"worker\"",
    scope=["read"]  # Reduced scope
)

# Check authorization with chain
result = client.authorize(
    principal="Agent::\"worker\"",
    action="read",
    resource="Document::\"data\"",
    delegation_chain=chain
)

Error Handling​

from wl_apdp import WatchlightError, AuthorizationDenied, PolicyValidationError

try:
    result = client.authorize(
        principal="Agent::\"bot\"",
        action="delete",
        resource="Document::\"important\""
    )
except AuthorizationDenied as e:
    print(f"Access denied: {e.reasons}")
    for reason in e.policy_reasons:
        print(f"  Policy {reason.policy_id}: {reason.description}")

except PolicyValidationError as e:
    print(f"Invalid request: {e}")

except WatchlightError as e:
    print(f"Watchlight error: {e}")

Context Manager​

from wl_apdp import WatchlightClient

# Automatic resource cleanup
with WatchlightClient("http://localhost:8081") as client:
    result = client.authorize(...)
    # Connection closed automatically

Configuration​

Environment Variables​

import os

# Set via environment
os.environ["WATCHLIGHT_URL"] = "http://localhost:8081"
os.environ["WATCHLIGHT_API_KEY"] = "your-key"

# Client auto-detects
client = WatchlightClient()

Configuration File​

# config.yaml
# watchlight:
#   url: http://localhost:8081
#   api_key: your-key
#   timeout: 30

client = WatchlightClient.from_config("config.yaml")

Health Checks​

# Check server health
health = client.health()
print(f"Status: {health.status}")  # "healthy" or "unhealthy"

# Check readiness
if client.is_ready():
    print("Server is ready")

Logging​

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

# Or configure specific logger
logger = logging.getLogger("wl_apdp")
logger.setLevel(logging.DEBUG)

Full Example​

from wl_apdp import WatchlightClient, Delegation

def check_agent_access():
    client = WatchlightClient("http://localhost:8081")

    # Create delegation from user
    delegation = Delegation(
        from_principal="User::\"data-analyst\"",
        to_principal="Agent::\"analysis-bot\"",
        scope=["read", "execute"],
        resources=["Database::analytics", "Tool::sql-query"]
    )

    # Check if agent can query database
    result = client.authorize(
        principal="Agent::\"analysis-bot\"",
        action="execute",
        resource="Tool::\"sql-query\"",
        context={
            "intent": "analyze_sales_trends",
            "goal": "monthly_report",
            "query_type": "read_only"
        },
        delegation_chain=[delegation]
    )

    if result.allowed:
        print("Agent authorized to run query")
        print(f"Matched policies: {[r.policy_id for r in result.reasons]}")
    else:
        print("Agent not authorized")
        print(f"Denial reasons: {result.denial_reasons}")

    return result.allowed

if __name__ == "__main__":
    check_agent_access()

Next Steps​