Guardrail API

Most deployments put the AI Gateway in the request path. Sometimes you cannot, or you want a ruling on something that is not an LLM call at all — a shell command an agent wants to run, a file it is about to read, an MCP tool invocation.

The Guardrail API answers that. You send text; the gateway evaluates your policies and returns a verdict. It is synchronous, it holds no state, and it runs the same policy engine as the proxy path, so one configuration governs both.

your caller  ──►  POST /beta/litellm_basic_guardrail_api  ──►  NONE
(LiteLLM, an IDE hook, your own service)                       BLOCKED
                                                               GUARDRAIL_INTERVENED

The path is named for LiteLLM because that integration came first, but nothing about the payload is LiteLLM-specific.

Enabling the endpoint

Set a shared secret on the gateway:

-e LEVOAI_GUARDRAIL_API_KEY="$(openssl rand -hex 24)"

The endpoint is disabled unless this is set. Opting in explicitly means a deployment that never configures this integration does not silently expose it. Callers present the same value as x-api-key (or Authorization: Bearer).

Requires gateway 0.13.8 or later. Pin an explicit image tag — a stale :latest on a host will return {"error":"not found"}.

Request

POST /beta/litellm_basic_guardrail_api
x-api-key: <your shared secret>
Content-Type: application/json

{
  "input_type": "request",
  "texts": ["What is our Q3 revenue forecast?"]
}

Field	Required	Meaning
texts	yes	Array of strings to scan. Plural, and a list — a text field is accepted and scans nothing.
input_type	no	request (default) evaluates requestGuards; response evaluates responseGuards.
model	no	Helps classify the call, e.g. anthropic/claude-sonnet-5.

input_type is the important one. Use request for text a person or your own agent authored, and response for text arriving from elsewhere — a tool result, a fetched page, a file — which is the indirect prompt-injection case.

Bodies are capped at 1 MiB; larger returns 400.

Response

{ "action": "NONE" }

{ "action": "BLOCKED",
  "blocked_reason": "Request blocked by inspection findings (policy: inspection)" }

{ "action": "GUARDRAIL_INTERVENED",
  "texts": ["my ssn is [SSN-REDACTED], say OK"] }

action	Meaning	What to do
NONE	Nothing found	Proceed
BLOCKED	A policy says stop	Reject, and show blocked_reason
GUARDRAIL_INTERVENED	Something was redacted	Use the returned texts, which are index-aligned with what you sent. If you cannot substitute them, proceed — the finding is still audited and alerted.

All three are HTTP 200. 401 means the key is wrong, 400 a malformed or oversized body.

Configuring what it detects

The endpoint evaluates your inspection guards. A minimal configuration covering injection, secrets and PII on both legs:

config:
  policies:
    rules:
      - name: allow-all
        match: "true"
        action: allow
        priority: 1000
    inspection:
      requestGuards:
        - type: llm_bastion
          action: block
          config:
            scanners:
              - scannerType: InjectionHeuristics
                action: block
        - type: pii_regex
          action: mask
        - type: secrets
          action: block
      responseGuards:
        - type: llm_bastion
          action: block
          config:
            scanners:
              - scannerType: InjectionHeuristics
                action: block
        - type: pii_regex
          action: mask
        - type: secrets
          action: block
    logFullAudit: true

Three things about this are easy to get wrong:

1. scanners belongs under config: inside an llm_bastion guard. One level up it is silently ignored and injection is never detected.
2. Configure injection on both guard lists. With it only on requestGuards, an input_type: response call returns NONE without scanning for injection — precisely the indirect case.
3. pii_regex with action: mask, not block. URLs match a PII pattern, so blocking denies any text containing one.

InjectionHeuristics is rule-based: no ML models to download, no GPU, and a p50 well under a millisecond. It catches base64- and unicode-obfuscated variants through a canonicalization pre-pass.

To drive this from the policies you manage in the Levo UI instead, set LEVOAI_BASE_URL, LEVOAI_AUTH_KEY and LEVOAI_ENVIRONMENT_ID. The gateway polls AI Policies every 30 seconds and hot-swaps the same engine this endpoint reads.

Coverage

Detects prompt injection (direct and indirect), secrets (9 patterns) and PII (19 patterns), plus the ML content-safety scanners — toxicity, bias, and others — when models are mounted.

It does not evaluate whether a shell command is destructive. rm -rf /, git push --force and DROP TABLE users return a clean verdict.

Use case: LiteLLM

LiteLLM's built-in generic_guardrail_api plugin calls this endpoint directly. In your LiteLLM config, at the top level — not under litellm_settings:

guardrails:
  - guardrail_name: "levo"
    litellm_params:
      guardrail: generic_guardrail_api
      api_key: os.environ/LEVOAI_GUARDRAIL_API_KEY
      mode: [pre_call, post_call]
      api_base: http://levo-gateway:8080
      unreachable_fallback: fail_open
      default_on: true
      extra_headers: ["x-forwarded-for"]

pre_call sends input_type: request, post_call sends response. Both legs of one call share a request ID so audit records join correctly.

Use case: Cursor hooks

Cursor can call an external command before it submits a prompt, runs a shell command, invokes an MCP tool, or reads a file, and act on the JSON that command prints. Everything you need is below — no repository access required.

Hook	Scanned	input_type
beforeSubmitPrompt	the prompt	request
beforeShellExecution	the command	request
beforeMCPExecution	tool name + input	request
beforeReadFile	file content	response

Map BLOCKED to deny and everything else to allow. GUARDRAIL_INTERVENED should not deny: a hook cannot substitute rewritten text back into a shell command, so denying would block legitimate work.

Note that beforeSubmitPrompt answers with {"continue": false} while the other three answer with {"permission": "deny"} — Cursor ignores the wrong one silently.

Because hooks run on each developer's machine, the gateway has to be reachable from laptops rather than only from inside your cluster.

Install

Save the script below as ~/.cursor/levo-guardrail-hook.py and make it executable. It needs only the Python 3 standard library — no pip, no curl, no jq.

chmod +x ~/.cursor/levo-guardrail-hook.py
export LEVO_GATEWAY_URL=https://your-gateway.example.com
export LEVO_API_KEY=<the same value as LEVOAI_GUARDRAIL_API_KEY>

Then register the hooks in ~/.cursor/hooks.json (or <project>/.cursor/hooks.json) and restart Cursor:

{
  "version": 1,
  "hooks": {
    "beforeSubmitPrompt": [
      { "command": "$HOME/.cursor/levo-guardrail-hook.py" }
    ],
    "beforeShellExecution": [
      { "command": "$HOME/.cursor/levo-guardrail-hook.py" }
    ],
    "beforeMCPExecution": [
      { "command": "$HOME/.cursor/levo-guardrail-hook.py" }
    ],
    "beforeReadFile": [
      { "command": "$HOME/.cursor/levo-guardrail-hook.py" }
    ]
  }
}

Configuration

Variable	Default	Meaning
LEVO_GATEWAY_URL	—	Gateway base URL, no trailing path
LEVO_API_KEY	—	Must equal LEVOAI_GUARDRAIL_API_KEY on the gateway
LEVO_TIMEOUT	5	Seconds to wait for a verdict
LEVO_FAIL_MODE	open	open allows when the gateway is unreachable; closed denies
LEVO_DEBUG	—	1 logs each decision to stderr

Choose the fail mode deliberately. open keeps developers working when the gateway is down and is the right default for a pilot; closed means a network blip stops shell commands. Cursor's own convention already fails open — it treats exit code 2 as "block" and any other non-zero exit as allow — so the script always exits 0 and expresses its decision in the JSON.

A 4xx from the gateway means misconfiguration rather than a transient outage, which under fail-open would silently allow everything. The script prints a MISCONFIGURED line to stderr in that case whether or not LEVO_DEBUG is set; Cursor surfaces hook stderr in its logs, so alert on that string if you deploy across a fleet.

Latency

The gateway answers in well under a millisecond for hook-sized payloads. What you feel is spawning a process per event: ~15 ms for bare python3, ~33 ms once json and urllib are imported, ~37 ms end to end — so roughly 4 ms is the round trip and the scan. That is inherent to Cursor's command-hook model.

The hook script

#!/usr/bin/env python3
"""Cursor hook -> Levo AI Gateway guardrail verdict.

Cursor invokes this once per hook event, passing the event as JSON on stdin
and reading a JSON decision from stdout. One script handles every event; the
`hook_event_name` field selects what text we scan and what shape the answer
takes.

Configure with environment variables (set them in hooks.json, or export them
in your shell profile):

    LEVO_GATEWAY_URL   required. Base URL of the gateway, e.g.
                       https://llm.example.com  (no trailing path)
    LEVO_API_KEY       required. Same value as LEVOAI_GUARDRAIL_API_KEY on
                       the gateway.
    LEVO_TIMEOUT       seconds to wait for a verdict. Default 5.
    LEVO_FAIL_MODE     open (default) or closed. What to do when the gateway
                       is unreachable, times out, or errors. `open` allows the
                       action, `closed` denies it. See README.
    LEVO_DEBUG         set to 1 to log decisions to stderr (Cursor surfaces
                       hook stderr in its logs).

Requires only the Python 3 standard library.
"""

import json
import os
import sys
import urllib.error
import urllib.request

ENDPOINT = "/beta/litellm_basic_guardrail_api"

# Events that ask a yes/no question and expect {"permission": ...}.
PERMISSION_EVENTS = {
    "beforeShellExecution",
    "beforeMCPExecution",
    "beforeReadFile",
}
# beforeSubmitPrompt is the odd one out: it expects {"continue": bool},
# not a permission. Cursor rejects a permission field here.
CONTINUE_EVENTS = {"beforeSubmitPrompt"}


def debug(msg):
    if os.environ.get("LEVO_DEBUG") == "1":
        print(f"levo-hook: {msg}", file=sys.stderr)


def extract(event):
    """Return (text, input_type) for an event, or (None, None) to skip.

    `input_type` picks which side of the policy the gateway evaluates:

      request  - text the developer or the agent authored. Scanned by
                 requestGuards.
      response - text that arrived from somewhere else (a file, a tool
                 result) and is about to enter the model's context. Scanned
                 by responseGuards. This is the indirect-injection case and
                 it is the reason both guard lists must be configured.
    """
    name = event.get("hook_event_name")

    if name == "beforeSubmitPrompt":
        return event.get("prompt", ""), "request"

    if name == "beforeShellExecution":
        return event.get("command", ""), "request"

    if name == "beforeMCPExecution":
        # tool_input is arbitrary JSON; scan the tool name alongside it so a
        # suspicious name is visible to the scanners too.
        tool_input = event.get("tool_input")
        rendered = tool_input if isinstance(tool_input, str) else json.dumps(tool_input or {})
        return f"{event.get('tool_name', '')} {rendered}".strip(), "request"

    if name == "beforeReadFile":
        # File content is untrusted input entering the context window.
        return event.get("content", ""), "response"

    return None, None


def scan(text, input_type):
    """POST one text to the gateway. Returns the parsed verdict dict.

    Raises on any transport or protocol failure so the caller can apply the
    configured fail mode rather than silently allowing.
    """
    base = os.environ.get("LEVO_GATEWAY_URL", "").rstrip("/")
    key = os.environ.get("LEVO_API_KEY", "")
    if not base or not key:
        raise RuntimeError("LEVO_GATEWAY_URL and LEVO_API_KEY must both be set")

    # The field is `texts`, plural, and it is a list. Sending `text` is
    # accepted by the server and scans nothing, returning a clean verdict.
    body = json.dumps({"input_type": input_type, "texts": [text]}).encode()
    req = urllib.request.Request(
        base + ENDPOINT,
        data=body,
        headers={"content-type": "application/json", "x-api-key": key},
        method="POST",
    )
    timeout = float(os.environ.get("LEVO_TIMEOUT", "5"))
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read())


def fail_decision(event_name, reason):
    """Apply LEVO_FAIL_MODE when we could not obtain a verdict."""
    closed = os.environ.get("LEVO_FAIL_MODE", "open").strip().lower() == "closed"
    debug(f"no verdict ({reason}); fail mode {'closed' if closed else 'open'}")
    if not closed:
        return allow(event_name)
    msg = f"Levo guardrail unavailable ({reason}); blocked by fail-closed policy."
    return deny(event_name, msg)


def allow(event_name):
    if event_name in CONTINUE_EVENTS:
        return {"continue": True}
    return {"permission": "allow"}


def deny(event_name, message):
    if event_name in CONTINUE_EVENTS:
        return {"continue": False, "user_message": message}
    return {
        "permission": "deny",
        "user_message": message,
        "agent_message": message,
    }


def main():
    try:
        event = json.loads(sys.stdin.read() or "{}")
    except json.JSONDecodeError:
        # Malformed input is our problem, not the developer's, and with no
        # hook_event_name we cannot know which contract to answer in. An
        # empty object means "no opinion" for every event, so it is the only
        # safe answer here — and it never wedges the IDE.
        print(json.dumps({}))
        return 0

    name = event.get("hook_event_name", "")
    text, input_type = extract(event)

    # Events we don't gate (after*, session*, ...) produce no decision. Cursor
    # treats an empty object as "no opinion".
    if text is None:
        print(json.dumps({}))
        return 0

    # Nothing to scan is not a reason to block.
    if not text.strip():
        print(json.dumps(allow(name)))
        return 0

    try:
        verdict = scan(text, input_type)
    except urllib.error.HTTPError as e:
        # A 4xx here is a misconfiguration, not a transient outage: 401 is the
        # wrong key, 404 means the URL is wrong or the gateway predates the
        # endpoint. Under the default fail-open that would silently allow
        # everything forever, which is the worst failure mode available — so
        # say so on stderr unconditionally, not just under LEVO_DEBUG.
        if 400 <= e.code < 500:
            print(
                f"levo-hook: MISCONFIGURED — gateway returned HTTP {e.code} for "
                f"{os.environ.get('LEVO_GATEWAY_URL', '')}{ENDPOINT}. "
                f"Nothing is being scanned. Check LEVO_GATEWAY_URL, LEVO_API_KEY, "
                f"and that the gateway is 0.13.8 or later.",
                file=sys.stderr,
            )
        decision = fail_decision(name, f"HTTP {e.code}")
        print(json.dumps(decision))
        return 0
    except Exception as e:  # timeout, DNS, connection refused, bad JSON
        decision = fail_decision(name, type(e).__name__)
        print(json.dumps(decision))
        return 0

    action = verdict.get("action", "NONE")
    debug(f"{name} -> {action}")

    # Only BLOCKED denies. GUARDRAIL_INTERVENED means the gateway redacted
    # something and handed back sanitized text — useful on the prompt path,
    # but a hook cannot substitute a rewritten shell command back into the
    # action, so treat it as allow. The finding is still audited and alerted
    # on the gateway side either way.
    if action == "BLOCKED":
        reason = verdict.get("blocked_reason") or "Blocked by Levo guardrail policy."
        print(json.dumps(deny(name, reason)))
        return 0

    print(json.dumps(allow(name)))
    return 0


if __name__ == "__main__":
    sys.exit(main())

Sending findings to a SIEM

With logFullAudit: true the gateway writes one JSON record per call:

{
  "request_id": "e0b68d10-ddbc-479b-bf44-f10bdbab6035",
  "source_ip": "10.0.4.19",
  "inspection": {
    "action": "Reject",
    "finding_count": 2,
    "highest_severity": "critical",
    "findings": [
      {
        "category": "secrets", "finding_type": "aws-access-key",
        "severity": "critical", "scanner": "secrets",
        "description": "aws-access-key detected in request body",
        "location": { "start": 30, "end": 50 }
      }
    ]
  }
}

Request leg only

input_type: request calls write this record. input_type: response calls do not — findings on the response leg are delivered to the Levo platform as alerts, but no ai_gateway_audit line is written. If you rely on the log for a SIEM, indirect-injection findings will not appear there.

Ship stdout with your existing log shipper. logFindingsMinSeverity and logFindingsMaxPerLine keep a large prompt from writing hundreds of findings onto one line; anything filtered still counts in finding_count and still raises an alert.

Troubleshooting

Symptom	Cause
{"error":"not found"}	Gateway older than 0.13.8, or a wrong base URL.
401 on every call	LEVOAI_GUARDRAIL_API_KEY unset on the gateway, or the caller's key differs.
Everything returns NONE	Sent text instead of texts, or no guards configured, or scanners nested one level too high.
Injection missed on tool output	No injection guard on responseGuards.
Legitimate commands intervened	pii_regex matching URLs. Expected — treat GUARDRAIL_INTERVENED as allow.
400 length limit exceeded	Body over 1 MiB.