fix(guardrails): scan Responses API input in Azure Text Moderation

litellm/proxy/guardrails/guardrail_hooks/azure/base.py

@@ -150,17 +150,3 @@ def get_user_prompt_from_request(self, data: Mapping[str, object], call_type: Ca
         if not isinstance(messages, list):
             return None
         return get_last_user_message(cast(list[AllMessageValues], messages))  # cast-ok: narrowed to list
-
-    def get_user_prompt(self, messages: list[AllMessageValues]) -> str | None:
-        """
-        Get the last consecutive block of messages from the user.
-
-        Example:
-        messages = [
-            {"role": "user", "content": "Hello, how are you?"},
-            {"role": "assistant", "content": "I'm good, thank you!"},
-            {"role": "user", "content": "What is the weather in Tokyo?"},
-        ]
-        get_user_prompt(messages) -> "What is the weather in Tokyo?"
-        """
-        return get_last_user_message(messages)

litellm/proxy/guardrails/guardrail_hooks/azure/text_moderation.py

@@ -21,7 +21,6 @@
 if TYPE_CHECKING:
     from litellm.caching.caching import DualCache
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
-    from litellm.types.llms.openai import AllMessageValues
     from litellm.types.proxy.guardrails.guardrail_hooks.azure.azure_text_moderation import (
         AzureTextModerationGuardrailResponse,
     )
@@ -232,14 +231,10 @@ async def async_pre_call_hook(
             "Azure Text Moderation: Running pre-call prompt scan, on call_type: %s",
             call_type,
         )
-        new_messages: Final[list[AllMessageValues] | None] = data.get("messages")
-        if new_messages is None:
-            verbose_proxy_logger.warning("Azure Text Moderation: not running guardrail. No messages in data")
-            return data
-        user_prompt: Final = self.get_user_prompt(new_messages)
+        user_prompt: Final = self.get_user_prompt_from_request(data, call_type)
 
         if user_prompt:
-            verbose_proxy_logger.info("Azure Text Moderation: User prompt: %s", user_prompt)
+            verbose_proxy_logger.debug("Azure Text Moderation: User prompt: %s", user_prompt)
             await self.async_make_request(
                 text=user_prompt,
             )

tests/integration/observability/test_azure_content_safety_audit.py

@@ -17,10 +17,13 @@
 from pydantic import JsonValue
 
 _ATTACK_MARKER: Final = "synthetic-attack-marker"
+_MODERATION_MARKER: Final = "synthetic-moderation-marker"
 
 _SHIELD_TARGET_PREFIX: Final = "/contentsafety/text:shieldPrompt?api-version="
+_ANALYZE_TARGET_PREFIX: Final = "/contentsafety/text:analyze?api-version="
 
 _OPT_IN_SHIELD: Final = "audit-shield-optin"
+_TEXT_MODERATION: Final = "audit-text-mod"
 
 
 def _chat_frame(identity: str, delta: dict[str, JsonValue], finish: str | None = None) -> bytes:
@@ -139,7 +142,23 @@ def respond(request: Request) -> Reply:
                     }
                 ).encode()
             )
-        return Reply(status=404)
+        assert request.target.startswith(_ANALYZE_TARGET_PREFIX), request.target
+        text: Final = body["text"]
+        assert isinstance(text, str)
+        severity: Final = 4 if _MODERATION_MARKER in text else 0
+        return Reply(
+            body=json.dumps(
+                {
+                    "blocklistsMatch": [],
+                    "categoriesAnalysis": [
+                        {"category": "Hate", "severity": severity},
+                        {"category": "Sexual", "severity": 0},
+                        {"category": "SelfHarm", "severity": 0},
+                        {"category": "Violence", "severity": 0},
+                    ],
+                }
+            ).encode()
+        )
 
     return respond
 
@@ -182,6 +201,16 @@ def audit_rig(
                     "guardrail_name": "audit-shield",
                     "litellm_params": _shield_params(azure, mode="pre_call", default_on=True),
                 },
+                {
+                    "guardrail_name": _TEXT_MODERATION,
+                    "litellm_params": {
+                        "guardrail": "azure/text_moderations",
+                        "mode": "pre_call",
+                        "default_on": False,
+                        "api_base": azure.url,
+                        "api_key": "synthetic-azure-key",
+                    },
+                },
             ],
         )
         owned: Final = stack.enter_context(owned_proxy_process(gateway, directory, {}, config=config, workers=2))
@@ -232,6 +261,16 @@ def chaos_rig(
                     "guardrail_name": "audit-shield",
                     "litellm_params": _shield_params(azure, mode="pre_call", default_on=True),
                 },
+                {
+                    "guardrail_name": _TEXT_MODERATION,
+                    "litellm_params": {
+                        "guardrail": "azure/text_moderations",
+                        "mode": "pre_call",
+                        "default_on": False,
+                        "api_base": azure.url,
+                        "api_key": "synthetic-azure-key",
+                    },
+                },
             ],
         )
         owned: Final = stack.enter_context(owned_proxy_process(gateway, directory, {}, config=config, workers=2))
@@ -281,6 +320,14 @@ def _shield_prompts(requests: tuple[Request, ...]) -> tuple[JsonValue, ...]:
     )
 
 
+def _analyze_texts(requests: tuple[Request, ...]) -> tuple[JsonValue, ...]:
+    return tuple(
+        object_value(json.loads(scan.body))["text"]
+        for scan in requests
+        if scan.target.startswith(_ANALYZE_TARGET_PREFIX)
+    )
+
+
 def _guardrail_entries(model: str, count: int = 1) -> list[JsonValue]:
     rows: Final = eventually(
         lambda: read_rows('SELECT metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)),
@@ -355,6 +402,60 @@ def test_responses_streaming_input_is_scanned_and_billed(
         assert entry["guardrail_cost"] == pytest.approx(0.38 / 1000), entry
 
 
+@pytest.mark.parametrize(
+    "body",
+    [
+        pytest.param(lambda prompt: {"input": prompt}, id="string-input"),
+        pytest.param(
+            lambda prompt: {"input": [{"role": "user", "content": [{"type": "input_text", "text": prompt}]}]},
+            id="list-input",
+        ),
+        pytest.param(lambda prompt: {"messages": [], "input": prompt}, id="empty-messages-stub"),
+    ],
+)
+def test_text_moderation_opt_in_scans_responses_input(
+    audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event],
+    request: pytest.FixtureRequest,
+    body: Callable[[str], dict[str, JsonValue]],
+) -> None:
+    owned, azure, provider, _ = audit_rig
+    prompt: Final = f"synthetic benign prompt {request.node.callspec.id} {uuid.uuid4().hex}"
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="openai/gpt-4.1-mini", api_base=provider.url + "/v1", api_key="synthetic-provider-key"
+        )
+        response: Final = owned.gateway.request(
+            "POST", "/v1/responses", {"model": model, "guardrails": [_TEXT_MODERATION], **body(prompt)}
+        )
+        assert response.status_code == 200, response.text
+        calls: Final = azure.drain()
+        assert _analyze_texts(calls) == (prompt,)
+        assert _shield_prompts(calls) == (prompt,)
+        assert len(_provider_calls(provider)) == 1
+        entries: Final = _guardrail_entries(model, count=2)
+        assert {object_value(entry)["guardrail_name"] for entry in entries} == {"audit-shield", _TEXT_MODERATION}, (
+            entries
+        )
+
+
+def test_text_moderation_opt_in_scans_chat_messages(audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event]) -> None:
+    owned, azure, provider, _ = audit_rig
+    prompt: Final = "synthetic benign prompt chat-optin " + uuid.uuid4().hex
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="openai/gpt-4.1-mini", api_base=provider.url + "/v1", api_key="synthetic-provider-key"
+        )
+        response: Final = owned.gateway.request(
+            "POST",
+            "/v1/chat/completions",
+            {"model": model, "guardrails": [_TEXT_MODERATION], "messages": [{"role": "user", "content": prompt}]},
+        )
+        assert response.status_code == 200, response.text
+        calls: Final = azure.drain()
+        assert _analyze_texts(calls) == (prompt,)
+        assert _shield_prompts(calls) == (prompt,)
+
+
 def test_chat_with_input_key_still_scans_messages_only(
     audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event],
 ) -> None:
@@ -529,6 +630,45 @@ def test_streaming_responses_attack_is_blocked_before_any_stream_bytes(
         assert _provider_calls(provider) == ()
 
 
+def test_text_moderation_opt_in_blocks_responses_input_above_threshold(
+    audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event],
+) -> None:
+    owned, azure, provider, _ = audit_rig
+    prompt: Final = f"synthetic prompt {_MODERATION_MARKER} " + uuid.uuid4().hex
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="openai/gpt-4.1-mini", api_base=provider.url + "/v1", api_key="synthetic-provider-key"
+        )
+        response: Final = owned.gateway.request(
+            "POST", "/v1/responses", {"model": model, "guardrails": [_TEXT_MODERATION], "input": prompt}
+        )
+        assert response.status_code == 400, response.text
+        assert _analyze_texts(azure.drain()) == (prompt,)
+        assert _provider_calls(provider) == ()
+
+
+def test_text_moderation_opt_in_blocks_streamed_responses_input_above_threshold(
+    audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event],
+) -> None:
+    owned, azure, provider, _ = audit_rig
+    prompt: Final = f"synthetic prompt {_MODERATION_MARKER} " + uuid.uuid4().hex
+    with owned.gateway.scenario() as scenario:
+        model: Final = scenario.model(
+            model="openai/gpt-4.1-mini", api_base=provider.url + "/v1", api_key="synthetic-provider-key"
+        )
+        with owned.gateway.client.stream(
+            "POST",
+            "/v1/responses",
+            json={"model": model, "guardrails": [_TEXT_MODERATION], "input": prompt, "stream": True},
+            headers={"Authorization": f"Bearer {owned.gateway.key}"},
+        ) as response:
+            body: Final = response.read().decode()
+        assert response.status_code == 400, body
+        assert "Prompt Shield" not in body, body
+        assert _analyze_texts(azure.drain()) == (prompt,)
+        assert _provider_calls(provider) == ()
+
+
 def test_azure_outage_produces_the_same_outcome_on_responses_and_chat(
     audit_rig: tuple[OwnedProxy, Wire, Wire, threading.Event],
 ) -> None: