Merge 78820e9794 into 2306816d40

litellm/proxy/guardrails/guardrail_hooks/levo/__init__.py

@@ -0,0 +1,75 @@
+from typing import TYPE_CHECKING, Final
+
+import litellm
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+from .levo import LevoGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+#: Forwarded to LevoGuardrail only when set, so unset values fall through to the
+#: constructor defaults rather than being pinned to None.
+_OPTIONAL_INIT_FIELDS: Final = (
+    "timeout",
+    "unreachable_fallback",
+    "extra_headers",
+    "buffer_streaming_until_moderated",
+    "additional_provider_specific_params",
+)
+
+
+def _get_config_value(litellm_params: "LitellmParams", optional_params: object, attribute_name: str) -> object:
+    """Read a field from optional_params if present, else from litellm_params.
+
+    The UI submits provider settings under optional_params while YAML puts them
+    at the top level, so both shapes have to resolve.
+    """
+    from_optional: Final = (
+        optional_params.get(attribute_name)
+        if isinstance(optional_params, dict)
+        else getattr(optional_params, attribute_name, None)
+        if optional_params is not None
+        else None
+    )
+    if from_optional is not None:
+        return from_optional
+    return getattr(litellm_params, attribute_name, None)
+
+
+def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
+    optional_params: Final = getattr(litellm_params, "optional_params", None)
+
+    api_base: Final = litellm_params.api_base
+    if not api_base:
+        raise ValueError(
+            "api_base is required for levo — point it at your Levo AI Gateway, e.g. http://levo-gateway:8080"
+        )
+
+    kwargs: Final[dict[str, object]] = {
+        field: value
+        for field in _OPTIONAL_INIT_FIELDS
+        for value in [_get_config_value(litellm_params, optional_params, field)]
+        if value is not None
+    }
+
+    _callback: Final = LevoGuardrail(
+        api_base=api_base,
+        api_key=litellm_params.api_key,
+        guardrail_name=guardrail.get("guardrail_name", "levo"),
+        event_hook=litellm_params.mode,
+        default_on=litellm_params.default_on,
+        **kwargs,
+    )
+
+    litellm.logging_callback_manager.add_litellm_callback(_callback)
+    return _callback
+
+
+guardrail_initializer_registry: Final = {
+    SupportedGuardrailIntegrations.LEVO.value: initialize_guardrail,
+}
+
+guardrail_class_registry: Final = {
+    SupportedGuardrailIntegrations.LEVO.value: LevoGuardrail,
+}

litellm/proxy/guardrails/guardrail_hooks/levo/levo.py

@@ -0,0 +1,96 @@
+"""Levo AI Gateway guardrail for LiteLLM.
+
+Sends prompts and completions to a self-hosted Levo AI Gateway, which runs
+Levo's policy engine — data-protection and content-safety scanners, CEL-based
+access control, MCP tool policies — and answers allow / block / rewrite.
+
+The gateway speaks LiteLLM's Basic Guardrail API, so the request and response
+bodies here are the same shape `generic_guardrail_api` uses. This integration
+exists on top of that for two reasons:
+
+1. Streamed responses are buffered until moderated by default. The generic
+   integration cannot enable that, so a response-side finding on a streaming
+   call arrives after the client already has the content.
+2. `guardrail: levo` with a typed config model, rather than a generic endpoint
+   plus provider-specific parameters.
+
+Gateway docs: https://docs.levo.ai/install-ai-gateway/ai-gateway-docker
+"""
+
+from typing import TYPE_CHECKING, Final, Literal
+
+from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import (
+    GenericGuardrailAPI,
+)
+from litellm.types.guardrails import GuardrailEventHooks
+
+if TYPE_CHECKING:
+    from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
+
+#: Path the gateway serves; appended to the configured api_base.
+LEVO_GUARDRAIL_PATH = "/beta/litellm_basic_guardrail_api"
+
+
+class LevoGuardrail(GenericGuardrailAPI):
+    """Guardrail backed by a self-hosted Levo AI Gateway."""
+
+    def __init__(
+        self,
+        api_base: str | None = None,
+        api_key: str | None = None,
+        buffer_streaming_until_moderated: bool | None = None,
+        unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
+        **kwargs: object,
+    ) -> None:
+        super().__init__(
+            api_base=api_base,
+            api_key=api_key,
+            unreachable_fallback=unreachable_fallback,
+            **kwargs,
+        )
+
+        # Read by UnifiedLLMGuardrails.async_post_call_streaming_iterator_hook via
+        # getattr(guardrail_to_apply, "streaming_*", default).
+        #
+        # Default to buffering: a response-side block is only meaningful if it
+        # lands before the client sees the content. Without this, chunks are
+        # emitted as they are produced and a violation is detected after the
+        # fact. Operators who need time-to-first-token more than response-side
+        # enforcement can set buffer_streaming_until_moderated=False.
+        buffer: Final = True if buffer_streaming_until_moderated is None else bool(buffer_streaming_until_moderated)
+        self.streaming_buffer_until_moderated = buffer
+        if buffer:
+            # Buffering can only moderate the assembled response, so it always
+            # implies end-of-stream evaluation.
+            self.streaming_end_of_stream_only = True
+
+    #: Scanning is the base implementation — the wire contract is identical, so
+    #: there is nothing to override. The binding itself is load-bearing: the
+    #: proxy selects the unified guardrail path with
+    #: ``"apply_guardrail" in type(callback).__dict__``, which inspects the
+    #: class's own attributes and does not see inherited methods. Without this
+    #: the guardrail is constructed and consulted, but never invoked, so every
+    #: request passes unscanned.
+    #:
+    #: Aliasing rather than wrapping in an ``async def`` that awaits ``super()``
+    #: keeps the single ``@log_guardrail_information`` layer the base method
+    #: already carries. A second decorated layer would log the call twice: the
+    #: inner wrapper's ``finally`` resets the "already recorded" ContextVar to
+    #: the value the outer wrapper had set, so the outer sees an unrecorded
+    #: call and emits its own span, Datadog record and spend-log entry.
+    apply_guardrail = GenericGuardrailAPI.apply_guardrail
+
+    @staticmethod
+    def get_config_model() -> type["GuardrailConfigModel"] | None:
+        from litellm.types.proxy.guardrails.guardrail_hooks.levo import (
+            LevoGuardrailConfigModel,
+        )
+
+        return LevoGuardrailConfigModel
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
+        # pre_call scans prompts, post_call scans completions. during_call is
+        # deliberately excluded: it duplicates the pre_call input event without
+        # adding a decision point.
+        return [GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call]

litellm/types/guardrails.py

@@ -84,6 +84,7 @@ class SupportedGuardrailIntegrations(Enum):
     DYNAMOAI = "dynamoai"
     GUARDRAILS_AI = "guardrails_ai"
     LAKERA = "lakera"
+    LEVO = "levo"
     LAKERA_V2 = "lakera_v2"
     PRESIDIO = "presidio"
     HIDE_SECRETS = "hide-secrets"

litellm/types/proxy/guardrails/guardrail_hooks/levo.py

@@ -0,0 +1,67 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from .base import GuardrailConfigModel
+
+
+class LevoGuardrailConfigModelOptionalParams(BaseModel):
+    timeout: float | None = Field(
+        default=None,
+        description="Per-request timeout in seconds for calls to the Levo AI Gateway.",
+    )
+
+    unreachable_fallback: Literal["fail_open", "fail_closed"] | None = Field(
+        default=None,
+        description=(
+            "Behaviour when the gateway cannot be reached. 'fail_closed' (default) "
+            "rejects the LLM call; 'fail_open' lets it through unscanned, trading "
+            "enforcement for availability."
+        ),
+    )
+
+    extra_headers: list[str] | None = Field(
+        default=None,
+        description=(
+            "Inbound header names whose values are forwarded to the gateway. "
+            "Values of headers outside LiteLLM's default allowlist are replaced "
+            "with a placeholder, so list anything the gateway must actually read "
+            "— e.g. 'x-forwarded-for' for the real client IP, or a JWT-claims "
+            "header used by identity policies."
+        ),
+    )
+
+    buffer_streaming_until_moderated: bool | None = Field(
+        default=None,
+        description=(
+            "Withhold streamed chunks until the assembled response has been "
+            "moderated. Defaults to true, so a blocked response cannot reach the "
+            "client after the fact. Set false to prioritise time-to-first-token, "
+            "accepting that response-side findings arrive too late to stop output."
+        ),
+    )
+
+
+class LevoGuardrailConfigModel(GuardrailConfigModel[LevoGuardrailConfigModelOptionalParams]):
+    api_base: str = Field(
+        min_length=1,
+        description=(
+            "Base URL of the Levo AI Gateway, e.g. http://levo-gateway:8080. "
+            "The /beta/litellm_basic_guardrail_api path is appended automatically."
+        ),
+    )
+
+    api_key: str | None = Field(
+        default=None,
+        description=(
+            "Shared secret presented to the gateway as x-api-key. Must match "
+            "LEVO_GUARDRAIL_API_KEY on the gateway, which serves the endpoint on "
+            "its data-plane port and refuses to enable it until that is set. "
+            "Env: LEVO_GUARDRAIL_API_KEY."
+        ),
+        json_schema_extra={"secret": True},
+    )
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "Levo AI Gateway"

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py

@@ -0,0 +1,154 @@
+from types import SimpleNamespace
+
+import pytest
+
+from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import (
+    GenericGuardrailAPI,
+)
+from litellm.proxy.guardrails.guardrail_hooks.levo import initialize_guardrail
+from litellm.proxy.guardrails.guardrail_hooks.levo.levo import (
+    LEVO_GUARDRAIL_PATH,
+    LevoGuardrail,
+)
+from litellm.proxy.guardrails.guardrail_registry import (
+    guardrail_class_registry,
+    guardrail_initializer_registry,
+)
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.proxy.guardrails.guardrail_hooks.levo import (
+    LevoGuardrailConfigModel,
+)
+
+
+def _params(**overrides: object) -> SimpleNamespace:
+    """LitellmParams-shaped stub, as the proxy passes to the initializer."""
+    base = dict(
+        guardrail="levo",
+        mode=["pre_call", "post_call"],
+        api_base="http://levo-gateway:8080",
+        api_key="s3cret",
+        default_on=True,
+        optional_params=None,
+    )
+    base.update(overrides)
+    return SimpleNamespace(**base)
+
+
+def _guardrail(name: str = "levo") -> dict[str, str]:
+    return {"guardrail_name": name}
+
+
+def test_registry_membership():
+    assert "levo" in guardrail_initializer_registry
+    assert guardrail_class_registry["levo"] is LevoGuardrail
+
+
+def test_config_model_wiring():
+    assert LevoGuardrail.get_config_model() is LevoGuardrailConfigModel
+    assert LevoGuardrailConfigModel.ui_friendly_name() == "Levo AI Gateway"
+
+
+def test_supported_hooks_limited_to_pre_and_post():
+    # during_call would duplicate the pre_call input event without adding a
+    # decision point.
+    assert LevoGuardrail.get_supported_event_hooks() == [
+        GuardrailEventHooks.pre_call,
+        GuardrailEventHooks.post_call,
+    ]
+
+
+def test_endpoint_path_appended_to_api_base():
+    g = LevoGuardrail(api_base="http://levo-gateway:8080", guardrail_name="levo")
+    assert g.api_base == f"http://levo-gateway:8080{LEVO_GUARDRAIL_PATH}"
+
+
+def test_api_key_sent_as_x_api_key():
+    # The gateway serves this endpoint on its data-plane port and rejects
+    # unauthenticated callers, so the shared secret must reach it.
+    g = LevoGuardrail(api_base="http://levo-gateway:8080", api_key="s3cret", guardrail_name="levo")
+    assert g.headers.get("x-api-key") == "s3cret"
+
+
+def test_initializer_requires_api_base():
+    with pytest.raises(ValueError, match="api_base is required"):
+        initialize_guardrail(_params(api_base=None), _guardrail())
+
+
+def test_initializer_builds_working_callback():
+    cb = initialize_guardrail(_params(), _guardrail())
+    assert isinstance(cb, LevoGuardrail)
+    assert cb.default_on is True
+    assert cb.api_base.endswith(LEVO_GUARDRAIL_PATH)
+
+
+def test_initializer_reads_optional_params_flattened_like_ui():
+    # The UI submits provider settings under optional_params rather than at the
+    # top level; both shapes must reach the constructor.
+    cb = initialize_guardrail(
+        _params(optional_params={"unreachable_fallback": "fail_open"}),
+        _guardrail(),
+    )
+    assert cb.unreachable_fallback == "fail_open"
+
+
+def test_unreachable_fallback_defaults_to_fail_closed():
+    cb = initialize_guardrail(_params(), _guardrail())
+    assert cb.unreachable_fallback == "fail_closed"
+
+
+# ── streaming ───────────────────────────────────────────────────────────────
+#
+# The reason this integration exists separately from generic_guardrail_api.
+
+
+def test_streaming_buffered_by_default():
+    # A response-side block is only meaningful if it lands before the client
+    # sees the content. Without buffering, chunks are emitted as they are
+    # produced and a violation is detected after the fact.
+    g = LevoGuardrail(api_base="http://levo-gateway:8080", guardrail_name="levo")
+    assert g.streaming_buffer_until_moderated is True
+    assert g.streaming_end_of_stream_only is True
+
+
+def test_streaming_buffering_can_be_disabled():
+    # Operators who need time-to-first-token more than response-side
+    # enforcement can opt out.
+    g = LevoGuardrail(
+        api_base="http://levo-gateway:8080",
+        guardrail_name="levo",
+        buffer_streaming_until_moderated=False,
+    )
+    assert g.streaming_buffer_until_moderated is False
+
+
+def test_streaming_flag_settable_via_optional_params():
+    cb = initialize_guardrail(
+        _params(optional_params={"buffer_streaming_until_moderated": False}),
+        _guardrail(),
+    )
+    assert cb.streaming_buffer_until_moderated is False
+
+
+def test_apply_guardrail_defined_on_the_class_not_inherited():
+    """Regression: the proxy selects the unified guardrail path with
+    ``"apply_guardrail" in type(callback).__dict__``, which inspects the
+    class's own attributes and does not see inherited methods.
+
+    A subclass that relies on inheritance is constructed, registered and even
+    consulted via ``should_run_guardrail`` — but never invoked, so every
+    request passes unscanned while the guardrail reports healthy. Guard the
+    binding so that failure mode cannot return silently.
+    """
+    assert "apply_guardrail" in LevoGuardrail.__dict__
+
+
+def test_apply_guardrail_is_the_base_method_not_a_second_wrapper():
+    """Regression: the binding above must alias the base method rather than
+    wrap it in another ``@log_guardrail_information`` layer.
+
+    Two decorated layers log the call twice — the inner wrapper's ``finally``
+    resets the "already recorded" ContextVar to the value the outer wrapper
+    set, so the outer sees an unrecorded call and emits its own span, Datadog
+    record and spend-log entry on top of the inner one.
+    """
+    assert LevoGuardrail.apply_guardrail is GenericGuardrailAPI.apply_guardrail

ui/litellm-dashboard/public/assets/logos/levo.svg

@@ -0,0 +1,6 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" width="64" height="64" role="img" aria-label="Levo AI">
+  <title>Levo AI</title>
+  <rect width="64" height="64" rx="14" fill="#0B1220"/>
+  <path d="M20 16h6v24h14v6H20V16z" fill="#4F8CFF"/>
+  <circle cx="44" cy="22" r="5" fill="#7FE3B0"/>
+</svg>

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts

@@ -306,6 +306,12 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
     mode: "pre_call",
     defaultOn: false,
   },
+  levo: {
+    provider: "Levo AI Gateway",
+    guardrailNameSuggestion: "Levo Guardrail",
+    mode: "pre_call",
+    defaultOn: false,
+  },
   straiker: {
     provider: "Straiker",
     guardrailNameSuggestion: "Straiker Guardrail",

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts

@@ -27,6 +27,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record<string, string> = {
   deepkeep: "deepkeep.svg",
   repelloai: "repelloai.png",
   straiker: "straiker.svg",
+  levo: "levo.svg",
 };
 
 describe("guardrail_garden_data logos", () => {

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts

@@ -454,6 +454,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
     tags: ["Security", "Policy", "Prompt Injection"],
     providerKey: "Repelloai",
   },
+  {
+    id: "levo",
+    name: "Levo AI Gateway",
+    description:
+      "Self-hosted AI gateway: PII and secret detection, prompt-injection heuristics, CEL access control and MCP tool policies, with streamed responses buffered until moderated",
+    category: "partner",
+    logo: guardrailLogoMap["Levo AI Gateway"],
+    tags: ["PII", "Secrets", "Prompt Injection", "MCP", "Self-hosted"],
+    providerKey: "Levo AI Gateway",
+  },
   {
     id: "straiker",
     name: "Straiker",

ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_info_helpers.tsx

@@ -23,6 +23,7 @@ import promptguardLogo from "../../../../../public/assets/logos/promptguard.svg"
 import qohashLogo from "../../../../../public/assets/logos/qohash.jpg";
 import repelloAiLogo from "../../../../../public/assets/logos/repelloai.png";
 import straikerLogo from "../../../../../public/assets/logos/straiker.svg";
+import levoLogo from "../../../../../public/assets/logos/levo.svg";
 import xecguardLogo from "../../../../../public/assets/logos/xecguard.svg";
 import zscalerLogo from "../../../../../public/assets/logos/zscaler.svg";
 
@@ -204,6 +205,7 @@ export const guardrailLogoMap = {
   "Qostodian Nexus": qohashLogo.src,
   "RepelloAI Argus": repelloAiLogo.src,
   Straiker: straikerLogo.src,
+  "Levo AI Gateway": levoLogo.src,
 } satisfies Record<string, string>;
 
 export const getGuardrailLogo = (displayName: string): string | undefined =>