CrewAI Integration

Multi-layer security for CrewAI agents — PII detection, unauthorized access blocking, and orchestration bypass protection.

Overview

This example demonstrates integrating Agent Control with CrewAI to provide multi-layer security for a customer support agent:

1. Agent Control for security/compliance — PII detection, unauthorized access blocking
2. CrewAI Guardrails for quality validation — length, tone, structure

Both systems work together: Agent Control provides non-negotiable security blocks, while CrewAI Guardrails improve quality through iterative retries.

To run the full example yourself, visit the source on GitHub: CrewAI Example

Why This Matters

Customer support agents (human or AI) can accidentally:

• Leak PII (emails, phones, SSNs, credit cards) in responses or logs
• Access other users’ data when they shouldn’t
• Disclose passwords, credentials, or admin information
• Violate GDPR, CCPA, PCI-DSS compliance requirements

Agent Control solves this with three-layer protection:

Layer	Stage	What It Does
Layer 1	PRE-execution	Block unauthorized data access requests before processing
Layer 2	POST-execution	Block PII that the LLM accidentally generates in tool responses
Layer 3	Final output	Catch PII in the final crew output (orchestration bypass protection)

Prerequisites

1. Start the Agent Control Server

make server-run
```text

Verify the server is running:

```bash
curl http://localhost:8000/health
```text

### 2. Set OpenAI API Key

```bash
export OPENAI_API_KEY="your-key-here"
```text

### 3. Setup Controls

```bash
cd examples/crewai
python setup_content_controls.py
```text

This creates three controls under a single policy assigned to the agent:

| Control                          | Type  | Stage | What It Checks                                  |
| -------------------------------- | ----- | ----- | ----------------------------------------------- |
| `unauthorized-access-prevention` | Regex | pre   | Blocks requests for other users' data           |
| `pii-detection-output`           | Regex | post  | Blocks SSN, credit cards, emails, phones in tool output |
| Final output validation          | Regex | post  | Catches PII in final crew output (bypass protection) |

## Running the Example

```bash
cd examples/crewai
python content_agent_protection.py

Expected Behavior

Scenario 1: Unauthorized Access — PRE blocks immediately

🔍 [LAYER 1: Agent Control PRE-execution]
  Control: 'unauthorized-access-prevention'
  Status: Sending to server for validation...

🚫 [LAYER 1: Agent Control PRE] BLOCKED
  Reason: Control 'unauthorized-access-prevention' matched
  Unauthorized access attempt detected in input

The request is blocked before the LLM is ever called.

Scenario 2: PII Leakage — POST blocks output

✅ [LAYER 1: Agent Control PRE] PASSED

🤖 [Tool Execution] Calling LLM to generate response...

🚫 [LAYER 2: Agent Control POST] BLOCKED
  Reason: Control 'pii-detection-output' matched
  Tool executed but output contained PII violations

The tool ran, but the output was blocked because it contained PII.

Scenario 2.5: Quality Issues — CrewAI Guardrails retry

✅ [LAYER 1: Agent Control PRE] PASSED
✅ [LAYER 2: Agent Control POST] PASSED

[CrewAI Guardrails Check - Attempt 1]
❌ Guardrail failed: Response too short (15 words).

[CrewAI automatically retries with feedback...]

[CrewAI Guardrails Check - Attempt 2]
✅ All guardrails passed

CrewAI guardrails handle quality — they retry up to 3 times with feedback.

Scenario 3: Orchestration Bypass — Final output validation

🚫 Tool POST-execution control blocks PII in tool response
CrewAI agent generates own response with PII...

🚫 [LAYER 3: Agent Control FINAL] BLOCKED
   PII detected in crew output

Even when CrewAI’s agent works around the tool-level block by generating its own response containing PII, the final output validation catches it.

Agent Control vs CrewAI Guardrails

Aspect	Agent Control	CrewAI Guardrails
Purpose	Security / Compliance	Quality / Format
Checks	PII, unauthorized access	Length, tone, structure
On Failure	Block immediately ❌	Retry with feedback ✅
Retries	No (0)	Yes (up to 3)
Examples	Email, SSN, unauthorized	Too short, unprofessional

How It Works

The @control() decorator with CrewAI tools

CrewAI tools are synchronous, but the @control() decorator is async. Use an asyncio.run() wrapper:

async def _handle_ticket(ticket: str) -> str:
    """Handle customer support ticket (protected by @control)."""
    llm = LLM(model="gpt-4o-mini")
    prompt = f"You are a customer support agent. Respond to: {ticket}"
    return llm.call([{"role": "user", "content": prompt}])

# Set tool name (REQUIRED for tool step detection)
_handle_ticket.name = "handle_ticket"
_handle_ticket.tool_name = "handle_ticket"

# Apply @control decorator
controlled_func = control()(_handle_ticket)

# CrewAI tool wrapper
@tool("handle_ticket")
def handle_ticket_tool(ticket: str) -> str:
    """Handle customer support ticket with PII protection."""
    try:
        return asyncio.run(controlled_func(ticket=ticket))
    except ControlViolationError as e:
        return f"🚫 SECURITY VIOLATION: {e.message}"

Three-layer validation flow

Why three layers?

• Layers 1 & 2 protect at the tool boundary (standard @control() usage)
• Layer 3 protects against orchestration bypass — when CrewAI’s agent generates its own response containing PII after a tool was blocked

Control configuration

Unauthorized Access (PRE)

{
    "scope": {
        "step_types": ["tool"],
        "step_names": ["handle_ticket"],
        "stages": ["pre"]
    },
    "selector": {"path": "input.ticket"},
    "evaluator": {
        "plugin": "regex",
        "config": {
            "pattern": r"(?i)(show\s+me|what\s+is|give\s+me|tell\s+me).*"
                       r"(other\s+user|another\s+user|admin|password|credential)"
        }
    },
    "action": {"decision": "deny"}
}

Architecture

Files

File	Description
content_agent_protection.py	Main CrewAI crew with @control() integration
setup_content_controls.py	One-time setup for controls and policy
pyproject.toml	Dependencies

Troubleshooting

"event loop already running" error

CrewAI tools are sync by default, but @control requires async. Use an asyncio.run() wrapper:

@tool("my_tool")
def my_tool(arg: str) -> str:
    return asyncio.run(controlled_async_func(arg))

Control not triggering

Most common cause: the setup script was not run.

cd examples/crewai
python setup_content_controls.py

Also verify the tool name is set on the async function:

_my_func.name = "tool_name"
_my_func.tool_name = "tool_name"

"Arguments validation failed" error

CrewAI may pass tool arguments in different formats. Make the tool handle both:

@tool("handle_ticket")
def handle_ticket_tool(ticket: str) -> str:
  if isinstance(ticket, dict):
    ticket_text = ticket.get('ticket') or ticket.get('description') or str(ticket)
  else:
    ticket_text = str(ticket)

Source Code

See the full example on GitHub: CrewAI Example