feat(guardrails): add aliyun security guardrail integration

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

@@ -0,0 +1,94 @@
+"""
+Aliyun AI Security Guardrail Integration for LiteLLM
+阿里云AI安全护栏集成
+This module provides integration with Aliyun's AI Security Guardrail service for:
+- ContentModeration 内容合规检测
+- PromptAttack 提示词攻击检测
+- SensitiveData 敏感内容检测
+- ModelHallucination 模型幻觉
+- MaliciousUrl 恶意URL检测
+...
+Documentation: https://help.aliyun.com/document_detail/2873209.html
+"""
+
+from typing import TYPE_CHECKING
+
+from litellm.types.guardrails import SupportedGuardrailIntegrations
+
+from .aliyun_ai_guardrail import AliyunAIGuardrail
+
+if TYPE_CHECKING:
+    from litellm.types.guardrails import Guardrail, LitellmParams
+
+
+def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail") -> AliyunAIGuardrail:
+    """
+    Initialize an Aliyun AI Guardrail instance.
+    Credentials are configured in config.yaml (litellm_params) and support
+    os.environ/ references:
+    - access_key_id: Aliyun Access Key ID
+    - access_key_secret: Aliyun Access Key Secret
+    Args:
+        litellm_params: The LiteLLM parameters for the guardrail
+        guardrail: The guardrail configuration
+    Returns:
+        AliyunAIGuardrail instance
+    """
+    import litellm
+    from litellm.secret_managers.main import get_secret_str
+
+    guardrail_name = 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)
+
+    # 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)
+
+    aliyun_guardrail = AliyunAIGuardrail(
+        guardrail_name=guardrail_name,
+        access_key_id=access_key_id,
+        access_key_secret=access_key_secret,
+        level=level,
+        max_text_length=max_text_length,
+        stream_window_size=stream_window_size,
+        stream_slide_step=stream_slide_step,
+        stream_first_check_step=stream_first_check_step,
+        region_id=region_id,
+        service_input=service_input,
+        service_output=service_output,
+        service_mcp=service_mcp,
+        default_on=litellm_params.default_on,
+        event_hook=litellm_params.mode,
+    )
+
+    litellm.logging_callback_manager.add_litellm_callback(aliyun_guardrail)
+
+    return aliyun_guardrail
+
+
+# Registry for guardrail initializers
+guardrail_initializer_registry = {
+    SupportedGuardrailIntegrations.ALIYUN_AI_GUARDRAIL.value: initialize_guardrail,
+}
+
+# Registry for guardrail classes
+guardrail_class_registry = {
+    SupportedGuardrailIntegrations.ALIYUN_AI_GUARDRAIL.value: AliyunAIGuardrail,
+}

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

@@ -0,0 +1,959 @@
+#!/usr/bin/env python3
+"""
+Aliyun AI Security Guardrail Integration for LiteLLM
+阿里云AI安全护栏集成
+This guardrail scans prompts and responses using the Aliyun AI Security Guardrail API to detect:
+- Content moderation violations
+- Sensitive data (PII)
+- Prompt injection attacks
+- Malicious URLs
+Documentation: https://help.aliyun.com/document_detail/2875413.html
+Credentials:
+Configured in config.yaml (litellm_params), support os.environ/ references:
+- access_key_id: Aliyun Access Key ID
+- access_key_secret: Aliyun Access Key Secret
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import re
+import uuid
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Literal
+from urllib.parse import quote
+
+from fastapi import HTTPException
+
+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 .base import AliyunGuardrailBase
+
+if TYPE_CHECKING:
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.types.llms.openai import AllMessageValues
+    from litellm.types.mcp import MCPPostCallResponseObject
+    from litellm.types.proxy.guardrails.guardrail_hooks.aliyun.aliyun_ai_guardrail import (
+        AliyunAIGuardrailResponse,
+    )
+    from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
+    from litellm.types.utils import EmbeddingResponse, ImageResponse, ModelResponse
+
+# Constants
+ENCODING = "UTF-8"
+ISO8601_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+ALGORITHM = "HmacSHA1"
+
+# Region to endpoint mapping
+REGION_ENDPOINTS = {
+    "cn-shanghai": "green-cip.cn-shanghai.aliyuncs.com",
+    "cn-beijing": "green-cip.cn-beijing.aliyuncs.com",
+    "cn-hangzhou": "green-cip.cn-hangzhou.aliyuncs.com",
+    "cn-shenzhen": "green-cip.cn-shenzhen.aliyuncs.com",
+    "cn-chengdu": "green-cip.cn-chengdu.aliyuncs.com",
+    "ap-southeast-1": "green-cip.ap-southeast-1.aliyuncs.com",
+    "eu-central-1": "green-cip.eu-central-1.aliyuncs.com",
+}
+
+# Service codes for domestic (China) regions
+SERVICE_INPUT_DOMESTIC = "query_security_check_pro"
+SERVICE_OUTPUT_DOMESTIC = "response_security_check_pro"
+
+# Service codes for international regions
+SERVICE_INPUT_INTERNATIONAL = "query_security_check_cb"
+SERVICE_OUTPUT_INTERNATIONAL = "response_security_check_cb"
+
+
+# Detection types
+CONTENT_MODERATION_TYPE = "contentModeration"
+PROMPT_ATTACK_TYPE = "promptAttack"
+SENSITIVE_DATA_TYPE = "sensitiveData"
+MALICIOUS_URL_TYPE = "maliciousUrl"
+MODEL_HALLUCINATION_TYPE = "modelHallucination"
+CUSTOM_LABEL_TYPE = "customLabel"
+
+# Suggestion returned by Aliyun when it has decided the content must be rejected
+BLOCK_SUGGESTION = "block"
+
+# An explicit upstream block that carries no parseable severity is treated as the most
+# severe level, so it is still weighed against the configured threshold rather than
+# being silently downgraded to "none".
+UNRESOLVED_BLOCK_LEVEL = "high"
+
+
+def level_to_int(risk_level: str) -> int:
+    """
+    Convert risk level string to integer for comparison.
+    Higher value = higher risk.
+    Supports both standard risk levels (none/low/medium/high)
+    and sensitive data levels (S0/S1/S2/S3/S4).
+    Mapping:
+    - none/S0 = 0 (no risk)
+    - low/S1 = 1 (low risk)
+    - medium/S2 = 2 (medium risk)
+    - high/S3/S4 = 3 (high risk)
+    """
+    level_lower = risk_level.lower() if risk_level else "none"
+    level_map = {
+        # Standard risk levels
+        "none": 0,
+        "low": 1,
+        "medium": 2,
+        "high": 3,
+        # Sensitive data levels (mapped to standard levels)
+        "s0": 0,  # No risk
+        "s1": 1,  # Low risk
+        "s2": 2,  # Medium risk
+        "s3": 3,  # High risk
+        "s4": 3,  # High risk (highest sensitive level)
+    }
+    return level_map.get(level_lower, 0)
+
+
+# Protection level thresholds
+# If detected_level >= threshold, then block
+PROTECTION_LEVEL_THRESHOLD = {
+    "low": 1,  # High protection: block low, medium, high (threshold=1, block if >=1)
+    "medium": 2,  # Medium protection: block medium, high (threshold=2, block if >=2)
+    "high": 3,  # Low protection: block high only (threshold=3, block if >=3)
+    "max": 99,  # Observation mode: never block (threshold very high)
+}
+
+
+class AliyunAIGuardrail(AliyunGuardrailBase, CustomGuardrail):
+    """
+    LiteLLM Built-in Guardrail for Aliyun AI Security Guardrail.
+    This guardrail scans prompts and responses using the Aliyun AI Security Guardrail API to detect
+    malicious content, injection attempts, sensitive data, and policy violations.
+    Configuration:
+        guardrail_name: Name of the guardrail instance
+        access_key_id: Aliyun Access Key ID
+        access_key_secret: Aliyun Access Key Secret
+        region_id: Aliyun region ID (default: cn-shanghai)
+        default_on: Whether to enable by default
+    """
+
+    def __init__(
+        self,
+        guardrail_name: str,
+        access_key_id: str | None = None,
+        access_key_secret: str | None = None,
+        region_id: str | None = None,
+        level: str | None = None,
+        max_text_length: int | None = None,
+        stream_window_size: int | None = None,
+        stream_slide_step: int | None = None,
+        stream_first_check_step: int | None = None,
+        service_input: str | None = None,
+        service_output: str | None = None,
+        service_mcp: str | None = None,
+        **kwargs,
+    ):
+        """
+        Initialize Aliyun AI Guardrail handler.
+        Credentials (access_key_id / access_key_secret) are passed in from config.yaml
+        via the guardrail loader.
+        Args:
+            region_id: Aliyun region ID (default: cn-shanghai)
+            level: Protection level for risk filtering
+                - "low": High protection, block all risks (low, medium, high, S1+)
+                - "medium": Medium protection, block medium and high risks (medium, high, S2+)
+                - "high": Low protection, block only high risks (high, S3+)
+                - "max": Observation mode, no blocking
+            service_input: Service code for input detection (default: query_security_check_pro)
+            service_output: Service code for output detection (default: response_security_check_pro)
+            service_mcp: Service code for MCP tool call detection, used by both
+                pre_mcp_call and post_mcp_call (default: query_security_check_pro)
+        """
+        super().__init__(
+            guardrail_name=guardrail_name,
+            **kwargs,
+        )
+        self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
+        self.access_key_id = access_key_id or ""
+        self.access_key_secret = access_key_secret or ""
+        self.region_id = region_id or "cn-shanghai"
+        if not self.access_key_id:
+            raise ValueError(
+                "Aliyun AI Guardrail: ak is required. Set access_key_id in config.yaml (supports os.environ/ reference)"
+            )
+        if not self.access_key_secret:
+            raise ValueError(
+                "Aliyun AI Guardrail: sk is required. "
+                "Set access_key_secret in config.yaml (supports os.environ/ reference)"
+            )
+        self.level = level or "medium"
+        if self.level not in PROTECTION_LEVEL_THRESHOLD:
+            raise ValueError(
+                f"Aliyun AI Guardrail: Invalid level '{self.level}'. "
+                f"Valid values are: {list(PROTECTION_LEVEL_THRESHOLD.keys())}"
+            )
+        self.max_text_length = max_text_length or 2000
+        self.endpoint = REGION_ENDPOINTS.get(self.region_id, REGION_ENDPOINTS["cn-shanghai"])
+        self.service_url = f"https://{self.endpoint}"
+        self.service_input = service_input or SERVICE_INPUT_DOMESTIC
+        self.service_output = service_output or SERVICE_OUTPUT_DOMESTIC
+        self.service_mcp = service_mcp or SERVICE_INPUT_DOMESTIC
+        self.stream_window_size = stream_window_size or 500
+        self.stream_slide_step = stream_slide_step or 300
+        self.stream_first_check_step = stream_first_check_step or 50
+        verbose_proxy_logger.info(
+            f"Initialized Aliyun AI Security Guardrail: {guardrail_name}, "
+            f"region: {self.region_id}, level: {self.level}, "
+            f"service_input: {self.service_input}, service_output: {self.service_output}, "
+            f"service_mcp: {self.service_mcp}"
+        )
+
+    @staticmethod
+    def get_config_model() -> type[GuardrailConfigModel] | None:
+        from litellm.types.proxy.guardrails.guardrail_hooks.aliyun.aliyun_ai_guardrail import (
+            AliyunAIGuardrailConfigModel,
+        )
+
+        return AliyunAIGuardrailConfigModel
+
+    @staticmethod
+    def _format_iso8601_date() -> str:
+        """Format current timestamp in ISO8601 format."""
+        return datetime.now(timezone.utc).strftime(ISO8601_DATE_FORMAT)
+
+    @staticmethod
+    def _percent_encode(value: str | None) -> str:
+        """URL encode a value according to Aliyun signature requirements."""
+        if value is None:
+            return ""
+        return quote(value.encode(ENCODING), safe="~").replace("+", "%20").replace("*", "%2A")
+
+    def _create_signature(self, string_to_sign: str) -> str:
+        """Create HMAC-SHA1 signature for API request."""
+        secret = self.access_key_secret + "&"
+        signature = hmac.new(
+            secret.encode(ENCODING),
+            string_to_sign.encode(ENCODING),
+            hashlib.sha1,
+        ).digest()
+        return base64.b64encode(signature).decode(ENCODING)
+
+    def _create_string_to_sign(self, http_method: str, parameters: dict[str, str]) -> str:
+        """Create the string to sign for API request."""
+        sorted_keys = sorted(parameters.keys())
+        canonicalized_query_string = ""
+        for key in sorted_keys:
+            canonicalized_query_string += "&" + self._percent_encode(key) + "=" + self._percent_encode(parameters[key])
+        string_to_sign = (
+            http_method + "&" + self._percent_encode("/") + "&" + self._percent_encode(canonicalized_query_string[1:])
+        )
+        return string_to_sign
+
+    def _split_text(self, text: str, max_length: int = 2000) -> list[str]:
+        """
+        Split text into segments of maximum length, trying to preserve sentence boundaries.
+        Args:
+            text: Text to split
+            max_length: Maximum length of each segment
+        Returns:
+            List of text segments
+        """
+        segments = []
+        while len(text) > max_length:
+            chunk = text[:max_length]
+            match = None
+            for pattern in [r"[。！？；:\.?!]+"]:
+                matches = list(re.finditer(pattern, chunk))
+                if matches:
+                    match = matches[-1]
+            if match:
+                cut_point = match.end()
+            else:
+                cut_point = max_length
+            segments.append(text[:cut_point])
+            text = text[cut_point:]
+        if text:
+            segments.append(text)
+        return segments
+
+    async def async_make_request(
+        self,
+        text: str | None = None,
+        service_type: Literal["input", "output", "mcp"] = "input",
+        image_urls: list[str] | None = None,
+    ) -> AliyunAIGuardrailResponse:
+        """
+        Make a request to the Aliyun AI Security Guardrail API.
+        Args:
+            text: Text to check (optional when only images are checked)
+            service_type: "input" for query_security_check, "output" for response_security_check,
+                "mcp" for MCP tool call check (uses service_mcp config)
+            image_urls: Public image URLs to check (optional)
+        Returns:
+            AliyunAIGuardrailResponse
+        """
+        from litellm.types.proxy.guardrails.guardrail_hooks.aliyun.aliyun_ai_guardrail import (
+            AliyunAIGuardrailResponse,
+        )
+
+        if service_type == "mcp":
+            service_code = self.service_mcp
+        elif service_type == "input":
+            service_code = self.service_input
+        else:
+            service_code = self.service_output
+        service_parameters: dict[str, Any] = {"requestFrom": "LiteLLM"}
+        if text:
+            service_parameters["content"] = text
+        if image_urls:
+            service_parameters["imageUrls"] = image_urls
+        parameters = {
+            "Action": "MultiModalGuard",
+            "Version": "2022-03-02",
+            "AccessKeyId": self.access_key_id,
+            "Timestamp": self._format_iso8601_date(),
+            "SignatureMethod": "HMAC-SHA1",
+            "SignatureVersion": "1.0",
+            "SignatureNonce": str(uuid.uuid4()),
+            "Format": "JSON",
+            "Service": service_code,
+            "ServiceParameters": json.dumps(service_parameters, ensure_ascii=False),
+        }
+        string_to_sign = self._create_string_to_sign("POST", parameters)
+        signature = self._create_signature(string_to_sign)
+        parameters["Signature"] = signature
+        verbose_proxy_logger.debug(
+            "Aliyun AI Guardrail request: service=%s, text_length=%d, image_count=%d",
+            service_code,
+            len(text) if text else 0,
+            len(image_urls) if image_urls else 0,
+        )
+        response = await self.async_handler.post(
+            url=self.service_url,
+            data=parameters,
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            timeout=30.0,
+        )
+        body = response.json()
+        verbose_proxy_logger.debug("Aliyun AI Guardrail response: %s", body)
+        if response.status_code != 200:
+            raise HTTPException(
+                status_code=response.status_code,
+                detail={"error": f"Aliyun AI Guardrail request failed. Status: {response.status_code}, Body: {body}"},
+            )
+        if body.get("Code") != 200:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error": f"Aliyun AI Guardrail API error. Code: {body.get('Code')}, Message: {body.get('Message')}"
+                },
+            )
+        return AliyunAIGuardrailResponse(
+            RequestId=body.get("RequestId", ""),
+            Code=body.get("Code", 0),
+            Message=body.get("Message"),
+            Data=body.get("Data"),
+        )
+
+    def _should_block_by_level(self, detected_level: str) -> bool:
+        """
+        Check if the detected risk level should trigger blocking based on protection level.
+        Logic: If detected_level_int >= threshold_int, then should block.
+        Args:
+            detected_level: Risk level from API response (none/low/medium/high or S0/S1/S2/S3/S4)
+        Returns:
+            True if should block, False otherwise
+        """
+        threshold = PROTECTION_LEVEL_THRESHOLD.get(self.level, 99)
+        detected_int = level_to_int(detected_level)
+        return detected_int >= threshold
+
+    def _resolve_detail_level(self, detail: dict[str, Any]) -> str:
+        """
+        Resolve the risk level of a single Detail entry.
+
+        MultiModalGuard reports severity in two shapes: the ``_pro`` service codes
+        return ``Detail[].Level``, while the documented response carries it as
+        ``Detail[].Result[].RiskLevel``. Honouring only the former downgrades the
+        latter to "none", which would let content Aliyun rejected pass through.
+        Args:
+            detail: A single entry of the response ``Detail`` list
+        Returns:
+            The risk level string to weigh against the configured threshold
+        """
+        level = detail.get("Level")
+        if isinstance(level, str) and level.strip():
+            return level
+        # Fall back to the highest RiskLevel reported across the individual results
+        resolved = ""
+        resolved_int = -1
+        for result in detail.get("Result") or []:
+            if not isinstance(result, dict):
+                continue
+            risk_level = result.get("RiskLevel")
+            if not isinstance(risk_level, str) or not risk_level.strip():
+                continue
+            risk_int = level_to_int(risk_level)
+            if risk_int > resolved_int:
+                resolved_int = risk_int
+                resolved = risk_level
+        if resolved:
+            return resolved
+        # No parseable severity: never downgrade an explicit block to "none"
+        if detail.get("Suggestion") == BLOCK_SUGGESTION:
+            return UNRESOLVED_BLOCK_LEVEL
+        return "none"
+
+    def _parse_response_and_check(
+        self,
+        response: AliyunAIGuardrailResponse,
+        check_type: Literal["input", "output"],
+    ) -> dict[str, Any]:
+        """
+        Parse the guardrail response and check if content should be blocked.
+        Blocking logic:
+        Check if detected level >= threshold based on protection level setting
+        Args:
+            response: The API response
+            check_type: "input" or "output"
+        Returns:
+            Dict with parsed results
+        Raises:
+            HTTPException if content should be blocked
+        """
+        data = response.get("Data", {})
+        if not data:
+            return {"flagged": False, "suggestion": "pass", "details": {}, "message": ""}
+        final_suggestion = data.get("Suggestion", "pass")
+        detail_list = data.get("Detail") or []
+        details: dict[str, dict[str, Any]] = {}
+        desensitization = ""
+        should_block = False
+        blocked_type = ""
+        blocked_level = ""
+        block_message = ""
+        for detail in detail_list:
+            detection_type = detail.get("Type", "")
+            detected_level = self._resolve_detail_level(detail)
+            suggestion = detail.get("Suggestion", "pass")
+            results = detail.get("Result", [])
+            details[detection_type] = {
+                "level": detected_level,
+                "suggestion": suggestion,
+                "results": results,
+            }
+            if detection_type == SENSITIVE_DATA_TYPE and results:
+                for result in results:
+                    ext = result.get("Ext", {})
+                    if ext and ext.get("Desensitization"):
+                        desensitization = ext.get("Desensitization", "")
+                        break
+            if not should_block and self._should_block_by_level(detected_level):
+                should_block = True
+                blocked_type = detection_type
+                blocked_level = detected_level
+                block_message = f"检测到{detection_type} (风险等级: {detected_level})"
+        # Aliyun rejected the content overall but no single detection could be attributed
+        # (e.g. an empty Detail list, or every entry reporting pass). Treat it as the most
+        # severe level so the decision is not lost, still subject to the threshold.
+        if not should_block and final_suggestion == BLOCK_SUGGESTION:
+            if self._should_block_by_level(UNRESOLVED_BLOCK_LEVEL):
+                should_block = True
+                blocked_level = UNRESOLVED_BLOCK_LEVEL
+                block_message = f"阿里云返回阻断建议 (Suggestion: {final_suggestion})"
+        verbose_proxy_logger.debug(
+            f"Aliyun AI Guardrail: level={self.level}, "
+            f"check_type={check_type}, should_block={should_block}, "
+            f"blocked_type={blocked_type}, blocked_level={blocked_level}"
+        )
+        if should_block:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error": f"Aliyun AI Guardrail: {block_message}",
+                    "type": check_type,
+                    "details": details,
+                },
+            )
+        return {
+            "flagged": final_suggestion != "pass",
+            "suggestion": final_suggestion,
+            "desensitization": desensitization,
+            "details": details,
+            "message": block_message,
+        }
+
+    @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",
+            "pass_through_endpoint",
+            "rerank",
+            "mcp_call",
+            "anthropic_messages",
+            "responses",
+        ],
+    ) -> dict[str, Any] | None:
+        """
+        Pre-call hook to scan user prompts before sending to LLM.
+        Raises HTTPException if content should be blocked.
+        """
+        verbose_proxy_logger.info(
+            "Aliyun AI Guardrail: Running pre-call prompt scan, on call_type: %s",
+            call_type,
+        )
+        if call_type == "call_mcp_tool":
+            return await self._mcp_pre_call_check(data)
+        new_messages: list[AllMessageValues] | None = data.get("messages")
+        if new_messages is None:
+            verbose_proxy_logger.warning("Aliyun AI Guardrail: not running guardrail. No messages in data")
+            return data
+        user_prompt = self.get_user_prompt(new_messages)
+        image_urls = self.get_image_urls(new_messages)
+        if not user_prompt and not image_urls:
+            verbose_proxy_logger.warning("Aliyun AI Guardrail: No user prompt or image found")
+            return None
+        verbose_proxy_logger.info(
+            "Aliyun AI Guardrail: Pre-call scan started, prompt length: %d, image count: %d",
+            len(user_prompt) if user_prompt else 0,
+            len(image_urls),
+        )
+        if user_prompt:
+            if len(user_prompt) > self.max_text_length:
+                segments = self._split_text(user_prompt, self.max_text_length)
+            else:
+                segments = [user_prompt]
+        else:
+            segments = []
+        # Attach all image URLs to the first text segment so content + images
+        # are checked together; remaining segments carry text only. When there
+        # is no text, send a single image-only request.
+        payloads: list[tuple] = []
+        if segments:
+            for idx, segment in enumerate(segments):
+                payloads.append((segment, image_urls if idx == 0 and image_urls else None))
+        elif image_urls:
+            payloads.append((None, image_urls))
+        semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests, MultiModalGuard API limit is 20
+
+        async def check_with_semaphore(segment_text: str | None, segment_images: list[str] | None):
+            async with semaphore:
+                return await self.async_make_request(text=segment_text, service_type="input", image_urls=segment_images)
+
+        responses = await asyncio.gather(*[check_with_semaphore(t, imgs) for t, imgs in payloads])
+        for response in responses:
+            self._parse_response_and_check(response, check_type="input")
+        verbose_proxy_logger.info("Aliyun AI Guardrail: Pre-call scan passed")
+        return None
+
+    # ================================================================
+    # MCP-specific guardrail methods
+    # ================================================================
+
+    async def _mcp_pre_call_check(self, data: dict) -> dict | None:
+        """MCP pre-call: audit tool name + arguments before execution."""
+        messages = data.get("messages", [])
+        content = messages[0].get("content", "") if messages else ""
+        verbose_proxy_logger.info(
+            "Aliyun AI Guardrail: ★ MCP pre-call check started, content: %s",
+            content,
+        )
+        if not content:
+            return None
+        if len(content) > self.max_text_length:
+            segments = self._split_text(content, self.max_text_length)
+        else:
+            segments = [content]
+        semaphore = asyncio.Semaphore(5)
+
+        async def check(text: str):
+            async with semaphore:
+                return await self.async_make_request(text=text, service_type="mcp")
+
+        responses = await asyncio.gather(*[check(s) for s in segments])
+        for resp in responses:
+            self._parse_response_and_check(resp, check_type="input")
+        verbose_proxy_logger.info("Aliyun AI Guardrail: ★ MCP pre-call check passed")
+        return None
+
+    def _should_run_post_mcp_call(self) -> bool:
+        """Check if post_mcp_call is configured in event_hook.
+
+        Since GuardrailEventHooks enum does not include post_mcp_call,
+        we cannot use should_run_guardrail(). This method manually checks
+        whether the user configured 'post_mcp_call' in the guardrail mode.
+
+        Returns True if post_mcp_call should run:
+        - event_hook is None → run for all events
+        - event_hook is a list containing 'post_mcp_call'
+        - event_hook is a string equal to 'post_mcp_call'
+        - event_hook is a Mode with 'post_mcp_call' in tags or default
+        """
+        from litellm.types.guardrails import Mode
+
+        if self.event_hook is None:
+            return True
+        if isinstance(self.event_hook, list):
+            return "post_mcp_call" in self.event_hook
+        if isinstance(self.event_hook, Mode):
+            for tag_value in self.event_hook.tags.values():
+                if isinstance(tag_value, list):
+                    if "post_mcp_call" in tag_value:
+                        return True
+                elif tag_value == "post_mcp_call":
+                    return True
+            if self.event_hook.default:
+                default_list = (
+                    self.event_hook.default if isinstance(self.event_hook.default, list) else [self.event_hook.default]
+                )
+                return "post_mcp_call" in default_list
+            return False
+        return self.event_hook == "post_mcp_call"
+
+    async def async_post_mcp_tool_call_hook(
+        self,
+        kwargs: dict,
+        response_obj: MCPPostCallResponseObject,
+        start_time: datetime,
+        end_time: datetime,
+    ) -> MCPPostCallResponseObject | None:
+        """MCP post-call: audit tool output after execution.
+
+        Raising from here does not block: the dispatcher treats every callback exception
+        as a non-blocking logging error and hands the untouched tool result back. Both
+        call sites also discard this hook's return value, so a violation has to be
+        written into the live tool result carried by ``kwargs["original_response"]``.
+        The replacement object is returned as well, to honour the hook's contract.
+        """
+        # Since GuardrailEventHooks enum has no post_mcp_call, the framework
+        # always invokes this hook if implemented. We check config manually.
+        if not self._should_run_post_mcp_call():
+            verbose_proxy_logger.info("Aliyun AI Guardrail: Skipping post_mcp_call — not configured in event_hook")
+            return None
+        verbose_proxy_logger.info("Aliyun AI Guardrail: ★ MCP post-call check started")
+        original_response = kwargs.get("original_response")
+        combined_text = ""
+        for candidate in (original_response, getattr(response_obj, "mcp_tool_call_response", None)):
+            if candidate is None:
+                continue
+            combined_text = self._extract_mcp_tool_text(candidate)
+            if combined_text:
+                break
+        if not combined_text:
+            return None
+        verbose_proxy_logger.info(
+            "Aliyun AI Guardrail: ★ MCP post-call check, response length: %d",
+            len(combined_text),
+        )
+        if len(combined_text) > self.max_text_length:
+            segments = self._split_text(combined_text, self.max_text_length)
+        else:
+            segments = [combined_text]
+        semaphore = asyncio.Semaphore(5)
+
+        async def check(text: str):
+            async with semaphore:
+                return await self.async_make_request(text=text, service_type="mcp")
+
+        try:
+            responses = await asyncio.gather(*[check(s) for s in segments])
+            for resp in responses:
+                self._parse_response_and_check(resp, check_type="output")
+        except HTTPException as e:
+            verbose_proxy_logger.warning(
+                "Aliyun AI Guardrail: ★ MCP post-call blocked — tool output replaced with the violation detail"
+            )
+            return self._block_mcp_tool_output(
+                detail=e.detail,
+                response_obj=response_obj,
+                original_response=original_response,
+            )
+        except Exception as e:
+            # Raising here is swallowed by the dispatcher as a non-blocking logging
+            # error, which would hand the unaudited tool output straight to the
+            # client. Fail closed, matching every other path of this integration.
+            verbose_proxy_logger.error(
+                f"Aliyun AI Guardrail: ★ MCP post-call check failed, blocking unaudited tool output: {str(e)}",
+                exc_info=True,
+            )
+            return self._block_mcp_tool_output(
+                detail={"error": f"Aliyun AI Guardrail 调用失败，工具输出未经审核，已拦截: {str(e)}"},
+                response_obj=response_obj,
+                original_response=original_response,
+            )
+        verbose_proxy_logger.info("Aliyun AI Guardrail: ★ MCP post-call check passed")
+        return None
+
+    @staticmethod
+    def _iter_mcp_content_items(payload: Any) -> list[Any]:
+        """
+        Normalise an MCP tool result into its list of content items.
+        Args:
+            payload: A CallToolResult, its content list, or a raw response body
+        Returns:
+            The content items to audit, empty when none can be located
+        """
+        if isinstance(payload, str):
+            return [payload]
+        content = getattr(payload, "content", None)
+        if isinstance(content, list):
+            return content
+        if isinstance(payload, dict):
+            inner = payload.get("content")
+            return inner if isinstance(inner, list) else [payload]
+        if isinstance(payload, list):
+            # MCPPostCallResponseObject declares mcp_tool_call_response as a list, so a
+            # CallToolResult handed to it is coerced by iterating the model into
+            # (field, value) pairs. Recover the real content instead of auditing reprs.
+            for item in payload:
+                if isinstance(item, tuple) and len(item) == 2 and item[0] == "content":
+                    if isinstance(item[1], list):
+                        return item[1]
+            return payload
+        return []
+
+    def _extract_mcp_tool_text(self, payload: Any) -> str:
+        """Collect the textual output of an MCP tool result."""
+        texts: list[str] = []
+        for item in self._iter_mcp_content_items(payload):
+            if isinstance(item, str):
+                text = item
+            elif isinstance(item, dict):
+                text = item.get("text", "") or ""
+            else:
+                text = getattr(item, "text", "") or ""
+            if text:
+                texts.append(text)
+        return "\n".join(texts)
+
+    @staticmethod
+    def _replace_tool_output_in_place(target: Any, blocked_content: list[Any]) -> bool:
+        """Overwrite an MCP tool result's content with ``blocked_content``, in place."""
+        if target is None:
+            return False
+        content = getattr(target, "content", None)
+        if isinstance(content, list):
+            content[:] = blocked_content
+            if hasattr(target, "isError"):
+                try:
+                    target.isError = True
+                except (AttributeError, TypeError, ValueError):
+                    pass
+            return True
+        if isinstance(target, list):
+            target[:] = blocked_content
+            return True
+        if isinstance(target, dict):
+            result = target.get("result")
+            if isinstance(result, dict) and isinstance(result.get("content"), list):
+                result["content"] = list(blocked_content)
+                return True
+            if isinstance(target.get("content"), list):
+                target["content"] = list(blocked_content)
+                return True
+        return False
+
+    def _block_mcp_tool_output(
+        self,
+        detail: Any,
+        response_obj: MCPPostCallResponseObject,
+        original_response: Any,
+    ) -> MCPPostCallResponseObject:
+        """Replace blocked MCP tool output, both in place and as the returned object."""
+        from litellm.types.llms.base import HiddenParams
+        from litellm.types.mcp import MCPPostCallResponseObject as _MCPPostCallResponseObject
+        from mcp.types import TextContent
+
+        payload = detail if isinstance(detail, dict) else {"error": str(detail)}
+        blocked_content: list[Any] = [TextContent(type="text", text=json.dumps(payload, ensure_ascii=False))]
+        for target in (original_response, getattr(response_obj, "mcp_tool_call_response", None)):
+            self._replace_tool_output_in_place(target, blocked_content)
+        hidden_params = getattr(response_obj, "hidden_params", None)
+        return _MCPPostCallResponseObject(
+            mcp_tool_call_response=blocked_content,
+            hidden_params=hidden_params if isinstance(hidden_params, HiddenParams) else HiddenParams(),
+        )
+
+    @log_guardrail_information
+    async def async_post_call_success_hook(
+        self,
+        data: dict,
+        user_api_key_dict: UserAPIKeyAuth,
+        response: Any | ModelResponse | EmbeddingResponse | ImageResponse,
+    ) -> Any:
+        """
+        Post-call hook to scan LLM responses.
+        Raises HTTPException if content should be blocked.
+        """
+        from litellm.types.utils import Choices, ModelResponse
+
+        if isinstance(response, ModelResponse) and response.choices:
+            # Scan every choice: n>1 responses would otherwise return unchecked content
+            content = "\n".join(
+                str(choice.message.content)
+                for choice in response.choices
+                if isinstance(choice, Choices) and choice.message.content
+            )
+            if content:
+                verbose_proxy_logger.info(
+                    "Aliyun AI Guardrail: Post-call scan started, response length: %d",
+                    len(content),
+                )
+                if len(content) > self.max_text_length:
+                    segments = self._split_text(content, self.max_text_length)
+                else:
+                    segments = [content]
+                semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests
+
+                async def check_with_semaphore(segment: str):
+                    async with semaphore:
+                        return await self.async_make_request(text=segment, service_type="output")
+
+                responses = await asyncio.gather(*[check_with_semaphore(segment) for segment in segments])
+                for guardrail_response in responses:
+                    self._parse_response_and_check(guardrail_response, check_type="output")
+                verbose_proxy_logger.info("Aliyun AI Guardrail: Post-call scan passed")
+        return response
+
+    @staticmethod
+    def _extract_stream_chunk_text(chunk: Any) -> str:
+        """
+        Collect every text field a streaming chunk can carry.
+        Covers both chat completion chunks and Responses API streaming events.
+        Auditing only ``delta.content`` would release tool call arguments,
+        reasoning text and every /v1/responses chunk to the client unchecked.
+        Args:
+            chunk: A streaming chunk
+        Returns:
+            The concatenated text to audit, empty when the chunk carries none
+        """
+        parts: list[str] = []
+
+        def collect_call(function: Any) -> None:
+            for field in ("name", "arguments"):
+                value = getattr(function, field, None)
+                if value:
+                    parts.append(str(value))
+
+        # Responses API events carry text outside of choices: either a plain
+        # string delta, or the assembled output of a terminal event.
+        event_delta = getattr(chunk, "delta", None)
+        if isinstance(event_delta, str) and event_delta:
+            parts.append(event_delta)
+        for output_item in getattr(getattr(chunk, "response", None), "output", None) or []:
+            content_parts = getattr(output_item, "content", None)
+            if content_parts:
+                for content_part in content_parts:
+                    text = getattr(content_part, "text", None)
+                    if text:
+                        parts.append(str(text))
+            else:
+                text = getattr(output_item, "text", None)
+                if text:
+                    parts.append(str(text))
+
+        for choice in getattr(chunk, "choices", None) or []:
+            delta = getattr(choice, "delta", None)
+            if delta is None:
+                continue
+            for field in ("content", "reasoning_content"):
+                value = getattr(delta, field, None)
+                if value:
+                    parts.append(str(value))
+            for call in getattr(delta, "tool_calls", None) or []:
+                collect_call(getattr(call, "function", None))
+            collect_call(getattr(delta, "function_call", None))
+        return "".join(parts)
+
+    async def async_post_call_streaming_iterator_hook(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        response: Any,
+        request_data: dict[str, Any],
+    ) -> AsyncGenerator[Any, None]:
+        """
+        Process streaming response with sliding window guardrail checks.
+        This method implements sliding window guardrail checks based on
+        `stream_window_size` and `stream_slide_step`.
+        It triggers a guardrail API call when:
+        1. Every `stream_slide_step` new chars accumulate since the last check.
+        2. Stream ends and there's remaining unchecked content.
+        For example, if stream_window_size=2000, stream_slide_step=300:
+        - At 300 chars: check chars 0-300 (window: last 2000)
+        - At 600 chars: check chars 0-600 (window: last 2000)
+        - At 2100 chars: check chars 100-2100 (window slides forward)
+        - At 2400 chars: check chars 400-2400 (window slides forward)
+        - When stream ends with 2500 chars: check chars 500-2500 (final window)
+        """
+        accumulated_text = ""
+        last_check_position = 0  # Position (total length) when last check was triggered
+        pending_chunks = []  # Buffer chunks until guardrail check passes
+        chunk_count = 0
+        is_first_check = True  # First check uses smaller threshold to reduce first-token latency
+        verbose_proxy_logger.info(
+            "Aliyun AI Guardrail: Streaming scan started, window=%d, step=%d, first_check_step=%d",
+            self.stream_window_size,
+            self.stream_slide_step,
+            self.stream_first_check_step,
+        )
+        try:
+            async for chunk in response:
+                chunk_text = self._extract_stream_chunk_text(chunk)
+                accumulated_text += chunk_text
+                chunk_count += 1
+                # Buffer the chunk, don't yield until guardrail check passes
+                pending_chunks.append(chunk)
+                current_length = len(accumulated_text)
+                new_chars_since_last_check = current_length - last_check_position
+                check_threshold = self.stream_first_check_step if is_first_check else self.stream_slide_step
+                if new_chars_since_last_check >= check_threshold:
+                    start = max(0, current_length - self.stream_window_size)
+                    text_to_check = accumulated_text[start:current_length]
+                    guardrail_response = await self.async_make_request(text=text_to_check, service_type="output")
+                    self._parse_response_and_check(guardrail_response, check_type="output")
+                    verbose_proxy_logger.info(
+                        "Aliyun AI Guardrail: Streaming check passed at position %d", current_length
+                    )
+                    for pending_chunk in pending_chunks:
+                        yield pending_chunk
+                    pending_chunks.clear()
+                    last_check_position = current_length
+                    is_first_check = False
+            # Stream ended - check any remaining unchecked content with a final window
+            if len(accumulated_text) > last_check_position:
+                start = max(0, len(accumulated_text) - self.stream_window_size)
+                remaining_text = accumulated_text[start:]
+                guardrail_response = await self.async_make_request(text=remaining_text, service_type="output")
+                self._parse_response_and_check(guardrail_response, check_type="output")
+            verbose_proxy_logger.info(
+                "Aliyun AI Guardrail: Streaming scan completed, total length: %d", len(accumulated_text)
+            )
+            for pending_chunk in pending_chunks:
+                yield pending_chunk
+            pending_chunks.clear()
+        except HTTPException as e:
+            error_detail = e.detail if isinstance(e.detail, dict) else {"message": str(e.detail)}
+            verbose_proxy_logger.info("Aliyun AI Guardrail: Streaming blocked at position %d", len(accumulated_text))
+            yield f"data: {json.dumps({'error': error_detail}, ensure_ascii=False)}\n\n"
+            return
+        except Exception as e:
+            verbose_proxy_logger.error(f"Aliyun AI Guardrail streaming error: {str(e)}", exc_info=True)
+            raise

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

@@ -0,0 +1,103 @@
+"""
+Base class for Aliyun guardrails
+阿里云护栏基类
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from litellm.types.llms.openai import AllMessageValues
+
+
+class AliyunGuardrailBase:
+    """
+    Base class for Aliyun guardrails.
+    """
+
+    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?"
+        """
+        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
+
+    def get_image_urls(self, messages: list[AllMessageValues]) -> list[str]:
+        """
+        Extract image URLs from the last consecutive block of user messages.
+        Only publicly accessible http(s) URLs are collected (in order,
+        de-duplicated). Uses the same message range as ``get_user_prompt``.
+        Example:
+        messages = [
+            {"role": "user", "content": [
+                {"type": "text", "text": "what is in this image?"},
+                {"type": "image_url", "image_url": {"url": "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

litellm/types/guardrails.py

@@ -134,6 +134,7 @@ class SupportedGuardrailIntegrations(Enum):
     HEADROOM = "headroom"
     COMPRESR = "compresr"
     STRAIKER = "straiker"
+    ALIYUN_AI_GUARDRAIL = "aliyun_ai_guardrail"
 
 
 class Role(Enum):
@@ -654,6 +655,55 @@ class JavelinGuardrailConfigModel(BaseModel):
     config: dict | None = Field(default=None, description="Additional configuration for the guardrail")
 
 
+class AliyunAIGuardrailConfigModel(BaseModel):
+    """Configuration parameters for the Aliyun AI Security guardrail."""
+
+    level: str | None = Field(
+        default=None,
+        description="Protection level. 'low': block all risks (high protection), 'medium': block medium+high risks, 'high': block high only, 'max': observe mode. Default: medium",
+    )
+    max_text_length: int | None = Field(
+        default=None,
+        description="Maximum text length for a single API call. Text longer than this will be split.",
+    )
+    stream_window_size: int | None = Field(
+        default=None,
+        description="Sliding window size (in chars) for streaming output guardrail checks. Default: 500",
+    )
+    stream_slide_step: int | None = Field(
+        default=None,
+        description="Sliding step (in chars) for streaming output guardrail checks. Default: 300",
+    )
+    stream_first_check_step: int | None = Field(
+        default=None,
+        description="First check threshold (in chars) to reduce first-token latency. Default: 50",
+    )
+    region_id: str | None = Field(
+        default=None,
+        description="Aliyun region ID. Default: cn-shanghai",
+    )
+    service_input: str | None = Field(
+        default=None,
+        description="Service code for input (pre-call) detection. Default: query_security_check",
+    )
+    service_output: str | None = Field(
+        default=None,
+        description="Service code for output (post-call) detection. Default: response_security_check",
+    )
+    service_mcp: str | None = Field(
+        default=None,
+        description="Service code for MCP tool call detection (pre/post MCP call). Default: query_security_check",
+    )
+    access_key_id: str | None = Field(
+        default=None,
+        description="Aliyun Access Key ID for the guardrail. Configure in config.yaml, supports os.environ/ reference",
+    )
+    access_key_secret: str | None = Field(
+        default=None,
+        description="Aliyun Access Key Secret for the guardrail. Configure in config.yaml, supports os.environ/ reference",
+    )
+
+
 class ContentFilterAction(str, Enum):
     """Action to take when content filter detects a match"""
 
@@ -999,6 +1049,7 @@ class LitellmParams(
     QostodianNexusConfigModel,
     VigilGuardGuardrailConfigModel,
     SingulrGuardrailConfigModel,
+    AliyunAIGuardrailConfigModel,
 ):
     guardrail: str = Field(description="The type of guardrail integration to use")
     mode: str | list[str] | Mode = Field(
@@ -1126,3 +1177,4 @@ 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

@@ -0,0 +1,165 @@
+"""
+Type definitions for Aliyun AI Security Guardrail
+阿里云AI安全护栏类型定义
+Aliyun AI Guardrail supports the following detection types:
+- contentModeration: Content safety moderation
+- sensitiveData: Sensitive data detection (PII, etc.)
+- promptAttack: Prompt injection attack detection
+- maliciousUrl: Malicious URL detection
+"""
+
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+from typing_extensions import NotRequired, TypedDict
+
+from ..base import GuardrailConfigModel
+
+
+# Response types
+class AliyunAIGuardrailResponseDetailResultExt(TypedDict, total=False):
+    """Extended information in result"""
+
+    Desensitization: Optional[str]  # Desensitized text when action is mask
+
+
+class AliyunAIGuardrailResponseDetailResult(TypedDict, total=False):
+    """Result item in detail"""
+
+    Confidence: Optional[float]
+    Label: Optional[str]
+    Ext: Optional[AliyunAIGuardrailResponseDetailResultExt]
+    # 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]
+
+
+class AliyunAIGuardrailResponseDetail(TypedDict):
+    """Detail item in response data"""
+
+    Type: str  # contentModeration, sensitiveData, promptAttack, maliciousUrl
+    Suggestion: str  # pass, block, mask
+    Result: List[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]
+
+
+class AliyunAIGuardrailResponseData(TypedDict, total=False):
+    """Response data from Aliyun AI Guardrail API"""
+
+    Suggestion: str  # Overall suggestion: pass, block, mask
+    Detail: Optional[List[AliyunAIGuardrailResponseDetail]]
+
+
+class AliyunAIGuardrailResponse(TypedDict):
+    """Response from Aliyun AI Guardrail API"""
+
+    RequestId: str
+    Code: int
+    Message: Optional[str]
+    Data: Optional[AliyunAIGuardrailResponseData]
+
+
+# Suggestion type
+AliyunAIGuardrailSuggestion = Literal["pass", "block", "watch"]
+
+# Detection type
+AliyunAIGuardrailDetectionType = 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
+
+
+# Risk level literals
+AliyunRiskLevel = Literal["none", "low", "medium", "high"]
+
+# Protection level literals
+AliyunProtectionLevel = Literal["low", "medium", "high", "max"]
+
+
+# Configuration models
+class AliyunAIGuardrailOptionalParams(BaseModel):
+    """
+    Optional parameters for Aliyun AI Guardrail.
+    Credentials (access_key_id / access_key_secret) are configured
+    in config.yaml on the AliyunAIGuardrailConfigModel and support os.environ/ references.
+    """
+
+    level: Optional[AliyunProtectionLevel] = 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(
+        default=2000,
+        description="Maximum text length for a single API call. Text longer than this will be split.",
+    )
+    stream_window_size: Optional[int] = 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(
+        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(
+        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(
+        default="cn-shanghai",
+        description="Aliyun region ID. Default: cn-shanghai",
+    )
+    service_input: Optional[str] = Field(
+        default="query_security_check_pro",
+        description="Service code for input (pre-call) detection. Default: query_security_check_pro",
+    )
+    service_output: Optional[str] = Field(
+        default="response_security_check_pro",
+        description="Service code for output (post-call) detection. Default: response_security_check_pro",
+    )
+    service_mcp: Optional[str] = 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",
+    )
+
+
+class AliyunAIGuardrailConfigModel(GuardrailConfigModel[AliyunAIGuardrailOptionalParams]):
+    """
+    Configuration model for Aliyun AI Guardrail.
+    Credentials are configured in config.yaml and support os.environ/ references:
+    - access_key_id: Aliyun Access Key ID
+    - access_key_secret: Aliyun Access Key Secret
+    """
+
+    access_key_id: Optional[str] = Field(
+        default=None,
+        description="Aliyun Access Key ID. Configure in config.yaml, supports os.environ/ reference",
+    )
+    access_key_secret: Optional[str] = Field(
+        default=None,
+        description="Aliyun Access Key Secret. Configure in config.yaml, supports os.environ/ reference",
+    )
+    optional_params: AliyunAIGuardrailOptionalParams = Field(
+        default_factory=AliyunAIGuardrailOptionalParams,
+        description="Optional parameters for the Aliyun AI Guardrail",
+    )
+
+    @staticmethod
+    def ui_friendly_name() -> str:
+        return "Aliyun AI Security Guardrail"