CrewAI Integration

from crewai import Crew, Agent, Task
from raxe import Raxe
from raxe import RaxeCrewGuard

# Create guard (default: log-only mode - safe for production)
guard = RaxeCrewGuard(Raxe())

# Wrap a crew for automatic scanning
protected_crew = guard.protect_crew(crew)
result = protected_crew.kickoff()

from crewai import Crew, Agent, Task
from raxe import Raxe
from raxe import RaxeCrewGuard

guard = RaxeCrewGuard(Raxe())

# Create agents
researcher = Agent(
    role="Researcher",
    goal="Research AI topics",
    backstory="Expert researcher",
)

writer = Agent(
    role="Writer",
    goal="Write articles",
    backstory="Technical writer",
)

# Create tasks
research_task = Task(
    description="Research AI safety",
    agent=researcher,
)

write_task = Task(
    description="Write an article",
    agent=writer,
)

# Create crew with RAXE callbacks for step-by-step scanning
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    step_callback=guard.step_callback,  # Scan each agent step
    task_callback=guard.task_callback,  # Scan task outputs
)

result = crew.kickoff()

from raxe import Raxe
from raxe import RaxeCrewGuard, CrewGuardConfig, ScanMode

# Blocking mode for high-severity threats
config = CrewGuardConfig(
    mode=ScanMode.BLOCK_ON_HIGH,  # Block HIGH and CRITICAL severity
    scan_step_outputs=True,       # Scan agent reasoning steps
    scan_task_outputs=True,       # Scan task completion outputs
    scan_tool_inputs=True,        # Scan tool call arguments
    scan_tool_outputs=True,       # Scan tool return values
    scan_agent_thoughts=True,     # Scan agent internal reasoning
    wrap_tools=True,              # Auto-wrap tools for scanning
)

guard = RaxeCrewGuard(Raxe(), config=config)

from crewai.tools import BaseTool
from raxe import Raxe
from raxe import RaxeCrewGuard, CrewGuardConfig, ScanMode

config = CrewGuardConfig(
    mode=ScanMode.BLOCK_ON_HIGH,
    wrap_tools=True,       # Enable automatic tool wrapping
    scan_tool_inputs=True, # Scan arguments passed to tools
    scan_tool_outputs=True,# Scan values returned from tools
)

guard = RaxeCrewGuard(Raxe(), config=config)

# Wrap individual tools for scanning
wrapped_tool = guard.wrap_tool(my_tool)

# Or wrap all tools in a list
wrapped_tools = guard.wrap_tools([tool1, tool2, tool3])

from raxe import CrewGuardConfig, ScanMode

def on_threat_detected(message: str, result):
    """Called when any threat is detected."""
    print(f"Threat: {result.severity}")
    send_alert(f"CrewAI threat detected: {result.severity}")

def on_blocked(message: str, result):
    """Called when a threat triggers blocking."""
    print(f"Blocked: {message[:50]}...")
    log_security_event(result)

config = CrewGuardConfig(
    mode=ScanMode.BLOCK_ON_HIGH,
    on_threat=on_threat_detected,  # Custom threat handling
    on_block=on_blocked,           # Custom block handling
)

from raxe import RaxeBlockedError, RaxeException

try:
    result = protected_crew.kickoff()
except RaxeBlockedError as e:
    # Threat was detected and blocked
    print(f"Crew blocked: {e.severity}")
    print(f"Rule: {e.rule_id}")
    # Handle blocked execution - return safe response
except RaxeException as e:
    # Other RAXE errors (config, initialization)
    logger.error(f"RAXE error: {e}")
    # Decide: fail open or fail closed

from raxe import CrewGuardConfig, ScanMode

config = CrewGuardConfig(
    # What to scan (all default to True)
    scan_step_outputs=True,     # Agent step outputs
    scan_task_outputs=True,     # Task completion outputs
    scan_tool_inputs=True,      # Tool call arguments
    scan_tool_outputs=True,     # Tool return values
    scan_agent_thoughts=True,   # Agent internal reasoning
    scan_crew_inputs=True,      # Crew kickoff inputs
    scan_crew_outputs=True,     # Final crew outputs

    # Context options for better threat detection
    include_agent_context=True,  # Include agent name/role in scan
    include_task_context=True,   # Include task description in scan
    max_thought_length=5000,     # Truncate thoughts longer than this

    # Tool-specific scanning settings
    wrap_tools=False,                           # Manual tool wrapping
    tool_scan_mode=ScanMode.BLOCK_ON_THREAT,   # Stricter mode for tools
)