feat(guardrails): add ConductGuard integration

Adds Conduct Guard as a first-class LiteLLM guardrail. Point any
LiteLLM proxy at Conduct and every LLM call routed through it is
policy-checked before the upstream request goes out — block, warn,
audit, or trigger a human-in-the-loop approval, with the same signed
configuration + hash-chained audit log Conduct exposes on its native
enforcement surfaces.

- litellm/types/guardrails.py: add CONDUCT to SupportedGuardrailIntegrations.
- litellm/proxy/guardrails/guardrail_hooks/conduct/__init__.py: registration
  via guardrail_initializer_registry and guardrail_class_registry, picked
  up by the auto-discovery in guardrail_registry.py.
- litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py: the adapter.
  CustomGuardrail subclass, async_pre_call_hook, response envelope parser
  for the five Conduct verdicts (ok / advisory / WARNING / BLOCKED /
  PENDING approval), fail-mode logic, session-ID resolution chain
  (litellm_metadata.trace_id → X-Conduct-Session-Id → hash fallback).
- tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py: envelope
  parsing, pre-call allow/block/approval, config precedence, missing-token
  construction error.

```yaml
guardrails:
  - guardrail_name: conduct-guard
    litellm_params:
      guardrail: conduct
      mode: pre_call
      api_base: https://api.conductai.ai      # optional, default
      api_key: os.environ/CONDUCT_AGENT_TOKEN # cond_agt_* token
      fail_mode: fail_closed                   # or fail_open
      tool_name: llm_call                      # scoped tool_name
```

A standalone PyPI package `conduct-litellm-guard` shipped ahead of this
PR for teams pinned to older LiteLLM versions. Once this integration
merges, the standalone README will point at the native support as the
preferred path.

- PyPI: https://pypi.org/project/conduct-litellm-guard/
- Product: https://conductai.ai/guard

Contact: sudhi@b2bsphere.com

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

@@ -0,0 +1,37 @@
+from typing import TYPE_CHECKING, Final
+
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+from .conduct import ConductGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
+    import litellm
+
+    _conduct_callback: Final = ConductGuardrail(
+        api_base=getattr(litellm_params, "api_base", None),
+        api_key=getattr(litellm_params, "api_key", None),
+        workspace_id=getattr(litellm_params, "workspace_id", None),
+        fail_mode=getattr(litellm_params, "fail_mode", "fail_closed"),
+        tool_name=getattr(litellm_params, "tool_name", "llm_call"),
+        timeout=getattr(litellm_params, "timeout", 8.0),
+        guardrail_name=guardrail.get("guardrail_name", ""),
+        event_hook=litellm_params.mode,
+        default_on=litellm_params.default_on,
+    )
+    litellm.logging_callback_manager.add_litellm_callback(_conduct_callback)
+
+    return _conduct_callback
+
+
+guardrail_initializer_registry: Final = {
+    SupportedGuardrailIntegrations.CONDUCT.value: initialize_guardrail,
+}
+
+
+guardrail_class_registry: Final = {
+    SupportedGuardrailIntegrations.CONDUCT.value: ConductGuardrail,
+}

litellm/proxy/guardrails/guardrail_hooks/conduct/conduct.py

@@ -0,0 +1,311 @@
+# +-------------------------------------------------------------+
+#
+#           Use Conduct Guard for your LLM calls
+#
+#     Runtime policy enforcement — block / warn / audit / approval
+#     Signed configuration + hash-chained audit + 20+ compliance packs
+#     Docs: https://conductai.ai/guard
+#
+# +-------------------------------------------------------------+
+from __future__ import annotations
+
+import hashlib
+import os
+import uuid
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Final, Literal
+
+from litellm._logging import verbose_proxy_logger
+from litellm.integrations.custom_guardrail import (
+    CustomGuardrail,
+    log_guardrail_information,
+)
+from litellm.llms.custom_httpx.http_handler import (
+    get_async_httpx_client,
+    httpxSpecialProvider,
+)
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.types.guardrails import GuardrailEventHooks
+
+GUARDRAIL_NAME: Final = "conduct"
+
+if TYPE_CHECKING:
+    from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
+
+
+Verdict = Literal["allow", "advisory", "warning", "block", "approval", "unknown"]
+FailMode = Literal["fail_open", "fail_closed"]
+
+
+@dataclass(frozen=True)
+class GuardDecision:
+    """Structured view of what ``guard_check`` returned. The raw text is
+    kept so audit / logging surfaces can quote it verbatim."""
+
+    verdict: Verdict
+    raw: str
+    rule_id: str | None = None
+    message: str | None = None
+
+    @classmethod
+    def parse(cls, text: str) -> "GuardDecision":
+        """Map the ``guard_check`` string envelope to a verdict.
+
+        Response contract from Conduct:
+          * ``"ok"`` or empty → allow silently
+          * ``"advisory: ..."`` → allow but log
+          * ``"WARNING — ..."`` → allow but surface
+          * ``"BLOCKED — ..."`` → hard block
+          * ``"PENDING approval — ..."`` → HITL — treat as block
+        """
+        stripped = (text or "").strip()
+        if not stripped or stripped.lower().startswith("ok"):
+            return cls(verdict="allow", raw=stripped)
+        if stripped.startswith("BLOCKED"):
+            return cls(
+                verdict="block",
+                raw=stripped,
+                rule_id=_extract_rule_id(stripped),
+                message=_strip_prefix(stripped, "BLOCKED"),
+            )
+        if stripped.startswith("PENDING approval"):
+            return cls(
+                verdict="approval",
+                raw=stripped,
+                rule_id=_extract_rule_id(stripped),
+                message=_strip_prefix(stripped, "PENDING approval"),
+            )
+        if stripped.startswith("WARNING"):
+            return cls(
+                verdict="warning",
+                raw=stripped,
+                rule_id=_extract_rule_id(stripped),
+                message=_strip_prefix(stripped, "WARNING"),
+            )
+        if stripped.startswith("advisory"):
+            return cls(
+                verdict="advisory",
+                raw=stripped,
+                rule_id=_extract_rule_id(stripped),
+                message=_strip_prefix(stripped, "advisory"),
+            )
+        return cls(verdict="unknown", raw=stripped)
+
+
+def _strip_prefix(text: str, prefix: str) -> str | None:
+    remainder = text[len(prefix):].strip()
+    return remainder.lstrip(":—- ").strip() or None
+
+
+def _extract_rule_id(text: str) -> str | None:
+    marker = "[rule:"
+    idx = text.find(marker)
+    if idx < 0:
+        return None
+    tail = text[idx + len(marker):]
+    end = tail.find("]")
+    return tail[:end].strip() if end >= 0 else None
+
+
+class ConductGuardrailBlocked(Exception):
+    """Raised inside the pre-call hook to abort a LiteLLM request. LiteLLM
+    surfaces the message to the caller."""
+
+    def __init__(self, decision: GuardDecision) -> None:
+        self.decision = decision
+        super().__init__(decision.message or decision.raw or "Blocked by Conduct Guard")
+
+
+class ConductGuardrail(CustomGuardrail):
+    """Conduct Guard as a LiteLLM ``CustomGuardrail``.
+
+    Reads config from LiteLLM's guardrail block. Every pre-call hook
+    invocation calls Conduct's ``guard_check`` MCP tool using the
+    supplied agent token. On block, raises so the LiteLLM proxy returns
+    an error to the caller instead of forwarding to the model."""
+
+    @classmethod
+    def get_supported_event_hooks(cls) -> list[GuardrailEventHooks]:
+        return [
+            GuardrailEventHooks.pre_call,
+            GuardrailEventHooks.during_call,
+        ]
+
+    def __init__(
+        self,
+        *,
+        api_base: str | None = None,
+        api_key: str | None = None,
+        workspace_id: str | None = None,
+        fail_mode: FailMode = "fail_closed",
+        tool_name: str = "llm_call",
+        timeout: float = 8.0,
+        **kwargs: Any,
+    ) -> None:
+        kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
+        super().__init__(**kwargs)
+
+        self._api_url = (
+            api_base or os.environ.get("CONDUCT_API_URL", "https://api.conductai.ai")
+        ).rstrip("/")
+        token = api_key or os.environ.get("CONDUCT_AGENT_TOKEN")
+        if not token:
+            raise ValueError(
+                "ConductGuardrail: agent token is required. Set CONDUCT_AGENT_TOKEN "
+                "in the environment or pass api_key in the guardrail config."
+            )
+        self._agent_token = token
+        self._workspace_id = workspace_id or os.environ.get("CONDUCT_WORKSPACE_ID")
+        self._fail_mode: FailMode = fail_mode
+        self._tool_name = tool_name
+        self._timeout = timeout
+        self._async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
+
+    # ── LiteLLM contract ───────────────────────────────────────────────
+
+    @log_guardrail_information
+    async def async_pre_call_hook(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        cache: Any,
+        data: dict[str, Any],
+        call_type: Literal[
+            "completion",
+            "text_completion",
+            "embeddings",
+            "image_generation",
+            "moderation",
+            "audio_transcription",
+        ],
+    ) -> dict[str, Any] | None:
+        decision = await self._check(data=data, call_type=call_type)
+
+        if decision.verdict in ("block", "approval"):
+            raise ConductGuardrailBlocked(decision)
+
+        data.setdefault("metadata", {}).setdefault("conduct_guard", {}).update(
+            {"verdict": decision.verdict, "rule_id": decision.rule_id}
+        )
+        return data
+
+    # ── Guard check ─────────────────────────────────────────────────
+
+    async def _check(self, *, data: dict[str, Any], call_type: str) -> GuardDecision:
+        tool_input = _build_tool_input(data, call_type)
+        session_id = _extract_session_id(data)
+        prompt = _extract_prompt_text(data)
+
+        arguments: dict[str, Any] = {
+            "tool_name": self._tool_name,
+            "tool_input": tool_input,
+        }
+        if prompt is not None:
+            arguments["prompt"] = prompt
+
+        payload = {
+            "jsonrpc": "2.0",
+            "id": str(uuid.uuid4()),
+            "method": "tools/call",
+            "params": {"name": "guard_check", "arguments": arguments},
+        }
+
+        headers = {
+            "Authorization": f"Bearer {self._agent_token}",
+            "Content-Type": "application/json",
+            "User-Agent": "litellm-conduct-guardrail/1.0",
+            "X-Claude-Surface": "litellm",
+        }
+        if self._workspace_id:
+            headers["X-Workspace-Id"] = self._workspace_id
+        if session_id:
+            headers["X-Conduct-Session-Id"] = session_id
+
+        try:
+            response = await self._async_handler.post(
+                f"{self._api_url}/guard/mcp",
+                json=payload,
+                headers=headers,
+                timeout=self._timeout,
+            )
+            response.raise_for_status()
+            body = response.json()
+
+            if "error" in body:
+                err = body["error"]
+                verbose_proxy_logger.warning(
+                    "conduct_guard: eval error %s — applying %s", err, self._fail_mode
+                )
+                if self._fail_mode == "fail_closed":
+                    return GuardDecision(
+                        verdict="block",
+                        raw=str(err),
+                        message="Conduct Guard policy-eval error (fail_closed).",
+                    )
+                return GuardDecision(verdict="allow", raw="fail_open")
+
+            result = body.get("result") or {}
+            for item in result.get("content", []) or []:
+                if item.get("type") == "text":
+                    return GuardDecision.parse(item.get("text", ""))
+            return GuardDecision(verdict="allow", raw="")
+        except Exception as e:
+            verbose_proxy_logger.warning(
+                "conduct_guard: transport error %s — applying %s", e, self._fail_mode
+            )
+            if self._fail_mode == "fail_closed":
+                return GuardDecision(
+                    verdict="block",
+                    raw=str(e),
+                    message="Conduct Guard is unreachable (fail_closed).",
+                )
+            return GuardDecision(verdict="allow", raw="fail_open")
+
+
+# ── Helpers ─────────────────────────────────────────────────────────
+
+
+def _extract_session_id(data: dict[str, Any]) -> str | None:
+    metadata = data.get("litellm_metadata") or data.get("metadata") or {}
+    for key in ("trace_id", "X-Conduct-Session-Id", "conduct_session_id"):
+        val = metadata.get(key)
+        if val:
+            return str(val)
+
+    user = data.get("user") or metadata.get("user") or ""
+    first_msg = ""
+    for m in data.get("messages") or []:
+        if isinstance(m, dict) and m.get("role") == "user":
+            first_msg = str(m.get("content", ""))[:512]
+            break
+    if not user and not first_msg:
+        return None
+    digest = hashlib.sha256((user + "|" + first_msg).encode("utf-8")).hexdigest()
+    return f"litellm-{digest[:16]}"
+
+
+def _extract_prompt_text(data: dict[str, Any]) -> str | None:
+    for m in reversed(data.get("messages") or []):
+        if isinstance(m, dict) and m.get("role") == "user":
+            content = m.get("content")
+            if isinstance(content, str):
+                return content[:4000]
+            if isinstance(content, list):
+                parts = [
+                    p.get("text", "") for p in content
+                    if isinstance(p, dict) and p.get("type") == "text"
+                ]
+                return " ".join(parts)[:4000] or None
+    return None
+
+
+def _build_tool_input(data: dict[str, Any], call_type: str) -> dict[str, Any]:
+    messages = data.get("messages") or []
+    return {
+        "model": data.get("model"),
+        "call_type": call_type,
+        "message_count": len(messages),
+        "temperature": data.get("temperature"),
+        "max_tokens": data.get("max_tokens"),
+        "stream": bool(data.get("stream")),
+        "content": _extract_prompt_text(data) or "",
+    }

litellm/types/guardrails.py

@@ -137,6 +137,7 @@ class SupportedGuardrailIntegrations(Enum):
     COMPRESR = "compresr"
     STRAIKER = "straiker"
     ALICE = "alice"
+    CONDUCT = "conduct"
 
 
 class Role(Enum):

tests/test_litellm/proxy/guardrails/test_conduct_guardrail.py

@@ -0,0 +1,122 @@
+"""Unit tests for the Conduct guardrail.
+
+Mocked transport — no real network. Verifies the response-envelope
+parser, pre-call hook behavior (allow / block / approval), fail-mode
+handling, and session-ID resolution chain.
+"""
+from __future__ import annotations
+
+import os
+from unittest.mock import AsyncMock
+
+import pytest
+
+from litellm.proxy.guardrails.guardrail_hooks.conduct.conduct import (
+    ConductGuardrail,
+    ConductGuardrailBlocked,
+    GuardDecision,
+)
+
+
+# ── Decision parsing ────────────────────────────────────────────────────
+
+
+class TestGuardDecisionParse:
+    @pytest.mark.parametrize("raw", ["ok", "OK", "", "  ok "])
+    def test_ok_variants_are_allow(self, raw: str) -> None:
+        assert GuardDecision.parse(raw).verdict == "allow"
+
+    def test_blocked_extracts_rule_id(self) -> None:
+        d = GuardDecision.parse(
+            "BLOCKED — command touches /etc/passwd  [rule: no-etc-passwd]"
+        )
+        assert d.verdict == "block"
+        assert d.rule_id == "no-etc-passwd"
+
+    def test_pending_approval_treated_as_block(self) -> None:
+        d = GuardDecision.parse(
+            "PENDING approval — HITL required [rule: prod-deploy-gate]"
+        )
+        assert d.verdict == "approval"
+        assert d.rule_id == "prod-deploy-gate"
+
+    def test_warning_is_warning(self) -> None:
+        d = GuardDecision.parse("WARNING — high-risk model [rule: model-tier]")
+        assert d.verdict == "warning"
+
+    def test_advisory_is_advisory(self) -> None:
+        d = GuardDecision.parse("advisory: policy eval error: boom")
+        assert d.verdict == "advisory"
+
+    def test_unknown_prefix_marked_unknown(self) -> None:
+        assert GuardDecision.parse("wat").verdict == "unknown"
+
+
+# ── Pre-call hook ──────────────────────────────────────────────────────
+
+
+@pytest.fixture(autouse=True)
+def _agent_token_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("CONDUCT_AGENT_TOKEN", "cond_agt_test_placeholder")
+
+
+def _guard() -> ConductGuardrail:
+    return ConductGuardrail()
+
+
+@pytest.mark.asyncio
+class TestPreCallHook:
+    async def test_allow_returns_data_with_metadata_tag(self) -> None:
+        g = _guard()
+        g._check = AsyncMock(return_value=GuardDecision(verdict="allow", raw="ok"))
+        data = {"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}
+        result = await g.async_pre_call_hook(None, None, data, "completion")
+        assert result is data
+        assert result["metadata"]["conduct_guard"]["verdict"] == "allow"
+
+    async def test_block_raises(self) -> None:
+        g = _guard()
+        g._check = AsyncMock(
+            return_value=GuardDecision(
+                verdict="block",
+                raw="BLOCKED — no secrets [rule: no-prod-secrets]",
+                rule_id="no-prod-secrets",
+            )
+        )
+        with pytest.raises(ConductGuardrailBlocked) as exc:
+            await g.async_pre_call_hook(None, None, {}, "completion")
+        assert exc.value.decision.rule_id == "no-prod-secrets"
+
+    async def test_pending_approval_also_raises(self) -> None:
+        g = _guard()
+        g._check = AsyncMock(
+            return_value=GuardDecision(verdict="approval", raw="PENDING approval — review")
+        )
+        with pytest.raises(ConductGuardrailBlocked):
+            await g.async_pre_call_hook(None, None, {}, "completion")
+
+
+class TestConfig:
+    def test_missing_token_raises_at_construction(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.delenv("CONDUCT_AGENT_TOKEN", raising=False)
+        with pytest.raises(ValueError, match="agent token"):
+            ConductGuardrail()
+
+    def test_config_api_key_wins_over_env(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setenv("CONDUCT_AGENT_TOKEN", "env-token")
+        g = ConductGuardrail(api_key="config-token")
+        assert g._agent_token == "config-token"
+
+    def test_config_api_base_wins_over_default(self) -> None:
+        g = ConductGuardrail(api_base="https://conduct.example.com/")
+        assert g._api_url == "https://conduct.example.com"
+
+
+if __name__ == "__main__":
+    import subprocess
+    import sys
+    raise SystemExit(subprocess.call([sys.executable, "-m", "pytest", __file__, "-v"]))