diff --git a/litellm/proxy/guardrails/guardrail_hooks/levo/__init__.py b/litellm/proxy/guardrails/guardrail_hooks/levo/__init__.py
new file mode 100644
index 00000000000..80053629cab
--- /dev/null
+++ b/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,
+}
diff --git a/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py
new file mode 100644
index 00000000000..0e916633778
--- /dev/null
+++ b/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]
diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py
index f77f8c280de..1c87a7a23b9 100644
--- a/litellm/types/guardrails.py
+++ b/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"
diff --git a/litellm/types/proxy/guardrails/guardrail_hooks/levo.py b/litellm/types/proxy/guardrails/guardrail_hooks/levo.py
new file mode 100644
index 00000000000..d9d720e5978
--- /dev/null
+++ b/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"
diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py
new file mode 100644
index 00000000000..36424d0feb5
--- /dev/null
+++ b/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
diff --git a/ui/litellm-dashboard/public/assets/logos/levo.svg b/ui/litellm-dashboard/public/assets/logos/levo.svg
new file mode 100644
index 00000000000..b1d8140b267
--- /dev/null
+++ b/ui/litellm-dashboard/public/assets/logos/levo.svg
@@ -0,0 +1,6 @@
+
diff --git a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts
index 03cfeed42ff..b6c1a1704a5 100644
--- a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts
+++ b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_configs.ts
@@ -306,6 +306,12 @@ export const GUARDRAIL_PRESETS: Record = {
mode: "pre_call",
defaultOn: false,
},
+ levo: {
+ provider: "Levo AI Gateway",
+ guardrailNameSuggestion: "Levo Guardrail",
+ mode: "pre_call",
+ defaultOn: false,
+ },
straiker: {
provider: "Straiker",
guardrailNameSuggestion: "Straiker Guardrail",
diff --git a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts
index 13909e48185..d190c887df7 100644
--- a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts
+++ b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.test.ts
@@ -27,6 +27,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record = {
deepkeep: "deepkeep.svg",
repelloai: "repelloai.png",
straiker: "straiker.svg",
+ levo: "levo.svg",
};
describe("guardrail_garden_data logos", () => {
diff --git a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts
index 744af89a357..e0602c12fab 100644
--- a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts
+++ b/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",
diff --git a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_info_helpers.tsx b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_info_helpers.tsx
index 83038b8e0e7..62322f73696 100644
--- a/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_info_helpers.tsx
+++ b/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;
export const getGuardrailLogo = (displayName: string): string | undefined =>