Python Integration

pip install requests

import requests

GUARDRAIL_API_KEY = "sk_your_api_key_here"
BASE_URL = "https://api.guardrail.ai"

def check_agent_output(agent_id: str, text: str) -> dict:
    """
    Evaluate AI agent output through the Guardrail firewall.
    Returns risk_score, status, flags, and redacted_text.
    """
    response = requests.post(
        f"{BASE_URL}/v1/guardrail/check",
        headers={
            "Authorization": f"Bearer {GUARDRAIL_API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "agent_id": agent_id,
            "proposed_text": text,
        },
        timeout=30
    )
    response.raise_for_status()
    return response.json()

# Example usage
result = check_agent_output(
    agent_id="customer-support-bot",
    proposed_text="Your SSN 123-45-6789 is confirmed in our records."
)

if result["status"] == "approved":
    print("Safe to send:", result["redacted_text"])
else:
    print(f"BLOCKED — Risk Score: {result['risk_score']}")
    for flag in result["flags"]:
        print(f"  • {flag}")

import requests
import logging
import time

logger = logging.getLogger(__name__)

class GuardrailClient:
    BASE_URL = "https://api.guardrail.ai"
    
    def __init__(self, api_key: str, max_retries: int = 3):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })
        self.max_retries = max_retries

    def check(self, agent_id: str, proposed_text: str) -> dict:
        """
        Evaluate text and raise on critical errors.
        Returns dict with risk_score, status, flags, redacted_text.
        """
        for attempt in range(self.max_retries):
            try:
                resp = self.session.post(
                    f"{self.BASE_URL}/v1/guardrail/check",
                    json={"agent_id": agent_id, "proposed_text": proposed_text},
                    timeout=30,
                )
                
                if resp.status_code == 200:
                    return resp.json()
                elif resp.status_code == 402:
                    raise Exception("Insufficient credits. Please top up your Guardrail balance.")
                elif resp.status_code == 429:
                    logger.warning("Rate limited. Retrying in 10s...")
                    time.sleep(10)
                elif resp.status_code in (500, 502, 503):
                    wait = 2 ** attempt
                    logger.warning(f"Server error. Retrying in {wait}s...")
                    time.sleep(wait)
                else:
                    resp.raise_for_status()
                    
            except requests.exceptions.Timeout:
                logger.error(f"Request timed out (attempt {attempt + 1})")
                
        raise Exception(f"Guardrail request failed after {self.max_retries} attempts")
    
    def is_safe(self, agent_id: str, text: str, max_risk: int = 30) -> bool:
        """Convenience method — returns True if text is safe to display."""
        result = self.check(agent_id, text)
        return result["risk_score"] <= max_risk


# Usage
guardrail = GuardrailClient(api_key="sk_your_api_key_here")

agent_reply = "Here is your full account password: hunter2"

if guardrail.is_safe("chatbot-v3", agent_reply):
    send_to_user(agent_reply)
else:
    send_to_user("I'm sorry, I cannot provide that information.")

from langchain.schema import BaseOutputParser
import requests

class GuardrailOutputParser(BaseOutputParser):
    """LangChain output parser that filters agent outputs through Guardrail."""
    
    api_key: str
    agent_id: str = "langchain-agent"
    block_on_risk_above: int = 60

    def parse(self, text: str) -> str:
        response = requests.post(
            "https://api.guardrail.ai/v1/guardrail/check",
            headers={"Authorization": f"Bearer {self.api_key}"},
            json={"agent_id": self.agent_id, "proposed_text": text}
        ).json()
        
        if response["risk_score"] > self.block_on_risk_above:
            return "[BLOCKED: Output contained policy violations]"
        return response["redacted_text"]

# Wire into your chain
from langchain.chains import LLMChain
from langchain.chat_models import ChatOpenAI

chain = LLMChain(
    llm=ChatOpenAI(),
    prompt=your_prompt,
    output_parser=GuardrailOutputParser(api_key="sk_your_key_here")
)