CrewAI + Tork Governance

Crew Governance

Wrap entire crews with one call

Agent Control

Govern individual agents

Task Validation

Check inputs and outputs

Communication

Monitor inter-agent messages

pip install tork-governance crewai crewai-tools

from crewai import Agent, Task, Crew
from tork.adapters.crewai import TorkCrewAIMiddleware
from tork.core.engine import GovernanceEngine

# Initialize governance
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkCrewAIMiddleware(engine=engine, 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)

# Now all task executions are governed
# - Inputs are checked before execution
# - Outputs are validated and can be redacted
# - Compliance receipts are generated

Middleware Methods

from crewai import Agent, Task
from tork.adapters.crewai import TorkCrewAIMiddleware, GovernedAgent
from tork.core.engine import GovernanceEngine

engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkCrewAIMiddleware(engine=engine)

# 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
# - Task description is validated before execution
# - Output is checked for PII, toxicity, policy violations
result = governed_writer.execute_task(writing_task)

from crewai import Agent, Task, Crew, Process
from tork.adapters.crewai import TorkCrewAIMiddleware, GovernedCrew
from tork.core.engine import GovernanceEngine

# Initialize
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkCrewAIMiddleware(engine=engine, 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 entire crew with governance
governed_crew = GovernedCrew(crew, middleware)

# All agents in the crew are now governed
# Inter-agent communication is monitored
# Task inputs/outputs are validated
result = governed_crew.kickoff(inputs={"topic": "AI Governance Best Practices"})

from crewai import Agent, Task
from tork.adapters.crewai import TorkGovernedTask
from tork.core.engine import GovernanceEngine

# Create engine (or use API key)
engine = GovernanceEngine(api_key="your_tork_api_key")

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

# Wrap with governance
governed_task = TorkGovernedTask(
    task=task,
    engine=engine,  # or api_key="your_key"
    agent_id="analysis-task"
)

# Validate before execution
validated_payload = governed_task.validate_before_execution()
# This checks the task description and expected output against policies

# ... execute the task ...

# Validate after execution
output = "Customer feedback analysis shows..."
safe_output = governed_task.validate_after_execution(output)
# This checks the output for PII, applies redaction if needed

from crewai import Agent, Task, Crew, Process
from tork.adapters.crewai import (
    TorkCrewAIMiddleware,
    GovernedCrew,
    GovernanceBlockedError,
    PIIDetectedError
)
from tork.core.engine import GovernanceEngine

def create_governed_crew(api_key: str):
    """Create a fully governed CrewAI crew."""

    engine = GovernanceEngine(api_key=api_key)
    middleware = TorkCrewAIMiddleware(
        engine=engine,
        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 full governance."""

    crew = create_governed_crew("your_tork_api_key")

    try:
        result = crew.kickoff(inputs={"inquiry": inquiry})
        return result

    except GovernanceBlockedError as e:
        # Policy violation - log and return safe response
        print(f"Blocked by governance: {e}")
        return "I apologize, but I cannot process this request."

    except PIIDetectedError as e:
        # PII in input - request clean input
        print(f"PII detected: {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 crewai import Agent, Task, Crew, Process
from tork.adapters.crewai import TorkCrewAIMiddleware, GovernedCrew
from tork.core.engine import GovernanceEngine

engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkCrewAIMiddleware(engine=engine, agent_id="collab-crew")

# Agents that can delegate to each other
manager = Agent(
    role="Project Manager",
    goal="Coordinate team efforts and ensure quality delivery",
    backstory="Experienced PM who delegates effectively.",
    allow_delegation=True  # Can delegate to other agents
)

developer = Agent(
    role="Developer",
    goal="Implement technical solutions",
    backstory="Skilled developer with broad expertise.",
    allow_delegation=True
)

tester = Agent(
    role="QA Tester",
    goal="Ensure code quality through testing",
    backstory="Thorough tester who catches edge cases.",
    allow_delegation=False
)

# When agents delegate, the middleware:
# 1. Validates the delegation request
# 2. Checks the delegated task content
# 3. Monitors the response before returning

crew = Crew(
    agents=[manager, developer, tester],
    tasks=[...],
    process=Process.hierarchical,  # Manager coordinates
    manager_agent=manager,
    verbose=True
)

# Governance applies to ALL inter-agent communication
governed_crew = GovernedCrew(crew, middleware)
result = governed_crew.kickoff(inputs={"project": "Build API endpoint"})

from tork.adapters.crewai import (
    TorkCrewAIMiddleware,
    GovernedCrew,
    GovernanceBlockedError,
    PIIDetectedError
)
from tork.core.engine import GovernanceEngine

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

    engine = GovernanceEngine(api_key="your_tork_api_key")
    middleware = TorkCrewAIMiddleware(engine=engine)
    governed_crew = GovernedCrew(crew, middleware)

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

    except GovernanceBlockedError as e:
        # Content blocked by policy
        return {
            "success": False,
            "error": "content_blocked",
            "message": str(e),
            "governance_passed": False
        }

    except PIIDetectedError as e:
        # PII detected and blocked
        return {
            "success": False,
            "error": "pii_detected",
            "message": "Personal information was detected and blocked",
            "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"] == "pii_detected":
        print("Please remove personal information from your request.")
    elif result["error"] == "content_blocked":
        print("Your request could not be processed due to policy restrictions.")
    else:
        print("An error occurred. Please try again.")

Exception Types

from tork.adapters.crewai import (
    TorkCrewAIMiddleware,   # Main middleware class
    GovernedAgent,          # Agent wrapper
    GovernedCrew,           # Crew wrapper
    TorkGovernedTask,       # Task wrapper
    GovernanceBlockedError, # Policy violation exception
    PIIDetectedError,       # PII detection exception
)