feat(guardrails): add aliyun security guardrail integration

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

@@ -11,7 +11,7 @@ This module provides integration with Aliyun's AI Security Guardrail service for
 Documentation: https://help.aliyun.com/document_detail/2873209.html
 """
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 
 from litellm.types.guardrails import SupportedGuardrailIntegrations
 
@@ -21,6 +21,19 @@ if TYPE_CHECKING:
     from litellm.types.guardrails import Guardrail, LitellmParams
 
 
+def _resolve_os_environ_reference(value: str | None) -> str | None:
+    """Resolve an ``os.environ/`` reference.
+
+    guardrail_registry.py only auto-resolves api_key/api_base, so the Aliyun
+    credential fields have to be resolved here.
+    """
+    from litellm.secret_managers.main import get_secret_str
+
+    if isinstance(value, str) and value.startswith("os.environ/"):
+        return get_secret_str(value)
+    return value
+
+
 def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> AliyunAIGuardrail:
     """
     Initialize an Aliyun AI Guardrail instance.
@@ -35,33 +48,27 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
         AliyunAIGuardrail instance
     """
     import litellm
-    from litellm.secret_managers.main import get_secret_str
 
-    guardrail_name = guardrail.get("guardrail_name")
+    guardrail_name: Final = guardrail.get("guardrail_name")
     if not guardrail_name:
         raise ValueError("Aliyun AI Guardrail: guardrail_name is required")
 
-    level = getattr(litellm_params, "level", None)
-    max_text_length = getattr(litellm_params, "max_text_length", None)
-    stream_window_size = getattr(litellm_params, "stream_window_size", None)
-    stream_slide_step = getattr(litellm_params, "stream_slide_step", None)
-    stream_first_check_step = getattr(litellm_params, "stream_first_check_step", None)
-    region_id = getattr(litellm_params, "region_id", None)
-    service_input = getattr(litellm_params, "service_input", None)
-    service_output = getattr(litellm_params, "service_output", None)
-    service_mcp = getattr(litellm_params, "service_mcp", None)
+    level: Final = getattr(litellm_params, "level", None)
+    max_text_length: Final = getattr(litellm_params, "max_text_length", None)
+    stream_window_size: Final = getattr(litellm_params, "stream_window_size", None)
+    stream_slide_step: Final = getattr(litellm_params, "stream_slide_step", None)
+    stream_first_check_step: Final = getattr(litellm_params, "stream_first_check_step", None)
+    region_id: Final = getattr(litellm_params, "region_id", None)
+    service_input: Final = getattr(litellm_params, "service_input", None)
+    service_output: Final = getattr(litellm_params, "service_output", None)
+    service_mcp: Final = getattr(litellm_params, "service_mcp", None)
 
-    # Get credentials from config. These custom fields are not auto-resolved by
-    # guardrail_registry.py (only api_key/api_base are), so resolve os.environ/
-    # references manually here.
-    access_key_id = getattr(litellm_params, "access_key_id", None)
-    access_key_secret = getattr(litellm_params, "access_key_secret", None)
-    if isinstance(access_key_id, str) and access_key_id.startswith("os.environ/"):
-        access_key_id = get_secret_str(access_key_id)
-    if isinstance(access_key_secret, str) and access_key_secret.startswith("os.environ/"):
-        access_key_secret = get_secret_str(access_key_secret)
+    # These custom credential fields are not auto-resolved by guardrail_registry.py
+    # (only api_key/api_base are), so os.environ/ references are resolved here.
+    access_key_id: Final = _resolve_os_environ_reference(getattr(litellm_params, "access_key_id", None))
+    access_key_secret: Final = _resolve_os_environ_reference(getattr(litellm_params, "access_key_secret", None))
 
-    aliyun_guardrail = AliyunAIGuardrail(
+    aliyun_guardrail: Final = AliyunAIGuardrail(
         guardrail_name=guardrail_name,
         access_key_id=access_key_id,
         access_key_secret=access_key_secret,
@@ -83,12 +90,14 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
     return aliyun_guardrail
 
 
-# Registry for guardrail initializers
-guardrail_initializer_registry = {
+# Registry for guardrail initializers.
+# Plain dicts: guardrail_registry.py gates discovery on `isinstance(registry, dict)`,
+# which a MappingProxyType would fail, silently skipping this guardrail's registration.
+guardrail_initializer_registry: Final = {  # mutable-ok: loader requires a real dict
     SupportedGuardrailIntegrations.ALIYUN_AI_GUARDRAIL.value: initialize_guardrail,
 }
 
 # Registry for guardrail classes
-guardrail_class_registry = {
+guardrail_class_registry: Final = {  # mutable-ok: loader requires a real dict
     SupportedGuardrailIntegrations.ALIYUN_AI_GUARDRAIL.value: AliyunAIGuardrail,
 }

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

@@ -5,7 +5,8 @@ Base class for Aliyun guardrails
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from collections.abc import Iterator, Sequence
+from typing import TYPE_CHECKING, Final
 
 if TYPE_CHECKING:
     from litellm.types.llms.openai import AllMessageValues
@@ -16,45 +17,69 @@ class AliyunGuardrailBase:
     Base class for Aliyun guardrails.
     """
 
-    def get_user_prompt(self, messages: list[AllMessageValues]) -> str | None:
+    @staticmethod
+    def _iter_user_messages(messages: Sequence[AllMessageValues]) -> Iterator[AllMessageValues]:
         """
-        Get the last consecutive block of messages from the user.
+        Yield every user message of the request, in order.
+        Restricting this to the trailing user block would let a caller hide a
+        prohibited turn behind an attacker-supplied assistant message.
+        """
+        return (message for message in messages if message.get("role") == "user")
+
+    @staticmethod
+    def _extract_image_url(part: object) -> str | None:
+        """
+        Return the URL of an ``image_url`` content part.
+        Args:
+            part: A single content part of a message
+        Returns:
+            The URL string, or None when the part carries no image URL
+        """
+        if not isinstance(part, dict) or part.get("type") != "image_url":
+            return None
+        image_url: Final = part.get("image_url")
+        if isinstance(image_url, dict):
+            url: Final = image_url.get("url")
+            return url if isinstance(url, str) else None
+        return image_url if isinstance(image_url, str) else None
+
+    def get_user_prompt(self, messages: Sequence[AllMessageValues]) -> str | None:
+        """
+        Collect the text of every user message in the request.
+        Scanning only the trailing user block would let a caller hide a
+        prohibited prompt behind an attacker-supplied assistant message, so all
+        user turns of the submitted request are audited.
         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?"
+        get_user_prompt(messages) -> "Hello, how are you?\nWhat is the weather in Tokyo?"
         """
         from litellm.litellm_core_utils.prompt_templates.common_utils import (
             convert_content_list_to_str,
         )
 
-        if not messages:
-            return None
-        # Iterate from the end to find the last consecutive block of user messages
-        user_messages = []
-        for message in reversed(messages):
-            if message.get("role") == "user":
-                user_messages.append(message)
-            else:
-                # Stop when we hit a non-user message
-                break
-        if not user_messages:
-            return None
-        # Reverse to get the messages in chronological order
-        user_messages.reverse()
-        user_prompt = ""
-        for message in user_messages:
-            text_content = convert_content_list_to_str(message)
-            user_prompt += text_content + "\n"
-        result = user_prompt.strip()
-        return result if result else None
+        user_prompt: Final = "\n".join(
+            convert_content_list_to_str(message) for message in self._iter_user_messages(messages)
+        ).strip()
+        return user_prompt or None
 
-    def get_image_urls(self, messages: list[AllMessageValues]) -> list[str]:
+    def _iter_public_image_urls(self, messages: Sequence[AllMessageValues]) -> Iterator[str]:
+        """Yield the publicly reachable image URLs of every user message, in order."""
+        for content in (message.get("content") for message in self._iter_user_messages(messages)):
+            if not isinstance(content, list):
+                continue
+            for url in (self._extract_image_url(part) for part in content):
+                # Only public http(s) URLs are reachable by the Aliyun API, so
+                # data: URIs and other inline payloads are skipped.
+                if url is not None and url.startswith(("http://", "https://")):
+                    yield url
+
+    def get_image_urls(self, messages: Sequence[AllMessageValues]) -> tuple[str, ...]:
         """
-        Extract image URLs from the last consecutive block of user messages.
+        Extract image URLs from every user message in the request.
         Only publicly accessible http(s) URLs are collected (in order,
         de-duplicated). Uses the same message range as ``get_user_prompt``.
         Example:
@@ -64,40 +89,8 @@ class AliyunGuardrailBase:
                 {"type": "image_url", "image_url": {"url": "https://a.com/x.png"}},
             ]},
         ]
-        get_image_urls(messages) -> ["https://a.com/x.png"]
+        get_image_urls(messages) -> ("https://a.com/x.png",)
         """
-        if not messages:
-            return []
-        # Iterate from the end to find the last consecutive block of user messages
-        user_messages = []
-        for message in reversed(messages):
-            if message.get("role") == "user":
-                user_messages.append(message)
-            else:
-                break
-        if not user_messages:
-            return []
-        user_messages.reverse()
-        image_urls: list[str] = []
-        seen = set()
-        for message in user_messages:
-            content = message.get("content")
-            if not isinstance(content, list):
-                continue
-            for part in content:
-                if not isinstance(part, dict) or part.get("type") != "image_url":
-                    continue
-                image_url = part.get("image_url")
-                url: str | None = None
-                if isinstance(image_url, dict):
-                    url = image_url.get("url")
-                elif isinstance(image_url, str):
-                    url = image_url
-                if not isinstance(url, str):
-                    continue
-                if not (url.startswith("http://") or url.startswith("https://")):
-                    continue
-                if url not in seen:
-                    seen.add(url)
-                    image_urls.append(url)
-        return image_urls
+        # dict.fromkeys is the order-preserving dedup; it is transient and the
+        # result is frozen into a tuple before it leaves this method.
+        return tuple(dict.fromkeys(self._iter_public_image_urls(messages)))  # mutable-ok: transient dedup, frozen here

litellm/types/guardrails.py

@@ -1177,4 +1177,3 @@ class PatchGuardrailRequest(BaseModel):
     guardrail_name: str | None = None
     litellm_params: BaseLitellmParams | None = None
     guardrail_info: dict[str, Any] | None = None
-

litellm/types/proxy/guardrails/guardrail_hooks/aliyun/aliyun_ai_guardrail.py

@@ -8,10 +8,11 @@ Aliyun AI Guardrail supports the following detection types:
 - maliciousUrl: Malicious URL detection
 """
 
-from typing import Any, Dict, List, Literal, Optional
+from collections.abc import Sequence
+from typing import Literal, TypeAlias
 
 from pydantic import BaseModel, Field
-from typing_extensions import NotRequired, TypedDict
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 from ..base import GuardrailConfigModel
 
@@ -20,77 +21,79 @@ from ..base import GuardrailConfigModel
 class AliyunAIGuardrailResponseDetailResultExt(TypedDict, total=False):
     """Extended information in result"""
 
-    Desensitization: Optional[str]  # Desensitized text when action is mask
+    Desensitization: ReadOnly[str | None]  # Desensitized text when action is mask
 
 
 class AliyunAIGuardrailResponseDetailResult(TypedDict, total=False):
     """Result item in detail"""
 
-    Confidence: Optional[float]
-    Label: Optional[str]
-    Ext: Optional[AliyunAIGuardrailResponseDetailResultExt]
+    Confidence: ReadOnly[float | None]
+    Label: ReadOnly[str | None]
+    Ext: ReadOnly[AliyunAIGuardrailResponseDetailResultExt | None]
     # Per-result risk level. This is the shape documented for MultiModalGuard; the
     # ``_pro`` service codes report the severity on the parent Detail as ``Level``
     # instead, so both have to be honoured when deciding whether to block.
-    RiskLevel: Optional[str]
+    RiskLevel: ReadOnly[str | None]
 
 
 class AliyunAIGuardrailResponseDetail(TypedDict):
     """Detail item in response data"""
 
-    Type: str  # contentModeration, sensitiveData, promptAttack, maliciousUrl
-    Suggestion: str  # pass, block, mask
-    Result: List[AliyunAIGuardrailResponseDetailResult]
+    Type: ReadOnly[str]  # contentModeration, sensitiveData, promptAttack, maliciousUrl
+    Suggestion: ReadOnly[str]  # pass, block, mask
+    Result: ReadOnly[Sequence[AliyunAIGuardrailResponseDetailResult]]
     # Risk level as returned by the ``_pro`` service codes (none/low/medium/high, or
     # S0-S4 for sensitiveData). Absent in the documented response shape, which carries
     # the severity as Result[].RiskLevel.
-    Level: NotRequired[str]
+    Level: ReadOnly[NotRequired[str]]
 
 
 class AliyunAIGuardrailResponseData(TypedDict, total=False):
     """Response data from Aliyun AI Guardrail API"""
 
-    Suggestion: str  # Overall suggestion: pass, block, mask
-    Detail: Optional[List[AliyunAIGuardrailResponseDetail]]
+    Suggestion: ReadOnly[str]  # Overall suggestion: pass, block, mask
+    Detail: ReadOnly[Sequence[AliyunAIGuardrailResponseDetail] | None]
 
 
 class AliyunAIGuardrailResponse(TypedDict):
     """Response from Aliyun AI Guardrail API"""
 
-    RequestId: str
-    Code: int
-    Message: Optional[str]
-    Data: Optional[AliyunAIGuardrailResponseData]
+    RequestId: ReadOnly[str]
+    Code: ReadOnly[int]
+    Message: ReadOnly[str | None]
+    Data: ReadOnly[AliyunAIGuardrailResponseData | None]
 
 
 # Suggestion type
-AliyunAIGuardrailSuggestion = Literal["pass", "block", "watch"]
+AliyunAIGuardrailSuggestion: TypeAlias = Literal["pass", "block", "watch"]
 
 # Detection type
-AliyunAIGuardrailDetectionType = Literal["contentModeration", "sensitiveData", "promptAttack", "maliciousUrl"]
+AliyunAIGuardrailDetectionType: TypeAlias = Literal[
+    "contentModeration", "sensitiveData", "promptAttack", "maliciousUrl"
+]
 
 
 class AliyunAIGuardrailRequestParams(TypedDict, total=False):
     """Request parameters for Aliyun AI Guardrail API"""
 
-    Action: str
-    Version: str
-    AccessKeyId: str
-    Timestamp: str
-    SignatureMethod: str
-    SignatureVersion: str
-    SignatureNonce: str
-    Format: str
-    Service: str
-    ServiceParameters: str
-    Signature: str
+    Action: ReadOnly[str]
+    Version: ReadOnly[str]
+    AccessKeyId: ReadOnly[str]
+    Timestamp: ReadOnly[str]
+    SignatureMethod: ReadOnly[str]
+    SignatureVersion: ReadOnly[str]
+    SignatureNonce: ReadOnly[str]
+    Format: ReadOnly[str]
+    Service: ReadOnly[str]
+    ServiceParameters: ReadOnly[str]
+    Signature: ReadOnly[str]
 
 
 # Risk level literals
-AliyunRiskLevel = Literal["none", "low", "medium", "high"]
+AliyunRiskLevel: TypeAlias = Literal["none", "low", "medium", "high"]
 
 # Protection level literals
-AliyunProtectionLevel = Literal["low", "medium", "high", "max"]
+AliyunProtectionLevel: TypeAlias = Literal["low", "medium", "high", "max"]
 
 
 # Configuration models
@@ -101,39 +104,39 @@ class AliyunAIGuardrailOptionalParams(BaseModel):
     in config.yaml on the AliyunAIGuardrailConfigModel and support os.environ/ references.
     """
 
-    level: Optional[AliyunProtectionLevel] = Field(
+    level: AliyunProtectionLevel | None = Field(
         default="medium",
         description="Protection level for risk filtering. 'low': block all risks (high protection), 'medium': block medium and high risks, 'high': block only high risks (low protection), 'max': observation mode (no blocking). Default: medium",
     )
-    max_text_length: Optional[int] = Field(
+    max_text_length: int | None = Field(
         default=2000,
         description="Maximum text length for a single API call. Text longer than this will be split.",
     )
-    stream_window_size: Optional[int] = Field(
+    stream_window_size: int | None = Field(
         default=500,
         description="Sliding window size (in chars) for streaming output guardrail checks. Each check sends the most recent N chars to the API.",
     )
-    stream_slide_step: Optional[int] = Field(
+    stream_slide_step: int | None = Field(
         default=300,
         description="Sliding step (in chars) for streaming output guardrail checks. A check is triggered every time N new chars accumulate since the last check.",
     )
-    stream_first_check_step: Optional[int] = Field(
+    stream_first_check_step: int | None = Field(
         default=50,
         description="First check threshold (in chars) for streaming output. The first guardrail check triggers earlier (at N chars) to reduce first-token latency, subsequent checks use stream_slide_step.",
     )
-    region_id: Optional[str] = Field(
+    region_id: str | None = Field(
         default="cn-shanghai",
         description="Aliyun region ID. Default: cn-shanghai",
     )
-    service_input: Optional[str] = Field(
+    service_input: str | None = Field(
         default="query_security_check_pro",
         description="Service code for input (pre-call) detection. Default: query_security_check_pro",
     )
-    service_output: Optional[str] = Field(
+    service_output: str | None = Field(
         default="response_security_check_pro",
         description="Service code for output (post-call) detection. Default: response_security_check_pro",
     )
-    service_mcp: Optional[str] = Field(
+    service_mcp: str | None = Field(
         default="query_security_check_pro",
         description="Service code for MCP tool call detection (pre_mcp_call and post_mcp_call). Default: query_security_check_pro",
     )
@@ -147,15 +150,15 @@ class AliyunAIGuardrailConfigModel(GuardrailConfigModel[AliyunAIGuardrailOptiona
     - access_key_secret: Aliyun Access Key Secret
     """
 
-    access_key_id: Optional[str] = Field(
+    access_key_id: str | None = Field(
         default=None,
         description="Aliyun Access Key ID. Configure in config.yaml, supports os.environ/ reference",
     )
-    access_key_secret: Optional[str] = Field(
+    access_key_secret: str | None = Field(
         default=None,
         description="Aliyun Access Key Secret. Configure in config.yaml, supports os.environ/ reference",
     )
-    optional_params: AliyunAIGuardrailOptionalParams = Field(
+    optional_params: AliyunAIGuardrailOptionalParams | None = Field(
         default_factory=AliyunAIGuardrailOptionalParams,
         description="Optional parameters for the Aliyun AI Guardrail",
     )

tests/test_litellm/proxy/guardrails/guardrail_hooks/aliyun/test_aliyun_ai_guardrail.py

@@ -295,7 +295,7 @@ class TestSplitText:
     def test_short_text_returns_single_segment(self):
         g = _make_guardrail()
         result = g._split_text("short text", max_length=100)
-        assert result == ["short text"]
+        assert result == ("short text",)
 
     def test_long_text_splits_at_sentence_boundary(self):
         g = _make_guardrail()
@@ -311,10 +311,10 @@ class TestSplitText:
         assert len(result) >= 3
         assert "".join(result) == text
 
-    def test_empty_text_returns_empty_list(self):
+    def test_empty_text_returns_no_segments(self):
         g = _make_guardrail()
         result = g._split_text("", max_length=100)
-        assert result == []
+        assert result == ()
 
 
 # ---------------------------------------------------------------------------
@@ -591,6 +591,62 @@ class TestConfigModel:
 # ---------------------------------------------------------------------------
 
 
+# ---------------------------------------------------------------------------
+# Trailing non-user message must not hide the prompt from scanning
+# ---------------------------------------------------------------------------
+
+
+class TestTrailingAssistantMessage:
+    def test_user_text_survives_trailing_assistant_message(self):
+        g = _make_guardrail()
+        messages = [
+            {"role": "user", "content": "违规的用户提问"},
+            {"role": "assistant", "content": "攻击者伪造的回复"},
+        ]
+        prompt = g.get_user_prompt(messages)
+        assert prompt is not None
+        assert "违规的用户提问" in prompt
+
+    def test_images_survive_trailing_assistant_message(self):
+        g = _make_guardrail()
+        messages = [
+            {
+                "role": "user",
+                "content": [{"type": "image_url", "image_url": {"url": IMG_A}}],
+            },
+            {"role": "assistant", "content": "攻击者伪造的回复"},
+        ]
+        assert g.get_image_urls(messages) == (IMG_A,)
+
+    @pytest.mark.asyncio
+    async def test_blocks_violation_despite_trailing_assistant_message(self):
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        blocked = _make_aliyun_api_response(
+            suggestion="block",
+            detail=[_make_detail(detection_type=CONTENT_MODERATION_TYPE, level="high")],
+        )
+
+        async def block_only_violating_text(*args, **kwargs):
+            scanned = json.loads(kwargs["data"]["ServiceParameters"]).get("content", "")
+            return blocked if "违规的用户提问" in scanned else clean
+
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, side_effect=block_only_violating_text):
+            with pytest.raises(HTTPException) as exc_info:
+                await g.async_pre_call_hook(
+                    user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                    cache=MagicMock(),
+                    data={
+                        "messages": [
+                            {"role": "user", "content": "违规的用户提问"},
+                            {"role": "assistant", "content": "攻击者伪造的回复"},
+                        ]
+                    },
+                    call_type="completion",
+                )
+            assert exc_info.value.status_code == 400
+
+
 class TestGetImageUrls:
     def test_extracts_http_and_https_urls(self):
         g = _make_guardrail()
@@ -604,7 +660,7 @@ class TestGetImageUrls:
                 ],
             }
         ]
-        assert g.get_image_urls(messages) == [IMG_A, IMG_B]
+        assert g.get_image_urls(messages) == (IMG_A, IMG_B)
 
     def test_skips_non_url_images(self):
         g = _make_guardrail()
@@ -618,12 +674,12 @@ class TestGetImageUrls:
                 ],
             }
         ]
-        assert g.get_image_urls(messages) == [IMG_A]
+        assert g.get_image_urls(messages) == (IMG_A,)
 
     def test_plain_text_returns_empty(self):
         g = _make_guardrail()
         messages = [{"role": "user", "content": "just text"}]
-        assert g.get_image_urls(messages) == []
+        assert g.get_image_urls(messages) == ()
 
     def test_deduplicates_across_messages(self):
         g = _make_guardrail()
@@ -631,20 +687,20 @@ class TestGetImageUrls:
             {"role": "user", "content": [{"type": "image_url", "image_url": {"url": IMG_A}}]},
             {"role": "user", "content": [{"type": "image_url", "image_url": {"url": IMG_A}}]},
         ]
-        assert g.get_image_urls(messages) == [IMG_A]
+        assert g.get_image_urls(messages) == (IMG_A,)
 
-    def test_only_last_consecutive_user_block(self):
+    def test_collects_images_from_every_user_message(self):
         g = _make_guardrail()
         messages = [
             {"role": "user", "content": [{"type": "image_url", "image_url": {"url": IMG_B}}]},
             {"role": "assistant", "content": "ok"},
             {"role": "user", "content": [{"type": "image_url", "image_url": {"url": IMG_A}}]},
         ]
-        assert g.get_image_urls(messages) == [IMG_A]
+        assert g.get_image_urls(messages) == (IMG_B, IMG_A)
 
     def test_empty_messages_returns_empty(self):
         g = _make_guardrail()
-        assert g.get_image_urls([]) == []
+        assert g.get_image_urls([]) == ()
 
 
 # ---------------------------------------------------------------------------
@@ -693,6 +749,67 @@ class TestServiceParametersConstruction:
 
 
 class TestPreCallHook:
+    @pytest.mark.asyncio
+    async def test_scans_responses_api_string_input(self):
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=clean) as mock_post:
+            await g.async_pre_call_hook(
+                user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                cache=MagicMock(),
+                data={"input": "违规的 responses 输入"},
+                call_type="responses",
+            )
+
+        scanned = "".join(
+            json.loads(call.kwargs["data"]["ServiceParameters"]).get("content", "") for call in mock_post.call_args_list
+        )
+        assert "违规的 responses 输入" in scanned
+
+    @pytest.mark.asyncio
+    async def test_blocks_violating_responses_api_input(self):
+        g = _make_guardrail(level="medium")
+        blocked = _make_aliyun_api_response(
+            suggestion="block",
+            detail=[_make_detail(detection_type=CONTENT_MODERATION_TYPE, level="high")],
+        )
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=blocked):
+            with pytest.raises(HTTPException) as exc_info:
+                await g.async_pre_call_hook(
+                    user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                    cache=MagicMock(),
+                    data={"input": "违规的 responses 输入"},
+                    call_type="responses",
+                )
+            assert exc_info.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_scans_responses_api_structured_input_with_image(self):
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        data = {
+            "input": [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "结构化输入文本"},
+                        {"type": "image_url", "image_url": {"url": IMG_A}},
+                    ],
+                }
+            ]
+        }
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=clean) as mock_post:
+            await g.async_pre_call_hook(
+                user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                cache=MagicMock(),
+                data=data,
+                call_type="responses",
+            )
+
+        sent = [json.loads(call.kwargs["data"]["ServiceParameters"]) for call in mock_post.call_args_list]
+        assert any("结构化输入文本" in (sp.get("content") or "") for sp in sent)
+        assert any(IMG_A in (sp.get("imageUrls") or []) for sp in sent)
+
     @pytest.mark.asyncio
     async def test_blocks_violation(self):
         g = _make_guardrail(level="medium")
@@ -982,6 +1099,161 @@ class TestPostCallHook:
         assert result is response
 
 
+def _make_tool_call_response(arguments: str, name: str = "send_email"):
+    """Build a non-streaming response whose only output is a tool call."""
+    import litellm
+    from litellm.types.utils import ChatCompletionMessageToolCall, Function
+
+    return litellm.ModelResponse(
+        id="test-id",
+        choices=[
+            litellm.Choices(
+                index=0,
+                message=litellm.Message(
+                    role="assistant",
+                    content=None,
+                    tool_calls=[
+                        ChatCompletionMessageToolCall(
+                            id="call_1",
+                            type="function",
+                            function=Function(name=name, arguments=arguments),
+                        )
+                    ],
+                ),
+            )
+        ],
+    )
+
+
+def _make_responses_api_response(text: str):
+    """Build a non-streaming /v1/responses body carrying assistant text."""
+    from litellm.types.llms.openai import ResponsesAPIResponse
+    from litellm.types.responses.main import GenericResponseOutputItem, OutputText
+
+    return ResponsesAPIResponse(
+        id="resp_1",
+        created_at=0,
+        model="gpt-4o",
+        object="response",
+        output=[
+            GenericResponseOutputItem(
+                type="message",
+                id="m1",
+                status="completed",
+                role="assistant",
+                content=[OutputText(type="output_text", text=text, annotations=[])],
+            )
+        ],
+        parallel_tool_calls=False,
+        tool_choice="auto",
+        tools=[],
+        temperature=1.0,
+        top_p=1.0,
+    )
+
+
+class TestPostCallStructuredFields:
+    """The streaming path already audits tool calls, reasoning text and
+    /v1/responses output. Auditing only ``message.content`` here would let the
+    very same content reach the client unchecked whenever stream=False."""
+
+    @pytest.mark.asyncio
+    async def test_scans_tool_call_arguments(self):
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        response = _make_tool_call_response('{"body": "违规的工具参数"}')
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=clean) as mock_post:
+            await g.async_post_call_success_hook(
+                data={"messages": [{"role": "user", "content": "hi"}]},
+                user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                response=response,
+            )
+        scanned = "".join(
+            json.loads(call.kwargs["data"]["ServiceParameters"])["content"] for call in mock_post.call_args_list
+        )
+        assert "违规的工具参数" in scanned
+
+    @pytest.mark.asyncio
+    async def test_blocks_violation_in_tool_call_arguments(self):
+        g = _make_guardrail(level="medium")
+        blocked = _make_aliyun_api_response(
+            suggestion="block",
+            detail=[_make_detail(detection_type=CONTENT_MODERATION_TYPE, level="high")],
+        )
+        response = _make_tool_call_response('{"body": "违规的工具参数"}')
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=blocked):
+            with pytest.raises(HTTPException) as exc_info:
+                await g.async_post_call_success_hook(
+                    data={"messages": [{"role": "user", "content": "hi"}]},
+                    user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                    response=response,
+                )
+            assert exc_info.value.status_code == 400
+
+    @pytest.mark.asyncio
+    async def test_scans_reasoning_content(self):
+        import litellm
+
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        response = litellm.ModelResponse(
+            id="test-id",
+            choices=[
+                litellm.Choices(
+                    index=0,
+                    message=litellm.Message(
+                        role="assistant",
+                        content="正常的回复内容",
+                        reasoning_content="推理过程里的违规内容",
+                    ),
+                )
+            ],
+        )
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=clean) as mock_post:
+            await g.async_post_call_success_hook(
+                data={"messages": [{"role": "user", "content": "hi"}]},
+                user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                response=response,
+            )
+        scanned = "".join(
+            json.loads(call.kwargs["data"]["ServiceParameters"])["content"] for call in mock_post.call_args_list
+        )
+        assert "推理过程里的违规内容" in scanned
+
+    @pytest.mark.asyncio
+    async def test_scans_responses_api_output(self):
+        g = _make_guardrail(level="medium")
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        response = _make_responses_api_response("响应体里的违规内容")
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=clean) as mock_post:
+            await g.async_post_call_success_hook(
+                data={"input": "hi"},
+                user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                response=response,
+            )
+        scanned = "".join(
+            json.loads(call.kwargs["data"]["ServiceParameters"])["content"] for call in mock_post.call_args_list
+        )
+        assert "响应体里的违规内容" in scanned
+
+    @pytest.mark.asyncio
+    async def test_blocks_violation_in_responses_api_output(self):
+        g = _make_guardrail(level="medium")
+        blocked = _make_aliyun_api_response(
+            suggestion="block",
+            detail=[_make_detail(detection_type=CONTENT_MODERATION_TYPE, level="high")],
+        )
+        response = _make_responses_api_response("响应体里的违规内容")
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, return_value=blocked):
+            with pytest.raises(HTTPException) as exc_info:
+                await g.async_post_call_success_hook(
+                    data={"input": "hi"},
+                    user_api_key_dict=UserAPIKeyAuth(api_key="test"),
+                    response=response,
+                )
+            assert exc_info.value.status_code == 400
+
+
 # ---------------------------------------------------------------------------
 # Post-MCP hook tests
 # ---------------------------------------------------------------------------
@@ -993,6 +1265,93 @@ def _make_call_tool_result(text: str = "tool output"):
     return CallToolResult(content=[TextContent(type="text", text=text)], isError=False)
 
 
+class TestExtractMcpToolText:
+    """A tool result carries text outside of ``content[].text``. Auditing only that
+    field would release structured payloads and embedded resources unchecked."""
+
+    def test_collects_structured_content(self):
+        from mcp.types import CallToolResult
+
+        g = _make_guardrail()
+        result = CallToolResult(content=[], structuredContent={"note": "结构化字段里的违规内容"}, isError=False)
+        assert "结构化字段里的违规内容" in g._extract_mcp_tool_text(result)
+
+    def test_collects_structured_content_alongside_text(self):
+        from mcp.types import CallToolResult, TextContent
+
+        g = _make_guardrail()
+        result = CallToolResult(
+            content=[TextContent(type="text", text="正常的工具输出")],
+            structuredContent={"note": "结构化字段里的违规内容"},
+            isError=False,
+        )
+        extracted = g._extract_mcp_tool_text(result)
+        assert "正常的工具输出" in extracted
+        assert "结构化字段里的违规内容" in extracted
+
+    def test_collects_embedded_resource_text(self):
+        from mcp.types import CallToolResult, EmbeddedResource, TextResourceContents
+
+        g = _make_guardrail()
+        result = CallToolResult(
+            content=[
+                EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="file:///tmp/note.txt",
+                        mimeType="text/plain",
+                        text="内嵌资源里的违规内容",
+                    ),
+                )
+            ],
+            isError=False,
+        )
+        assert "内嵌资源里的违规内容" in g._extract_mcp_tool_text(result)
+
+    def test_collects_structured_content_from_dict_payload(self):
+        g = _make_guardrail()
+        payload = {"content": [], "structuredContent": {"note": "结构化字段里的违规内容"}}
+        assert "结构化字段里的违规内容" in g._extract_mcp_tool_text(payload)
+
+    def test_collects_structured_content_from_coerced_tuple_list(self):
+        """MCPPostCallResponseObject coerces a CallToolResult into (field, value) pairs."""
+        g = _make_guardrail()
+        payload = [("content", []), ("structuredContent", {"note": "结构化字段里的违规内容"}), ("isError", False)]
+        assert "结构化字段里的违规内容" in g._extract_mcp_tool_text(payload)
+
+    @pytest.mark.asyncio
+    async def test_blocks_violation_in_structured_content(self):
+        from mcp.types import CallToolResult, TextContent
+
+        g = _make_guardrail(level="medium")
+        tool_result = CallToolResult(
+            content=[TextContent(type="text", text="正常的工具输出")],
+            structuredContent={"note": "结构化字段里的违规内容"},
+            isError=False,
+        )
+        kwargs, response_obj = _make_post_mcp_hook_args(tool_result)
+        clean = _make_aliyun_api_response(suggestion="pass", detail=[])
+        blocked = _make_aliyun_api_response(
+            suggestion="block",
+            detail=[_make_detail(detection_type=CONTENT_MODERATION_TYPE, level="high")],
+        )
+
+        async def block_only_structured_content(*args, **kwargs):
+            scanned = json.loads(kwargs["data"]["ServiceParameters"]).get("content", "")
+            return blocked if "结构化字段里的违规内容" in scanned else clean
+
+        with patch.object(g.async_handler, "post", new_callable=AsyncMock, side_effect=block_only_structured_content):
+            await g.async_post_mcp_tool_call_hook(
+                kwargs=kwargs,
+                response_obj=response_obj,
+                start_time=datetime.now(),
+                end_time=datetime.now(),
+            )
+        remaining = " ".join(getattr(item, "text", "") for item in tool_result.content)
+        assert CONTENT_MODERATION_TYPE in remaining
+        assert tool_result.isError is True
+
+
 # ---------------------------------------------------------------------------
 # Streaming hook tests
 # ---------------------------------------------------------------------------
@@ -1162,35 +1521,35 @@ class TestShouldRunPostMcpCall:
 class TestIterMcpContentItems:
     def test_plain_string_is_wrapped(self):
         g = _make_guardrail()
-        assert g._iter_mcp_content_items("hello") == ["hello"]
+        assert g._iter_mcp_content_items("hello") == ("hello",)
 
     def test_object_with_content_list(self):
         g = _make_guardrail()
         payload = MagicMock()
         payload.content = ["a", "b"]
-        assert g._iter_mcp_content_items(payload) == ["a", "b"]
+        assert g._iter_mcp_content_items(payload) == ("a", "b")
 
     def test_dict_with_content_list(self):
         g = _make_guardrail()
-        assert g._iter_mcp_content_items({"content": ["a"]}) == ["a"]
+        assert g._iter_mcp_content_items({"content": ["a"]}) == ("a",)
 
     def test_dict_without_content_returns_itself(self):
         g = _make_guardrail()
         payload = {"text": "no content key"}
-        assert g._iter_mcp_content_items(payload) == [payload]
+        assert g._iter_mcp_content_items(payload) == (payload,)
 
     def test_coerced_tuple_pairs_recover_real_content(self):
         g = _make_guardrail()
         payload = [("meta", None), ("content", ["real"]), ("isError", False)]
-        assert g._iter_mcp_content_items(payload) == ["real"]
+        assert g._iter_mcp_content_items(payload) == ("real",)
 
     def test_plain_list_passes_through(self):
         g = _make_guardrail()
-        assert g._iter_mcp_content_items(["a", "b"]) == ["a", "b"]
+        assert g._iter_mcp_content_items(["a", "b"]) == ("a", "b")
 
     def test_unsupported_payload_returns_empty(self):
         g = _make_guardrail()
-        assert g._iter_mcp_content_items(123) == []
+        assert g._iter_mcp_content_items(123) == ()
 
 
 class TestReplaceToolOutputInPlace: