CrewAI Integration Guide¶

Integrate ACGP with CrewAI for task-based team workflow governance.

Community Integration

This integration is community-maintained and may not track the latest SDK changes. Contributions welcome — see Contributing.

Overview¶

CrewAI organizes agents into crews with roles and tasks. ACGP provides governance at the task and agent level, ensuring safe execution of team workflows.

Installation¶

PythonTypeScript

pip install acgp-crewai

For production runtime posture in the underlying Python steward:

pip install "acgp-sdk[postgres]"

Coming Soon

TypeScript CrewAI integration is planned for a future release. Python is available now.

Basic Usage¶

Python

from crewai import Crew, Agent, Task
from acgp_crewai import GovernedCrew
from acgp import GovernanceSteward, PostgresStateStorage

# Your CrewAI setup
researcher = Agent(role="Researcher", goal="Research topics")
writer = Agent(role="Writer", goal="Write content")

task1 = Task(description="Research AI safety", agent=researcher)
task2 = Task(description="Write article", agent=writer)

crew = Crew(agents=[researcher, writer], tasks=[task1, task2])

# Add governance
steward = GovernanceSteward.production(
    blueprint_file="blueprint.yaml",
    state_storage=PostgresStateStorage(connection_string="postgresql://runtime/acgp"),
)
governed_crew = GovernedCrew(
    crew,
    steward,
    agent_ids={
        "Researcher": "urn:acgp:agent:researcher:prod:7f4c9d2a",
        "Writer": "urn:acgp:agent:writer:prod:9b2a41cc",
    },
)

# Execute with governance
result = governed_crew.kickoff()

Features¶

Crew-Level Governance¶

Wraps entire crew workflows:

governed_crew = GovernedCrew(
    crew,
    steward,
    crew_name="content_team",
    agent_ids={
        "Researcher": "urn:acgp:agent:researcher:prod:7f4c9d2a",
        "Writer": "urn:acgp:agent:writer:prod:9b2a41cc",
    },
)
result = governed_crew.kickoff(inputs={"topic": "AI safety"})

Task-Level Governance¶

Govern individual tasks:

from acgp_crewai import GovernedTask

governed_task = GovernedTask(
    task,
    steward,
    agent_id="urn:acgp:agent:researcher:prod:7f4c9d2a",
)
result = governed_task.execute()

Agent-Level Governance¶

Govern specific agents:

from acgp_crewai import GovernedAgent

governed_agent = GovernedAgent(
    agent,
    steward,
    agent_id="urn:acgp:agent:researcher:prod:7f4c9d2a",
)
result = governed_agent.execute_task(task)

Async Support¶

from acgp_crewai import AsyncGovernedCrew

async_crew = AsyncGovernedCrew(
    crew,
    steward,
    agent_ids={
        "Researcher": "urn:acgp:agent:researcher:prod:7f4c9d2a",
        "Writer": "urn:acgp:agent:writer:prod:9b2a41cc",
    },
)
result = await async_crew.kickoff()

Advanced Features¶

Governed Principal Identity

CrewAI wrappers now require explicit governed principal identifiers. Pass agent_id for single-agent or task wrappers, and use agent_ids for crew wrappers keyed by each agent's display label or role.

Get Governance Metrics¶

metrics = governed_crew.get_metrics()
print(f"Crew: {metrics['crew_name']}")
print(f"Total interventions: {metrics['total_interventions']}")

Exception Handling¶

from acgp_crewai import InterventionError, CrewGovernanceError

try:
    result = governed_crew.kickoff()
except InterventionError as e:
    print(f"Crew blocked: {e.message}")
    print(f"Trace ID: {e.trace_id}")

Control Intervention Behavior¶

# Don't raise exceptions on interventions
governed_crew = GovernedCrew(
    crew,
    steward,
    raise_on_intervention=False  # Log instead of raising
)

Best Practices¶

Role-Based Policies

Define different governance rules for different agent roles (researcher, writer, reviewer).

Task Dependencies

Governance respects CrewAI's task dependencies - blocked tasks halt the workflow.

Automatic Wrapping

GovernedCrew automatically wraps all agents and tasks in the crew with governance.

Next: Autogen Integration