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 the GovernedCrew wrapper 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​

GovernedCrew Approach​

The simplest integration wraps your existing crew with GovernedCrew:

from controlzero import Client
from controlzero.integrations.crewai import GovernedCrew
from crewai import Agent, Task, Crew

cz = Client(api_key="cz_live_your_api_key_here")

# Create agents and tasks as normal
researcher = Agent(
    role="Senior Researcher",
    goal="Find and summarize relevant information",
    backstory="Expert research analyst",
)

research_task = Task(
    description="Research the topic and extract key findings",
    expected_output="A list of key findings",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[research_task],
)

# Wrap with governance enforcement
governed_crew = GovernedCrew(
    crew=crew,
    client=cz,
    crew_name="docs-crew",
)

result = governed_crew.kickoff()

GovernedCrew enforces a policy check at kickoff() and automatically wraps each agent and task with governance.

Tool Wrapper Approach​

Wrap individual CrewAI tools with policy enforcement:

from controlzero import Client
from crewai_tools import SerperDevTool, FileReadTool

cz = Client(api_key="cz_live_your_api_key_here")

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.guard(tool.name, args={"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​

CrewAI v1.x callback changes

As of CrewAI v1.x, the step_callback parameter on Crew was replaced with the EventListener / TraceCollectionListener pattern for observability hooks. The GovernedCrew wrapper approach shown above uses .kickoff() directly and remains compatible. For per-step event hooks, use CrewAI's EventListener API alongside Control Zero's GovernedCrew.

from controlzero import Client
from controlzero.integrations.crewai import GovernedCrew
from crewai import Agent, Task, Crew, Process

# [Control Zero]
cz = Client(api_key="cz_live_your_api_key_here")

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

writer = Agent(
    role="Report Writer",
    goal="Write clear, concise reports based on research findings",
    backstory="Professional technical writer",
    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,
)

# Wrap with governance enforcement and run
governed_crew = GovernedCrew(
    crew=crew,
    client=cz,
    crew_name="document-analysis-crew",
)

try:
    result = governed_crew.kickoff()
    print(result)
except Exception as e:
    print(f"Crew blocked by policy: {e}")

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.guard("search_web", args={"agent_id": "researcher"})  # allowed

# The writer is restricted
cz.guard("search_web", args={"agent_id": "writer"})  # denied by policy

Next Steps​