Skip to main content

CrewAI Integration

Enforce Control Zero policies across multi-agent CrewAI crews and tasks.

Overview​

CrewAI orchestrates multiple AI agents working together on complex tasks. Control Zero integrates with CrewAI through step callbacks and tool wrappers, giving you policy enforcement over which tools each agent can use, which models they can call, and what actions they can take.

Installation​

pip install controlzero crewai crewai-tools

Setup​

Step Callback Approach​

The simplest integration uses the built-in create_step_callback() from the SDK:

import controlzero
from controlzero.integrations.crewai import create_step_callback
from crewai import Agent, Task, Crew

cz = controlzero.init()
callback = create_step_callback(cz, agent_id="docs-crew")

# Create agents with policy enforcement
researcher = Agent(
    role="Senior Researcher",
    goal="Find and summarize relevant information",
    backstory="Expert research analyst",
)

crew = Crew(
    agents=[researcher],
    tasks=[...],
    step_callback=callback,
)

The callback enforces both agent.step (with the agent role as resource) and tool.call (with the tool name as resource) at every step.

Tool Wrapper Approach​

Wrap individual CrewAI tools with policy enforcement:

import controlzero
from crewai_tools import SerperDevTool, FileReadTool

cz = controlzero.ControlZero()
cz.initialize()

def enforced_tool(tool, agent_id: str = ""):
    """Wrap a CrewAI tool with Control Zero policy enforcement."""
    original_run = tool._run

    def wrapped_run(*args, **kwargs):
        cz.enforce(
            action="tool.call",
            resource=f"tool/{tool.name}",
            context={"agent_id": agent_id},
        )
        return original_run(*args, **kwargs)

    tool._run = wrapped_run
    return tool


# Wrap tools with enforcement
search_tool = enforced_tool(SerperDevTool(), agent_id="researcher")
file_tool = enforced_tool(FileReadTool(), agent_id="researcher")

Full Example: Document Analysis Crew​

import controlzero
from crewai import Agent, Task, Crew, Process

# -- Control Zero --
cz = controlzero.ControlZero()
cz.initialize()

def enforce_step(step_output):
    """Policy enforcement callback for all agent steps."""
    if hasattr(step_output, "tool") and step_output.tool:
        cz.enforce(
            action="tool.call",
            resource=f"tool/{step_output.tool}",
            context={"agent_id": str(step_output.agent)},
        )

# -- Agents --
researcher = Agent(
    role="Document Researcher",
    goal="Analyze documents and extract key findings",
    backstory="Expert at reading and summarizing technical documents",
    step_callback=enforce_step,
    verbose=True,
)

writer = Agent(
    role="Report Writer",
    goal="Write clear, concise reports based on research findings",
    backstory="Professional technical writer",
    step_callback=enforce_step,
    verbose=True,
)

# -- Tasks --
research_task = Task(
    description="Analyze the quarterly sales report and extract key metrics",
    expected_output="A list of key findings with supporting data",
    agent=researcher,
)

writing_task = Task(
    description="Write an executive summary based on the research findings",
    expected_output="A one-page executive summary",
    agent=writer,
)

# -- Crew --
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
)

# Run with policy enforcement active
try:
    result = crew.kickoff()
    print(result)
except controlzero.PolicyDeniedError as e:
    print(f"Crew blocked by policy: {e.message}")

Example Policy​

Allow the researcher to use search tools but deny file writes. Allow the writer to use text tools but deny web access:

{
  "name": "document-crew-policy",
  "rules": [
    {
      "effect": "allow",
      "action": "llm.generate",
      "resource": "model/*"
    },
    {
      "effect": "allow",
      "action": "tool.call",
      "resource": "tool/search_web"
    },
    {
      "effect": "allow",
      "action": "tool.call",
      "resource": "tool/read_file"
    },
    {
      "effect": "deny",
      "action": "tool.call",
      "resource": "tool/write_file"
    },
    {
      "effect": "deny",
      "action": "tool.call",
      "resource": "tool/execute_code"
    }
  ]
}

Per-Agent Policies​

Control Zero supports context-based policy conditions. Use the agent_id in your context to apply different rules to different agents in the same crew:

# The researcher gets broader tool access
cz.enforce(
    action="tool.call",
    resource="tool/search_web",
    context={"agent_id": "researcher"},  # allowed
)

# The writer is restricted
cz.enforce(
    action="tool.call",
    resource="tool/search_web",
    context={"agent_id": "writer"},  # denied by policy
)

Next Steps​