CrewAI Integration

Wrap CrewAI tasks with Nomotic governance using the GovernedCrewTask pattern.

Installation

pip install nomotic crewai

Pattern

GovernedCrewTask wraps a task function with governance evaluation. If governance denies the action, the task raises a RuntimeError with the verdict and UCS score instead of executing.

Example

from nomotic.governed_agent import GovernedAgentBase, GovernanceVetoError
from typing import Any, Callable


class GovernedCrewTask:
    """Wraps a CrewAI task with Nomotic governance."""

    def __init__(self, governed_agent: GovernedAgentBase):
        self.agent = governed_agent

    def execute(
        self,
        action_type: str,
        target: str,
        task_fn: Callable[[], str],
        parameters: dict[str, Any] | None = None,
    ) -> str:
        """Execute a task with governance.

        Raises RuntimeError with verdict details if denied.
        """
        try:
            return self.agent.governed_run(
                action_type=action_type,
                target=target,
                parameters=parameters,
                execute_fn=task_fn,
            )
        except GovernanceVetoError as e:
            raise RuntimeError(
                f"Governance denied: {e.verdict} (UCS: {e.ucs_score:.3f})"
            ) from e


# Usage with CrewAI
governed_task = GovernedCrewTask(governed_agent)

try:
    result = governed_task.execute(
        action_type="analyze",
        target="market/report",
        task_fn=lambda: crew_agent.run("Analyze Q2 market trends"),
    )
except RuntimeError as e:
    print(f"Task blocked: {e}")

The GovernedCrewTask wrapper can be applied to any task in a CrewAI crew, giving you per-task governance evaluation with full audit trail coverage.