Custom Guardrails

Synchrone oder asynchrone Python-Prüfungen für Validierung, Autorisierung und anwendungsspezifische Sicherheitsregeln.

Guardrail erstellen

Lege Python-Dateien in einem guardrails/-Verzeichnis ab und folge dabei demselben Projekt-Muster wie bei Middleware. Eigene Checks verwenden die Modi und die Ausführungsreihenfolge aus dem Guardrails-Guide.

from connic import GuardrailResult
import re

def check(content: str, context: dict) -> GuardrailResult:
    """Verify the input contains a valid ticket ID format."""
    if not re.search(r'TICKET-\d{4,8}', content):
        return GuardrailResult(
            passed=False,
            message="Please include a valid ticket ID (e.g., TICKET-12345)"
        )
    return GuardrailResult(passed=True)

Signatur und Rückgabewert

• Der Dateiname entspricht dem name in der Konfiguration, zum Beispiel guardrails/validate-ticket-id.py für name: validate-ticket-id.
• Die Datei muss eine check(content: str, context: dict) -> GuardrailResult-Funktion exportieren, synchron oder asynchron.
• content ist der geprüfte Text, entweder Input oder Output.
• context ist dasselbe Kontext-Dictionary, das der Middleware zur Verfügung steht, einschließlich run_id, agent_name, connector_id, timestamp und aller selbst gesetzten Werte.
• Gibt GuardrailResult(passed=True) oder GuardrailResult(passed=False, message="...") zurück.

Async-Beispiel

from connic import GuardrailResult
import httpx
import re

async def check(content: str, context: dict) -> GuardrailResult:
    """Check if the user has permission to use this agent.

    Input guardrails run before middleware, so request data is read from
    content. If you need middleware-enriched context, use an output guardrail
    or enforce permissions in middleware instead.
    """
    match = re.search(r'["\']user_id["\']\s*:\s*["\']([^"\']+)', content)
    user_id = match.group(1) if match else None
    if not user_id:
        return GuardrailResult(passed=False, message="User not authenticated")

    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.example.com/users/{user_id}/permissions")
        if resp.status_code != 200:
            return GuardrailResult(passed=False, message="Could not verify permissions")

        data = resp.json()
        if not data.get("allowed"):
            return GuardrailResult(passed=False, message="Insufficient permissions")

    return GuardrailResult(passed=True)

Logging aus einem Custom Guardrail

print, Ausgaben nach sys.stderr und Aufrufe von logging aus der Python-Standardbibliothek werden pro Run erfasst und im Projekt im Tab Logs und in der Run-Detailansicht mit der Source guardrail.<name> angezeigt. Wirft die Prüffunktion eine Exception, zeichnet Connic den Traceback mit dieser Quelle und dem Log-Level Error auf und behandelt die Exception entsprechend dem Modus der Regel als Guardrail-Verstoß: warn setzt die Verarbeitung fort, der Standardmodus block stoppt sie dagegen und gibt die konfigurierte Ablehnungsnachricht zurück.