CrewAI Integration

New to RAXE? Start with the Quickstart and learn how detection works.

Overview

RAXE provides think-time security for CrewAI multi-agent crews — real-time threat detection during agent inference, before task execution. Protect agent-to-agent communications, task handoffs, and tool invocations. What RAXE scans:

Agent messages and reasoning
Task outputs and handoffs
Tool inputs and outputs
Inter-agent communications

Installation

pip install raxe[crewai]

Quick Start

quick_start.py

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()

Using Callbacks

Alternatively, use callbacks directly in your crew:

callbacks.py

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()

Configuration

config.py

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)

Available Modes

Mode	Description
`ScanMode.LOG_ONLY`	Log threats, allow execution (default)
`ScanMode.BLOCK_ON_THREAT`	Block any detected threat
`ScanMode.BLOCK_ON_HIGH`	Block HIGH and CRITICAL severity
`ScanMode.BLOCK_ON_CRITICAL`	Block only CRITICAL severity

Tool Scanning

Automatically scan tool inputs and outputs:

tool_scanning.py

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])

Threat Callbacks

Handle detected threats with custom callbacks:

threat_callbacks.py

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
)

Error Handling

error_handling.py

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

Advanced Options

advanced.py

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
)

Best Practices

Start with log-only mode

Monitor threats before enabling blocking:

log_only.py

from raxe import Raxe
from raxe import RaxeCrewGuard

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

Wrap tools for comprehensive coverage

Enable tool wrapping for full protection:

wrap_tools.py

from raxe import Raxe
from raxe import RaxeCrewGuard, CrewGuardConfig

config = CrewGuardConfig(wrap_tools=True)  # Auto-wrap all tools
guard = RaxeCrewGuard(Raxe(), config=config)

Use appropriate blocking threshold

Choose blocking threshold based on risk tolerance:

thresholds.py

from raxe import CrewGuardConfig, ScanMode

# Strict: Block any threat (for high-security environments)
config = CrewGuardConfig(mode=ScanMode.BLOCK_ON_THREAT)

# Balanced: Block HIGH and CRITICAL (recommended)
config = CrewGuardConfig(mode=ScanMode.BLOCK_ON_HIGH)

# Minimal: Block only CRITICAL (low false positive tolerance)
config = CrewGuardConfig(mode=ScanMode.BLOCK_ON_CRITICAL)

Supported Versions

CrewAI Version	Status
crewai >= 0.28.0	Supported

Getting Started

How It Works

Protect Your AI

Guides

Advanced

Enterprise

Reference

Overview

Installation

Quick Start

Using Callbacks

Configuration

Available Modes

Tool Scanning

Threat Callbacks

Error Handling

Advanced Options

Best Practices

Supported Versions

​Overview

​Installation

​Quick Start

​Using Callbacks

​Configuration

​Available Modes

​Tool Scanning

​Threat Callbacks

​Error Handling

​Advanced Options

​Best Practices

​Supported Versions

Overview

Installation

Quick Start

Using Callbacks

Configuration

Available Modes

Tool Scanning

Threat Callbacks

Error Handling

Advanced Options

Best Practices

Supported Versions