CrewAI

CrewAI

Wrap any CrewAI tool with a Xybern authorisation check. The guard function calls /enforce/intercept before the tool executes and raises PermissionError on a DENY decision.

# crewai_xybern.py
import requests
from crewai import Agent, Task, Crew, Tool

XYBERN_API_KEY = "xb_your_api_key"

def xybern_intercept(agent_id: str, tool_name: str, tool_input: dict) -> dict:
    """Call Xybern before every tool execution."""
    return requests.post(
        "https://www.xybern.com/api/v1/enforce/intercept",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_id": agent_id,
            "action": {"type": "tool_call", "tool": tool_name, "parameters": tool_input},
            "context": {"framework": "crewai"}
        }
    ).json()

def guarded_tool(agent_id: str, tool_fn, tool_name: str):
    """Wrap any CrewAI tool with Xybern authorisation."""
    def wrapper(input_data: dict):
        check = xybern_intercept(agent_id, tool_name, input_data)
        if check["decision"] == "DENY":
            raise PermissionError(f"Xybern denied: {check['reason']}")
        return tool_fn(input_data)
    return wrapper

def raw_send_email(data: dict):
    print(f"Sending email to {data['to']}")

guarded_email = guarded_tool("finance_agent_001", raw_send_email, "send_email")

finance_agent = Agent(
    role="Financial Analyst",
    goal="Generate quarterly reports",
    backstory="Expert in financial analysis",
    tools=[Tool(name="send_email", func=guarded_email, description="Send email with Xybern authorisation")]
)

task = Task(description="Generate and send the Q4 report", agent=finance_agent)
crew = Crew(agents=[finance_agent], tasks=[task])
result = crew.kickoff()