From 16b4b3f91d85bebb13d80e912d581846cf5f7740 Mon Sep 17 00:00:00 2001 From: Yamparala-Venkata-Gopi Date: Sun, 16 Aug 2026 13:53:15 -0500 Subject: [PATCH 1/4] feat: add Levo AI Gateway guardrail integration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `guardrail: levo`, backed by a self-hosted Levo AI Gateway. The gateway runs Levo's policy engine — PII and secret detection, prompt-injection heuristics, CEL-based access control, MCP tool policies — and answers allow / block / rewrite. ```yaml guardrails: - guardrail_name: "levo" litellm_params: guardrail: levo api_base: http://levo-gateway:8080 api_key: os.environ/LEVO_GUARDRAIL_API_KEY mode: [pre_call, post_call] default_on: true ``` The gateway already speaks the Basic Guardrail API, so this subclasses `GenericGuardrailAPI` and reuses its wire contract. It exists separately for one behavioural reason: streamed responses are buffered until moderated by default. `generic_guardrail_api` never sets `streaming_buffer_until_moderated`, so a response-side finding on a streaming call lands after the client already holds the content. Operators who need time-to-first-token more than response-side enforcement can opt out with `buffer_streaming_until_moderated: false`. Verified against a running gateway, comparing the two integrations on the same image and prompt with a completion containing PII: generic_guardrail_api : 19 content chunks emitted, PII reached the client guardrail: levo : 0 content chunks, 400 before any output Clean streamed responses are unaffected and still deliver content. `apply_guardrail` is overridden in the class body even though it only delegates to the parent. 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 the override the guardrail is constructed, registered and consulted via `should_run_guardrail` — but never invoked, so every request passes unscanned while the guardrail reports healthy. A test asserts the override stays present, since that failure is silent. Tests cover registry membership, config-model wiring, endpoint path construction, the api_key header, initializer validation, optional_params flattening as the UI submits it, the streaming defaults and their opt-out, and the `__dict__` regression above. --- .../guardrail_hooks/levo/__init__.py | 75 ++++++++++ .../guardrails/guardrail_hooks/levo/levo.py | 106 +++++++++++++ litellm/types/guardrails.py | 1 + .../proxy/guardrails/guardrail_hooks/levo.py | 67 +++++++++ .../guardrails/guardrail_hooks/test_levo.py | 139 ++++++++++++++++++ .../public/assets/logos/levo.svg | 6 + .../_components/guardrail_garden_configs.ts | 6 + .../_components/guardrail_garden_data.ts | 10 ++ .../_components/guardrail_info_helpers.tsx | 2 + 9 files changed, 412 insertions(+) create mode 100644 litellm/proxy/guardrails/guardrail_hooks/levo/__init__.py create mode 100644 litellm/proxy/guardrails/guardrail_hooks/levo/levo.py create mode 100644 litellm/types/proxy/guardrails/guardrail_hooks/levo.py create mode 100644 tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py create mode 100644 ui/litellm-dashboard/public/assets/logos/levo.svg 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..7594322f13c --- /dev/null +++ b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py @@ -0,0 +1,106 @@ +"""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 +from litellm.types.utils import GenericGuardrailAPIInputs + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj + 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 + + async def apply_guardrail( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict, + input_type: Literal["request", "response"], + logging_obj: "LiteLLMLoggingObj | None" = None, + ) -> GenericGuardrailAPIInputs: + """Scan a request or response via the Levo AI Gateway. + + Delegates to the base implementation — the wire contract is identical. + This override must exist **in this class body**: 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 it + the guardrail is constructed and consulted, but never invoked, so every + request passes unscanned. + """ + return await super().apply_guardrail( + inputs=inputs, + request_data=request_data, + input_type=input_type, + logging_obj=logging_obj, + ) + + @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 c7cdfaad780..1d003db9e66 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..0f3d9e054fc --- /dev/null +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py @@ -0,0 +1,139 @@ +from types import SimpleNamespace + +import pytest + +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): + """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="levo"): + 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 + override so that failure mode cannot return silently. + """ + assert "apply_guardrail" in LevoGuardrail.__dict__ 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 @@ + + Levo AI + + + + 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.ts b/ui/litellm-dashboard/src/app/(dashboard)/guardrails/_components/guardrail_garden_data.ts index 744af89a357..00bac6096a6 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: `${ASSET_PREFIX}levo.svg`, + 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 12aaba0d696..071cb91ec1b 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"; @@ -193,6 +194,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 => From fabfe6961886470fb2b6a2f658c17289ab95967b Mon Sep 17 00:00:00 2001 From: Yamparala-Venkata-Gopi Date: Sun, 16 Aug 2026 14:02:17 -0500 Subject: [PATCH 2/4] fix: parameterize dict and annotate test helpers Addresses review feedback on incomplete typing at the new boundaries: `request_data` was an unparameterized `dict`, and the `_params` / `_guardrail` test helpers carried no parameter or return annotations, so neither could be statically checked. `make lint` exit 0, 13 tests pass. --- litellm/proxy/guardrails/guardrail_hooks/levo/levo.py | 2 +- .../proxy/guardrails/guardrail_hooks/test_levo.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py index 7594322f13c..63bed0932fd 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py +++ b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py @@ -69,7 +69,7 @@ class LevoGuardrail(GenericGuardrailAPI): async def apply_guardrail( self, inputs: GenericGuardrailAPIInputs, - request_data: dict, + request_data: dict[str, object], input_type: Literal["request", "response"], logging_obj: "LiteLLMLoggingObj | None" = None, ) -> GenericGuardrailAPIInputs: diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py index 0f3d9e054fc..be67d17a7f2 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py @@ -17,7 +17,7 @@ from litellm.types.proxy.guardrails.guardrail_hooks.levo import ( ) -def _params(**overrides): +def _params(**overrides: object) -> SimpleNamespace: """LitellmParams-shaped stub, as the proxy passes to the initializer.""" base = dict( guardrail="levo", @@ -31,7 +31,7 @@ def _params(**overrides): return SimpleNamespace(**base) -def _guardrail(name="levo"): +def _guardrail(name: str = "levo") -> dict[str, str]: return {"guardrail_name": name} From 2988ebe5a949a67f1b196528763924ff9865d3f9 Mon Sep 17 00:00:00 2001 From: Yamparala-Venkata-Gopi Date: Sun, 16 Aug 2026 14:11:40 -0500 Subject: [PATCH 3/4] fix: use guardrailLogoMap for the Levo card logo The card referenced an undefined ASSET_PREFIX identifier, which broke the UI type check and three vitest suites. Every other partner card resolves its logo through guardrailLogoMap, which already carries the levo.svg import; use that. Register levo in the partner logo-file test so the card set and the expected-logo map stay in sync. --- .../guardrails/_components/guardrail_garden_data.test.ts | 1 + .../(dashboard)/guardrails/_components/guardrail_garden_data.ts | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) 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 00bac6096a6..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 @@ -460,7 +460,7 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [ 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: `${ASSET_PREFIX}levo.svg`, + logo: guardrailLogoMap["Levo AI Gateway"], tags: ["PII", "Secrets", "Prompt Injection", "MCP", "Self-hosted"], providerKey: "Levo AI Gateway", }, From 78820e979455f4be178016352d1607789b6e8076 Mon Sep 17 00:00:00 2001 From: Yamparala-Venkata-Gopi Date: Sun, 16 Aug 2026 14:40:00 -0500 Subject: [PATCH 4/4] fix: alias the inherited apply_guardrail instead of wrapping it check_guardrail_apply_decorator requires every apply_guardrail defined in guardrail_hooks to carry @log_guardrail_information, which the override did not. Adding the decorator would have logged each call twice: the base method is already decorated, and the inner wrapper's finally resets the already-recorded ContextVar to the value the outer wrapper set, so the outer wrapper sees an unrecorded call and emits a second span, Datadog record and spend-log entry. The override only ever delegated to super(), so bind the base method into the class namespace directly. That keeps the single decorator layer while still satisfying the proxy's dispatch check, which looks for the name in type(callback).__dict__ and does not see inherited methods. --- .../guardrails/guardrail_hooks/levo/levo.py | 40 +++++++------------ .../guardrails/guardrail_hooks/test_levo.py | 17 +++++++- 2 files changed, 31 insertions(+), 26 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py index 63bed0932fd..0e916633778 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py +++ b/litellm/proxy/guardrails/guardrail_hooks/levo/levo.py @@ -23,10 +23,8 @@ from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guar GenericGuardrailAPI, ) from litellm.types.guardrails import GuardrailEventHooks -from litellm.types.utils import GenericGuardrailAPIInputs if TYPE_CHECKING: - from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel #: Path the gateway serves; appended to the configured api_base. @@ -66,29 +64,21 @@ class LevoGuardrail(GenericGuardrailAPI): # implies end-of-stream evaluation. self.streaming_end_of_stream_only = True - async def apply_guardrail( - self, - inputs: GenericGuardrailAPIInputs, - request_data: dict[str, object], - input_type: Literal["request", "response"], - logging_obj: "LiteLLMLoggingObj | None" = None, - ) -> GenericGuardrailAPIInputs: - """Scan a request or response via the Levo AI Gateway. - - Delegates to the base implementation — the wire contract is identical. - This override must exist **in this class body**: 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 it - the guardrail is constructed and consulted, but never invoked, so every - request passes unscanned. - """ - return await super().apply_guardrail( - inputs=inputs, - request_data=request_data, - input_type=input_type, - logging_obj=logging_obj, - ) + #: 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: diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py index be67d17a7f2..36424d0feb5 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/test_levo.py @@ -2,6 +2,9 @@ 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, @@ -134,6 +137,18 @@ def test_apply_guardrail_defined_on_the_class_not_inherited(): 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 - override so that failure mode cannot return silently. + 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