feat(guardrails): add NeuralTrust TrustGuard as a native option

Native hook, Garden tile, and mocked tests. Fail-closed on unusable
verdicts, empty transforms, timeouts, and non-availability HTTP errors
so the LiteLLM path matches TrustGate.

litellm/proxy/guardrails/guardrail_hooks/neuraltrust/README.md

@@ -0,0 +1,45 @@
+# NeuralTrust TrustGuard
+
+Native LiteLLM guardrail. Sends chat input and output to TrustGuard `POST /v1/evaluate`.
+
+## Config
+
+```yaml
+guardrails:
+  - guardrail_name: neuraltrust-trustguard
+    litellm_params:
+      guardrail: neuraltrust
+      mode: [pre_call, post_call]
+      api_key: os.environ/TRUSTGUARD_API_KEY
+      api_base: os.environ/TRUSTGUARD_API_BASE  # default https://trustguard.neuraltrust.ai
+      collector_key: os.environ/TRUSTGUARD_COLLECTOR_KEY  # tgcol_… ; optional if the API key is bound
+      unreachable_fallback: fail_closed
+      timeout: 5
+      default_on: true
+```
+
+## Auth
+
+Bearer `tgk_…` API key. Address the collector with `collector_key`, or omit it when the key is already bound to one.
+
+## Verdicts
+
+| TrustGuard `status` | LiteLLM |
+| --- | --- |
+| `block` | HTTP 400 (trace_id / request_id only; findings are not echoed) |
+| `transform` | rewrite the last user message / last text from `transformed_payload` |
+| `report` / `allow` | pass through (`report` is logged by trace_id) |
+
+Unknown verdicts, malformed bodies, and `transform` without a usable payload fail closed.
+
+## Fail-open vs fail-closed
+
+`unreachable_fallback` applies only to transport failures: connect errors, timeouts, HTTP 502/504.
+
+HTTP 503 entitlements, 401/403, other 4xx/5xx, and unusable TrustGuard verdicts always fail closed.
+
+`fail_open` means the request bypasses TrustGuard entirely when the endpoint is unreachable. It is off by default.
+
+## Streaming
+
+LiteLLM streaming guardrails default to `block_only`. `block` still fires on streamed calls. `transform` rewrites are not applied to the streamed tokens; use non-streaming requests when DLP redaction must reach the client.

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

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Final
+
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+from .neuraltrust import NeuralTrustGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def initialize_guardrail(litellm_params: LitellmParams, guardrail: Guardrail) -> NeuralTrustGuardrail:
+    import litellm
+
+    _callback: Final = NeuralTrustGuardrail(
+        api_base=litellm_params.api_base,
+        api_key=litellm_params.api_key,
+        collector_key=litellm_params.collector_key,
+        unreachable_fallback=litellm_params.unreachable_fallback,
+        timeout=litellm_params.timeout,
+        guardrail_name=guardrail.get("guardrail_name", ""),
+        event_hook=litellm_params.mode,
+        default_on=litellm_params.default_on,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(_callback)
+    return _callback
+
+
+guardrail_initializer_registry: Final = {  # mutable-ok: guardrail_registry discovers dict registries
+    SupportedGuardrailIntegrations.NEURALTRUST.value: initialize_guardrail,
+}
+
+guardrail_class_registry: Final = {  # mutable-ok: guardrail_registry discovers dict registries
+    SupportedGuardrailIntegrations.NEURALTRUST.value: NeuralTrustGuardrail,
+}

litellm/proxy/guardrails/guardrail_hooks/neuraltrust/neuraltrust.py

@@ -0,0 +1,330 @@
+"""NeuralTrust TrustGuard native LiteLLM guardrail.
+
+Calls TrustGuard POST /v1/evaluate on pre_call (input) and post_call (output).
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any, Final, Literal
+
+import httpx
+from fastapi import HTTPException
+
+from litellm._logging import verbose_proxy_logger
+from litellm.exceptions import Timeout
+from litellm.integrations.custom_guardrail import (
+    CustomGuardrail,
+    get_session_id_from_request_data,
+    log_guardrail_information,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.proxy.guardrails.guardrail_hooks.neuraltrust import DEFAULT_API_BASE
+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
+
+EVALUATE_PATH: Final = "/v1/evaluate"
+DEFAULT_TIMEOUT: Final = 5.0
+STATUS_BLOCK: Final = "block"
+STATUS_TRANSFORM: Final = "transform"
+STATUS_REPORT: Final = "report"
+STATUS_ALLOW: Final = "allow"
+KNOWN_STATUSES: Final = frozenset({STATUS_ALLOW, STATUS_BLOCK, STATUS_TRANSFORM, STATUS_REPORT})
+UNREACHABLE_HTTP_STATUSES: Final = frozenset({502, 504})
+
+
+class _TrustGuardUnreachable(Exception):
+    """Transport or availability failure; eligible for unreachable_fallback."""
+
+
+def _message_text(message: Mapping[str, object]) -> str | None:
+    content: Final = message.get("content")
+    return content if isinstance(content, str) and content else None
+
+
+def _copy_messages(messages: list[object]) -> list[dict[str, object]] | None:
+    copied: list[dict[str, object]] = []
+    for message in messages:
+        if not isinstance(message, dict):
+            return None
+        copied.append(dict(message))
+    return copied
+
+
+def _texts_from_messages(messages: list[dict[str, object]]) -> list[str]:
+    return [text for message in messages if (text := _message_text(message)) is not None]
+
+
+def _rewrite_last_user_message(
+    messages: list[dict[str, object]],
+    redacted: str,
+) -> list[dict[str, object]]:
+    rewritten: Final = [dict(message) for message in messages]
+    last_user: int | None = None
+    for index, message in enumerate(rewritten):
+        if message.get("role") == "user":
+            last_user = index
+    target: Final = last_user if last_user is not None else len(rewritten) - 1
+    if target < 0:
+        return [{"role": "user", "content": redacted}]
+    rewritten[target] = {**rewritten[target], "content": redacted}
+    return rewritten
+
+
+def _model_name(
+    inputs: GenericGuardrailAPIInputs,
+    logging_obj: LiteLLMLoggingObj | None,
+) -> str:
+    if logging_obj is not None and logging_obj.model:
+        return str(logging_obj.model)
+    return str(inputs.get("model") or "")
+
+
+class NeuralTrustGuardrail(CustomGuardrail):
+    """LiteLLM hook that evaluates prompts and completions with TrustGuard."""
+
+    @staticmethod
+    def get_config_model() -> type[GuardrailConfigModel]:
+        from litellm.types.proxy.guardrails.guardrail_hooks.neuraltrust import (
+            NeuralTrustGuardrailConfigModel,
+        )
+
+        return NeuralTrustGuardrailConfigModel
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
+        return [
+            GuardrailEventHooks.pre_call,
+            GuardrailEventHooks.post_call,
+        ]
+
+    def __init__(
+        self,
+        api_base: str | None = None,
+        api_key: str | None = None,
+        collector_key: str | None = None,
+        unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
+        timeout: float | None = None,
+        **kwargs: Any,
+    ) -> None:
+        self.async_handler = get_async_httpx_client(
+            llm_provider=httpxSpecialProvider.GuardrailCallback,
+        )
+        self.api_base = (api_base or os.environ.get("TRUSTGUARD_API_BASE") or DEFAULT_API_BASE).rstrip("/")
+        self.api_key = api_key or os.environ.get("TRUSTGUARD_API_KEY") or ""
+        if not self.api_key:
+            raise ValueError(
+                "TrustGuard API key is required. Set TRUSTGUARD_API_KEY or pass api_key in litellm_params."
+            )
+        self.collector_key = collector_key or os.environ.get("TRUSTGUARD_COLLECTOR_KEY") or ""
+        self.unreachable_fallback: Literal["fail_closed", "fail_open"] = unreachable_fallback
+        resolved_timeout: Final = DEFAULT_TIMEOUT if timeout is None else float(timeout)
+        self.timeout = resolved_timeout
+        kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
+        super().__init__(**kwargs)
+
+    @log_guardrail_information
+    async def apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,  # mutable-ok: CustomGuardrail.apply_guardrail contract
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None = None,
+    ) -> GenericGuardrailAPIInputs:
+        body: Final = self._evaluate_body(inputs, request_data, input_type, logging_obj)
+        try:
+            result: Final = await self._call_evaluate(body)
+        except HTTPException:
+            raise
+        except _TrustGuardUnreachable as exc:
+            return self._handle_unreachable(inputs, exc)
+
+        status: Final = result["status"]
+        if status == STATUS_BLOCK:
+            raise HTTPException(
+                status_code=400,
+                detail={  # mutable-ok: FastAPI HTTPException.detail is a JSON object
+                    "error": "Violated guardrail policy",
+                    "neuraltrust_guardrail_response": "Blocked by NeuralTrust TrustGuard.",
+                    "trace_id": result.get("trace_id"),
+                    "request_id": result.get("request_id"),
+                },
+            )
+        if status == STATUS_TRANSFORM:
+            return self._apply_transform(inputs, result.get("transformed_payload"))
+        if status == STATUS_REPORT:
+            verbose_proxy_logger.info("TrustGuard report-only findings trace_id=%s", result.get("trace_id"))
+        return inputs
+
+    def _evaluate_body(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,
+        input_type: Literal["request", "response"],
+        logging_obj: LiteLLMLoggingObj | None,
+    ) -> dict[str, object]:
+        body: dict[str, object] = {  # mutable-ok: outbound JSON
+            "payload": self._payload(inputs, input_type),
+            "direction": "input" if input_type == "request" else "output",
+            "protocol": "llm",
+            "attributes": {
+                "content_type": "application/json",
+                "model": {"name": _model_name(inputs, logging_obj)},
+            },
+        }
+        if self.collector_key:
+            body["collector_key"] = self.collector_key
+        session_id: Final = get_session_id_from_request_data(request_data)
+        if session_id:
+            body["session_id"] = session_id
+        return body
+
+    @staticmethod
+    def _payload(
+        inputs: GenericGuardrailAPIInputs,
+        input_type: Literal["request", "response"],
+    ) -> dict[str, object]:
+        if input_type == "request":
+            structured: Final = inputs.get("structured_messages")
+            payload: dict[str, object] = {  # mutable-ok: outbound JSON
+                "messages": structured
+                if structured
+                else [{"role": "user", "content": text} for text in (inputs.get("texts") or ())],
+            }
+            tools: Final = inputs.get("tools")
+            if tools:
+                payload["tools"] = tools
+            return payload
+
+        texts: Final = list(inputs.get("texts") or ())
+        tool_calls: Final = inputs.get("tool_calls")
+        messages: list[dict[str, object]] = [{"role": "assistant", "content": text} for text in texts]
+        if tool_calls:
+            if messages:
+                messages[-1] = {**messages[-1], "tool_calls": tool_calls}
+            else:
+                messages = [{"role": "assistant", "content": None, "tool_calls": tool_calls}]
+        if not messages:
+            messages = [{"role": "assistant", "content": ""}]
+        return {"messages": messages}
+
+    async def _call_evaluate(self, body: dict[str, object]) -> dict[str, object]:
+        url: Final = f"{self.api_base}{EVALUATE_PATH}"
+        headers: Final = MappingProxyType(
+            {
+                "Authorization": f"Bearer {self.api_key}",
+                "Content-Type": "application/json",
+            }
+        )
+        try:
+            response: Final = await self.async_handler.post(
+                url,
+                json=body,
+                headers=headers,
+                timeout=self.timeout,
+            )
+            response.raise_for_status()
+        except Timeout as exc:
+            raise _TrustGuardUnreachable(exc) from exc
+        except httpx.HTTPStatusError as exc:
+            status_code: Final = exc.response.status_code
+            if status_code == 503:
+                raise HTTPException(
+                    status_code=503,
+                    detail="TrustGuard entitlements unavailable",
+                ) from exc
+            if status_code in (401, 403):
+                raise HTTPException(
+                    status_code=status_code,
+                    detail="TrustGuard authentication failed",
+                ) from exc
+            if status_code in UNREACHABLE_HTTP_STATUSES:
+                raise _TrustGuardUnreachable(exc) from exc
+            raise HTTPException(
+                status_code=503,
+                detail="TrustGuard request failed",
+            ) from exc
+        except httpx.RequestError as exc:
+            raise _TrustGuardUnreachable(exc) from exc
+
+        try:
+            parsed: Final[object] = response.json()
+        except ValueError as exc:
+            raise _TrustGuardUnreachable("TrustGuard returned non-JSON body") from exc
+        if not isinstance(parsed, dict):
+            raise HTTPException(status_code=503, detail="TrustGuard returned an invalid response")
+        status: Final = parsed.get("status")
+        if not isinstance(status, str) or status.lower() not in KNOWN_STATUSES:
+            raise HTTPException(status_code=503, detail="TrustGuard returned an unknown verdict")
+        parsed["status"] = status.lower()
+        return parsed
+
+    def _handle_unreachable(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        error: Exception,
+    ) -> GenericGuardrailAPIInputs:
+        if self.unreachable_fallback == "fail_open":
+            verbose_proxy_logger.critical(
+                "TrustGuard unreachable (fail-open): %s",
+                error,
+                exc_info=error,
+            )
+            return inputs
+        verbose_proxy_logger.error("TrustGuard unreachable (fail-closed): %s", error)
+        raise HTTPException(
+            status_code=503,
+            detail="TrustGuard guardrail service unreachable",
+        ) from error
+
+    @staticmethod
+    def _apply_transform(
+        inputs: GenericGuardrailAPIInputs,
+        transformed: object,
+    ) -> GenericGuardrailAPIInputs:
+        if not isinstance(transformed, Mapping):
+            raise HTTPException(status_code=400, detail="TrustGuard transform missing payload")
+
+        raw_messages: Final = transformed.get("messages")
+        if isinstance(raw_messages, list) and raw_messages:
+            rewritten_messages: Final = _copy_messages(raw_messages)
+            if rewritten_messages is None:
+                raise HTTPException(status_code=400, detail="TrustGuard transform missing payload")
+            texts_from_messages: Final = _texts_from_messages(rewritten_messages)
+            return {  # mutable-ok: GenericGuardrailAPIInputs is a TypedDict
+                **inputs,
+                "structured_messages": rewritten_messages,
+                "texts": texts_from_messages or inputs.get("texts"),
+            }
+
+        raw_input: Final = transformed.get("input")
+        if not isinstance(raw_input, str) or not raw_input:
+            raise HTTPException(status_code=400, detail="TrustGuard transform missing payload")
+
+        original_messages: Final = inputs.get("structured_messages")
+        if isinstance(original_messages, list) and original_messages:
+            copied: Final = _copy_messages(original_messages)
+            if copied is None:
+                raise HTTPException(status_code=400, detail="TrustGuard transform missing payload")
+            rewritten: Final = _rewrite_last_user_message(copied, raw_input)
+            return {  # mutable-ok: GenericGuardrailAPIInputs is a TypedDict
+                **inputs,
+                "structured_messages": rewritten,
+                "texts": _texts_from_messages(rewritten) or inputs.get("texts"),
+            }
+
+        original_texts: Final = list(inputs.get("texts") or ())
+        if not original_texts:
+            raise HTTPException(status_code=400, detail="TrustGuard transform missing payload")
+        rewritten_texts: Final = list(original_texts)
+        rewritten_texts[-1] = raw_input
+        return {**inputs, "texts": rewritten_texts}  # mutable-ok: GenericGuardrailAPIInputs is a TypedDict

litellm/types/guardrails.py

@@ -36,6 +36,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.ibm import (
 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
     ContentFilterCategoryConfig,
 )
+from litellm.types.proxy.guardrails.guardrail_hooks.neuraltrust import (
+    NeuralTrustGuardrailConfigModel,
+)
 from litellm.types.proxy.guardrails.guardrail_hooks.ovalix import (
     OvalixGuardrailConfigModel,
 )
@@ -70,7 +73,7 @@ Pydantic object defining how to set guardrails on litellm proxy
 guardrails:
   - guardrail_name: "bedrock-pre-guard"
     litellm_params:
-      guardrail: bedrock  # supported values: "akto", "aporia", "bedrock", "lakera", "zscaler_ai_guard"
+      guardrail: bedrock  # supported values: "akto", "aporia", "bedrock", "lakera", "neuraltrust", "zscaler_ai_guard"
       mode: "during_call"
       guardrailIdentifier: ff6ujrregl1q
       guardrailVersion: "DRAFT"
@@ -88,6 +91,7 @@ class SupportedGuardrailIntegrations(Enum):
     PRESIDIO = "presidio"
     HIDE_SECRETS = "hide-secrets"
     HIDDENLAYER = "hiddenlayer"
+    NEURALTRUST = "neuraltrust"
     AIM = "aim"
     CATO_NETWORKS = "cato_networks"
     PANGEA = "pangea"
@@ -872,7 +876,7 @@ class BaseLitellmParams(ContentFilterConfigModel):  # works for new and patch up
         default="fail_closed",
         description=(
             "Behavior when a guardrail endpoint is unreachable due to network errors. "
-            "Implemented by guardrail='generic_guardrail_api', 'akto', 'vigil_guard', 'repelloai', 'headroom', and 'compresr'. "
+            "Implemented by guardrail='generic_guardrail_api', 'akto', 'vigil_guard', 'repelloai', 'headroom', 'compresr', and 'neuraltrust'. "
             "'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
         ),
     )
@@ -996,6 +1000,7 @@ class LitellmParams(
     QualifireGuardrailConfigModel,
     BlockCodeExecutionGuardrailConfigModel,
     HiddenlayerGuardrailConfigModel,
+    NeuralTrustGuardrailConfigModel,
     QostodianNexusConfigModel,
     VigilGuardGuardrailConfigModel,
     SingulrGuardrailConfigModel,

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

@@ -0,0 +1,44 @@
+from typing import Final, Literal
+
+from pydantic import Field
+
+from .base import GuardrailConfigModel
+
+DEFAULT_API_BASE: Final = "https://trustguard.neuraltrust.ai"
+
+
+class NeuralTrustGuardrailConfigModel(GuardrailConfigModel):
+    """Config for the NeuralTrust TrustGuard native LiteLLM hook."""
+
+    api_key: str | None = Field(
+        default=None,
+        description=("TrustGuard API key (tgk_...). If not provided, TRUSTGUARD_API_KEY is checked."),
+    )
+
+    api_base: str | None = Field(
+        default=None,
+        description=("TrustGuard API base URL. Default https://trustguard.neuraltrust.ai. Env: TRUSTGUARD_API_BASE."),
+    )
+
+    collector_key: str | None = Field(
+        default=None,
+        description=(
+            "TrustGuard collector key (tgcol_...). Optional when the API key is bound to a "
+            "collector. Env: TRUSTGUARD_COLLECTOR_KEY."
+        ),
+    )
+
+    unreachable_fallback: Literal["fail_closed", "fail_open"] = Field(
+        default="fail_closed",
+        description=(
+            "What to do on transport failures (connect errors, timeouts, HTTP 502/504). "
+            "'fail_closed' blocks the request; 'fail_open' allows it. "
+            "HTTP 503 entitlements, 401/403, other 4xx/5xx, unknown verdicts, and "
+            "unusable transform payloads always fail closed. "
+            "'fail_open' means the request bypasses TrustGuard entirely."
+        ),
+    )
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "NeuralTrust"

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

@@ -0,0 +1,466 @@
+import os
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import pytest
+from fastapi import HTTPException
+from httpx import Request, Response
+
+from litellm.exceptions import Timeout
+from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+from litellm.proxy.guardrails.guardrail_hooks.neuraltrust.neuraltrust import (
+    NeuralTrustGuardrail,
+)
+from litellm.types.utils import GenericGuardrailAPIInputs
+
+
+def _response(payload: object, status_code: int = 200) -> Response:
+    request = Request("POST", "https://trustguard.neuraltrust.ai/v1/evaluate")
+    return Response(status_code, request=request, json=payload)
+
+
+def _logging() -> LiteLLMLoggingObj:
+    return LiteLLMLoggingObj(
+        model="gpt-4o-mini",
+        messages=[{"role": "user", "content": "hello"}],
+        stream=False,
+        call_type="completion",
+        litellm_call_id="call-1",
+        function_id="fn-1",
+        start_time=None,
+    )
+
+
+def _guardrail(**kwargs: object) -> NeuralTrustGuardrail:
+    params: dict[str, object] = {
+        "api_key": "tgk_test",
+        "collector_key": "tgcol_test",
+        "guardrail_name": "neuraltrust",
+        "event_hook": "pre_call",
+    }
+    params.update(kwargs)
+    return NeuralTrustGuardrail(**params)  # type: ignore[arg-type]
+
+
+class TestNeuralTrustGuardrail:
+    def setup_method(self) -> None:
+        for key in ("TRUSTGUARD_API_KEY", "TRUSTGUARD_API_BASE", "TRUSTGUARD_COLLECTOR_KEY"):
+            os.environ.pop(key, None)
+
+    def teardown_method(self) -> None:
+        for key in ("TRUSTGUARD_API_KEY", "TRUSTGUARD_API_BASE", "TRUSTGUARD_COLLECTOR_KEY"):
+            os.environ.pop(key, None)
+
+    def test_missing_api_key_raises(self) -> None:
+        with pytest.raises(ValueError, match="API key is required"):
+            NeuralTrustGuardrail(guardrail_name="neuraltrust", event_hook="pre_call")
+
+    def test_initialization_defaults(self) -> None:
+        guardrail = _guardrail(default_on=True)
+        assert guardrail.api_base == "https://trustguard.neuraltrust.ai"
+        assert guardrail.collector_key == "tgcol_test"
+        assert guardrail.unreachable_fallback == "fail_closed"
+        assert guardrail.timeout == 5.0
+
+    @pytest.mark.asyncio
+    async def test_allow_request(self) -> None:
+        guardrail = _guardrail()
+        inputs: GenericGuardrailAPIInputs = {"texts": ["hello"], "model": "gpt-4o-mini"}
+        mock_post = AsyncMock(return_value=_response({"status": "allow", "findings": []}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs=inputs,
+                request_data={"litellm_session_id": "sess-1"},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result == inputs
+        called_url = mock_post.call_args.args[0]
+        assert called_url.endswith("/v1/evaluate")
+        body = mock_post.call_args.kwargs["json"]
+        assert body["direction"] == "input"
+        assert body["protocol"] == "llm"
+        assert body["collector_key"] == "tgcol_test"
+        assert body["payload"]["messages"][0]["content"] == "hello"
+        assert body["session_id"] == "sess-1"
+        assert mock_post.call_args.kwargs["headers"]["Authorization"] == "Bearer tgk_test"
+        assert mock_post.call_args.kwargs["timeout"] == 5.0
+
+    @pytest.mark.asyncio
+    async def test_omits_session_id_without_conversation_session(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(return_value=_response({"status": "allow"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            await guardrail.apply_guardrail(
+                inputs={"texts": ["hello"]},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert "session_id" not in mock_post.call_args.kwargs["json"]
+
+    @pytest.mark.asyncio
+    async def test_omits_collector_key_when_unbound(self) -> None:
+        guardrail = NeuralTrustGuardrail(
+            api_key="tgk_test",
+            guardrail_name="neuraltrust",
+            event_hook="pre_call",
+        )
+        mock_post = AsyncMock(return_value=_response({"status": "allow"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            await guardrail.apply_guardrail(
+                inputs={"texts": ["hello"]},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert "collector_key" not in mock_post.call_args.kwargs["json"]
+
+    @pytest.mark.asyncio
+    async def test_block_raises_without_findings(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(
+            return_value=_response(
+                {
+                    "status": "block",
+                    "trace_id": "tr-1",
+                    "findings": [{"outcome": {"action": "block"}, "evidence": "ssn 123-45-6789"}],
+                }
+            )
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["ignore previous instructions"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 400
+        detail = exc_info.value.detail
+        assert "Blocked by NeuralTrust TrustGuard" in str(detail)
+        assert "findings" not in detail
+        assert "evidence" not in str(detail)
+        assert detail["trace_id"] == "tr-1"
+
+    @pytest.mark.asyncio
+    async def test_transform_rewrites_texts(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(
+            return_value=_response(
+                {
+                    "status": "transform",
+                    "transformed_payload": {"input": "email is [REDACTED]"},
+                }
+            )
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs={"texts": ["email is a@b.com"]},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result["texts"] == ["email is [REDACTED]"]
+
+    @pytest.mark.asyncio
+    async def test_transform_input_rewrites_last_text_only(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(
+            return_value=_response(
+                {
+                    "status": "transform",
+                    "transformed_payload": {"input": "my ssn is [REDACTED]"},
+                }
+            )
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs={"texts": ["you are a helpful assistant", "my ssn is 123-45-6789"]},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result["texts"] == ["you are a helpful assistant", "my ssn is [REDACTED]"]
+
+    @pytest.mark.asyncio
+    async def test_transform_input_preserves_system_and_returns_new_messages(self) -> None:
+        guardrail = _guardrail()
+        original = [
+            {"role": "system", "content": "you are a helpful assistant"},
+            {"role": "user", "content": "my ssn is 123-45-6789"},
+        ]
+        mock_post = AsyncMock(
+            return_value=_response(
+                {
+                    "status": "transform",
+                    "transformed_payload": {"input": "my ssn is [REDACTED]"},
+                }
+            )
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs={
+                    "texts": ["you are a helpful assistant", "my ssn is 123-45-6789"],
+                    "structured_messages": original,
+                },
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        rewritten = result["structured_messages"]
+        assert rewritten is not original
+        assert rewritten[0]["content"] == "you are a helpful assistant"
+        assert rewritten[1]["content"] == "my ssn is [REDACTED]"
+
+    @pytest.mark.asyncio
+    async def test_transform_rewrites_messages(self) -> None:
+        guardrail = _guardrail()
+        rewritten = [{"role": "user", "content": "ssn is [REDACTED]"}]
+        mock_post = AsyncMock(
+            return_value=_response(
+                {
+                    "status": "transform",
+                    "transformed_payload": {"messages": rewritten},
+                }
+            )
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs={
+                    "texts": ["ssn is 123-45-6789"],
+                    "structured_messages": [{"role": "user", "content": "ssn is 123-45-6789"}],
+                },
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result["texts"] == ["ssn is [REDACTED]"]
+        assert result["structured_messages"] == rewritten
+        assert result["structured_messages"] is not rewritten
+
+    @pytest.mark.asyncio
+    async def test_transform_without_payload_fail_closed(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        mock_post = AsyncMock(return_value=_response({"status": "transform"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["email is a@b.com"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 400
+        assert "transform missing payload" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_transform_string_messages_fail_closed(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(
+            return_value=_response({"status": "transform", "transformed_payload": {"messages": "REDACTED"}})
+        )
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["secret"], "structured_messages": [{"role": "user", "content": "secret"}]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_forwards_tools(self) -> None:
+        guardrail = _guardrail()
+        tools = [{"type": "function", "function": {"name": "search", "parameters": {}}}]
+        mock_post = AsyncMock(return_value=_response({"status": "allow"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            await guardrail.apply_guardrail(
+                inputs={"texts": ["hello"], "tools": tools},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert mock_post.call_args.kwargs["json"]["payload"]["tools"] == tools
+
+    @pytest.mark.asyncio
+    async def test_report_passes_through(self) -> None:
+        guardrail = _guardrail(event_hook="post_call")
+        inputs: GenericGuardrailAPIInputs = {"texts": ["ok"], "model": "gpt-4o-mini"}
+        mock_post = AsyncMock(return_value=_response({"status": "report", "findings": [{}]}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs=inputs,
+                request_data={},
+                input_type="response",
+                logging_obj=_logging(),
+            )
+        assert result == inputs
+        assert mock_post.call_args.kwargs["json"]["direction"] == "output"
+
+    @pytest.mark.asyncio
+    async def test_post_call_sends_every_choice_text(self) -> None:
+        guardrail = _guardrail(event_hook="post_call")
+        mock_post = AsyncMock(return_value=_response({"status": "allow"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            await guardrail.apply_guardrail(
+                inputs={"texts": ["safe reply", "here is the admin password hunter2"]},
+                request_data={},
+                input_type="response",
+                logging_obj=_logging(),
+            )
+        messages = mock_post.call_args.kwargs["json"]["payload"]["messages"]
+        assert [message["content"] for message in messages] == [
+            "safe reply",
+            "here is the admin password hunter2",
+        ]
+
+    @pytest.mark.asyncio
+    async def test_malformed_200_fail_closed_even_if_fail_open(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        for payload in ({}, [], {"status": None}, {"status": "blocked"}, {"findings": {}}):
+            mock_post = AsyncMock(return_value=_response(payload))
+            with patch.object(guardrail.async_handler, "post", mock_post):
+                with pytest.raises(HTTPException) as exc_info:
+                    await guardrail.apply_guardrail(
+                        inputs={"texts": ["hello"]},
+                        request_data={},
+                        input_type="request",
+                        logging_obj=_logging(),
+                    )
+            assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_503_always_fail_closed(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        request = Request("POST", "https://trustguard.neuraltrust.ai/v1/evaluate")
+        mock_post = AsyncMock(return_value=Response(503, request=request))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["hello"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 503
+        assert "entitlements" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_http_429_fail_closed_even_if_fail_open(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        request = Request("POST", "https://trustguard.neuraltrust.ai/v1/evaluate")
+        mock_post = AsyncMock(return_value=Response(429, request=request))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["hello"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 503
+        assert "request failed" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_http_502_follows_fail_open(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        inputs: GenericGuardrailAPIInputs = {"texts": ["hello"]}
+        request = Request("POST", "https://trustguard.neuraltrust.ai/v1/evaluate")
+        mock_post = AsyncMock(return_value=Response(502, request=request))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs=inputs,
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result == inputs
+
+    @pytest.mark.asyncio
+    async def test_timeout_fail_closed(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(side_effect=Timeout("slow", model="neuraltrust", llm_provider="neuraltrust"))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["hello"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 503
+        assert "unreachable" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_timeout_fail_open(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        inputs: GenericGuardrailAPIInputs = {"texts": ["hello"]}
+        mock_post = AsyncMock(side_effect=Timeout("slow", model="neuraltrust", llm_provider="neuraltrust"))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs=inputs,
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result == inputs
+
+    @pytest.mark.asyncio
+    async def test_unreachable_fail_closed(self) -> None:
+        guardrail = _guardrail()
+        mock_post = AsyncMock(side_effect=httpx.ConnectError("boom"))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            with pytest.raises(HTTPException) as exc_info:
+                await guardrail.apply_guardrail(
+                    inputs={"texts": ["hello"]},
+                    request_data={},
+                    input_type="request",
+                    logging_obj=_logging(),
+                )
+        assert exc_info.value.status_code == 503
+
+    @pytest.mark.asyncio
+    async def test_unreachable_fail_open(self) -> None:
+        guardrail = _guardrail(unreachable_fallback="fail_open")
+        inputs: GenericGuardrailAPIInputs = {"texts": ["hello"]}
+        mock_post = AsyncMock(side_effect=httpx.ConnectError("boom"))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            result = await guardrail.apply_guardrail(
+                inputs=inputs,
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert result == inputs
+
+    @pytest.mark.asyncio
+    async def test_custom_timeout_is_passed_to_client(self) -> None:
+        guardrail = _guardrail(timeout=12)
+        mock_post = AsyncMock(return_value=_response({"status": "allow"}))
+        with patch.object(guardrail.async_handler, "post", mock_post):
+            await guardrail.apply_guardrail(
+                inputs={"texts": ["hello"]},
+                request_data={},
+                input_type="request",
+                logging_obj=_logging(),
+            )
+        assert mock_post.call_args.kwargs["timeout"] == 12.0
+
+    def test_get_config_model(self) -> None:
+        model = NeuralTrustGuardrail.get_config_model()
+        assert model is not None
+        assert model.ui_friendly_name() == "NeuralTrust"
+
+    def test_registry_contains_neuraltrust(self) -> None:
+        from litellm.proxy.guardrails.guardrail_hooks.neuraltrust import (
+            NeuralTrustGuardrail as Registered,
+        )
+        from litellm.proxy.guardrails.guardrail_registry import (
+            guardrail_class_registry,
+            guardrail_initializer_registry,
+        )
+
+        assert "neuraltrust" in guardrail_initializer_registry
+        assert guardrail_class_registry["neuraltrust"] is Registered

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

@@ -0,0 +1,22 @@
+<svg xmlns="http://www.w3.org/2000/svg" width="32" height="32" viewBox="0 0 32 32" fill="none">
+  <g clip-path="url(#neuraltrustClip)">
+    <path fill="url(#neuraltrustGrad)" d="M32 0H0v32h32z" />
+    <path
+      fill="#fff"
+      d="M18.092 20.06a.67.67 0 0 1-.55.3.7.7 0 0 1-.565-.286l-2.704-3.814-1.45 2.103 2.197 3.098a3.08 3.08 0 0 0 2.51 1.297h.038a3.06 3.06 0 0 0 2.502-1.342l8.02-11.477h-2.926z"
+    />
+    <path
+      fill="#fff"
+      d="M14.292 11.518a.63.63 0 0 1 .552.286l2.652 3.74 1.449-2.103-2.145-3.024a3.08 3.08 0 0 0-2.509-1.297h-.039a3.06 3.06 0 0 0-2.506 1.35L3.925 21.85l-.085.123h2.91l6.98-10.155a.68.68 0 0 1 .562-.3"
+    />
+  </g>
+  <defs>
+    <linearGradient id="neuraltrustGrad" x1="30.667" x2="6.667" y1="0" y2="32" gradientUnits="userSpaceOnUse">
+      <stop stop-color="#03AFFF" />
+      <stop offset="1" stop-color="#9B29FF" />
+    </linearGradient>
+    <clipPath id="neuraltrustClip">
+      <path fill="#fff" d="M0 0h32v32H0z" />
+    </clipPath>
+  </defs>
+</svg>

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

@@ -216,6 +216,12 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
     mode: "pre_call",
     defaultOn: false,
   },
+  neuraltrust: {
+    provider: "Neuraltrust",
+    guardrailNameSuggestion: "NeuralTrust TrustGuard",
+    mode: "pre_call",
+    defaultOn: false,
+  },
   noma: {
     provider: "Noma",
     guardrailNameSuggestion: "Noma Security",

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

@@ -12,6 +12,7 @@ const EXPECTED_PARTNER_LOGO_FILES: Record<string, string> = {
   panw: "palo_alto_networks.jpeg",
   cisco_ai_defense: "cisco.png",
   noma: "noma_security.png",
+  neuraltrust: "neuraltrust.svg",
   aporia: "aporia.png",
   aim: "aim_security.jpeg",
   cato_networks: "cato_networks.svg",
@@ -51,4 +52,10 @@ describe("guardrail_garden_data logos", () => {
       expect(card.logo, `card ${card.id}`).not.toContain("/ui/assets/logos/");
     }
   });
+
+  it("does not publish unsourced NeuralTrust eval numbers", () => {
+    const card = PARTNER_GUARDRAIL_CARDS.find((c) => c.id === "neuraltrust");
+    expect(card).toBeDefined();
+    expect(card?.eval).toBeUndefined();
+  });
 });

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

@@ -319,6 +319,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
     tags: ["Enterprise", "Security", "Prompt Injection", "PII"],
     providerKey: "CiscoAiDefense",
   },
+  {
+    id: "neuraltrust",
+    name: "NeuralTrust",
+    description:
+      "TrustGuard runtime guardrails: prompt injection, toxicity, DLP, and policy enforcement on LLM input and output.",
+    category: "partner",
+    logo: guardrailLogoMap["NeuralTrust"],
+    tags: ["Security", "Prompt Injection", "DLP"],
+    providerKey: "Neuraltrust",
+  },
   {
     id: "noma",
     name: "Noma Security",

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

@@ -195,6 +195,20 @@ describe("guardrail_info_helpers", () => {
       expect(result.logo).toContain("noma_security.png");
     });
 
+    it("should resolve NeuralTrust logo and display name", () => {
+      populateGuardrailProviders({
+        neuraltrust: { ui_friendly_name: "NeuralTrust" },
+      });
+      populateGuardrailProviderMap({
+        neuraltrust: { ui_friendly_name: "NeuralTrust" },
+      });
+
+      const result = getGuardrailLogoAndName("neuraltrust");
+
+      expect(result.displayName).toBe("NeuralTrust");
+      expect(result.logo).toContain("neuraltrust.svg");
+    });
+
     it("should resolve RepelloAI Argus logo and display name", () => {
       populateGuardrailProviders({
         repelloai: { ui_friendly_name: "RepelloAI Argus" },

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

@@ -13,6 +13,7 @@ import lakeraAiLogo from "../../../../../public/assets/logos/lakeraai.jpeg";
 import lassoLogo from "../../../../../public/assets/logos/lasso.png";
 import litellmLogo from "../../../../../public/assets/logos/litellm_logo.jpg";
 import microsoftAzureLogo from "../../../../../public/assets/logos/microsoft_azure.svg";
+import neuraltrustLogo from "../../../../../public/assets/logos/neuraltrust.svg";
 import nomaSecurityLogo from "../../../../../public/assets/logos/noma_security.png";
 import openaiSmallLogo from "../../../../../public/assets/logos/openai_small.svg";
 import paloAltoNetworksLogo from "../../../../../public/assets/logos/palo_alto_networks.jpeg";
@@ -172,6 +173,7 @@ export const guardrailLogoMap = {
   "Aporia AI": aporiaLogo.src,
   "PANW Prisma AIRS": paloAltoNetworksLogo.src,
   "Cisco AI Defense": ciscoLogo.src,
+  NeuralTrust: neuraltrustLogo.src,
   "Noma Security": nomaSecurityLogo.src,
   "Javelin Guardrails": javelinLogo.src,
   "Pillar Guardrail": pillarLogo.src,