Xybern API Documentation

The Authorisation API for AI agents. Intercept, evaluate, and enforce every agent action before it runs.

ℹ️

Base URL: https://www.xybern.com/api/v1

The Xybern Authorisation API sits between your AI agents and the world. Every tool call, every delegation, every action — evaluated against your policies and either allowed or denied in real time. Full audit trail included.

Authentication

All API requests require your API key in the X-API-Key header.

X-API-Key: xb_your_api_key_here

Get your API key from the Authorisation Layer under API Keys.

⚠️

Keep your API key secure. Never expose it in client-side code or public repositories.

SDK Installation New

Download our official SDKs for Python and JavaScript - no package manager required.

📦

How it works: Download the SDK file to your project folder, then import it. No pip/npm install needed.

Python SDK

Step 1: Install the required dependency

pip install requests

Step 2: Download the SDK to your project folder

# Navigate to your project directory first
cd /path/to/your/project

# Download the SDK file
curl -O https://www.xybern.com/static/sdk/xybern.py

# Now you have xybern.py in your project folder

Step 3: Import and use in your Python script (must be in same folder)

# your_script.py (in the same folder as xybern.py)
from xybern import Xybern

client = Xybern(api_key="xb_your_api_key")

result = client.verify(
    content="AI generated text to verify...",
    source={"type": "llm", "model": "gpt-4", "provider": "openai"}
)

print(f"Trust Score: {result['trust_score']}%")
print(f"Fairness Score: {result['bias']['fairness_score']}%")

Project structure:

my_project/
├── xybern.py          # Downloaded SDK
├── your_script.py     # Your code that imports xybern
└── ...

JavaScript SDK

Option A: Browser (easiest)

Just add a script tag - no download needed:

<!DOCTYPE html>
<html>
<head>
    <script src="https://www.xybern.com/static/sdk/xybern.js"></script>
</head>
<body>
    <script>
        const client = new Xybern({ apiKey: 'xb_your_api_key' });
        
        async function verify() {
            const result = await client.verify({
                content: 'AI generated text...',
                source: { type: 'llm', model: 'gpt-4' }
            });
            console.log('Trust Score:', result.trust_score);
        }
        
        verify();
    </script>
</body>
</html>

Option B: Node.js

Step 1: Download the SDK to your project folder

# Navigate to your project directory
cd /path/to/your/project

# Download the SDK file
curl -O https://www.xybern.com/static/sdk/xybern.js

Step 2: Require and use in your script

// app.js (in the same folder as xybern.js)
const { Xybern } = require('./xybern.js');

const client = new Xybern({ apiKey: 'xb_your_api_key' });

async function main() {
    const result = await client.verify({
        content: 'AI generated text...',
        source: { type: 'llm', model: 'gpt-4' }
    });
    
    console.log(`Trust Score: ${result.trust_score}%`);
}

main();

Project structure:

my_project/
├── xybern.js          # Downloaded SDK
├── app.js             # Your code that requires xybern
└── package.json

✓

That's it! The SDK file must be in the same directory as your script. No global installation needed.

⚠️

Important: Keep your API key secure. For production, use environment variables instead of hardcoding.

Quick Start (Direct API)

Or use the API directly without the SDK:

import requests

API_KEY = "xb_your_api_key_here"

response = requests.post(
    "https://www.xybern.com/api/v1/enforce/intercept",
    headers={"X-API-Key": API_KEY},
    json={
        "agent_id": "agent_crewai_001",
        "action": {
            "type": "tool_call",
            "tool": "send_email",
            "parameters": {"to": "cfo@company.com", "subject": "Q4 Report"}
        },
        "context": {"task": "financial_reporting", "session_id": "sess_abc123"}
    }
)

result = response.json()
print(f"Decision: {result['decision']}")   # ALLOW or DENY
print(f"Reason: {result['reason']}")

const API_KEY = "xb_your_api_key_here";

const response = await fetch("https://www.xybern.com/api/v1/enforce/intercept", {
    method: "POST",
    headers: { "X-API-Key": API_KEY, "Content-Type": "application/json" },
    body: JSON.stringify({
        agent_id: "agent_autogen_001",
        action: {
            type: "tool_call",
            tool: "write_file",
            parameters: { path: "/reports/q4.csv", content: "..." }
        },
        context: { task: "report_generation", session_id: "sess_xyz789" }
    })
});

const result = await response.json();
console.log(`Decision: ${result.decision}`);  // ALLOW or DENY

curl -X POST https://www.xybern.com/api/v1/enforce/intercept \
  -H "X-API-Key: xb_your_api_key_here" \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "agent_langgraph_001",
    "action": {
      "type": "tool_call",
      "tool": "query_database",
      "parameters": {"table": "customers", "query": "SELECT * FROM customers"}
    },
    "context": {"task": "data_analysis", "session_id": "sess_lan001"}
  }'

POST /v1/verify LLM

Verify LLM outputs (GPT-4, Claude, Llama, etc.) with bias detection, regulatory compliance, and factual analysis.

POST /v1/verify

Analyzes content, extracts claims, detects bias, checks compliance, and provides trust scoring.

Request Body

Field	Type	Required	Description
`content`	string	Yes	The AI-generated text to verify
`source`	object	Yes	Source metadata (type, model, provider)
`original_prompt`	string	No	The prompt used to generate the content
`jurisdictions`	array	No	Regulatory jurisdictions to check ["US", "EU"]
`domains`	array	No	Regulatory domains ["finance", "healthcare"]

Example

import requests

response = requests.post(
    "https://www.xybern.com/api/v1/verify",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "content": "The quarterly revenue increased by 15% year-over-year.",
        "source": {"type": "llm", "model": "gpt-4", "provider": "openai"},
        "original_prompt": "Summarize the financial report",
        "jurisdictions": ["US", "EU"],
        "domains": ["finance"]
    }
)

result = response.json()
print(f"Trust Score: {result['trust_score']}%")
print(f"Bias Fairness: {result['bias']['fairness_score']}%")
print(f"Regulatory Score: {result['regulatory']['compliance_score']}%")

POST /v1/verify/agent Agent

Verify AI agent behavior, tool usage, and multi-step decision-making for LangChain, AutoGen, CrewAI, and custom agents.

POST /v1/verify/agent

Analyzes agent execution traces for safety, governance, and risk assessment.

🤖

What we analyze: Tool calls, action sequences, data access patterns, output safety, and policy compliance.

Supported Agent Frameworks

LangChain - Agents, chains, and tool calls
AutoGen - Multi-agent conversations
CrewAI - Crew tasks and agent collaboration
OpenAI Assistants - Function calls and code interpreter
Custom - Any agent with step/action traces

Request Body

Field	Type	Required	Description
`agent_trace`	object	Yes	Agent execution trace with steps and tool calls
`agent_trace.agent_id`	string	Yes	Unique identifier for the agent
`agent_trace.steps`	array	Yes	Array of step objects (action, tool, input, output)
`agent_trace.final_output`	string	Yes	The agent's final response/output
`agent_type`	string	No	langchain, autogen, crewai, openai_assistant, custom
`policies`	array	No	Custom policy rules to check ["no_pii", "no_external_api"]

Agent Trace Structure

{
  "agent_trace": {
    "agent_id": "my_agent_001",
    "steps": [
      {
        "step": 1,
        "action": "tool_call",
        "tool": "web_search",
        "input": "search query here",
        "output": "search results...",
        "timestamp": "2026-01-30T10:00:00Z"
      },
      {
        "step": 2,
        "action": "reasoning",
        "tool": "llm",
        "input": "analyze the search results",
        "output": "based on the results..."
      }
    ],
    "final_output": "Here is my analysis..."
  },
  "agent_type": "langchain"
}

Example

import requests

response = requests.post(
    "https://www.xybern.com/api/v1/verify/agent",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "agent_trace": {
            "agent_id": "agent_123",
            "steps": [
                {
                    "step": 1,
                    "action": "search",
                    "tool": "web_search",
                    "input": "latest stock prices AAPL",
                    "output": "AAPL: $185.50"
                },
                {
                    "step": 2,
                    "action": "analyze",
                    "tool": "calculator",
                    "input": "185.50 * 1000",
                    "output": "185500"
                }
            ],
            "final_output": "Your AAPL portfolio is worth $185,500"
        },
        "agent_type": "langchain"
    }
)

result = response.json()
print(f"Agent Safety Score: {result['trust_score']}%")
print(f"Tool Usage Valid: {result['governance']['passed']}")

POST /v1/verify/agent/stream

Real-time verification of agent steps as they execute - verify each action before proceeding.

POST /v1/verify/agent/stream

Stream verification for live agent monitoring and step-by-step approval.

Request Body

Field	Type	Required	Description
`agent_id`	string	Yes	Unique identifier for the agent session
`step`	object	Yes	Current step to verify (action, tool, input, output)
`context`	object	No	Previous steps context for continuity
`block_on_risk`	boolean	No	If true, returns block recommendation for risky actions

Example: Real-time Agent Guardrails

import requests

XYBERN_API_KEY = "xb_your_api_key"

class GuardedAgent:
    """Agent with real-time Xybern verification guardrails"""
    
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.step_count = 0
        self.context = {"previous_steps": []}
    
    def verify_step(self, action, tool, tool_input, tool_output):
        """Verify each step before proceeding"""
        self.step_count += 1
        
        step = {
            "step_number": self.step_count,
            "action": action,
            "tool": tool,
            "input": str(tool_input),
            "output": str(tool_output)
        }
        
        response = requests.post(
            "https://www.xybern.com/api/v1/verify/agent/stream",
            headers={"X-API-Key": XYBERN_API_KEY},
            json={
                "agent_id": self.agent_id,
                "step": step,
                "context": self.context,
                "block_on_risk": True
            }
        ).json()
        
        # Update context for next step
        self.context["previous_steps"].append(step)
        
        return {
            "approved": response.get("step_approved", True),
            "risk_level": response.get("risk_level", "low"),
            "warnings": response.get("warnings", []),
            "should_block": response.get("block_recommended", False)
        }
    
    def execute_with_guardrails(self, tool_name, tool_input):
        """Execute tool only if Xybern approves"""
        
        # Pre-execution check
        pre_check = self.verify_step(
            action="pre_execute",
            tool=tool_name,
            tool_input=tool_input,
            tool_output="pending"
        )
        
        if pre_check["should_block"]:
            return {
                "blocked": True,
                "reason": pre_check["warnings"],
                "risk": pre_check["risk_level"]
            }
        
        # Execute the tool
        result = self._run_tool(tool_name, tool_input)
        
        # Post-execution verification
        post_check = self.verify_step(
            action="post_execute",
            tool=tool_name,
            tool_input=tool_input,
            tool_output=result
        )
        
        return {
            "blocked": False,
            "result": result,
            "verification": post_check
        }

# Usage
agent = GuardedAgent("financial_agent_001")

# This will be verified before and after execution
result = agent.execute_with_guardrails(
    tool_name="database_query",
    tool_input="SELECT * FROM customer_data WHERE balance > 100000"
)

if result["blocked"]:
    print(f"⚠️ Action blocked: {result['reason']}")
else:
    print(f"✓ Executed: {result['result']}")

Response Fields

Field	Type	Description
`step_approved`	boolean	Whether the step passes verification
`risk_level`	string	low, medium, high, critical
`block_recommended`	boolean	True if action should be blocked
`warnings`	array	List of detected issues or concerns
`governance_flags`	array	Policy violations detected

⚠️

High-risk actions like database writes, API calls with side effects, or file system access should always use block_on_risk: true.

POST /v1/verify/multimodal

Verify content extracted from PDFs, images, and audio files.

POST /v1/verify/multimodal

Extracts text from files and verifies the content.

Supported Formats

PDF: .pdf
Images: .jpg, .png, .gif, .webp (OCR extraction)
Audio: .mp3, .wav, .m4a (speech-to-text)

Example

import requests

# Upload a PDF for verification
with open("financial_report.pdf", "rb") as f:
    response = requests.post(
        "https://www.xybern.com/api/v1/verify/multimodal",
        headers={"X-API-Key": "xb_your_api_key"},
        files={"file": ("report.pdf", f, "application/pdf")},
        data={
            "source_type": "llm",
            "source_model": "gpt-4",
            "jurisdictions": "US,EU",
            "domains": "finance"
        }
    )

result = response.json()
print(f"Extracted Text Length: {len(result.get('extracted_text', ''))}")
print(f"Trust Score: {result['trust_score']}%")

// JavaScript - Upload image for OCR verification
const formData = new FormData();
formData.append('file', imageFile);
formData.append('source_type', 'llm');
formData.append('source_model', 'gpt-4');

const response = await fetch('https://www.xybern.com/api/v1/verify/multimodal', {
    method: 'POST',
    headers: { 'X-API-Key': 'xb_your_api_key' },
    body: formData
});

const result = await response.json();
console.log(`Trust Score: ${result.trust_score}%`);

POST /v1/verify/hallucination Phase 8

Detect when LLMs confidently state false or fabricated information. Uses multiple detection methods to catch hallucinations that simple verification might miss.

POST /v1/verify/hallucination

Multi-signal hallucination detection: semantic entropy, self-consistency, knowledge grounding, confidence calibration, and fabrication pattern scanning.

💡

Also integrated into POST /v1/verify — every verification response now includes a hallucination field automatically. Use this standalone endpoint for deeper analysis or when you only need hallucination detection.

Detection Methods

Method	What It Detects
Semantic Entropy	Measures claim-level uncertainty — high entropy indicates the model is uncertain about what it's stating
Self-Consistency	Checks whether claim status, confidence, and entropy signals agree — contradictions flag potential fabrication
Knowledge Grounding	Verifies claims against known facts (dates, founders, locations) to catch fabricated details
Confidence Calibration	Flags claims where the model is over-confident relative to evidence — the classic hallucination signature
Fabrication Pattern Scan	Detects fake citations, fabricated statistics, made-up DOIs, and unverifiable authority claims

Request Body

Field	Type	Required	Description
`content`	string	Yes	AI-generated text to check for hallucinations
`source`	object	No	Source metadata: `{ type, model, provider }`
`original_prompt`	string	No	The prompt that generated the content
`claims`	array	No	Pre-parsed claims from `/v1/verify` (if omitted, claims are extracted automatically)
`sensitivity`	string	No	`low`, `balanced` (default), or `high`
`include_verification`	boolean	No	If `true`, also runs full `/v1/verify` and returns results together

Python Example

import requests

# Standalone hallucination detection
response = requests.post(
    "https://www.xybern.com/api/v1/verify/hallucination",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "content": gpt4_output,
        "source": {"type": "llm", "model": "gpt-4", "provider": "openai"},
        "sensitivity": "high",
        "include_verification": True
    }
)

result = response.json()
hal = result["hallucination"]

print(f"Hallucination Score: {hal['hallucination_score']}/100")
print(f"Risk Level: {hal['risk_level']}")
print(f"Flagged Claims: {hal['stats']['hallucinated_count']}/{hal['stats']['total_claims_analyzed']}")

# Check each flagged claim
for claim in hal["flagged_claims"]:
    print(f"  ⚠️ [{claim['risk_level']}] {claim['text'][:60]}...")
    print(f"     Method: {claim['detection_method']}")
    print(f"     Reason: {claim['reasoning']}")
    print(f"     Fix: {claim['suggestion']}")

Response

{
  "ok": true,
  "verification_id": "hal_abc123xyz",
  "hallucination": {
    "hallucination_score": 35,
    "risk_level": "medium",
    "flagged_claims": [
      {
        "claim_id": "c2",
        "text": "According to a 2024 study by MIT...",
        "risk_level": "high",
        "confidence_score": 0.85,
        "detection_method": "confidence_calibration",
        "reasoning": "Confidence calibration gap (0.55): model is over-confident relative to evidence",
        "category": "authority_claim",
        "grounding_score": 0.3,
        "entropy_score": 0.65,
        "suggestion": "Add a specific, verifiable citation for this authority claim."
      }
    ],
    "stats": {
      "total_claims_analyzed": 5,
      "hallucinated_count": 2,
      "grounded_count": 3,
      "uncertain_count": 0
    },
    "detection_methods_used": [
      "confidence_calibration",
      "fabrication_pattern_scan",
      "knowledge_grounding",
      "self_consistency",
      "semantic_entropy"
    ],
    "metrics": {
      "semantic_entropy": 0.42,
      "self_consistency_score": 0.75,
      "knowledge_grounding_score": 68.5,
      "calibration_score": 0.72
    },
    "summary": "Detected 2 potential hallucination(s) across 5 claims (moderate risk, score: 35/100)."
  }
}

Hallucination Risk Levels

Score	Risk	Action
0-9	🟢 None	Content appears well-grounded
10-24	🟡 Low	Minor uncertainty detected, generally safe
25-49	🟠 Medium	Some claims flagged — review before publishing
50-74	🔴 High	Significant hallucination risk — manual review required
75-100	⛔ Critical	Content likely contains fabricated information — do not publish

Response Structure

All verification endpoints return a comprehensive response with deterministic scoring - the same input always produces the same score:

{
  "ok": true,
  "verification_id": "ver_abc123xyz",
  "trust_score": 78,
  "verification_status": "verified",
  "claims": [...],
  
  // NEW: Deterministic Scoring (v1)
  "scoring": {
    "verification_score": 78,
    "confidence_band": "medium-high",
    "confidence_range": { "low": 68, "high": 85 },
    "claims": {
      "verified": 4,
      "total": 5,
      "ratio": "4/5",
      "percentage": 80
    },
    "issues": {
      "penalty_applied": -5,
      "high_severity": 0,
      "medium_severity": 1,
      "low_severity": 0
    },
    "web_evidence_boost": 3,
    "llm_trust_score": 72,  // Original LLM score (for reference)
    "scoring_method": "deterministic_v1"
  },
  
  // NEW: Confidence Display
  "confidence": {
    "band": "medium-high",
    "range": { "low": 68, "high": 85 },
    "display": {
      "score_display": "78%",
      "band_display": "Medium-High Confidence",
      "range_display": "68-85%",
      "claims_display": "4/5 claims verified"
    }
  },
  
  "bias": {
    "fairness_score": 92,
    "status": "fair",
    "biases_detected": []
  },
  "regulatory": {
    "compliance_score": 88,
    "flags": []
  },
  "governance": {
    "passed": true,
    "violations": []
  },
  "trace_id": "tr_xxx",
  "created_at": "2026-01-30T12:00:00Z"
}

Confidence Bands

Instead of a single fluctuating score, Xybern provides confidence bands that honestly represent the certainty level:

Band	Score Range	Status	Meaning
🟢 High	80-100	`verified`	Strong evidence, most claims verified
🟢 Medium-High	65-79	`verified`	Good confidence, majority claims verified
🟡 Medium	50-64	`partial`	Mixed results, review recommended
🟠 Medium-Low	35-49	`partial`	Significant gaps in verification
🔴 Low	0-34	`unverified`	Low confidence, manual review required

💡

Deterministic Scoring: The verification_score is calculated from claims and issues, not LLM judgment. Running the same content multiple times produces the same score, making results reproducible and trustworthy.

OpenAI / GPT Integration

Verify GPT-4, GPT-3.5, or any OpenAI model output before displaying to users.

Basic Integration

from openai import OpenAI
import requests

openai_client = OpenAI(api_key="sk-...")
XYBERN_API_KEY = "xb_your_api_key"

def chat_with_verification(user_message):
    # Step 1: Get response from OpenAI
    completion = openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}]
    )
    ai_response = completion.choices[0].message.content
    
    # Step 2: Verify the response with Xybern
    verification = requests.post(
        "https://www.xybern.com/api/v1/verify",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "content": ai_response,
            "original_prompt": user_message,
            "source": {"type": "llm", "model": "gpt-4", "provider": "openai"}
        }
    ).json()
    
    # Step 3: Return response with trust metadata
    return {
        "response": ai_response,
        "trust_score": verification["trust_score"],
        "verified": verification["verification_status"] == "verified",
        "bias_check": verification.get("bias", {}).get("status", "unknown")
    }

# Usage
result = chat_with_verification("What are the benefits of cloud computing?")
print(f"Response: {result['response']}")
print(f"Trust Score: {result['trust_score']}%")

With Streaming

def stream_with_post_verification(user_message):
    # Collect streamed response
    full_response = ""
    stream = openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_message}],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            content = chunk.choices[0].delta.content
            full_response += content
            yield content  # Stream to user
    
    # Verify complete response (async/background)
    verification = requests.post(
        "https://www.xybern.com/api/v1/verify",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "content": full_response,
            "source": {"type": "llm", "model": "gpt-4", "provider": "openai"}
        }
    ).json()
    
    # Log or store verification results
    yield f"\n\n[Trust Score: {verification['trust_score']}%]"

Anthropic / Claude Integration

Verify Claude responses for compliance and accuracy.

from anthropic import Anthropic
import requests

anthropic_client = Anthropic(api_key="sk-ant-...")
XYBERN_API_KEY = "xb_your_api_key"

def claude_with_verification(user_message, system_prompt=None):
    # Step 1: Get response from Claude
    message = anthropic_client.messages.create(
        model="claude-3-opus-20240229",
        max_tokens=1024,
        system=system_prompt or "You are a helpful assistant.",
        messages=[{"role": "user", "content": user_message}]
    )
    ai_response = message.content[0].text
    
    # Step 2: Verify with Xybern
    verification = requests.post(
        "https://www.xybern.com/api/v1/verify",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "content": ai_response,
            "original_prompt": user_message,
            "source": {"type": "llm", "model": "claude-3-opus", "provider": "anthropic"},
            "domains": ["finance", "healthcare"]  # Optional: regulatory checks
        }
    ).json()
    
    return {
        "response": ai_response,
        "trust_score": verification["trust_score"],
        "regulatory_score": verification.get("regulatory", {}).get("compliance_score"),
        "bias_fairness": verification.get("bias", {}).get("fairness_score")
    }

# Usage
result = claude_with_verification("Explain HIPAA compliance requirements")
if result["trust_score"] < 70:
    print("⚠️ Low confidence - consider manual review")

LangChain Integration

Add Xybern verification as a callback or chain component.

As a Custom Callback

from langchain.callbacks.base import BaseCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import requests

class XybernVerificationCallback(BaseCallbackHandler):
    def __init__(self, api_key):
        self.api_key = api_key
        self.last_verification = None
    
    def on_llm_end(self, response, **kwargs):
        """Verify LLM output after generation"""
        content = response.generations[0][0].text
        
        result = requests.post(
            "https://www.xybern.com/api/v1/verify",
            headers={"X-API-Key": self.api_key},
            json={
                "content": content,
                "source": {"type": "llm", "model": "langchain", "provider": "langchain"}
            }
        ).json()
        
        self.last_verification = result
        
        if result["trust_score"] < 50:
            print(f"⚠️ Low trust score: {result['trust_score']}%")

# Usage
xybern_callback = XybernVerificationCallback("xb_your_api_key")
llm = ChatOpenAI(model="gpt-4", callbacks=[xybern_callback])

response = llm.invoke([HumanMessage(content="What is quantum computing?")])
print(f"Verification: {xybern_callback.last_verification['trust_score']}%")

As a Runnable in Chain

from langchain_core.runnables import RunnableLambda

def verify_output(text):
    """Xybern verification as a chain step"""
    result = requests.post(
        "https://www.xybern.com/api/v1/verify",
        headers={"X-API-Key": "xb_your_api_key"},
        json={"content": text, "source": {"type": "llm", "model": "gpt-4"}}
    ).json()
    
    return {
        "content": text,
        "trust_score": result["trust_score"],
        "verified": result["verification_status"] == "verified"
    }

# Build chain with verification
chain = llm | RunnableLambda(lambda x: x.content) | RunnableLambda(verify_output)

result = chain.invoke([HumanMessage(content="Explain machine learning")])
print(f"Content: {result['content'][:100]}...")
print(f"Verified: {result['verified']}")

AI Agent Integration

Verify agent decisions, tool usage, and multi-step reasoning.

🤖

When to verify agents: After each tool call, before executing actions, and at final output.

LangChain Agent with Verification

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import requests

XYBERN_API_KEY = "xb_your_api_key"

@tool
def search_database(query: str) -> str:
    """Search the company database"""
    return f"Results for: {query}"

@tool  
def calculate(expression: str) -> str:
    """Perform calculations"""
    return str(eval(expression))

def run_agent_with_verification(user_input):
    # Create and run agent
    llm = ChatOpenAI(model="gpt-4")
    tools = [search_database, calculate]
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, return_intermediate_steps=True)
    
    result = executor.invoke({"input": user_input})
    
    # Build agent trace for Xybern
    agent_trace = {
        "agent_id": "langchain_agent_001",
        "steps": []
    }
    
    for i, (action, observation) in enumerate(result["intermediate_steps"]):
        agent_trace["steps"].append({
            "step": i + 1,
            "action": action.tool,
            "tool": action.tool,
            "input": action.tool_input,
            "output": observation
        })
    
    agent_trace["final_output"] = result["output"]
    
    # Verify agent behavior
    verification = requests.post(
        "https://www.xybern.com/api/v1/verify/agent",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_trace": agent_trace,
            "agent_type": "langchain"
        }
    ).json()
    
    return {
        "output": result["output"],
        "trust_score": verification["trust_score"],
        "governance_passed": verification.get("governance", {}).get("passed", False),
        "steps_verified": len(agent_trace["steps"])
    }

AutoGen Agent Verification

import autogen
import requests

XYBERN_API_KEY = "xb_your_api_key"

def verify_autogen_conversation(chat_history):
    """Verify an AutoGen multi-agent conversation"""
    
    # Extract steps from AutoGen chat
    steps = []
    for i, msg in enumerate(chat_history):
        steps.append({
            "step": i + 1,
            "action": "message",
            "tool": msg.get("name", "assistant"),
            "input": msg.get("role", ""),
            "output": msg.get("content", "")
        })
    
    verification = requests.post(
        "https://www.xybern.com/api/v1/verify/agent",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_trace": {
                "agent_id": "autogen_group",
                "steps": steps,
                "final_output": steps[-1]["output"] if steps else ""
            },
            "agent_type": "autogen"
        }
    ).json()
    
    return verification

# After AutoGen conversation
# verification = verify_autogen_conversation(chat_result.chat_history)

Custom Agent Loop

class VerifiedAgent:
    """Agent with built-in Xybern verification"""
    
    def __init__(self, xybern_key):
        self.xybern_key = xybern_key
        self.trace = {"agent_id": "custom_agent", "steps": []}
    
    def execute_tool(self, tool_name, tool_input):
        """Execute tool and record for verification"""
        # Your tool execution logic
        result = self._run_tool(tool_name, tool_input)
        
        # Record step
        self.trace["steps"].append({
            "step": len(self.trace["steps"]) + 1,
            "action": "tool_call",
            "tool": tool_name,
            "input": str(tool_input),
            "output": str(result)
        })
        
        return result
    
    def finalize(self, final_output):
        """Complete agent run and verify"""
        self.trace["final_output"] = final_output
        
        verification = requests.post(
            "https://www.xybern.com/api/v1/verify/agent",
            headers={"X-API-Key": self.xybern_key},
            json={"agent_trace": self.trace, "agent_type": "custom"}
        ).json()
        
        return {
            "output": final_output,
            "verification": verification
        }

Common Integration Patterns

Best practices for integrating Xybern into your AI workflows.

Pattern 1: Verify-Before-Display (Chatbots)

def chatbot_response(user_message):
    """Always verify before showing to user"""
    ai_response = get_ai_response(user_message)
    
    verification = xybern_verify(ai_response)
    
    if verification["trust_score"] >= 70:
        return {"response": ai_response, "status": "verified"}
    elif verification["trust_score"] >= 40:
        return {
            "response": ai_response,
            "status": "caution",
            "warning": "This response may contain unverified claims"
        }
    else:
        return {
            "response": "I'm not confident in my answer. Let me connect you with a human.",
            "status": "escalate",
            "original": ai_response
        }

Pattern 2: Batch Verification (Content Pipelines)

import concurrent.futures

def verify_batch(contents):
    """Verify multiple pieces of content in parallel"""
    
    def verify_single(content):
        return requests.post(
            "https://www.xybern.com/api/v1/verify",
            headers={"X-API-Key": XYBERN_API_KEY},
            json={"content": content, "source": {"type": "llm", "model": "gpt-4"}}
        ).json()
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(verify_single, contents))
    
    return results

# Usage: verify 100 AI-generated articles
articles = ["Article 1...", "Article 2...", ...]
verifications = verify_batch(articles)
approved = [a for a, v in zip(articles, verifications) if v["trust_score"] >= 70]

Pattern 3: Middleware (FastAPI/Flask)

from fastapi import FastAPI, Request
from functools import wraps

app = FastAPI()

def verify_ai_response(func):
    """Decorator to verify AI responses"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        result = await func(*args, **kwargs)
        
        if "ai_response" in result:
            verification = requests.post(
                "https://www.xybern.com/api/v1/verify",
                headers={"X-API-Key": XYBERN_API_KEY},
                json={
                    "content": result["ai_response"],
                    "source": {"type": "llm", "model": "gpt-4"}
                }
            ).json()
            
            result["verification"] = {
                "trust_score": verification["trust_score"],
                "bias_check": verification.get("bias", {}).get("status")
            }
        
        return result
    return wrapper

@app.post("/chat")
@verify_ai_response
async def chat(request: Request):
    data = await request.json()
    ai_response = generate_response(data["message"])
    return {"ai_response": ai_response}

Pattern 4: Real-time Monitoring Dashboard

import logging

class XybernMonitor:
    """Log all verifications for monitoring"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.logger = logging.getLogger("xybern")
    
    def verify_and_log(self, content, source, metadata=None):
        verification = requests.post(
            "https://www.xybern.com/api/v1/verify",
            headers={"X-API-Key": self.api_key},
            json={"content": content, "source": source}
        ).json()
        
        # Log for monitoring/alerting
        self.logger.info({
            "verification_id": verification.get("verification_id"),
            "trust_score": verification["trust_score"],
            "bias_status": verification.get("bias", {}).get("status"),
            "model": source.get("model"),
            "metadata": metadata
        })
        
        # Alert on low scores
        if verification["trust_score"] < 40:
            self.logger.warning(f"LOW TRUST SCORE: {verification['trust_score']}%")
        
        return verification

# Usage
monitor = XybernMonitor("xb_your_api_key")
result = monitor.verify_and_log(
    content=ai_response,
    source={"type": "llm", "model": "gpt-4"},
    metadata={"user_id": "123", "session": "abc"}
)

✓

Best Practice: Always verify content before storing in databases or displaying to end users in regulated industries.

Authorisation New

Pre-execution AI governance layer that intercepts every AI action before it reaches any real system. Enforces policies, manages agent trust, and records immutable audit trails.

ℹ️

How it works: Your AI agents call /v1/enforce/intercept before executing any action. The control plane evaluates policies, checks agent trust, and returns allow, block, or escalate. Every decision is recorded in the Authorisation Layer Vault hash chain.

Architecture

Agent wants to execute_trade
    ↓
POST /v1/enforce/intercept  (+ optional signed_assertion for identity)
    ↓
┌──────────────────────────────────────┐
│  Identity Verification  (< 1ms)       │  ← Ed25519 sig + Redis nonce dedup
└──────────────────────────────────────┘
    ↓
┌──────────────────────────────────────┐
│  Policy Engine                        │  ← All active policies evaluated
│  • action_type  — name matching       │
│  • threshold    — trust level         │
│  • content      — regex patterns      │
│  • temporal     — hours / days        │
│  • chain        — delegation depth    │
│  • identity     — DID / scope rules   │
│  • metadata     — field comparisons   │  ← notional_usd > 100k, ticker == X, …
│  • composite    — AND / OR combiner   │
└──────────────────────────────────────┘
    ↓
┌──────────────────────────────────────┐
│  Decision                             │
│  allow    → fast path (~5ms)          │
│  block    → fast path (~5ms)          │
│  escalate → human review queue        │
└──────────────────────────────────────┘
    ↓
Vault entry + decision returned to SDK / caller

Decision Paths

Path	Latency	Description
`fast`	~5ms	Policy directly blocks or allows — no LLM call
`standard`	3-5s	Full LLM verification runs against the action content
`escalation`	—	Action held for human review in the escalation queue

POST /v1/enforce/intercept Core

Intercept an AI action before execution. This is the primary endpoint — every action must pass through here.

POST /v1/enforce/intercept

Pre-execution interception with policy evaluation, trust scoring, and vault recording.

Request Body

Field	Type	Required	Description
`action_type`	string	Yes	Action name: `execute_trade`, `send_email`, `query_database`, etc.
`action_content`	string	No	Content being actioned (e.g. email body, SQL query)
`metadata`	object	No	Arbitrary metadata (amount, recipient, etc.)
`agent_id`	string	No	Registered agent performing the action
`chain_id`	string	No	Multi-step chain identifier
`chain_step`	integer	No	Step number within the chain
`parent_decision_id`	string	No	Previous decision in this chain

Example

import requests

API_KEY = "xb_your_api_key"

# Before executing a trade, intercept it
result = requests.post(
    "https://www.xybern.com/api/v1/enforce/intercept",
    headers={"X-API-Key": API_KEY},
    json={
        "action_type": "execute_trade",
        "action_content": "Buy 1000 shares of AAPL at market price",
        "metadata": {"symbol": "AAPL", "quantity": 1000, "side": "buy"},
        "agent_id": "agent_abc123"
    }
).json()

if result["decision"] == "allow":
    execute_trade()  # proceed
elif result["decision"] == "block":
    log_blocked(result["reasoning"])
elif result["decision"] == "escalate":
    notify_human(result["decision_id"])

Response

{
  "ok": true,
  "decision": "allow",
  "decision_id": "enf_7c8f8c26e62c",
  "decision_path": "fast",
  "trust_score": 85,
  "reasoning": "No policies triggered — default allow",
  "policies_evaluated": ["policy_123", "policy_456"],
  "policies_triggered": [],
  "vault_entry_id": "ve_abc123",
  "latency_ms": 5,
  "created_at": "2026-03-13T21:48:54Z"
}

POST /v1/enforce/agent-comm

Intercept agent-to-agent instructions. When Agent A tells Agent B to do something, the instruction flows through the control plane first.

POST /v1/enforce/agent-comm

Agent-to-agent communication interception with full audit trail.

Request Body

Field	Type	Required	Description
`source_agent_id`	string	Yes	Agent sending the instruction
`target_agent_id`	string	Yes	Agent receiving the instruction
`instruction`	string	Yes	The instruction content
`action_type`	string	No	Defaults to `agent_instruction`
`chain_id`	string	No	Chain identifier for multi-step flows

Example

result = requests.post(
    "https://www.xybern.com/api/v1/enforce/agent-comm",
    headers={"X-API-Key": API_KEY},
    json={
        "source_agent_id": "agent_research_01",
        "target_agent_id": "agent_trading_01",
        "instruction": "Execute buy order for AAPL based on analysis",
        "chain_id": "chain_research_trade_001"
    }
).json()

if result["decision"] == "allow":
    forward_instruction_to_target_agent()

Enforcement Policies

Create, read, update, and delete enforcement policies programmatically. Policies are evaluated in priority order (highest first) and the most restrictive decision wins.

GET /v1/enforce/policies

List all policies for the workspace, ordered by priority.

POST /v1/enforce/policies

Create a new enforcement policy.

PUT /v1/enforce/policies/{id}

Update an existing policy.

DELETE /v1/enforce/policies/{id}

Delete a policy.

Policy Types

Type	What It Does	Conditions
`action_type`	Matches actions by name	Uses `action_types` field (supports wildcards: `delete_*`)
`threshold`	Blocks agents below a trust level	Uses `trust_threshold` field
`content_pattern`	Regex match on action content	`conditions.patterns`: array of regex strings
`temporal`	Time/day restrictions	`conditions.blocked_hours`: [0-23], `conditions.blocked_days`: [1-7]
`chain_of_custody`	Agent chain depth/agent rules	`conditions.max_chain_depth`, `conditions.forbidden_agents`

Create Policy Example

# Block trades containing sensitive keywords
policy = requests.post(
    "https://www.xybern.com/api/v1/enforce/policies",
    headers={"X-API-Key": API_KEY},
    json={
        "name": "Block Insider Trading Keywords",
        "description": "Block any trade with insider-related language",
        "policy_type": "content_pattern",
        "decision": "block",
        "priority": 200,
        "action_types": ["execute_trade", "modify_order"],
        "conditions": {
            "patterns": [
                "insider.*info",
                "material.*non-public",
                "tip.*from.*executive"
            ]
        }
    }
).json()

# Block all activity on weekends
weekend_policy = requests.post(
    "https://www.xybern.com/api/v1/enforce/policies",
    headers={"X-API-Key": API_KEY},
    json={
        "name": "Weekend Trading Lockout",
        "policy_type": "temporal",
        "decision": "block",
        "priority": 300,
        "action_types": ["execute_trade"],
        "conditions": {"blocked_days": [6, 7]}
    }
).json()

Agent Registry

Register AI agents with the control plane. Each agent accumulates an adaptive trust level — clean records earn fast-path clearance, blocks reduce trust.

POST /v1/enforce/agents

GET /v1/enforce/agents

List all registered agents.

GET /v1/enforce/agents/{agent_id}

Get agent details including trust level and action history.

GET /v1/enforce/agents/{agent_id}/history

Get trust history and decision timeline for an agent.

Adaptive Trust

Event	Trust Change	Description
Action allowed	+0.2	Clean record builds trust over time
Action blocked	-2.0	Policy violations significantly reduce trust
Action escalated	-0.5	Uncertain decisions slightly reduce trust

Register Agent Example

# Register a trading agent
agent = requests.post(
    "https://www.xybern.com/api/v1/enforce/agents",
    headers={"X-API-Key": API_KEY},
    json={
        "name": "Trading Agent Alpha",
        "framework": "langchain",
        "description": "Automated equity trading agent",
        "capabilities": ["execute_trade", "read_data", "send_email"],
        "permissions": {
            "allowed_action_types": ["execute_trade", "read_data"],
            "denied_action_types": ["delete_data", "admin_action"]
        }
    }
).json()

agent_id = agent["agent"]["agent_id"]
print(f"Registered: {agent_id}, trust: {agent['agent']['trust_level']}")

Decisions & Escalations

Query the immutable decision log and manage the human review queue.

GET /v1/enforce/decisions

List decisions with pagination and filtering by action_type or decision.

GET /v1/enforce/decisions/{decision_id}

Get a single decision with full details.

GET /v1/enforce/escalations

List pending escalations (actions awaiting human review).

POST /v1/enforce/escalations/{id}/resolve

Approve or reject an escalated action. Body: {"resolution": "approved"|"rejected"}. Immediately unblocks any SDK call polling this escalation.

GET /v1/enforce/escalations/{id}/status

Poll the resolution status of a pending escalation. Returns {"ok": true, "status": "pending"|"approved"|"rejected"}. Called automatically by the SDK's wait_for_escalation() method every escalation_poll_interval seconds.

Query Decisions

# Get all blocked decisions
blocked = requests.get(
    "https://www.xybern.com/api/v1/enforce/decisions",
    headers={"X-API-Key": API_KEY},
    params={"decision": "block", "per_page": 20}
).json()

for d in blocked["decisions"]:
    print(f"{d['decision_id']}: {d['action_type']} → {d['decision']} "
          f"(trust: {d['trust_score']}, {d['latency_ms']}ms)")

# Resolve an escalation
requests.post(
    f"https://www.xybern.com/api/v1/enforce/escalations/{esc_id}/resolve",
    headers={"X-API-Key": API_KEY},
    json={"resolution": "approved", "reason": "Reviewed and safe to proceed"}
)

Stats Endpoint

GET /v1/enforce/stats

Aggregate enforcement statistics: decisions, block rate, avg latency, agent counts.

Custom Policy Builder

Build enforcement policies tailored to your organization's needs. The policy engine supports six built-in types and a composite type for complex AND/OR logic. All types are available in the Authorisation Layer UI under Control Plane → Add Policy.

Content Pattern Policy

Block or escalate actions whose content matches regex patterns. Useful for PII detection, insider trading keywords, or prohibited content.

{
    "name": "PII Detection",
    "policy_type": "content_pattern",
    "decision": "escalate",
    "action_types": ["send_email", "export_data"],
    "conditions": {
        "patterns": [
            "\\b\\d{3}-\\d{2}-\\d{4}\\b",
            "\\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\\.[A-Z]{2,}\\b",
            "password|secret|credential|api[_-]?key"
        ]
    }
}

Temporal Policy

Restrict actions to specific hours or days. Blocked hours are in UTC (0-23), blocked days use ISO format (1=Monday, 7=Sunday).

{
    "name": "After-Hours Lockout",
    "policy_type": "temporal",
    "decision": "block",
    "action_types": ["execute_trade", "wire_transfer"],
    "conditions": {
        "blocked_hours": [0, 1, 2, 3, 4, 5, 22, 23],
        "blocked_days": [6, 7]
    }
}

Chain of Custody Policy

Limit agent delegation depth and block specific agents from chains.

{
    "name": "Max Delegation Depth",
    "policy_type": "chain_of_custody",
    "decision": "escalate",
    "conditions": {
        "max_chain_depth": 3,
        "forbidden_agents": ["agent_untrusted_001"],
        "required_agents": ["agent_compliance_reviewer"]
    }
}

✓

Dashboard support: All policy types can be created from the Authorisation Layer UI under Control Plane → Add Policy, with dynamic condition fields that appear based on the selected type.

Metadata Field Policy New

Evaluate field-level conditions against the metadata object passed on each intercept call. Combine rules with AND/OR logic to build precise, data-driven policies — such as blocking high-notional trades or flagging specific tickers.

ℹ️

How it works: When you call /enforce/intercept with a metadata object, the engine evaluates every active metadata policy against those fields. Rules can reference any key in the metadata object with numeric or string comparisons.

Supported Operators

Operator	Type	Example
`>` `<` `>=` `<=`	Numeric	`notional_usd > 100000`
`==` `!=`	String / Numeric	`ticker == "TSLA"`
`contains` `not_contains`	String	`strategy contains "pre-earnings"`
`exists` `not_exists`	Presence	`insider_flag exists`

Example: Block High-Risk Financial Transactions

requests.post(f"{BASE}/enforce/policies", headers=HEADERS, json={
    "name": "High-Risk Financial Transactions",
    "policy_type": "metadata",
    "decision": "block",
    "action_types": ["execute_trade", "wire_transfer", "fund_transfer"],
    "conditions": {
        "operator": "AND",          # ALL rules must match to trigger
        "rules": [
            {"field": "notional_usd", "operator": ">",       "value": 100000},
            {"field": "strategy",     "operator": "contains", "value": "pre-earnings"}
        ]
    }
})

Example: OR Logic — Flag Any Large or Sensitive Action

{
    "name": "Sensitive Trade Escalation",
    "policy_type": "metadata",
    "decision": "escalate",
    "conditions": {
        "operator": "OR",           # ANY rule matching triggers the policy
        "rules": [
            {"field": "notional_usd", "operator": ">=", "value": 500000},
            {"field": "ticker",       "operator": "==", "value": "GME"},
            {"field": "insider_flag", "operator": "exists"}
        ]
    }
}

Intercept Call with Metadata

result = requests.post(f"{BASE}/enforce/intercept", headers=HEADERS, json={
    "action_type": "execute_trade",
    "action_content": "Buy $4.2M block of TSLA ahead of earnings",
    "agent_id": "agent_trading_01",
    "metadata": {
        "ticker":       "TSLA",
        "notional_usd": 4200000,
        "strategy":     "pre-earnings",
        "order_type":   "market"
    }
}).json()

# → decision: "block"
# → reasoning: "All metadata conditions met [metadata.notional_usd > 100000; ...]"

💡

SDK auto-extract: When using the @guard decorator or XybernToolkit, numeric and string keyword arguments are automatically extracted as metadata and evaluated against metadata policies — no manual metadata construction needed.

Okta for AI Agents — Cryptographic Identity v1.22

Every AI agent is assigned an Ed25519 cryptographic identity — a verifiable credential that defines what the agent is, what it's authorised to do, and under what conditions. The control plane verifies identity on every action before any policy evaluation. Replay protection is enforced via Redis-backed nonce deduplication across all workers.

Architecture: Ed25519 keypairs → W3C DIDs → Signed assertions → Redis replay guard → Permission boundaries → Credential lifecycle → Vault-bound proofs

Key Concepts

Concept	Description
Ed25519 Keypair	Every agent gets a public/private key. Private key returned once at registration, never stored server-side.
DID	W3C-aligned `did:xybern:{workspace}:{agent}` identifier.
Signed Assertion	JSON payload signed by agent's private key, verified by control plane in < 1 ms.
Permission Boundary	Scopes, resource rules, temporal windows, delegation controls, rate limits.
Nonce + Timestamp	Replay protection: each assertion must have a unique nonce and timestamp within 5 minutes. Nonces are deduplicated in Redis across all API workers.

Register Agent (via SDK — recommended)

from xybern import Xybern, AgentCredential

client = Xybern(api_key="xb_your_key")

# Generates Ed25519 keypair, registers agent, issues credential — one call
cred = client.agents.register(
    name="Trading Agent Alpha",
    framework="langchain",
    capabilities=["execute_trade", "query_portfolio"],
)
cred.save("./trading_agent.cred")  # chmod 600, private key never leaves your server
print(cred.did)                    # did:xybern:{workspace}:agent_abc123
print(cred.key_fingerprint)        # SHA-256 hex fingerprint

Register Agent (raw API)

resp = requests.post(f"{BASE}/enforce/agents", headers=HEADERS, json={
    "name": "Trading Agent Alpha",
    "framework": "langchain",
    "scopes": ["trade:write", "db:read"],
    "permission_boundary": {
        "resource_rules": [{"action_types": ["execute_trade"], "max_amount": 50000}],
        "temporal": {"allowed_hours": {"start": 9, "end": 17}},
        "rate_limit": {"max_actions": 100, "window_minutes": 60}
    }
})
cred = resp.json()["credential"]
private_key = cred["private_key"]  # SAVE THIS — shown only once
did = cred["did"]                   # did:xybern:ws1:agent_abc123

Intercept with Identity (SDK — auto-signing)

# SDK signs every assertion automatically — no crypto code needed
result = client.agents.intercept(
    action_type="execute_trade",
    action_content="Buy 100 AAPL at market",
    credential="./trading_agent.cred",   # or AgentCredential object
    metadata={"ticker": "AAPL", "notional_usd": 19000},
)
# response includes identity_verified=True when signature checks pass
print(result.identity_verified)  # True

Intercept with Identity (raw API)

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from base64 import b64encode, b64decode
import json, uuid

nonce = uuid.uuid4().hex
payload = {"agent_id": "agent_abc", "action": "execute_trade",
           "nonce": nonce, "timestamp": "2026-01-25T12:00:00Z"}
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
pk = Ed25519PrivateKey.from_private_bytes(b64decode(PRIVATE_KEY))
sig = b64encode(pk.sign(canonical)).decode()

resp = requests.post(f"{BASE}/enforce/intercept", headers=HEADERS, json={
    "action_type": "execute_trade",
    "agent_id": "agent_abc",
    "signed_assertion": payload,
    "assertion_signature": sig
})
# → response: {"identity_verified": true, "identity": {"did": "...", "fingerprint": "..."}}

SDK Auto-Capture v1.19

Integrate once — every AI agent action is automatically captured on the Authorisation Layer dashboard. No manual intercept() call per action. Three patterns available depending on your stack.

Option 1: @guard Decorator

Wrap any Python function. Every call is intercepted, signed, and recorded. Blocked actions raise AgentBlockedError and the wrapped function never executes. When an action is escalated and wait_on_escalate=True (default), the agent blocks until a human approves or rejects on the Authorisation Layer dashboard.

from xybern import Xybern, AgentCredential, AgentBlockedError

client = Xybern(api_key="xb_your_key")
cred   = AgentCredential.load("./trading_agent.cred")

@client.agents.guard(credential=cred)
def execute_trade(ticker, quantity, notional_usd=0):
    place_order(ticker, quantity)   # only runs if Authorisation Layer allows it

@client.agents.guard(credential=cred)
def query_portfolio(account):
    return get_holdings(account)

# Every call → captured on Authorisation Layer dashboard, identity verified
execute_trade("AAPL", 100, notional_usd=19000)   # → ALLOW
execute_trade("TSLA", 1000, notional_usd=5000000) # → BLOCK (raises AgentBlockedError)

The @guard decorator accepts the following optional parameters for human-in-the-loop escalation gating:

Parameter	Default	Description
`credential`	required	Path to `.cred` file or `AgentCredential` object
`wait_on_escalate`	`True`	Block execution until a human approves or rejects on the dashboard
`escalation_timeout`	`300`	Maximum seconds to wait for a human decision before raising `AgentBlockedError`
`escalation_poll_interval`	`5`	Seconds between each poll of the escalation status endpoint
`metadata_fn`	`None`	Callable `(args, kwargs) → dict` for fine-grained policy metadata

Provide a metadata_fn for fine-grained policy evaluation:

@client.agents.guard(
    credential=cred,
    metadata_fn=lambda args, kw: {
        "ticker":       kw.get("ticker"),
        "notional_usd": kw.get("notional_usd", 0),
        "strategy":     kw.get("strategy", ""),
    }
)
def execute_trade(ticker, quantity, notional_usd=0, strategy=""):
    place_order(ticker, quantity)

Option 2: LangChain Callback Handler

Zero changes to tool code. Attach XybernLangChainCallback to any LangChain agent — every tool call is intercepted before execution.

from xybern import Xybern, XybernLangChainCallback, AgentCredential
from langchain.agents import initialize_agent

client  = Xybern(api_key="xb_your_key")
cred    = AgentCredential.load("./trading_agent.cred")
handler = XybernLangChainCallback(client=client, credential=cred)

agent = initialize_agent(tools, llm, callbacks=[handler])
# Done — every tool call appears on the Authorisation Layer dashboard

Option 3: XybernToolkit

Wrap a list of plain callables for non-LangChain agents. Exposes tools as attributes with full auto-interception.

from xybern import Xybern, XybernToolkit, AgentCredential

client = Xybern(api_key="xb_your_key")
cred   = AgentCredential.load("./trading_agent.cred")

def execute_trade(ticker, quantity, notional_usd=0): ...
def query_portfolio(account): ...
def risk_analysis(sector): ...

tools = XybernToolkit(client=client, credential=cred,
                      tools=[execute_trade, query_portfolio, risk_analysis])

# All calls auto-captured — same API as the original functions
tools.execute_trade("AAPL", 100, notional_usd=19000)
tools.query_portfolio("ACC-001")
print(tools.list_tools())  # ["execute_trade", "query_portfolio", "risk_analysis"]

AgentCredential — Key Management

Method	Description
`cred.save(path)`	Write JSON credential file, chmod 600
`AgentCredential.load(path)`	Load credential from file
`AgentCredential.from_env("PREFIX")`	Load from env vars: `PREFIX_AGENT_ID`, `PREFIX_PRIVATE_KEY`, etc.
`cred.to_env_snippet("PREFIX")`	Generate `export` lines for secrets managers / CI
`client.agents.rotate_credential(agent_id)`	Rotate keypair — old key immediately invalid

✓

Every decision is live on the dashboard. Whether via @guard, XybernLangChainCallback, or direct intercept(), every enforcement decision is recorded in real time under Authorisation Layer → Control Plane → Enforcement Decisions with full metadata, trust score, and policy reasoning.

Credential Lifecycle

Method	Endpoint	Description
`GET`	`/v1/enforce/agents/{id}/credentials`	List all credentials for an agent
`GET`	`/v1/enforce/agents/{id}/credentials/active`	Get active credential
`POST`	`/v1/enforce/agents/{id}/credentials/rotate`	Rotate: revoke old, issue new (returns private key)
`POST`	`/v1/enforce/credentials/{id}/revoke`	Permanently revoke a credential
`GET`	`/v1/enforce/credentials`	List all workspace credentials
`POST`	`/v1/enforce/credentials/{id}/verify`	Verify a signed assertion

Identity Policy Type

requests.post(f"{BASE}/enforce/policies", headers=HEADERS, json={
    "name": "Require Agent Identity",
    "policy_type": "identity",
    "decision": "block",
    "conditions": {
        "require_identity": True,
        "required_scopes": ["trade:write"],
        "blocked_dids": ["did:xybern:ws1:agent_compromised"]
    }
})

Agent-to-Agent Delegation (A2A Auth) NEW

When Agent A tells Agent B to perform an action, both sides must be authorised. Xybern intermediates every agent-to-agent interaction with bidirectional authorisation, scope attenuation, and cryptographic delegation grants recorded in the Provenance Vault.

Why this matters: Multi-agent frameworks (CrewAI, AutoGen, LangGraph) let agents freely call each other with zero authorisation. Xybern ensures Agent A is authorised to delegate, Agent B is authorised to accept, and the delegated scopes are the intersection of both — not a superset.

Core Concepts

Concept	Description
`delegation_policy`	Per-agent config: `can_delegate`, `can_accept_delegation`, `delegable_scopes`, `acceptable_scopes`, `max_delegation_depth`
Scope Attenuation	Granted scopes = source's delegable ∩ target's acceptable ∩ requested. Wildcards supported (`trade:*`).
Delegation Grant	First-class verifiable token (`dlg_...`) that the target agent presents when acting on behalf of the source.
Delegation Chains	Agent B can re-delegate to Agent C if depth allows. Max depth enforced, cascading revocation propagates down the chain.

Register Agents with Delegation Policy

import requests

BASE = "https://xybern.com/api/v1"
HEADERS = {"X-API-Key": "xb_...", "Content-Type": "application/json"}

# Source agent — can delegate trade:read and db:read
requests.post(f"{BASE}/enforce/agents", headers=HEADERS, json={
    "name": "finance-agent",
    "framework": "crewai",
    "permissions": {"allowed_action_types": ["execute_trade", "query_database"]},
    "scopes": ["trade:write", "trade:read", "db:read", "agent:delegate"],
    "delegation_policy": {
        "can_delegate": True,
        "can_accept_delegation": False,
        "delegable_scopes": ["trade:read", "db:read"],
        "max_delegation_depth": 3
    }
})

# Target agent — accepts trade:read and db:read
requests.post(f"{BASE}/enforce/agents", headers=HEADERS, json={
    "name": "analyst-agent",
    "framework": "crewai",
    "permissions": {"allowed_action_types": ["query_database", "read_data"]},
    "scopes": ["trade:read", "db:read"],
    "delegation_policy": {
        "can_delegate": False,
        "can_accept_delegation": True,
        "acceptable_scopes": ["trade:read", "db:read"]
    }
})

POST /v1/enforce/delegate

Request a delegation grant from one agent to another.

resp = requests.post(f"{BASE}/enforce/delegate", headers=HEADERS, json={
    "source_agent_id": "agent_abc123",
    "target_agent_id": "agent_def456",
    "scopes": ["trade:read", "db:read"],
    "action_types": ["query_database"],
    "constraints": {"max_amount": 100000},
    "instruction": "Analyse Q1 trading performance",
    "ttl_hours": 12,
    "max_uses": 5
})

grant = resp.json()["grant"]
# grant["grant_id"]          → "dlg_42eb33adab69"
# grant["attenuated_scopes"] → ["db:read", "trade:read"]
# grant["delegation_depth"]  → 1
# grant["vault_entry_id"]    → "ve_590dba34..."

POST /v1/enforce/delegate/verify

Verify a delegation grant before executing an action.

resp = requests.post(f"{BASE}/enforce/delegate/verify", headers=HEADERS, json={
    "grant_id": "dlg_42eb33adab69",
    "agent_id": "agent_def456",
    "action_type": "query_database"
})
# resp.json()["valid"]  → True
# resp.json()["reason"] → "Grant verified"

Using a Grant with Intercept

Pass grant_id to /enforce/intercept so the control plane verifies the delegation before policy evaluation.

resp = requests.post(f"{BASE}/enforce/intercept", headers=HEADERS, json={
    "action_type": "query_database",
    "action_content": "SELECT * FROM trades WHERE quarter='Q1'",
    "agent_id": "agent_def456",
    "grant_id": "dlg_42eb33adab69",
    "metadata": {"table": "trades", "operation": "read"}
})
# If grant is invalid or revoked → decision: "block"
# If grant is valid → normal policy evaluation proceeds

POST /v1/enforce/delegate/{grant_id}/revoke

Revoke a grant with cascading — all child grants in the delegation chain are revoked automatically.

resp = requests.post(f"{BASE}/enforce/delegate/dlg_42eb33adab69/revoke",
    headers=HEADERS, json={"reason": "task_complete"})
# resp.json()["revoked_count"]  → 1
# resp.json()["revoked_grants"] → ["dlg_42eb33adab69"]

GET /v1/enforce/delegations

List all delegation grants for the workspace. Filter by status (active, revoked, expired) or agent_id.

curl -H "X-API-Key: xb_..." \
  "https://xybern.com/api/v1/enforce/delegations?status=active"

Delegation Policies

Create enforcement policies of type delegation to control A2A behaviour at the workspace level.

requests.post(f"{BASE}/enforce/policies", headers=HEADERS, json={
    "name": "Block Finance→Marketing delegation",
    "policy_type": "delegation",
    "decision": "block",
    "conditions": {
        "blocked_pairs": [["agent_finance", "agent_marketing"]],
        "max_depth": 2,
        "blocked_scopes": ["payments:execute"]
    }
})

Policy Shadow Mode NEW

Test new policies against live production traffic without affecting any decisions. Shadow policies are evaluated on every incoming action but their verdicts are logged separately — the actual authorisation decision is never changed.

Use case: Before deploying a policy that blocks all trades over $100K, run it in shadow mode for 24 hours. The Policy Simulation dashboard shows exactly which actions would have been blocked, which agents would be affected, and whether any outcomes would change — all with zero production risk.

How It Works

Step	What Happens
1. Create	Create a policy with `"mode": "shadow"`. It's active but observe-only.
2. Evaluate	Every `/enforce/intercept` call evaluates shadow policies in a separate pass after the real decision.
3. Record	Shadow results are stored in `shadow_results` on the decision record: what the shadow policy would have decided, which policies triggered, whether the outcome would differ.
4. Report	The `/enforce/shadow/report` endpoint and the Policy Simulation dashboard aggregate the impact.
5. Promote	When satisfied, promote to live with a single API call or one click in the dashboard.

Create a Shadow Policy

resp = requests.post(f"{BASE}/enforce/policies", headers=HEADERS, json={
    "name": "Block Trades Over 500K",
    "policy_type": "threshold",
    "action_types": ["execute_trade"],
    "decision": "block",
    "priority": 200,
    "trust_threshold": 95,
    "mode": "shadow"        # ← observe-only
})
# resp.json()["policy"]["mode"] → "shadow"

Shadow Results on Decisions

After sending traffic, each decision record includes a shadow_results field:

{
  "decision": "allow",
  "shadow_results": {
    "shadow_decision": "block",
    "would_have_changed": true,
    "live_decision": "allow",
    "triggered": [
      {
        "policy_id": "abc123",
        "name": "Block Trades Over 500K",
        "decision": "block",
        "reason": "Agent trust 50.2 < threshold 95"
      }
    ],
    "reasoning": "Agent trust 50.2 < threshold 95"
  }
}

GET /v1/enforce/shadow/report

Aggregate impact report for all active shadow policies over a time window.

curl -H "X-API-Key: xb_..." \
  "https://xybern.com/api/v1/enforce/shadow/report?hours=24"

{
  "shadow_policies": 2,
  "hours": 24,
  "report": {
    "total_evaluated": 847,
    "would_block": 23,
    "would_escalate": 8,
    "would_change_outcome": 15,
    "affected_agents": ["agent_abc", "agent_def"],
    "per_policy": [
      {
        "policy_id": "abc123",
        "name": "Block Trades Over 500K",
        "would_block": 23,
        "would_escalate": 0,
        "total_triggered": 23
      }
    ]
  }
}

POST /v1/enforce/shadow/{policy_id}/promote

Promote a shadow policy to live mode — it will start enforcing immediately.

resp = requests.post(
    f"{BASE}/enforce/shadow/{policy_id}/promote",
    headers=HEADERS
)
# resp.json()["policy"]["mode"] → "live"

Toggle Mode via Update

You can also switch any policy between live and shadow using the standard update endpoint:

requests.put(f"{BASE}/enforce/policies/{policy_id}",
    headers=HEADERS,
    json={"mode": "shadow"}  # or "live"
)

Dashboard

The Authorisation Layer dashboard includes:

Policies tab — every policy shows a LIVE or SHADOW badge with a one-click mode toggle.
Policy Simulation tab — impact stats (would-block, would-escalate, outcome changes), per-policy breakdown, live vs shadow comparison table, and a "Promote to Live" button.

Temporal Permission Windows NEW

Grant time-bounded permissions to AI agents that auto-expire. Modeled after Just-In-Time (JIT) access patterns in human IAM — AWS STS temporary credentials, CyberArk JIT provisioning, HashiCorp Vault dynamic secrets — but purpose-built for AI agent authorization.

No standing access. An agent gets exactly the permissions it needs, for exactly the duration of its workflow, and those permissions auto-revoke when the window closes. Every lifecycle event (creation, use, extension, revocation) is anchored in the Provenance Vault.

Concept	Description
`TemporalPermissionWindow`	A time-bounded authorization grant with scopes, action types, constraints, and auto-expiry.
Duration	1 minute to 24 hours. Configurable per window.
Scopes	Same scope system as credentials — `trade:write`, `db:read`, etc.
Max Uses	Optional limit on how many times the window can be used within its time boundary.
Extensions	Windows can be extended up to `max_extensions` times (default: 3). Each extension is vault-recorded.
Workflow Binding	Optional `workflow_id` ties the window to a specific workflow execution.
Lazy Expiry	No background sweeper — expiry is checked at intercept time for zero overhead.

Create a Temporal Permission Window

import requests

BASE = "https://xybern.com/api/v1"
HEADERS = {"Authorization": "Bearer xb_your_key"}

resp = requests.post(f"{BASE}/enforce/temporal-windows", headers=HEADERS, json={
    "agent_id": "agent_finance_01",
    "scopes": ["payments.read", "payments.execute", "db:read"],
    "duration_minutes": 30,
    "action_types": ["execute_trade", "query_database"],
    "constraints": {"max_amount": 50000},
    "reason": "Processing customer order #4821",
    "workflow_id": "wf_order_4821",
    "max_uses": 10,
    "max_extensions": 2
})

window = resp.json()["window"]
# window["window_id"]         → "tw_a1b2c3d4e5f6"
# window["expires_at"]        → "2026-04-01T14:30:00Z"
# window["remaining_seconds"] → 1800
# window["is_active"]         → True

How It Works at Intercept Time

When an agent makes a request through POST /v1/enforce/intercept, the control plane automatically checks for active temporal windows. If the agent has an active window that covers the requested action, the response includes a temporal_window field:

{
  "decision": "allow",
  "decision_id": "enf_abc123",
  "temporal_window": {
    "window_id": "tw_a1b2c3d4e5f6",
    "remaining_seconds": 1247,
    "reason": "Temporal window 'tw_a1b2c3d4e5f6' grants access (expires in 1247s)"
  }
}

Extend a Window

If a workflow is still running and needs more time, extend the window (up to max_extensions):

resp = requests.post(
    f"{BASE}/enforce/temporal-windows/{window_id}/extend",
    headers=HEADERS,
    json={
        "additional_minutes": 15,
        "reason": "Workflow still processing batch"
    }
)
# resp.json()["window"]["extensions"]      → 1
# resp.json()["window"]["remaining_seconds"] → updated

Revoke a Window

Immediately terminate a window before it expires:

resp = requests.post(
    f"{BASE}/enforce/temporal-windows/{window_id}/revoke",
    headers=HEADERS,
    json={"reason": "Workflow completed early"}
)
# resp.json()["window"]["status"] → "revoked"

Pre-flight Check

Check if an agent has an active window without consuming a use:

resp = requests.post(f"{BASE}/enforce/temporal-windows/check", headers=HEADERS, json={
    "agent_id": "agent_finance_01",
    "action_type": "execute_trade",
    "metadata": {"amount": 25000}
})
# resp.json()["has_active_window"] → True
# resp.json()["remaining_seconds"] → 847

List & Stats

# List all windows (filter by agent, status, workflow)
requests.get(f"{BASE}/enforce/temporal-windows?active_only=true", headers=HEADERS)

# Agent-specific windows
requests.get(f"{BASE}/enforce/agents/{agent_id}/temporal-windows", headers=HEADERS)

# Aggregate stats
requests.get(f"{BASE}/enforce/temporal-windows/stats", headers=HEADERS)
# → { active_windows, expired_windows, revoked_windows, total_uses, avg_duration_minutes, ... }

Dashboard

The Authorisation Layer dashboard includes a dedicated Temporal Windows tab showing:

Active windows with real-time countdown timers, scope badges, and one-click extend/revoke
Window history — expired and revoked windows with usage stats
Aggregate stats — active count, total uses, average duration, agents with active windows

Breakglass Protocol NEW

Emergency override mechanism for blocked enforcement decisions. When a critical action is blocked by policy but must proceed, operators trigger a breakglass override — a time-limited bypass with mandatory justification, immutable audit trail, and post-incident review requirements.

⚠️

Use sparingly. Breakglass overrides bypass normal authorization. Every event is recorded in the Provenance Vault and requires post-incident review. A per-agent cooldown (max 3 per 30 minutes) prevents abuse.

Concept	Description
`BreakglassEvent`	A time-limited authorization override with justification, severity, and audit trail
`Justification`	Mandatory written rationale (min 10 chars) explaining why the override is needed
`Severity`	`critical`, `high`, or `medium` — controls visibility and review priority
`Cooldown`	Max 3 breakglass events per agent per 30-minute window
`Auto-expire`	Override auto-expires after configured duration (default 15 min, max 2 hours)
`Post-incident review`	Events require review annotation after the fact for compliance

Trigger a Breakglass Override

import requests

BASE = "https://www.xybern.com/api/v1"
HEADERS = {"X-API-Key": "xb_your_key"}

override = requests.post(f"{BASE}/enforce/breakglass", headers=HEADERS, json={
    "agent_id": "agent_deploy_01",
    "action_type": "deploy:production",
    "justification": "Critical hotfix for payment processing outage — approved by VP Eng",
    "triggered_by": "oncall_engineer_42",
    "severity": "critical",
    "duration_minutes": 15,
    "max_actions": 5,
})

bg = override.json()["event"]
print(f"Override active: {bg['breakglass_id']}")
print(f"Expires in: {bg['remaining_seconds']}s")

How It Works at Intercept Time

When an action is blocked and the agent has an active breakglass override, the decision is automatically flipped to allow with a breakglass proof attached:

{
  "decision": "allow",
  "decision_id": "enf_abc123",
  "decision_path": "breakglass",
  "reasoning": "Policy violation detected | Breakglass override active (bg_xyz789)",
  "breakglass": {
    "breakglass_id": "bg_xyz789",
    "remaining_seconds": 842,
    "reason": "Breakglass override 'bg_xyz789' active (expires in 842s)"
  }
}

When a decision is blocked and no active override exists, the response includes a hint:

{
  "decision": "block",
  "breakglass_available": true
}

Close an Override Early

requests.post(f"{BASE}/enforce/breakglass/{bg_id}/close", headers=HEADERS, json={
    "reason": "Hotfix deployed successfully, no longer needed"
})

Add Post-Incident Review

requests.post(f"{BASE}/enforce/breakglass/{bg_id}/review", headers=HEADERS, json={
    "reviewed_by": "security_lead_01",
    "review_notes": "Override was justified — payment outage affected 12k users. "
                    "Root cause: stale policy blocking deploy:production for this agent. "
                    "Action: Updated policy to allow during incident windows."
})

List & Stats

# List all breakglass events
requests.get(f"{BASE}/enforce/breakglass", headers=HEADERS)

# Active overrides only
requests.get(f"{BASE}/enforce/breakglass?active_only=true", headers=HEADERS)

# Aggregate stats
requests.get(f"{BASE}/enforce/breakglass/stats", headers=HEADERS)
# → { total_events, active_overrides, pending_review, reviewed, by_severity, ... }

Dashboard

The Authorisation Layer dashboard includes a dedicated Breakglass Protocol tab showing:

Active overrides with real-time countdown timers, severity badges, and one-click close
Event history — closed and expired events with review status indicators
Severity breakdown — critical/high/medium counts at a glance
Post-incident review — review button on unreviewed events, notes visible inline

Policy-as-Code SDK NEW

Define authorization policies as Python code, version them in Git, and deploy atomically via the SDK. Xybern diffs your policy definitions against the current state and creates, updates, or removes policies automatically — with full Provenance Vault tracking.

Concept	Description
`Policy`	A fluent builder for a single enforcement policy — `.on()`, `.when_*()`, `.block()`
`PolicyPack`	A versioned collection of policies deployed as one atomic unit
`PolicyClient`	SDK sub-client at `client.policies` for deploy, rollback, validate operations
Deploy	Create/update/delete policies from a pack — `POST /v1/enforce/policy-packs`
Rollback	Revert to previous pack version — `POST /v1/enforce/policy-packs/:name/rollback`
Validate	Dry-run a pack without deploying — `POST /v1/enforce/policy-packs/validate`
Source Hash	SHA-256 fingerprint of policy definitions — idempotent redeployments

Define Policies with the Python DSL

from xybern import Xybern, PolicyPack, Policy

client = Xybern(api_key="xb_your_key")

pack = PolicyPack("finance-controls", version="2.0.0",
                  description="Production trading controls")

# Block low-trust agents from executing trades
pack.add(Policy("Block Untrusted Traders")
    .on("execute_trade")
    .when_threshold(trust_below=50)
    .block("Agents with trust < 50 cannot trade"))

# No deployments on weekends
pack.add(Policy("Weekend Deploy Freeze")
    .on("deploy:*")
    .when_time(blocked_days=[6, 7])
    .escalate("Weekend deploys require human approval"))

# Detect PII in prompts
pack.add(Policy("PII Scanner")
    .on("send_prompt")
    .when_content_matches(r"\b\d{3}-\d{2}-\d{4}\b", r"\bSSN\b")
    .block("PII detected in prompt content"))

# Limit delegation chain depth
pack.add(Policy("Chain Depth Guard")
    .when_chain(max_depth=3)
    .escalate("Delegation chain too deep"))

Deploy a Policy Pack

# Deploy to live enforcement
result = client.policies.deploy(pack)
print(result)
# → {"ok": true, "pack": {...}, "summary": {"created": 4, "updated": 0, "deleted": 0}}

# Deploy in shadow mode (observe only, never enforced)
result = client.policies.deploy(pack, deploy_mode="shadow")

# Redeploy with changes — Xybern auto-diffs
pack_v2 = PolicyPack("finance-controls", version="2.1.0")
pack_v2.add(Policy("Block Untrusted Traders")
    .on("execute_trade")
    .when_threshold(trust_below=60)  # raised threshold
    .block("Agents with trust < 60 cannot trade"))

result = client.policies.deploy(pack_v2)
# → {"summary": {"created": 0, "updated": 1, "deleted": 3, "unchanged": 0}}

Validate Without Deploying

# Dry-run validation
validation = client.policies.validate(pack)
print(validation)
# → {"ok": true, "valid": true, "policy_count": 4, "policies": [...]}

Rollback to Previous Version

# Something went wrong — rollback instantly
result = client.policies.rollback("finance-controls")
# → redeploys previous version's policies

Delete a Pack

# Remove a pack and all its managed policies
result = client.policies.delete("finance-controls")
# → {"ok": true, "policies_removed": 4}

REST API Reference

POST /v1/enforce/policy-packs — Deploy a policy pack
GET /v1/enforce/policy-packs — List all packs
GET /v1/enforce/policy-packs/:name — Get a specific pack
POST /v1/enforce/policy-packs/:name/rollback — Rollback to previous version
DELETE /v1/enforce/policy-packs/:name — Delete a pack
POST /v1/enforce/policy-packs/validate — Dry-run validation
GET /v1/enforce/policy-packs/stats — Pack statistics

Dashboard

The Authorisation Layer Control Plane includes a dedicated Policy-as-Code view showing all active packs, their version, policy count, deploy mode, source hash, and deployment timestamp. From the dashboard you can rollback or delete packs with one click.

Federation NEW

Enable secure, policy-controlled interactions between AI agents across different Xybern workspaces and organizations. Federation links establish bilateral trust, and short-lived tokens allow scoped cross-org actions.

Concept	Description
`Federation Link`	A trust relationship between two workspaces (source → target) with configurable guardrails
`Federation Token`	A short-lived, scope-limited, use-counted token for cross-org agent calls
`Trust Cap`	Maximum trust level for external agents (prevents foreign agents from exceeding local thresholds)
`Direction`	`outbound` (you → partner) or `inbound` (partner → you)

Propose a Federation Link

from xybern import Xybern

client = Xybern(api_key="xb_your_key")

# Propose federation to a partner organization
link = client.federation.propose(
    target_workspace_id="partner_workspace_id",
    source_org_name="Acme Corp",
    target_org_name="Partner Inc",
    allowed_action_types=["data:read", "data:query"],
    allowed_scopes=["read:market_data"],
    max_trust_level=40.0,        # External agents capped at 40
    expires_in_days=90,           # Link expires after 90 days
)
print(link["link_id"])           # fed_a1b2c3d4e5f6
print(link["shared_secret"])     # Exchange this securely

Accept an Inbound Link (Partner Side)

# Partner accepts the pending link
partner_client = Xybern(api_key="xb_partner_key")
result = partner_client.federation.accept(
    link_id="fed_a1b2c3d4e5f6",
    approved_by="security-team",
)
# Both sides now show status: "active"

Issue a Cross-Org Token

# Source org issues a token for one of its agents
token = client.federation.issue_token(
    link_id="fed_a1b2c3d4e5f6",
    agent_id="research-agent-01",
    scopes=["read:market_data"],
    action_types=["data:read"],
    ttl_seconds=300,     # 5-minute lifetime
    max_uses=10,         # Up to 10 uses
)
print(token["token"])    # Short-lived credential
print(token["expires_in"])  # 300

Use Token in Cross-Org Intercept

# Agent includes the federation token in its intercept call
result = partner_client.agents.intercept(
    action_type="data:read",
    action_content="Query market data for AAPL",
    credential="./my_agent.cred",
    federation_token=token["token"],
)
# Response includes federation proof:
# result.raw["federation"]["source_org"]  → "Acme Corp"
# result.raw["federation"]["trust_cap"]   → 40.0

REST API Reference

Method	Endpoint	Description
`POST`	`/v1/enforce/federation/links`	Propose a new federation link
`GET`	`/v1/enforce/federation/links`	List all federation links
`POST`	`/v1/enforce/federation/links/:id/accept`	Accept a pending link
`POST`	`/v1/enforce/federation/links/:id/suspend`	Suspend an active link
`POST`	`/v1/enforce/federation/links/:id/revoke`	Permanently revoke a link
`POST`	`/v1/enforce/federation/tokens`	Issue a cross-org token
`GET`	`/v1/enforce/federation/stats`	Get federation statistics

Authorisation Layer

The Authorisation Layer Control Plane includes a dedicated Federation view showing all active links, pending invites, token counts, and federated action totals. You can accept, suspend, or revoke links directly from the dashboard.

Human-in-the-Loop Escalation Gating NEW

When a policy decision is escalate, the agent can now block and wait until a human reviews the action on the Authorisation Layer dashboard. This is controlled by the wait_on_escalate parameter on the @guard decorator (default: True).

Full Flow

Agent calls the guarded function → Authorisation Layer intercepts and returns decision: "escalate" with an escalation_id.
SDK polls GET /v1/enforce/escalations/{id}/status every escalation_poll_interval seconds.
Human opens Authorisation Layer → Control Plane, sees the pending escalation card with Approve and Reject buttons.
If approved: agent unblocks and the wrapped function executes normally.
If rejected (or timeout exceeded): AgentBlockedError is raised and the function never runs.

Code Example

from xybern import Xybern, AgentCredential, AgentBlockedError

client = Xybern(api_key="xb_your_key")
cred   = AgentCredential.load("./bot.cred")

@client.agents.guard(
    credential=cred,
    wait_on_escalate=True,      # Block until human decides (default)
    escalation_timeout=300,     # Wait up to 5 minutes
    escalation_poll_interval=5, # Check every 5 seconds
)
def wire_transfer(amount_usd: float, recipient: str):
    # Only executes if a human approves on the Authorisation Layer dashboard
    execute_wire(amount_usd, recipient)

try:
    wire_transfer(50000, "vendor@example.com")
except AgentBlockedError as e:
    print(f"Transfer rejected or timed out: {e}")

InterceptResult Attributes

Attribute	Type	Description
`.decision`	`str`	`"allow"`, `"block"`, or `"escalate"`
`.trust_score`	`float`	Agent trust score at time of decision (0-100)
`.decision_id`	`str`	Immutable decision record ID
`.escalation_id`	`str \| None`	Set when `decision == "escalate"`; used to poll for human resolution
`.policy_name`	`str \| None`	Name of the policy that triggered this decision
`.latency_ms`	`int`	Decision latency in milliseconds

wait_for_escalation() Method

The @guard decorator calls this automatically when wait_on_escalate=True. You can also call it directly when handling escalations manually.

# Manual escalation polling (when wait_on_escalate=False)
result = client.agents.intercept(
    action_type="wire_transfer",
    action_content="Transfer $50,000 to vendor",
    agent_id=cred.agent_id,
)

if result.decision == "escalate":
    resolution = client.agents.wait_for_escalation(
        escalation_id=result.escalation_id,
        timeout=300,        # seconds
        poll_interval=5,    # seconds
    )
    # resolution == "approved" or "rejected"
    if resolution == "approved":
        execute_wire(50000, "vendor@example.com")

Escalation Status Endpoint

GET /v1/enforce/escalations/{escalation_id}/status

Poll the resolution status of a pending escalation. Returns {"ok": true, "status": "pending"|"approved"|"rejected"}. Used internally by the SDK's wait_for_escalation() method.

Dashboard: Approve / Reject

Pending escalations appear as cards in Authorisation Layer → Control Plane. Each card shows the action details, agent identity, and two buttons: Approve and Reject. Resolving a card immediately unblocks any waiting SDK call.

The underlying endpoint is: POST /api/sentinel/enforcement/escalations/{escalation_id}/resolve with body {"resolution": "approved"|"rejected"}.

💡

Set wait_on_escalate=False if you want fire-and-forget escalation behaviour — the SDK returns immediately with decision="escalate" and you handle the escalation_id asynchronously.

Category	Examples Detected
Gender	salesman → salesperson, chairman → chairperson
Race	blacklist → blocklist, racial generalizations
Age	"young and energetic", "digital native"
Disability	crazy → unexpected, handicapped → person with disability

More Authorisation Examples

Common patterns for working with the Authorisation API beyond basic intercept calls.

Registering an Agent Identity

import requests

response = requests.post(
    "https://www.xybern.com/api/v1/enforce/agents",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "agent_id": "finance_agent_001",
        "name": "Finance Reporting Agent",
        "role": "financial_analyst",
        "allowed_tools": ["read_database", "generate_report", "send_email"],
        "denied_tools": ["delete_records", "modify_schema"],
        "max_delegation_depth": 2
    }
)
print(response.json()["status"])  # "registered"

Creating a Policy

response = requests.post(
    "https://www.xybern.com/api/v1/enforce/policies",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "policy_id": "no_external_email_without_approval",
        "name": "Block external email without human approval",
        "rules": [
            {
                "condition": {
                    "action.tool": "send_email",
                    "action.parameters.to": {"not_contains": "@company.com"}
                },
                "effect": "DENY",
                "reason": "External emails require human-in-the-loop approval"
            }
        ],
        "shadow_mode": False
    }
)

A2A Delegation

response = requests.post(
    "https://www.xybern.com/api/v1/enforce/delegate",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "delegator_agent_id": "orchestrator_agent",
        "delegate_agent_id": "sub_agent_001",
        "scoped_tools": ["read_database", "generate_report"],
        "expires_in_seconds": 1800,
        "context": {"task": "q4_report", "initiated_by": "orchestrator_agent"}
    }
)
print(response.json()["delegation_token"])

Breakglass — Emergency Override

response = requests.post(
    "https://www.xybern.com/api/v1/enforce/breakglass",
    headers={"X-API-Key": "xb_your_api_key"},
    json={
        "agent_id": "incident_response_agent",
        "reason": "Production incident — database migration requires elevated access",
        "requested_tools": ["modify_schema", "delete_records"],
        "duration_seconds": 900,
        "approver_id": "admin_user_007"
    }
)
# Breakglass events are always logged regardless of outcome
print(response.json()["breakglass_token"])

Framework Integrations

Drop Xybern authorisation into your existing agent stack. Each integration wraps your tools or callbacks so every action is intercepted before execution — no changes to your agent logic required.

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()

AutoGen

Wrap AutoGen's function_map so every function call is authorised before execution. DENY decisions return a safe error string rather than raising, keeping the conversation intact.

# autogen_xybern.py
import requests
import autogen

XYBERN_API_KEY = "xb_your_api_key"

def authorise_action(agent_name: str, tool: str, params: dict) -> bool:
    resp = requests.post(
        "https://www.xybern.com/api/v1/enforce/intercept",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_id": agent_name,
            "action": {"type": "tool_call", "tool": tool, "parameters": params},
            "context": {"framework": "autogen"}
        }
    ).json()
    return resp["decision"] == "ALLOW"

config_list = [{"model": "gpt-4", "api_key": "your_openai_key"}]

def guarded_function_map(agent_name: str, function_map: dict) -> dict:
    """Wrap an AutoGen function_map with Xybern checks."""
    guarded = {}
    for fn_name, fn in function_map.items():
        def make_guarded(name, original_fn):
            def guarded_fn(**kwargs):
                if not authorise_action(agent_name, name, kwargs):
                    return f"Action '{name}' was denied by Xybern authorisation."
                return original_fn(**kwargs)
            return guarded_fn
        guarded[fn_name] = make_guarded(fn_name, fn)
    return guarded

def query_database(table: str, query: str):
    return f"Results from {table}: [...]"

analyst = autogen.AssistantAgent(
    name="analyst_agent",
    llm_config={"config_list": config_list}
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    function_map=guarded_function_map("analyst_agent", {"query_database": query_database})
)

user_proxy.initiate_chat(analyst, message="Query the customers table and summarise the data.")

LangGraph

Add a Xybern gate as a dedicated node in your LangGraph state graph. The gate runs before any execution node and sets an xybern_approved flag in state — the execution node checks this flag before proceeding.

# langgraph_xybern.py
import requests
from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage

XYBERN_API_KEY = "xb_your_api_key"

class AgentState(TypedDict):
    messages: list
    next_action: str
    xybern_approved: bool

def xybern_gate(state: AgentState) -> AgentState:
    """LangGraph node: authorise the next action before execution."""
    next_action = state.get("next_action", "")
    if not next_action:
        return {**state, "xybern_approved": True}

    resp = requests.post(
        "https://www.xybern.com/api/v1/enforce/intercept",
        headers={"X-API-Key": XYBERN_API_KEY},
        json={
            "agent_id": "langgraph_agent_001",
            "action": {"type": "tool_call", "tool": next_action, "parameters": {}},
            "context": {"framework": "langgraph", "state_step": len(state["messages"])}
        }
    ).json()

    return {**state, "xybern_approved": resp["decision"] == "ALLOW"}

def execute_action(state: AgentState) -> AgentState:
    if not state["xybern_approved"]:
        msg = AIMessage(content="Action denied by Xybern authorisation policy.")
        return {**state, "messages": state["messages"] + [msg]}
    msg = AIMessage(content=f"Executing {state['next_action']}...")
    return {**state, "messages": state["messages"] + [msg], "next_action": ""}

graph = StateGraph(AgentState)
graph.add_node("xybern_gate", xybern_gate)
graph.add_node("execute", execute_action)
graph.set_entry_point("xybern_gate")
graph.add_edge("xybern_gate", "execute")
graph.add_edge("execute", END)

app = graph.compile()
result = app.invoke({
    "messages": [HumanMessage(content="Run report")],
    "next_action": "generate_report",
    "xybern_approved": False
})

LlamaIndex

Use make_guarded_tool to wrap any LlamaIndex FunctionTool with a Xybern check. Denied actions return a safe string that the agent can interpret without crashing.

# llamaindex_xybern.py
import requests
from llama_index.core.tools import FunctionTool
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

XYBERN_API_KEY = "xb_your_api_key"
AGENT_ID = "llamaindex_agent_001"

def xybern_check(tool_name: str, params: dict) -> bool:
    resp = 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": params},
            "context": {"framework": "llamaindex"}
        }
    ).json()
    return resp["decision"] == "ALLOW"

def make_guarded_tool(name: str, description: str, fn):
    """Wrap a LlamaIndex tool function with Xybern authorisation."""
    def guarded(**kwargs):
        if not xybern_check(name, kwargs):
            return f"Xybern denied action: {name}. Check your authorisation policies."
        return fn(**kwargs)
    return FunctionTool.from_defaults(fn=guarded, name=name, description=description)

def search_documents(query: str) -> str:
    return f"Documents matching '{query}': [doc1, doc2]"

def write_report(content: str, filename: str) -> str:
    return f"Report written to {filename}"

tools = [
    make_guarded_tool("search_documents", "Search internal documents", search_documents),
    make_guarded_tool("write_report", "Write a report to disk", write_report),
]

llm = OpenAI(model="gpt-4")
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)
response = agent.chat("Find all documents about Q4 and write a summary report.")
print(response)

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"])

Code Examples

Using the SDK

from xybern import Xybern

client = Xybern(api_key="xb_your_key")

result = client.verify(
    content="AI generated content...",
    source={"type": "llm", "model": "gpt-4"}
)

print(f"Trust: {result['trust_score']}%")
print(f"Fairness: {result['bias']['fairness_score']}%")

for bias in result['bias']['biases_detected']:
    print(f"[{bias['severity']}] {bias['suggestion']}")

Error Handling

Status	Meaning
`200`	Success
`400`	Bad request
`401`	Invalid API key
`429`	Rate limited
`500`	Server error

Component	Technology	Impact
Worker Model	Gevent (cooperative I/O)	~2,000 concurrent requests per worker
Post-Processing	ThreadPoolExecutor(3)	Bias + Regulatory + Hallucination run in parallel
Response Cache	Redis (15s TTL)	Dashboard reads served from cache under load
Content Dedup	SHA-256 hash + Redis (5 min)	Identical content skips LLM entirely
Rate Limiting	Redis sliding-window counters	Per-key RPM + daily enforcement
DB Pool	SQLAlchemy (30 + 20 overflow)	Handles burst DB concurrency

Instance	Workers	Concurrent Requests	Verifications/Hour
2 vCPU / 4GB	4	~8,000	~5,000
4 vCPU / 16GB	8	~16,000	~10,000
8 vCPU / 16GB	16	~32,000	~25,000

Caching & Deduplication

Identical content verified within a 5-minute window returns the cached result instantly — no LLM call, no latency, no cost.

⚡

Cached responses include "cached": true in the response body so you can distinguish them from fresh verifications.

How It Works

Content is hashed with SHA-256 on arrival
The hash is checked against a per-workspace Redis cache
If a match is found within the 5-minute TTL, the cached result is returned with a fresh verification_id
If no match, the full verification pipeline runs and the result is cached

Agent RBAC (Role-Based Access Control)

Define reusable roles with granular permissions and assign them to agents. Roles are evaluated during every intercept call — if any active role denies the action type, the request is blocked before policies run.

Concept	Description
Role	A named permission bundle with allowed/denied action types, scopes, and a minimum trust threshold.
Assignment	A many-to-many link between roles and agents. One agent can hold multiple roles.
Inheritance	A role can inherit from a parent role, composing permissions up the hierarchy.
Wildcards	Action type patterns like `payment:*` match any action starting with `payment:`.

Create a Role

from xybern import Xybern

client = Xybern(api_key="xb_your_key")

role = client.roles.create(
    name="finance-agent",
    description="Can read and transfer payments, but not admin actions",
    allowed_action_types=["payment:*", "report:read"],
    denied_action_types=["admin:*"],
    min_trust_level=60.0,
)
print(role["role"]["role_id"])

Assign a Role to an Agent

client.roles.assign(
    role_id="role_abc123",
    agent_id="agent_xyz789",
)

Role Inheritance

# Create a base "reader" role
reader = client.roles.create(
    name="reader",
    allowed_action_types=["*.read"],
)

# Create "analyst" that inherits from "reader" and adds more
analyst = client.roles.create(
    name="analyst",
    allowed_action_types=["report:generate", "data:query"],
    inherits_from=reader["role"]["role_id"],
)

How Enforcement Works

Agent submits an action via client.agents.intercept(...)
Control plane resolves the agent and loads all active roles
If any role's denied list matches the action type → block
If no role's allowed list matches → block
If the agent's trust level is below the role's min_trust_level → block
Otherwise, proceed to policy evaluation

REST API Reference

Method	Endpoint	Description
`POST`	`/v1/enforce/roles`	Create a role
`GET`	`/v1/enforce/roles`	List roles
`GET`	`/v1/enforce/roles/:id`	Get a role
`PUT`	`/v1/enforce/roles/:id`	Update a role
`DELETE`	`/v1/enforce/roles/:id`	Delete a role
`POST`	`/v1/enforce/roles/:id/assign`	Assign role to agent
`POST`	`/v1/enforce/roles/:id/unassign`	Unassign role from agent
`GET`	`/v1/enforce/roles/:id/agents`	List agents in role
`GET`	`/v1/enforce/roles/stats`	RBAC statistics

Dashboard

The Roles (RBAC) tab in the Authorisation Layer dashboard displays all roles, their agent counts, allowed/denied action types, trust thresholds, and inheritance chains. You can delete roles directly from the UI.

Webhooks & Real-Time Event Streaming

Subscribe to authorization events and have them delivered to any HTTP endpoint in real time. Every payload is HMAC-SHA256 signed with your webhook secret.

Supported Event Types

Event	Fires when...
`decision.block`	Any agent action is blocked
`decision.escalate`	A decision requires human review
`decision.allow`	An action is approved (opt-in, high volume)
`breakglass.triggered`	Emergency override activated
`breakglass.deactivated`	Emergency override ended
`federation.link_proposed`	New cross-org trust link proposed
`federation.token_issued`	Cross-org token minted
`policy.deployed`	A policy pack goes live
`policy.rollback`	A policy pack is rolled back
`agent.registered`	New agent registered
`agent.trust_changed`	Agent trust score crosses a threshold
`rbac.role_assigned`	A role is assigned to an agent
`temporal.window_opened`	A JIT access window opens
`temporal.window_expired`	A JIT access window closes

Create a Webhook

from xybern import Xybern

client = Xybern(api_key="xb_your_key")

hook = client.webhooks.create(
    url="https://hooks.slack.com/services/T0/B0/xxx",
    events=["decision.block", "breakglass.triggered"],
    description="Slack alerts for security events",
)
# Save hook["webhook"]["secret"] to verify signatures later
print(hook["webhook"]["webhook_id"])

Payload Format

{
  "event": "decision.block",
  "timestamp": "2026-04-01T19:32:14.003Z",
  "workspace_id": "f5300764...",
  "data": {
    "decision_id": "enf_a1b2c3d4e5f6",
    "agent_id": "agent_35c86445e7e2",
    "action_type": "admin:delete_workspace",
    "decision": "block",
    "decision_path": "rbac",
    "reasoning": "Action not allowed by agent roles",
    "trust_score": 22.8
  }
}

Signature Verification

Every delivery includes an X-Xybern-Signature header. Verify it server-side:

import hmac, hashlib

def verify_webhook(payload_bytes, signature_header, secret):
    expected = hmac.new(
        secret.encode(), payload_bytes, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature_header)

Retry Policy

Failed deliveries are retried with exponential backoff (1s, 2s, 4s, 8s, 16s). After 50 consecutive failures, the webhook is auto-disabled. Re-enable it via the API or dashboard.

REST API Reference

Method	Endpoint	Description
`POST`	`/v1/enforce/webhooks`	Create a subscription
`GET`	`/v1/enforce/webhooks`	List subscriptions
`GET`	`/v1/enforce/webhooks/:id`	Get a subscription
`PUT`	`/v1/enforce/webhooks/:id`	Update a subscription
`DELETE`	`/v1/enforce/webhooks/:id`	Delete a subscription
`POST`	`/v1/enforce/webhooks/:id/test`	Send a test event
`GET`	`/v1/enforce/webhooks/:id/deliveries`	List delivery history
`POST`	`/v1/enforce/webhooks/:id/rotate-secret`	Rotate signing secret
`GET`	`/v1/enforce/webhooks/stats`	Webhook statistics

Rate Limits

Plan	Requests/Minute	Requests/Day
Free	10	100
Pro	60	5,000
Enterprise	Custom	Custom

Rate limits are enforced per API key using sliding-window counters. When exceeded, the API returns:

{
  "ok": false,
  "error": "rate_limit_exceeded",
  "message": "Rate limit exceeded: 60 requests/minute",
  "retry_after": 45
}

📧

Need higher limits? Contact info@xybern.com

SDK Enforcement Client

The Python and JavaScript SDKs expose the full enforcement control plane via client.enforce — no raw HTTP required. Every endpoint in the Authorisation API has a corresponding SDK method.

📦

Version: Available from xybern>=1.18.0 (Python) and @xybern/sdk>=1.18.0 (JavaScript).

Python

from xybern import Xybern

client = Xybern(api_key="xb_your_key")

# ── Core intercept ──────────────────────────────────────────────
result = client.enforce.intercept(
    action_type="execute_trade",
    action_content="Buy 500 AAPL @ market",
    agent_id="agent_abc123",
    metadata={"symbol": "AAPL", "qty": 500},
)
print(result["decision"])   # allow | block | escalate

# Block until a human resolves the escalation
if result["decision"] == "escalate":
    esc_id = result.get("escalation_id")
    status = client.enforce.wait_for_escalation(esc_id, timeout=3600.0)
    # status["status"] == "approved" | "rejected"

# ── Agents ──────────────────────────────────────────────────────
agent = client.enforce.register_agent(
    name="TradeBot", framework="langchain",
    scopes=["execute_trade", "query_database"],
)
print(agent["credential"]["private_key"])  # one-time — store securely

# ── Delegation ──────────────────────────────────────────────────
grant = client.enforce.delegate(
    source_agent_id="agent_A",
    target_agent_id="agent_B",
    scopes=["trade:read"],
    duration_minutes=60,
    max_uses=10,
)

# ── JIT temporal window ─────────────────────────────────────────
window = client.enforce.create_temporal_window(
    agent_id="agent_abc123",
    scopes=["database_write"],
    duration_minutes=30,
    reason="Emergency schema migration",
)

# ── Breakglass ──────────────────────────────────────────────────
bg = client.enforce.trigger_breakglass(
    agent_id="agent_abc123",
    justification="Critical prod incident — P0",
    severity="critical",
    duration_minutes=60,
)

# ── Policy packs (policy-as-code) ───────────────────────────────
client.enforce.deploy_policy_pack(
    name="finance-v2",
    version="2.1.0",
    policy_definitions=[
        {
            "name": "Block high-value trades",
            "policy_type": "threshold",
            "action_types": ["execute_trade"],
            "decision": "escalate",
            "trust_threshold": 80,
        }
    ],
)

# ── Federation ──────────────────────────────────────────────────
link = client.enforce.propose_federation(
    target_workspace_id="ws_partner_org",
    allowed_action_types=["query_database"],
    max_trust_level=70,
)
token = client.enforce.issue_federation_token(
    link_id=link["link"]["link_id"],
    source_agent_id="agent_abc123",
    scopes=["query_database"],
    ttl_seconds=300,
)

# ── Webhooks ────────────────────────────────────────────────────
client.enforce.create_webhook(
    url="https://your-system.com/xybern-events",
    events=["decision.block", "breakglass.*", "federation.*"],
)

# ── Roles (RBAC) ────────────────────────────────────────────────
role = client.enforce.create_role(
    name="read-only-agent",
    allowed_action_types=["query_database", "read_file"],
    denied_action_types=["execute_trade", "send_email"],
    min_trust_level=40,
)
client.enforce.assign_role(role["role"]["role_id"], "agent_abc123")

JavaScript / TypeScript

import { Xybern } from '@xybern/sdk';

const client = new Xybern({ apiKey: 'xb_your_key' });

// ── Core intercept ──────────────────────────────────────────────
const result = await client.enforce.intercept({
  actionType: 'execute_trade',
  actionContent: 'Buy 500 AAPL @ market',
  agentId: 'agent_abc123',
  metadata: { symbol: 'AAPL', qty: 500 },
});
// result.decision === 'allow' | 'block' | 'escalate'

// Block until resolved (with timeout)
if (result.decision === 'escalate') {
  const status = await client.enforce.waitForEscalation(result.escalationId, {
    pollIntervalMs: 5000,
    timeoutMs: 3_600_000,
  });
  // status.status === 'approved' | 'rejected'
}

// ── Register agent ──────────────────────────────────────────────
const agent = await client.enforce.registerAgent({
  name: 'TradeBot',
  framework: 'langchain',
  scopes: ['execute_trade', 'query_database'],
});

// ── Batch intercept (up to 500 actions) ────────────────────────
const batch = await client.enforce.batchIntercept([
  { ref: 'step-1', actionType: 'read_file',     actionContent: 'Read config.json' },
  { ref: 'step-2', actionType: 'execute_trade', actionContent: 'Buy 500 AAPL' },
  { ref: 'step-3', actionType: 'send_email',    actionContent: 'Notify compliance' },
]);
// batch.allowed / batch.blocked / batch.escalated

// ── Policy pack deployment ──────────────────────────────────────
await client.enforce.deployPolicyPack({
  name: 'finance-v2',
  version: '2.1.0',
  policyDefinitions: [
    {
      name: 'Block high-value trades',
      policyType: 'threshold',
      actionTypes: ['execute_trade'],
      decision: 'escalate',
      trustThreshold: 80,
    },
  ],
});

// ── Temporal window ─────────────────────────────────────────────
await client.enforce.createTemporalWindow({
  agentId: 'agent_abc123',
  scopes: ['database_write'],
  durationMinutes: 30,
  reason: 'Emergency schema migration',
});

Full method reference

All methods are available on both client.enforce (Python) and client.enforce (JavaScript). The complete list covers:

Category	Methods
Core	`intercept`, `batch_intercept`, `intercept_agent_comm`
Agents	`register_agent`, `get_agent`, `update_agent`, `deactivate_agent`, `get_agent_history`, `get_agent_communications`
Credentials	`list_credentials`, `get_active_credential`, `rotate_credential`, `revoke_credential`
Policies	`list_policies`, `create_policy`, `update_policy`, `delete_policy`, `promote_shadow_policy`, `get_shadow_report`
Decisions	`list_decisions`, `get_decision`
Escalations	`list_escalations`, `get_escalation_status`, `resolve_escalation`, `wait_for_escalation`
Delegations	`delegate`, `verify_delegation`, `revoke_delegation`, `list_delegations`
Temporal Windows	`create_temporal_window`, `extend_temporal_window`, `revoke_temporal_window`, `check_temporal_window`
Breakglass	`trigger_breakglass`, `close_breakglass`, `review_breakglass`, `get_breakglass_stats`
Roles (RBAC)	`create_role`, `assign_role`, `unassign_role`, `list_role_agents`
Federation	`propose_federation`, `accept_federation`, `issue_federation_token`, `revoke_federation`
Webhooks	`create_webhook`, `test_webhook`, `rotate_webhook_secret`, `get_webhook_deliveries`
Policy Packs	`deploy_policy_pack`, `rollback_policy_pack`, `validate_policy_pack`

MCP Server NEW

Xybern ships a native Model Context Protocol server that exposes the entire enforcement control plane as tools any MCP-compatible agent can call — Claude Code, Cursor, Windsurf, Codex, Gemini CLI, and more.

No SDK installation required on the agent side. No custom integration. Connect once and every AI agent in your organisation operates under your Xybern policies automatically.

⚡

Supported clients: Claude Desktop, Claude Code, Cursor, Windsurf, Zed, Cline, Continue.dev, OpenAI Codex, Gemini CLI, and any MCP-compatible agent.

Zero-Friction Setup

Add Xybern to your MCP config with a single API key. The server auto-registers the agent session on first connect — no pre-registration step required.

Claude Desktop / Claude Code

Edit ~/.claude/claude_desktop_config.json (Claude Desktop) or your project's .mcp.json (Claude Code):

{
  "mcpServers": {
    "xybern": {
      "command": "python",
      "args": ["-m", "mcp.server"],
      "cwd": "/path/to/xybern-v1",
      "env": {
        "XYBERN_API_KEY": "xb_your_key_here",
        "XYBERN_AGENT_NAME": "claude-code-prod"
      }
    }
  }
}

Cursor / Windsurf

Add to your MCP settings (Settings → MCP Servers → Add):

{
  "name": "xybern",
  "command": "python",
  "args": ["-m", "mcp.server"],
  "cwd": "/path/to/xybern-v1",
  "env": {
    "XYBERN_API_KEY": "xb_your_key_here"
  }
}

Environment variables

Variable	Required	Description
`XYBERN_API_KEY`	Yes	Your Xybern API key (starts with `xb_`)
`XYBERN_BASE_URL`	No	API base URL. Defaults to `https://www.xybern.com/api/v1`
`XYBERN_AGENT_ID`	No	Pin to a specific registered agent. Auto-registers and caches if unset.
`XYBERN_AGENT_NAME`	No	Display name for auto-registered agent. Defaults to `mcp-agent`

💾

The auto-registered agent ID is cached in ~/.xybern/mcp_agent_id so the same identity is reused across sessions without re-registering.

Install dependencies

pip install mcp httpx

MCP Tools Reference

Once connected, Claude and other agents have these tools available. The server also injects enforcement rules into Claude's context automatically on startup.

xybern_intercept

Check whether an action is allowed before executing it. Call this before any significant action — file writes, database changes, API calls, financial operations.

# Claude calls this automatically before sensitive actions
xybern_intercept(
    action_type="execute_trade",
    action_content="Buy 500 AAPL at market price",
    metadata='{"symbol": "AAPL", "qty": 500, "notional_usd": 95000}',
    chain_id="workflow_xyz",
    chain_step=2,
)
# Returns: DECISION: ALLOW / BLOCK / ESCALATE
#          Trust score: 87/100
#          Reasoning: All checks passed
#          Decision ID: dec_abc123
#          Vault entry: ve_xyz789

xybern_preflight

Check a multi-step plan before committing. Claude describes what it intends to do and Xybern returns which steps would be blocked or escalated under current policies.

xybern_preflight(
    plan="""
    1. Query prod DB for all users created this week
    2. Export records to CSV
    3. Email CSV to analytics@company.com
    4. Delete records older than 90 days
    """
)
# Returns: PREFLIGHT RESULT
#          Overall assessment: ESCALATE
#          Analysis: Step 3 (PII export) would be blocked.
#          Step 4 (bulk delete) requires human approval.

xybern_wait_for_approval

Block until a human approves or rejects an escalated action in the Authorisation Layer dashboard. Claude pauses and waits — it will not proceed until resolution.

xybern_wait_for_approval(
    escalation_id="esc_abc123",
    timeout_seconds=300,
)
# Returns: APPROVED — You may proceed.
# or:      REJECTED — Do not proceed.

xybern_get_context

Read current workspace policies, agent trust level, active JIT windows, and delegation grants. Use this to understand what is currently allowed.

xybern_explain_block

Get a plain-English explanation of why an action was blocked, which policy triggered, and what options the user has to proceed.

xybern_request_temporal_window

Request a time-limited JIT permission window for elevated access. Recorded immutably in the vault.

xybern_request_temporal_window(
    scopes="database_write,file_delete",
    duration_minutes=30,
    reason="Emergency schema migration — ticket ENG-4521",
)

xybern_get_recent_decisions

Read recent enforcement decisions from the vault. Lets managers ask Claude "what did my agents do today?" directly.

xybern_get_recent_decisions(limit=20, decision_filter="block")

Resources & Prompts

Beyond tools, the Xybern MCP server exposes live data as readable resources and an enforcement context prompt that Claude can load to understand all active policies.

Resources

URI	Description
`xybern://policies`	All active live and shadow policies for the workspace
`xybern://decisions/recent`	Last 50 enforcement decisions from the Authorisation Layer vault
`xybern://agent/status`	Current agent identity, trust level, and active permissions

Prompts

Load enforcement_context to inject live policy rules into Claude's system prompt. Claude then knows exactly which action types require pre-approval without needing to call xybern_get_context first.

// In your Claude system prompt setup:
{
  "role": "system",
  "content": ""
}
// Resolves to the current workspace's active policies, blocked/escalated
// action types, and the agent's trust level — updated on every session.

Proxy Gateway Mode

The most powerful deployment: route any MCP client's connections to other MCP servers through Xybern. Every tool call — GitHub, Slack, database, internal APIs — passes through enforcement before execution. Zero changes to your agent or MCP server code.

Without Xybern:
  Claude Code → GitHub MCP      (no auth, no policies)
  Claude Code → Database MCP    (no auth, no policies)
  Claude Code → Slack MCP       (no auth, no policies)

With Xybern MCP Gateway:
  Claude Code → Xybern MCP → [policy check] → GitHub MCP   ✓ allowed
                            → [policy check] → Database MCP ⏸ escalated
                            → [policy check] → Slack MCP    ✗ blocked

Xybern becomes the identity and authorisation perimeter for the entire MCP ecosystem.

Step 1 — Register your MCP server

curl -X POST https://www.xybern.com/api/sentinel/mcp/servers \
  -H "Cookie: <session>" \
  -H "Content-Type: application/json" \
  -d '{
    "workspace_id": "your-workspace-id",
    "server_name": "filesystem",
    "upstream_url": "http://localhost:3000",
    "transport_type": "streamable_http",
    "auth_type": "bearer",
    "auth_secret": "my-mcp-server-token",
    "blocked_tools": ["delete_file", "overwrite_file"]
  }'

transport_type: streamable_http (MCP spec 2025-03-26) or sse (legacy HTTP+SSE).
auth_type: none, bearer, header, or basic — auth secret stored encrypted.

Step 2 — Point your client at the proxy URL

The proxy URL for a registered server is:

https://www.xybern.com/gateway/mcp/<server_name>

Claude Desktop / Claude Code — config.json:

{
  "mcpServers": {
    "filesystem": {
      "url": "https://www.xybern.com/gateway/mcp/filesystem",
      "headers": { "X-Xybern-API-Key": "xb_live_..." }
    }
  }
}

Any MCP client with HTTP transport:

MCP_SERVER_URL=https://www.xybern.com/gateway/mcp/filesystem
MCP_SERVER_HEADERS='{"X-Xybern-API-Key": "xb_live_..."}'

SSE transport (legacy clients):

# SSE stream endpoint
GET https://www.xybern.com/gateway/mcp/<server_name>/sse
X-Xybern-API-Key: xb_live_...

# Message endpoint (returned in first SSE event)
POST https://www.xybern.com/gateway/mcp/<server_name>/message?sessionId=<id>
X-Xybern-API-Key: xb_live_...

What gets enforced automatically

Method	Enforcement
`tools/call`	Full ControlPlane — policies, trust score, agent tool policies, chain detection
`resources/read`	Sensitive URI patterns enforced; others pass through
`tools/list`	Response filtered — blocked / non-allowed tools stripped before client sees them
`initialize`	Server validated against registry — unknown servers rejected

Agent Tool Policies

Define per-agent, per-server tool permission matrices — including regex constraints on arguments (e.g. SELECT-only SQL queries):

curl -X POST https://www.xybern.com/api/sentinel/mcp/agent-policies \
  -H "Cookie: <session>" \
  -H "Content-Type: application/json" \
  -d '{
    "workspace_id": "your-workspace-id",
    "agent_id": "agent_abc123",
    "server_name": "database",
    "allowed_tools": ["execute_query", "list_tables"],
    "blocked_tools": ["drop_table", "delete_rows"],
    "argument_constraints": {
      "execute_query": [
        {"arg": "sql", "pattern": "^\\s*SELECT", "description": "SELECT queries only"}
      ]
    }
  }'

LLM Gateway — Overview

The Xybern LLM Gateway is a transparent reverse proxy that sits between your AI agent and any LLM provider. Every completion request passes through the enforcement engine — policies, trust scoring, escalation, and vault logging — without requiring any changes to your agent code.

Without Xybern:
  Agent → OpenAI        (no policies, no audit, no control)
  Agent → Anthropic     (no policies, no audit, no control)

With Xybern LLM Gateway:
  Agent → Xybern Gateway → [enforce] → OpenAI     ✓ allowed
                          → [enforce] → Anthropic  ⏸ escalated
                          → [enforce] → Azure      ✗ blocked

All you change is the base_url (or endpoint_url for Bedrock). Your existing provider credentials pass through. Xybern identifies your workspace via the X-Xybern-API-Key header.

Supported Providers

Xybern supports all major LLM providers. Route calls by pointing your agent at the corresponding proxy URL:

Provider	Proxy URL	Models
OpenAI	`https://www.xybern.com/gateway/openai`	GPT-4o, GPT-4-turbo, o1, o3
Anthropic	`https://www.xybern.com/gateway/anthropic`	Claude 3.5, Claude 3, Claude 4
Azure OpenAI	`https://www.xybern.com/gateway/azure`	GPT-4o, GPT-4-turbo (Azure-hosted)
AWS Bedrock	`https://www.xybern.com/gateway/bedrock`	Claude, Titan, Llama, Mistral, Cohere
Google Gemini	`https://www.xybern.com/gateway/gemini`	Gemini 2.0 Flash, Gemini 1.5 Pro

Quick Start

Each provider requires two changes: set the base URL to the Xybern proxy, and add your Xybern API key as a header. Your original provider API key passes through unchanged.

OpenAI SDK

from openai import OpenAI

client = OpenAI(
    api_key="sk-...",           # your real OpenAI key — passed through
    base_url="https://www.xybern.com/gateway/openai",
    default_headers={"X-Xybern-API-Key": "xb_live_..."},
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

OpenAI environment variables (works with any OpenAI-compatible framework)

OPENAI_BASE_URL=https://www.xybern.com/gateway/openai
OPENAI_DEFAULT_HEADERS='{"X-Xybern-API-Key": "xb_live_..."}'

LangChain

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o",
    openai_api_key="sk-...",
    openai_api_base="https://www.xybern.com/gateway/openai",
    default_headers={"X-Xybern-API-Key": "xb_live_..."},
)

Anthropic SDK

import anthropic

client = anthropic.Anthropic(
    api_key="sk-ant-...",
    base_url="https://www.xybern.com/gateway/anthropic",
    default_headers={"X-Xybern-API-Key": "xb_live_..."},
)

Azure OpenAI

from openai import AzureOpenAI

client = AzureOpenAI(
    api_key="your-azure-key",
    azure_endpoint="https://www.xybern.com/gateway/azure",
    api_version="2024-02-01",
    default_headers={
        "X-Xybern-API-Key": "xb_live_...",
        "X-Azure-Upstream": "https://<your-resource>.openai.azure.com",
    },
)

AWS Bedrock (boto3)

import boto3

bedrock = boto3.client(
    service_name="bedrock-runtime",
    region_name="us-east-1",
    endpoint_url="https://www.xybern.com/gateway/bedrock/us-east-1",
)

# Pass AWS credentials + Xybern key via custom headers using a botocore event:
def inject_xybern_key(request, **kwargs):
    request.headers["X-Xybern-API-Key"] = "xb_live_..."

bedrock.meta.events.register("before-send.bedrock-runtime.*", inject_xybern_key)

Google Gemini (OpenAI-compatible endpoint)

from openai import OpenAI

client = OpenAI(
    api_key="AIza...",          # your Google API key
    base_url="https://www.xybern.com/gateway/gemini/openai",
    default_headers={"X-Xybern-API-Key": "xb_live_..."},
)

response = client.chat.completions.create(
    model="gemini-2.0-flash",
    messages=[{"role": "user", "content": "Hello"}],
)

What Gets Enforced

Every request through the gateway is evaluated by the Xybern enforcement engine before being forwarded to the provider. The same policies, trust scoring, and escalation flow used by the SDK apply here — zero configuration changes needed.

Signal	Description
Active policies	Every active policy in the workspace is evaluated against the request content and agent identity.
Trust scoring	Metadata, LLM signal, and behavioural history combine into a real-time trust score per request.
Behavioural anomaly	Rate spikes, novel action types, unusual hours, and magnitude outliers are flagged automatically.
Human-in-the-loop	High-risk requests are held and return an `escalate` decision until a human approves or rejects in Authorisation Layer.
Vault logging	Every decision is written to the tamper-evident vault with full request metadata.

Decision responses

When a request is blocked or escalated, the gateway returns a response in the provider's native format so the agent can handle it without any special-case logic:

// Blocked — returned as a normal chat completion with refusal content
{
  "id": "chatcmpl-xybern-block-...",
  "choices": [{
    "message": {
      "role": "assistant",
      "content": "[BLOCKED by Xybern policy] Action blocked: ...",
      "xybern_decision_id": "dec_..."
    }
  }]
}

// Escalated — held for human review
{
  "choices": [{
    "message": {
      "content": "[ESCALATED] Action held for human approval. Escalation ID: esc_...",
      "xybern_decision_id": "dec_...",
      "xybern_escalation_id": "esc_..."
    }
  }]
}

Active Connections

The Authorisation Layer dashboard (Gateway tab → Active Connections) shows which providers your workspace has routed traffic through. Connections are created automatically the first time a request passes through the gateway for that provider.

You can disconnect a provider to pause enforcement and block all new requests through that provider route. Reconnect at any time. This is useful when temporarily suspending a provider without removing your integration.

API

# List all connections for a workspace
GET /api/sentinel/enforcement/gateway/connections?workspace_id=<id>

# Disconnect (pause) a provider
POST /api/sentinel/enforcement/gateway/connections/openai/toggle
Content-Type: application/json
{ "workspace_id": "<id>", "is_active": false }

# Reconnect a provider
POST /api/sentinel/enforcement/gateway/connections/openai/toggle
Content-Type: application/json
{ "workspace_id": "<id>", "is_active": true }

Valid provider slugs: openai, anthropic, azure, bedrock, gemini.

Dry-Run / Observe Mode

Send requests through the gateway without enforcing decisions — ideal for testing policies before enabling them in production. Add the X-Xybern-Dry-Run: true header to any gateway request.

from openai import OpenAI

client = OpenAI(
    api_key="sk-...",
    base_url="https://www.xybern.com/gateway/openai",
    default_headers={
        "X-Xybern-API-Key": "xb_live_...",
        "X-Xybern-Dry-Run": "true",   # observe only — no enforcement
    },
)
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

In dry-run mode the enforcement result is appended to the response as a xybern_dry_run field — you can inspect what decision would have been taken without blocking or escalating the request.

Xybern API Documentation v1.23 · info@xybern.com