Skip to main content

CrewAI Integration

Integrate Watchlight authorization with CrewAI multi-agent systems.

Installation​

pip install wl-apdp[crewai]

Quick Start​

from crewai import Agent, Task, Crew
from wl_apdp.crewai import AuthorizedAgent, AuthorizedTool, WatchlightCrewHook

# Initialize Watchlight integration
hook = WatchlightCrewHook(
    watchlight_url="http://localhost:8081",
    default_delegator="User::\"crew-manager\""
)

# Create an authorized agent
researcher = AuthorizedAgent(
    name="Researcher",
    role="Senior Research Analyst",
    goal="Conduct thorough research on given topics",
    backstory="Expert researcher with access to various data sources",
    # Watchlight authorization settings
    principal_id="Agent::\"researcher\"",
    allowed_intents=["search_*", "analyze_*", "summarize_*"],
    allowed_goals=["research", "analysis", "reporting"]
)

# Create a task
research_task = Task(
    description="Research the latest AI trends",
    agent=researcher,
    # Task metadata becomes authorization context
    context={
        "intent": "search_ai_trends",
        "goal": "research"
    }
)

# Create and run the crew
crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    callbacks=[hook]  # Attach Watchlight hook
)

result = crew.kickoff()

AuthorizedAgent​

AuthorizedAgent extends CrewAI's Agent with authorization:

from wl_apdp.crewai import AuthorizedAgent

agent = AuthorizedAgent(
    # Standard CrewAI agent parameters
    name="Data Analyst",
    role="Senior Data Analyst",
    goal="Analyze data and provide insights",
    backstory="...",

    # Watchlight authorization parameters
    principal_id="Agent::\"data-analyst\"",
    delegated_by="User::\"analytics-manager\"",
    delegated_scope=["read", "execute"],

    # Allowed intents (patterns)
    allowed_intents=[
        "query_*",
        "analyze_*",
        "visualize_*"
    ],

    # Allowed goals
    allowed_goals=[
        "data_analysis",
        "reporting",
        "dashboard_creation"
    ],

    # Trust level (affects tool access)
    trust_level=3
)

AuthorizedTool​

Wrap CrewAI tools with authorization:

from crewai_tools import SerperDevTool
from wl_apdp.crewai import AuthorizedTool

# Wrap existing tool
search_tool = AuthorizedTool(
    tool=SerperDevTool(),
    resource_id="Tool::\"web-search\"",
    required_scope=["execute"],
    risk_level=1
)

# Use with agent
agent = AuthorizedAgent(
    name="Researcher",
    tools=[search_tool],
    principal_id="Agent::\"researcher\""
)

Creating Custom Authorized Tools​

from wl_apdp.crewai import AuthorizedTool, authorized_tool

@authorized_tool(
    resource_id="Tool::\"database-query\"",
    required_scope=["read", "execute"],
    risk_level=2
)
def query_database(query: str) -> str:
    """Execute a read-only database query."""
    # Tool implementation
    return execute_query(query)

WatchlightCrewHook​

The hook intercepts agent actions for authorization:

from wl_apdp.crewai import WatchlightCrewHook

hook = WatchlightCrewHook(
    watchlight_url="http://localhost:8081",
    api_key="your-api-key",

    # Default delegator for all agents
    default_delegator="User::\"system\"",

    # Behavior on authorization failure
    on_denied="raise",  # "raise", "skip", "log"

    # Include task context in authorization
    include_task_context=True,

    # Log authorization decisions
    log_decisions=True
)

crew = Crew(
    agents=[...],
    tasks=[...],
    callbacks=[hook]
)

Hook Events​

class CustomHook(WatchlightCrewHook):
    def on_authorization_granted(self, agent, action, resource, result):
        """Called when authorization is granted."""
        print(f"Granted: {agent.name} -> {action} on {resource}")

    def on_authorization_denied(self, agent, action, resource, result):
        """Called when authorization is denied."""
        print(f"Denied: {agent.name} -> {action} on {resource}")
        print(f"Reasons: {result.denial_reasons}")

    def on_delegation_created(self, from_agent, to_agent, scope):
        """Called when one agent delegates to another."""
        print(f"Delegation: {from_agent} -> {to_agent}")

Multi-Agent Delegation​

Hierarchical Crews​

from wl_apdp.crewai import AuthorizedAgent, DelegatingAgent

# Manager agent that can delegate
manager = DelegatingAgent(
    name="Project Manager",
    principal_id="Agent::\"manager\"",
    delegated_by="User::\"executive\"",
    can_delegate_to=["Agent::\"researcher\"", "Agent::\"writer\""],
    delegation_scope=["read", "write"]
)

# Worker agents
researcher = AuthorizedAgent(
    name="Researcher",
    principal_id="Agent::\"researcher\"",
    # Will receive delegation from manager
)

writer = AuthorizedAgent(
    name="Writer",
    principal_id="Agent::\"writer\""
)

# Crew with delegation
crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[...],
    process=Process.hierarchical,
    manager_agent=manager
)

Automatic Delegation Chains​

hook = WatchlightCrewHook(
    watchlight_url="http://localhost:8081",
    # Build delegation chain from crew hierarchy
    auto_delegation=True,
    # Root delegator
    root_delegator="User::\"admin\""
)

Task Authorization​

Task-Level Context​

task = Task(
    description="Analyze customer feedback",
    agent=analyst,
    # Authorization context
    watchlight_context={
        "intent": "analyze_sentiment",
        "goal": "customer_insights",
        "data_classification": "internal"
    }
)

Conditional Task Execution​

from wl_apdp.crewai import ConditionalTask

# Task only runs if authorized
task = ConditionalTask(
    description="Access sensitive data",
    agent=analyst,
    # Required authorization
    requires_authorization={
        "action": "read",
        "resource": "Database::\"customer-pii\"",
        "context": {
            "purpose": "authorized_analysis"
        }
    },
    # Fallback if denied
    fallback_task=alternative_task
)

Full Example​

from crewai import Crew, Process
from crewai_tools import SerperDevTool, FileReadTool
from wl_apdp.crewai import (
    AuthorizedAgent,
    AuthorizedTool,
    WatchlightCrewHook,
    ConditionalTask
)

# Initialize hook
hook = WatchlightCrewHook(
    watchlight_url="http://localhost:8081",
    default_delegator="User::\"research-lead\"",
    on_denied="skip",
    log_decisions=True
)

# Authorized tools
search = AuthorizedTool(
    tool=SerperDevTool(),
    resource_id="Tool::\"web-search\"",
    risk_level=1
)

file_reader = AuthorizedTool(
    tool=FileReadTool(),
    resource_id="Tool::\"file-reader\"",
    risk_level=2
)

# Agents
researcher = AuthorizedAgent(
    name="Research Specialist",
    role="Senior Researcher",
    goal="Find comprehensive information on topics",
    backstory="Experienced researcher with strong analytical skills",
    principal_id="Agent::\"researcher\"",
    allowed_intents=["search_*", "find_*", "analyze_*"],
    allowed_goals=["research", "competitive_analysis"],
    tools=[search, file_reader],
    trust_level=3
)

writer = AuthorizedAgent(
    name="Content Writer",
    role="Technical Writer",
    goal="Create clear, well-structured content",
    backstory="Expert at synthesizing research into readable content",
    principal_id="Agent::\"writer\"",
    allowed_intents=["write_*", "summarize_*", "format_*"],
    allowed_goals=["content_creation", "documentation"],
    tools=[file_reader],
    trust_level=2
)

# Tasks
research_task = ConditionalTask(
    description="Research the latest developments in AI agents",
    agent=researcher,
    watchlight_context={
        "intent": "search_ai_developments",
        "goal": "research"
    }
)

writing_task = ConditionalTask(
    description="Write a summary report based on research",
    agent=writer,
    watchlight_context={
        "intent": "write_summary",
        "goal": "content_creation"
    },
    context=[research_task]  # Depends on research
)

# Create crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    callbacks=[hook]
)

# Run
result = crew.kickoff()
print(result)

Configuration​

Environment Variables​

export WATCHLIGHT_URL=http://localhost:8081
export WATCHLIGHT_API_KEY=your-key
export WATCHLIGHT_DEFAULT_DELEGATOR="User::\"admin\""

YAML Configuration​

# crewai_config.yaml
watchlight:
  url: http://localhost:8081
  api_key: ${WATCHLIGHT_API_KEY}
  default_delegator: "User::\"admin\""
  on_denied: skip
  log_decisions: true

agents:
  researcher:
    principal_id: "Agent::\"researcher\""
    allowed_intents: ["search_*", "analyze_*"]
    trust_level: 3
hook = WatchlightCrewHook.from_config("crewai_config.yaml")

Next Steps​