CrewAI + Tork Governance

Crew Governance

Wrap entire crews with one call

Agent Control

Govern individual agents

Task Validation

Check inputs and outputs

Local Receipts

A receipt for every check, on-device

pip install tork-governance crewai crewai-tools

from crewai import Agent, Task, Crew
from tork_governance.adapters.crewai import TorkCrewAIMiddleware

# On-device governance. No api_key = zero network calls.
# Pass api_key="tork_live_..." to also report metadata-only attestations.
middleware = TorkCrewAIMiddleware(agent_id="my-crew")

# Create a standard CrewAI agent
researcher = Agent(
    role="Research Analyst",
    goal="Find accurate information about given topics",
    backstory="You are an expert researcher with attention to detail.",
    verbose=True
)

# Wrap the agent with governance
governed_researcher = middleware.wrap_agent(researcher)

# governed_researcher.execute_task(task) now:
# - governs the task description before execution (PII redacted, or the
#   task raises ValueError when the configured action is DENY)
# - governs the agent's output before it is returned
# - records a local receipt for each check in middleware.receipts

Middleware Methods

from crewai import Agent, Task
from tork_governance.adapters.crewai import TorkCrewAIMiddleware, GovernedAgent

middleware = TorkCrewAIMiddleware(agent_id="content-team")

# Create your CrewAI agents
writer = Agent(
    role="Content Writer",
    goal="Write engaging and accurate content",
    backstory="Expert content creator with SEO knowledge.",
    verbose=True
)

editor = Agent(
    role="Editor",
    goal="Review and improve content quality",
    backstory="Experienced editor focused on clarity and accuracy.",
    verbose=True
)

# Wrap agents individually for granular control
governed_writer = GovernedAgent(writer, middleware)
governed_editor = GovernedAgent(editor, middleware)

# Create a task
writing_task = Task(
    description="Write a blog post about AI safety best practices",
    expected_output="A 500-word blog post with actionable tips",
    agent=governed_writer
)

# Execute with governance:
# - the task description is governed before execution
# - the output is scanned for PII and redacted before it is returned
result = governed_writer.execute_task(writing_task)

from crewai import Agent, Task, Crew, Process
from tork_governance.adapters.crewai import TorkCrewAIMiddleware, GovernedCrew

middleware = TorkCrewAIMiddleware(agent_id="content-crew")

# Define agents
researcher = Agent(
    role="Research Analyst",
    goal="Gather comprehensive information",
    backstory="Expert at finding and synthesizing information.",
    allow_delegation=False
)

writer = Agent(
    role="Content Writer",
    goal="Create engaging content from research",
    backstory="Skilled writer who turns research into readable content.",
    allow_delegation=False
)

reviewer = Agent(
    role="Quality Reviewer",
    goal="Ensure content accuracy and quality",
    backstory="Detail-oriented reviewer with high standards.",
    allow_delegation=False
)

# Define tasks
research_task = Task(
    description="Research the topic: {topic}",
    expected_output="Comprehensive research notes with sources",
    agent=researcher
)

writing_task = Task(
    description="Write an article based on the research",
    expected_output="A well-structured article",
    agent=writer
)

review_task = Task(
    description="Review the article for accuracy and quality",
    expected_output="Final approved article with any corrections",
    agent=reviewer
)

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

# Wrap the crew with governance
governed_crew = GovernedCrew(crew, middleware)

# kickoff() governs every string value in inputs before the crew runs
# and governs the crew's final output before returning it.
# It does not intercept messages the agents exchange with each other.
result = governed_crew.kickoff(inputs={"topic": "AI Governance Best Practices"})

from crewai import Task
from tork_governance import GovernanceAction
from tork_governance.adapters.crewai import TorkCrewAIMiddleware

middleware = TorkCrewAIMiddleware(agent_id="analysis-task")

task = Task(
    description="Analyze customer feedback and summarize key themes",
    expected_output="A summary report with top 5 themes and recommendations"
)

# Govern the task description before you run it
check = middleware.process_input(task.description)
if check.action == GovernanceAction.DENY:
    raise ValueError(f"Task blocked: {check.receipt.receipt_id}")
task.description = check.output  # redacted if PII was present

# ... execute the task with your crew ...

# Govern the output after execution
output = "Customer feedback analysis shows..."
result = middleware.process_output(output)
safe_output = result.output             # PII redacted
print(result.pii.has_pii, result.pii.types)
print(result.receipt.receipt_id)

# Govern a tool call before it runs: the tool name and arguments are
# serialised and scanned for PII like any other text
tool_check = middleware.check_tool_call("send_email", {"to": "jane@example.com"})
print(tool_check.action, tool_check.pii.types)

from crewai import Agent, Task, Crew, Process
from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.crewai import TorkCrewAIMiddleware, GovernedCrew

def create_governed_crew() -> GovernedCrew:
    """Create a fully governed CrewAI crew."""

    # DENY instead of the default REDACT: inputs containing PII raise
    # ValueError from kickoff() rather than being redacted.
    tork = Tork(default_action=GovernanceAction.DENY)
    middleware = TorkCrewAIMiddleware(
        tork=tork,
        agent_id="customer-service-crew"
    )

    # Customer service agents
    intake_agent = Agent(
        role="Intake Specialist",
        goal="Understand and categorize customer inquiries",
        backstory="Expert at understanding customer needs quickly.",
        allow_delegation=True
    )

    resolver_agent = Agent(
        role="Resolution Specialist",
        goal="Provide accurate solutions to customer issues",
        backstory="Knowledgeable support specialist with extensive training.",
        allow_delegation=False
    )

    qa_agent = Agent(
        role="Quality Assurance",
        goal="Ensure responses meet quality and compliance standards",
        backstory="Detail-oriented QA specialist focused on compliance.",
        allow_delegation=False
    )

    # Task pipeline
    intake_task = Task(
        description="Analyze the customer inquiry: {inquiry}",
        expected_output="Categorized inquiry with key details extracted",
        agent=intake_agent
    )

    resolution_task = Task(
        description="Provide a helpful response to the customer's issue",
        expected_output="Clear, actionable response for the customer",
        agent=resolver_agent
    )

    qa_task = Task(
        description="Review the response for accuracy and compliance",
        expected_output="Approved response or feedback for revision",
        agent=qa_agent
    )

    crew = Crew(
        agents=[intake_agent, resolver_agent, qa_agent],
        tasks=[intake_task, resolution_task, qa_task],
        process=Process.sequential,
        verbose=True
    )

    return GovernedCrew(crew, middleware)

def handle_customer_inquiry(inquiry: str) -> str:
    """Process a customer inquiry with governance on inputs and output."""

    crew = create_governed_crew()

    try:
        return crew.kickoff(inputs={"inquiry": inquiry})
    except ValueError as e:
        # The adapter raises ValueError when governance denies an input.
        # The message carries the receipt id, never the blocked text.
        print(f"Blocked by governance: {e}")
        return "Please remove any personal information and try again."

# Usage
response = handle_customer_inquiry(
    "I need help with my account password reset"
)
print(response)

from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.crewai import TorkCrewAIMiddleware, GovernedCrew

def safe_crew_execution(crew, inputs: dict) -> dict:
    """Execute crew with error handling."""

    # Only a DENY action raises. With the default REDACT action, PII is
    # replaced in place and no exception is thrown.
    tork = Tork(default_action=GovernanceAction.DENY)
    middleware = TorkCrewAIMiddleware(tork=tork)
    governed_crew = GovernedCrew(crew, middleware)

    try:
        result = governed_crew.kickoff(inputs=inputs)
        return {
            "success": True,
            "result": result,
            "governance_passed": True
        }

    except ValueError as e:
        # Input denied by governance (PII detected, action=DENY).
        # str(e) is "Input blocked: <receipt_id>".
        return {
            "success": False,
            "error": "input_blocked",
            "message": str(e),
            "governance_passed": False
        }

    except Exception as e:
        # Other errors (LLM, network, etc.)
        return {
            "success": False,
            "error": "execution_error",
            "message": str(e),
            "governance_passed": None  # Unknown
        }

# Usage with fallback
result = safe_crew_execution(my_crew, {"query": "Help with account"})

if not result["success"]:
    if result["error"] == "input_blocked":
        print("Please remove personal information from your request.")
    else:
        print("An error occurred. Please try again.")

Exception Types

from tork_governance import Tork, TorkConfig, GovernanceAction, GovernanceResult
from tork_governance.adapters.crewai import (
    TorkCrewAIMiddleware,   # Main middleware class
    GovernedAgent,          # Agent wrapper
    GovernedCrew,           # Crew wrapper
)
# Denied inputs raise the built-in ValueError; there are no custom exceptions.