CrewAI Agent Guardrails with Veto

from crewai import Agent, Task, Crew, Tool
from veto import VetoClient

# Initialize Veto client
veto = VetoClient(api_key="veto_live_xxx")

# Wrap tools with authorization
@veto.guard(
    policy="file_operations",
    agent_role="researcher"  # Context passed to policy
)
def read_documents(query: str) -> str:
    """Read documents from the knowledge base."""
    return document_store.search(query)

@veto.guard(
    policy="api_calls",
    agent_role="writer"
)
def publish_content(content: str) -> dict:
    """Publish content to external platforms."""
    return api_client.publish(content)

# Define agents with wrapped tools
researcher = Agent(
    role="Researcher",
    goal="Find and analyze relevant information",
    tools=[Tool(name="read_documents", func=read_documents)],
    backstory="Expert at finding accurate information"
)

writer = Agent(
    role="Writer",
    goal="Create engaging content based on research",
    tools=[Tool(name="publish_content", func=publish_content)],
    backstory="Skilled at crafting compelling narratives"
)

# Create and run the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[...],
    verbose=True
)

result = crew.kickoff()

# Role-based policies for CrewAI agents
policies:
  file_operations:
    description: "File system access controls"
    rules:
      - name: "researcher_read_only"
        condition:
          agent_role: "researcher"
        allow:
          actions: ["read"]
        deny:
          actions: ["write", "delete"]
          paths: ["/etc/*", "/.env*"]

      - name: "manager_full_access"
        condition:
          agent_role: "manager"
        allow:
          actions: ["read", "write"]
        deny:
          actions: ["delete"]
          paths: ["/production/*"]

  api_calls:
    description: "External API access controls"
    rules:
      - name: "writer_limited_publish"
        condition:
          agent_role: "writer"
        allow:
          endpoints: ["api.example.com/publish"]
          methods: ["POST"]
        deny:
          endpoints: ["api.example.com/delete", "api.example.com/admin/*"]

      - name: "researcher_no_external"
        condition:
          agent_role: "researcher"
        deny:
          endpoints: ["*"]

  financial:
    description: "Transaction and payment controls"
    rules:
      - name: "manager_approval_required"
        condition:
          agent_role: "manager"
          amount: {greater_than: 1000}
        require_approval: true

      - name: "accountant_limited"
        condition:
          agent_role: "accountant"
        allow:
          actions: ["read", "create"]
        deny:
          actions: ["delete"]

from crewai import Agent, Task, Crew, Process
from veto import VetoClient, ApprovalRequest

veto = VetoClient(api_key="veto_live_xxx")

# Manager with approval workflow for high-value actions
@veto.guard(
    policy="manager_actions",
    on_deny="request_approval"  # Route to human approval
)
def approve_transaction(amount: float, recipient: str) -> dict:
    """Approve financial transactions."""
    return payment_service.process(amount, recipient)

# Hierarchical crew with manager
manager = Agent(
    role="Manager",
    goal="Oversee crew operations and approve critical decisions",
    tools=[Tool(name="approve_transaction", func=approve_transaction)],
    allow_delegation=True,
    backstory="Experienced team lead with approval authority"
)

workers = [
    Agent(role="Analyst", goal="Analyze data", tools=[...]),
    Agent(role="Executor", goal="Execute approved tasks", tools=[...]),
]

crew = Crew(
    agents=[manager, *workers],
    tasks=[...],
    process=Process.hierarchical,  # Manager delegates to workers
    manager_llm="gpt-4"
)

# Policies automatically apply based on delegating agent
result = crew.kickoff()