feat(guardrails): configurable Reco fail modes

Expose unreachable_fallback and fail_on_error on the Reco guardrail
and keep fail-open as the default. Values under optional_params win.
Top-level litellm_params values apply only when set explicitly, since
their defaults are fail-closed and would otherwise flip the default

Also type RecoGuardrail's forwarded kwargs with an Unpack'd TypedDict
instead of silencing ANN003

Signed-off-by: MickeyShnaiderman-RecoLabs <mickeys@reco.ai>
Co-authored-by: Cursor <cursoragent@cursor.com>

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

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING, Final
 
 from litellm.secret_managers.main import get_secret_str
 from litellm.types.guardrails import SupportedGuardrailIntegrations
+from litellm.types.proxy.guardrails.guardrail_hooks.reco import RecoOptionalParams
 
 from .reco import RecoGuardrail
 
@@ -9,13 +10,17 @@ if TYPE_CHECKING:
     from litellm.types.guardrails import Guardrail, LitellmParams
 
 
-def _get_config_value(optional_params: object, attribute_name: str) -> str | None:
-    """Read one param from a dict or a pydantic object, resolving an
-    ``os.environ/<VAR>`` reference the way guardrail api_key/api_base are resolved."""
-    if isinstance(optional_params, dict):
-        value = optional_params.get(attribute_name)
-    else:
-        value = getattr(optional_params, attribute_name, None)
+def _get_config_value(litellm_params: "LitellmParams", optional_params: object, attribute_name: str) -> object:
+    """Read one param from optional_params (a dict or a pydantic object), falling back to a top-level
+    litellm_params field only when it was set explicitly, since those carry fail-closed defaults.
+    Resolves an ``os.environ/<VAR>`` reference the way guardrail api_key/api_base are resolved."""
+    nested: Final = (
+        optional_params.get(attribute_name)
+        if isinstance(optional_params, dict)
+        else getattr(optional_params, attribute_name, None)
+    )
+    explicitly_set_top_level: Final = attribute_name in getattr(litellm_params, "model_fields_set", frozenset())
+    value: Final = getattr(litellm_params, attribute_name) if nested is None and explicitly_set_top_level else nested
     if isinstance(value, str) and value.startswith("os.environ/"):
         return get_secret_str(value)
     return value
@@ -25,11 +30,20 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
     import litellm
 
     optional_params: Final = getattr(litellm_params, "optional_params", None)
+    configured: Final = (
+        (field_name, _get_config_value(litellm_params, optional_params, field_name))
+        for field_name in RecoOptionalParams.model_fields
+    )
+    reco_params: Final = RecoOptionalParams.model_validate(
+        {field_name: value for field_name, value in configured if value is not None}
+    )
 
     _reco_callback: Final = RecoGuardrail(
         guardrail_name=guardrail.get("guardrail_name", ""),
-        reco_tenant_id=_get_config_value(optional_params, "reco_tenant_id"),
-        api_base=_get_config_value(optional_params, "api_base"),
+        reco_tenant_id=reco_params.reco_tenant_id,
+        api_base=reco_params.api_base,
+        unreachable_fallback=reco_params.unreachable_fallback,
+        fail_on_error=reco_params.fail_on_error,
         event_hook=litellm_params.mode,
         default_on=litellm_params.default_on,
     )

litellm/proxy/guardrails/guardrail_hooks/reco/reco.py

@@ -1,6 +1,8 @@
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Literal, Optional
 
+from typing_extensions import ReadOnly, TypedDict, Unpack
+
 from litellm.exceptions import GuardrailRaisedException
 from litellm.integrations.custom_guardrail import log_guardrail_information
 from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import (
@@ -15,6 +17,13 @@ if TYPE_CHECKING:
     from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
 
 
+class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
+    """Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
+
+    guardrail_name: ReadOnly[str | None]
+    default_on: ReadOnly[bool]
+
+
 class RecoGuardrail(GenericGuardrailAPI):
     """Reco guardrail integration for LiteLLM, built on the Generic Guardrail API wire contract."""
 
@@ -23,7 +32,9 @@ class RecoGuardrail(GenericGuardrailAPI):
         reco_tenant_id: str | None,
         api_base: str | None,
         headers: Mapping[str, str] | None = None,
-        **kwargs,  # noqa: ANN003  # kwargs-ok: forwards guardrail_name/event_hook/default_on to CustomGuardrail, whose types are narrower than LitellmParams' own field types
+        unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_open",
+        fail_on_error: bool = False,
+        **kwargs: Unpack[_CustomGuardrailOptions],
     ) -> None:
         if not reco_tenant_id:
             raise ValueError("reco_tenant_id is required for the Reco guardrail")
@@ -37,8 +48,8 @@ class RecoGuardrail(GenericGuardrailAPI):
         super().__init__(
             headers=merged_headers,
             api_base=api_base,
-            unreachable_fallback="fail_open",
-            fail_on_error=False,
+            unreachable_fallback=unreachable_fallback,
+            fail_on_error=fail_on_error,
             **kwargs,
         )
 

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

@@ -1,4 +1,5 @@
 import uuid
+from typing import Literal
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -22,6 +23,14 @@ class RecoOptionalParams(BaseModel):
     api_base: str = Field(
         description="Base URL for the Reco guardrail API. Reco's endpoint is per-region and per-silo, so this must be set explicitly rather than defaulted.",
     )
+    unreachable_fallback: Literal["fail_closed", "fail_open"] = Field(
+        default="fail_open",
+        description="Behavior when Reco is unreachable. 'fail_open' (default) lets the request through uninspected; 'fail_closed' blocks it, but only takes effect when fail_on_error is true.",
+    )
+    fail_on_error: bool = Field(
+        default=False,
+        description="Behavior on any Reco error, including unreachability. False (default) lets the request through uninspected; True blocks it.",
+    )
 
     @field_validator("reco_tenant_id")
     @classmethod

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

@@ -5,6 +5,7 @@ Tests for the Reco guardrail integration.
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import httpx
 import pytest
 from pydantic import ValidationError
 
@@ -46,10 +47,6 @@ class TestRecoGuardrailConfiguration:
             "X-Reco-Tenant-Id": "11111111-1111-1111-1111-111111111111",
         }
 
-    def test_unreachable_fallback_and_fail_on_error_are_hardcoded(self, reco_guardrail):
-        assert reco_guardrail.unreachable_fallback == "fail_open"
-        assert reco_guardrail.fail_on_error is False
-
     def test_missing_reco_tenant_id_raises(self):
         with pytest.raises(ValueError, match="reco_tenant_id"):
             RecoGuardrail(reco_tenant_id=None, api_base="https://edge1.us.reco.ai")
@@ -128,6 +125,81 @@ class TestRecoGuardrailInitializer:
         assert guardrail.headers == {"X-Reco-Tenant-Id": "44444444-4444-4444-4444-444444444444"}
 
 
+def _litellm_params(top_level: dict[str, object], nested: dict[str, object]) -> LitellmParams:
+    return LitellmParams(
+        guardrail="reco",
+        mode="pre_call",
+        optional_params={
+            "reco_tenant_id": "22222222-2222-2222-2222-222222222222",
+            "api_base": "https://edge2.eu.reco.ai",
+            **nested,
+        },
+        **top_level,
+    )
+
+
+async def _apply_with_reco_unreachable(guardrail: RecoGuardrail):
+    with patch.object(guardrail.async_handler, "post", side_effect=httpx.ConnectError("connection refused")):
+        return await guardrail.apply_guardrail(
+            inputs={"texts": ["hello"]},
+            request_data={"metadata": {}},
+            input_type="request",
+        )
+
+
+class TestRecoGuardrailFailureModes:
+    @pytest.mark.asyncio
+    async def test_unreachable_reco_lets_request_through_by_default(self, reco_guardrail):
+        result = await _apply_with_reco_unreachable(reco_guardrail)
+
+        assert result["texts"] == ["hello"]
+
+    @pytest.mark.asyncio
+    async def test_fail_closed_blocks_request_when_reco_unreachable(self):
+        guardrail = RecoGuardrail(
+            reco_tenant_id="11111111-1111-1111-1111-111111111111",
+            api_base="https://edge1.us.reco.ai",
+            unreachable_fallback="fail_closed",
+            fail_on_error=True,
+        )
+
+        with pytest.raises(Exception, match="connection refused"):
+            await _apply_with_reco_unreachable(guardrail)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("top_level", "nested"),
+        [
+            pytest.param({}, {"unreachable_fallback": "fail_closed", "fail_on_error": True}, id="optional-params"),
+            pytest.param({"unreachable_fallback": "fail_closed", "fail_on_error": True}, {}, id="top-level"),
+        ],
+    )
+    async def test_initialize_guardrail_fails_closed_when_configured(self, top_level, nested):
+        guardrail = initialize_guardrail(_litellm_params(top_level, nested), {"guardrail_name": "prod-reco"})
+
+        with pytest.raises(Exception, match="connection refused"):
+            await _apply_with_reco_unreachable(guardrail)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("top_level", "nested"),
+        [
+            pytest.param({}, {}, id="unset-ignores-top-level-fail-closed-defaults"),
+            pytest.param(
+                {"unreachable_fallback": "fail_closed", "fail_on_error": True},
+                {"fail_on_error": False},
+                id="optional-params-wins-over-top-level",
+            ),
+        ],
+    )
+    async def test_initialize_guardrail_stays_fail_open(self, top_level, nested):
+        guardrail = initialize_guardrail(_litellm_params(top_level, nested), {"guardrail_name": "prod-reco"})
+
+        result = await _apply_with_reco_unreachable(guardrail)
+
+        assert result["texts"] == ["hello"]
+
+
 class TestRecoGuardrailBlocking:
     @pytest.mark.asyncio
     async def test_action_blocked_raises_exception_with_configured_guardrail_name(self, reco_guardrail):