Custom Pipelines

Custom Pipelines

Use the XybernClient class and its @guard decorator to wrap any Python function in your own agent pipeline. Works with any orchestration system, no framework dependency required.

# custom_pipeline_xybern.py
import requests
from functools import wraps

XYBERN_API_KEY = "xb_your_api_key"

class XybernClient:
    """Minimal Xybern client for custom agent pipelines."""

    def __init__(self, api_key: str, agent_id: str):
        self.api_key = api_key
        self.agent_id = agent_id
        self.base_url = "https://www.xybern.com/api/v1"

    def _headers(self):
        return {"X-API-Key": self.api_key, "Content-Type": "application/json"}

    def intercept(self, tool: str, params: dict, context: dict = None) -> dict:
        payload = {
            "agent_id": self.agent_id,
            "action": {"type": "tool_call", "tool": tool, "parameters": params},
            "context": context or {}
        }
        return requests.post(
            f"{self.base_url}/enforce/intercept",
            headers=self._headers(),
            json=payload
        ).json()

    def guard(self, tool_name: str):
        """Decorator: wrap any function with Xybern authorisation."""
        def decorator(fn):
            @wraps(fn)
            def wrapper(*args, **kwargs):
                result = self.intercept(tool_name, kwargs)
                if result["decision"] == "DENY":
                    raise PermissionError(f"[Xybern] Denied: {result['reason']}")
                return fn(*args, **kwargs)
            return wrapper
        return decorator

# Usage
xybern = XybernClient(api_key=XYBERN_API_KEY, agent_id="custom_pipeline_agent")

@xybern.guard("send_payment")
def send_payment(amount: float, recipient: str, currency: str = "USD"):
    print(f"Sending {currency} {amount} to {recipient}")
    return {"status": "sent"}

@xybern.guard("read_customer_data")
def read_customer_data(customer_id: str, fields: list):
    return {"id": customer_id, "data": "..."}

# These will be authorised by Xybern before executing
send_payment(amount=50000.00, recipient="vendor@external.com", currency="USD")
read_customer_data(customer_id="cust_001", fields=["name", "email", "balance"])