scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 89

feat(guardrails): add Cisco AI Defense integration (#28249) (#30338)

Browse source
This commit is contained in:
Yassin Kortam 2026-06-12 23:21:23 -07:00 • committed by GitHub
parent c90eb7e96f
commit ec9353cb69
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 7382 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

BIN
litellm/proxy/_experimental/out/assets/logos/cisco.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 1.9 KiB

108
litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/__init__.py Normal file
View file

@ -0,0 +1,108 @@
"""Cisco AI Defense Guardrail Integration for LiteLLM."""
from typing import TYPE_CHECKING
from litellm.types.guardrails import SupportedGuardrailIntegrations
from .cisco_ai_defense import (
CiscoAIDefenseGuardrail,
CiscoAIDefenseGuardrailAPIError,
CiscoAIDefenseGuardrailMissingSecrets,
)
if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"):
import litellm
guardrail_name = guardrail.get("guardrail_name")
if not guardrail_name:
raise ValueError("Cisco AI Defense: guardrail_name is required")
optional_params = getattr(litellm_params, "optional_params", None)
_callback = CiscoAIDefenseGuardrail(
guardrail_name=guardrail_name,
api_key=litellm_params.api_key,
api_base=litellm_params.api_base,
inspection_type=_get_optional_value(
litellm_params, optional_params, "inspection_type"
),
inspect_path=_get_optional_value(
litellm_params, optional_params, "inspect_path"
),
enabled_rules=_get_optional_value(
litellm_params, optional_params, "enabled_rules"
),
integration_profile_id=_get_optional_value(
litellm_params, optional_params, "integration_profile_id"
),
integration_profile_version=_get_optional_value(
litellm_params, optional_params, "integration_profile_version"
),
integration_tenant_id=_get_optional_value(
litellm_params, optional_params, "integration_tenant_id"
),
integration_type=_get_optional_value(
litellm_params, optional_params, "integration_type"
),
on_flagged_action=_get_optional_value(
litellm_params, optional_params, "on_flagged_action"
),
fallback_on_error=_get_optional_value(
litellm_params, optional_params, "fallback_on_error"
),
timeout=_get_optional_value(litellm_params, optional_params, "timeout"),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on or False,
)
litellm.logging_callback_manager.add_litellm_callback(_callback)
# MCP post-tool-call hooks are dispatched through success callbacks.
litellm.logging_callback_manager.add_litellm_success_callback(_callback)
return _callback
def _get_optional_value(litellm_params, optional_params, attribute_name):
"""Resolve Cisco optional params without inheriting sibling defaults."""
if optional_params is not None:
if isinstance(optional_params, dict):
if attribute_name in optional_params:
return optional_params[attribute_name]
else:
nested_fields_set = getattr(optional_params, "model_fields_set", None)
if nested_fields_set is None or attribute_name in nested_fields_set:
value = getattr(optional_params, attribute_name, None)
if value is not None:
return value
if litellm_params is None:
return None
# Only accept flattened values the caller explicitly set.
fields_set = getattr(litellm_params, "model_fields_set", None)
if fields_set is None or attribute_name not in fields_set:
return None
return getattr(litellm_params, attribute_name, None)
guardrail_initializer_registry = {
SupportedGuardrailIntegrations.CISCO_AI_DEFENSE.value: initialize_guardrail,
}
guardrail_class_registry = {
SupportedGuardrailIntegrations.CISCO_AI_DEFENSE.value: CiscoAIDefenseGuardrail,
}
__all__ = [
"CiscoAIDefenseGuardrail",
"CiscoAIDefenseGuardrailAPIError",
"CiscoAIDefenseGuardrailMissingSecrets",
"initialize_guardrail",
"guardrail_initializer_registry",
"guardrail_class_registry",
]

2358
litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense.py Normal file
View file

File diff suppressed because it is too large Load diff

704
litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense_mcp.py Normal file
View file

@ -0,0 +1,704 @@
"""MCP-specific inspection logic for the Cisco AI Defense guardrail.
The public guardrail class imports this private mixin from
``cisco_ai_defense.py``. Keeping MCP logic here avoids circular imports
while preserving the existing public import path.
"""
from datetime import datetime
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from fastapi import HTTPException
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.common_utils.callback_utils import (
add_guardrail_to_applied_guardrails_header,
)
from litellm.types.guardrails import GuardrailEventHooks
if TYPE_CHECKING:
from litellm.types.mcp import MCPPostCallResponseObject
from .cisco_ai_defense import _ScanContext
def _serialize_mcp_content_item(item: object) -> Dict[str, Any]:
"""Serialize an MCP content item to a JSON-friendly dict.
Handles raw dicts, MCP SDK Pydantic models, and simple ``.text`` objects.
"""
if isinstance(item, dict):
return dict(item)
model_dump = getattr(item, "model_dump", None)
if callable(model_dump):
try:
return dict(model_dump(exclude_none=True))
except TypeError:
return dict(model_dump())
text = getattr(item, "text", None)
if isinstance(text, str):
return {"type": getattr(item, "type", "text"), "text": text}
return {"type": "text", "text": str(item)}
class _CiscoAIDefenseMcpMixin:
"""MCP-specific instance methods for ``CiscoAIDefenseGuardrail``.
Holds the MCP hooks, JSON-RPC payload builders, and redaction helpers.
"""
if TYPE_CHECKING:
api_base: str
inspect_path: str
inspection_type: str
_PROVIDER_NAME: str
guardrail_name: Optional[str]
def should_run_guardrail(
self, data: dict, event_type: GuardrailEventHooks
) -> bool: ...
async def _post_inspection(
self, url: str, payload: Dict[str, Any], surface: str
) -> Dict[str, Any]: ...
def _handle_api_error(
self,
error: Exception,
*,
request_data: Optional[dict] = ...,
start_time: Optional[datetime] = ...,
surface: str = ...,
direction: str = ...,
) -> Dict[str, Any]: ...
def _finalize_inspection(
self,
inspect_response: Dict[str, Any],
request_data: dict,
context: "_ScanContext",
start_time: datetime,
response_obj: object = ...,
) -> Dict[str, Any]: ...
# ------------------------------------------------------------------
# MCP post-tool hook (dispatcher contract)
# ------------------------------------------------------------------
async def async_post_mcp_tool_call_hook(
self,
kwargs: dict,
response_obj: "MCPPostCallResponseObject",
start_time: datetime,
end_time: datetime,
) -> Optional["MCPPostCallResponseObject"]:
"""Scan MCP tool output and return a replacement object on block."""
del start_time, end_time
if self.inspection_type != "mcp":
return None
request_data: Dict[str, Any] = {}
for key in (
"name",
"litellm_call_id",
"id",
"user",
"mcp_tool_name",
"tool_name",
"mcp_arguments",
"arguments",
"mcp_server_name",
"server_name",
"metadata",
"litellm_metadata",
"mcp_tool_call_metadata",
"guardrails",
):
if key in kwargs and kwargs[key] is not None:
request_data[key] = kwargs[key]
self._hydrate_mcp_tool_context(request_data)
if not (
self.should_run_guardrail(
data=request_data,
event_type=GuardrailEventHooks.during_mcp_call,
)
or self.should_run_guardrail(
data=request_data,
event_type=GuardrailEventHooks.pre_mcp_call,
)
):
verbose_proxy_logger.debug(
"Cisco AI Defense guardrail (%s): no MCP mode configured "
"— skipping MCP response scan.",
self.guardrail_name,
)
return None
mcp_tool_response = self._extract_mcp_tool_call_response(response_obj)
if mcp_tool_response is None:
verbose_proxy_logger.debug(
"Cisco AI Defense guardrail: no MCP tool response payload "
"to scan, skipping"
)
return None
original_response = kwargs.get("original_response")
try:
await self._inspect_mcp_response(
request_data=request_data,
response=mcp_tool_response,
redact_response_obj=(
original_response
if original_response is not None
else mcp_tool_response
),
)
except HTTPException as exc:
blocking_response = self._build_blocking_mcp_response(
detail=exc.detail, original_response_obj=response_obj
)
self._replace_mcp_tool_response(response_obj, blocking_response)
if original_response is not None:
self._replace_mcp_tool_response(original_response, blocking_response)
add_guardrail_to_applied_guardrails_header(
request_data=request_data, guardrail_name=self.guardrail_name
)
verbose_proxy_logger.warning(
"Cisco AI Defense guardrail (%s): MCP response blocked — "
"tool output replaced with synthesized violation message.",
self.guardrail_name,
)
return blocking_response
add_guardrail_to_applied_guardrails_header(
request_data=request_data, guardrail_name=self.guardrail_name
)
return None
def _build_blocking_mcp_response(
self,
detail: object,
original_response_obj: object,
) -> "MCPPostCallResponseObject":
"""Build a synthetic MCPPostCallResponseObject for blocked output."""
import json as _json
from litellm.types.llms.base import HiddenParams
from litellm.types.mcp import MCPPostCallResponseObject
from mcp.types import TextContent
if isinstance(detail, dict):
payload = detail
else:
payload = {
"error": "Blocked by Cisco AI Defense Guardrail",
"message": (
str(detail) if detail else "Blocked by Cisco AI Defense Guardrail"
),
"provider": self._PROVIDER_NAME,
"guardrail": self.guardrail_name,
"surface": "mcp",
"direction": "output",
"action": "block",
}
original_hidden = getattr(original_response_obj, "hidden_params", None)
if isinstance(original_hidden, HiddenParams):
hidden_params: Any = original_hidden
else:
response_cost = getattr(original_hidden, "response_cost", None)
hidden_params = (
HiddenParams(response_cost=response_cost)
if response_cost is not None
else HiddenParams()
)
return MCPPostCallResponseObject(
mcp_tool_call_response=[
TextContent(type="text", text=_json.dumps(payload))
],
hidden_params=hidden_params,
)
@staticmethod
def _replace_mcp_tool_response(
response_obj: object, replacement_obj: object
) -> bool:
replacement = getattr(replacement_obj, "mcp_tool_call_response", None)
if replacement is None:
return False
inner = getattr(response_obj, "mcp_tool_call_response", None)
if inner is not None:
if _CiscoAIDefenseMcpMixin._replace_mcp_tool_response(
inner, replacement_obj
):
return True
try:
setattr(response_obj, "mcp_tool_call_response", replacement)
return True
except (AttributeError, TypeError, ValueError):
return False
content = getattr(response_obj, "content", None)
if isinstance(content, list):
content[:] = replacement
structured_replacement = (
_CiscoAIDefenseMcpMixin._replacement_structured_content(replacement)
)
if hasattr(response_obj, "structuredContent"):
try:
setattr(response_obj, "structuredContent", structured_replacement)
except (AttributeError, TypeError, ValueError):
pass
if hasattr(response_obj, "isError"):
try:
setattr(response_obj, "isError", True)
except (AttributeError, TypeError, ValueError):
pass
return True
if isinstance(response_obj, list):
response_obj[:] = replacement
return True
if isinstance(response_obj, dict):
result = response_obj.get("result")
if isinstance(result, dict):
result["content"] = replacement
result["structuredContent"] = (
_CiscoAIDefenseMcpMixin._replacement_structured_content(replacement)
)
result["isError"] = True
return True
response_obj["result"] = {
"content": replacement,
"structuredContent": _CiscoAIDefenseMcpMixin._replacement_structured_content(
replacement
),
"isError": True,
}
return True
return False
@staticmethod
def _replacement_structured_content(
replacement: object,
) -> Optional[Dict[str, str]]:
if not isinstance(replacement, list) or not replacement:
return None
first = replacement[0]
text = (
first.get("text")
if isinstance(first, dict)
else getattr(first, "text", None)
)
return {"result": text} if isinstance(text, str) else None
@staticmethod
def _extract_mcp_tool_call_response(response_obj: object) -> object:
"""Pull the raw tool-call response off a MCPPostCallResponseObject."""
inner = getattr(response_obj, "mcp_tool_call_response", None)
if inner is None and isinstance(response_obj, dict):
inner = response_obj.get("mcp_tool_call_response")
return inner if inner is not None else response_obj
# ------------------------------------------------------------------
# MCP request / response inspection
# ------------------------------------------------------------------
async def _inspect_mcp_request(
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
) -> Dict[str, Any]:
del user_api_key_dict # carried via logging metadata, not the wire payload
url = f"{self.api_base}{self.inspect_path}"
payload = self._build_mcp_request_payload(data=data)
if payload is None:
verbose_proxy_logger.debug(
"Cisco AI Defense guardrail: could not build MCP request "
"payload, skipping"
)
return {}
start_time = datetime.now()
try:
inspect_response = await self._post_inspection(
url=url, payload=payload, surface="mcp"
)
except HTTPException:
raise
except Exception as exc:
return self._handle_api_error(
exc,
request_data=data,
start_time=start_time,
surface="mcp",
direction="input",
)
from .cisco_ai_defense import _ScanContext
return self._finalize_inspection(
inspect_response=inspect_response,
request_data=data,
context=_ScanContext(surface="mcp", direction="input"),
start_time=start_time,
)
async def _inspect_mcp_response(
self,
request_data: dict,
response: object,
user_api_key_dict: Optional[UserAPIKeyAuth] = None,
redact_response_obj: object = None,
) -> Dict[str, Any]:
del user_api_key_dict # carried via logging metadata, not the wire payload
url = f"{self.api_base}{self.inspect_path}"
payload = self._build_mcp_response_payload(
request_data=request_data,
response=response,
)
if payload is None:
verbose_proxy_logger.debug(
"Cisco AI Defense guardrail: could not build MCP response "
"payload, skipping"
)
return {}
start_time = datetime.now()
try:
inspect_response = await self._post_inspection(
url=url, payload=payload, surface="mcp"
)
except HTTPException:
raise
except Exception as exc:
return self._handle_api_error(
exc,
request_data=request_data,
start_time=start_time,
surface="mcp",
direction="output",
)
from .cisco_ai_defense import _ScanContext
return self._finalize_inspection(
inspect_response=inspect_response,
request_data=request_data,
context=_ScanContext(surface="mcp", direction="output"),
start_time=start_time,
response_obj=(
response if redact_response_obj is None else redact_response_obj
),
)
def _build_mcp_request_payload(
self,
data: dict,
) -> Optional[Dict[str, Any]]:
"""Build the JSON-RPC ``tools/call`` envelope sent to ``/inspect/mcp``.
The Cisco AI Defense MCP inspect endpoint expects the JSON-RPC
envelope itself as the request body — *not* wrapped under a
``request`` key with sibling ``metadata`` / ``config`` keys. Policies
are applied based on the API key linked to the request. Operator
metadata (user, call id, src/dst app, etc.) is carried out-of-band
via the standard logging payload so the wire contract stays
identical to a hand-rolled ``curl`` against ``/inspect/mcp``.
"""
if data.get("jsonrpc") == "2.0":
return {
"jsonrpc": "2.0",
"id": (data.get("id") or data.get("litellm_call_id") or "litellm-mcp"),
"method": data.get("method") or "tools/call",
"params": data.get("params") or {},
}
tool_name = (
data.get("mcp_tool_name") or data.get("tool_name") or data.get("name")
)
if not tool_name:
return None
arguments = data.get("mcp_arguments")
if arguments is None:
arguments = data.get("arguments")
return {
"jsonrpc": "2.0",
"id": data.get("litellm_call_id") or "litellm-mcp",
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": (arguments if isinstance(arguments, dict) else {}),
},
}
def _build_mcp_response_payload(
self,
request_data: dict,
response: object,
) -> Optional[Dict[str, Any]]:
"""Build the MCP response-inspection body sent to ``/inspect/mcp``."""
request_payload = self._build_mcp_request_payload(data=request_data)
if request_payload is None:
return None
normalized = self._normalize_mcp_response(response)
if normalized is None:
return None
payload = dict(request_payload)
response_id = normalized.get("id")
if response_id not in (None, "litellm-mcp"):
payload["id"] = response_id
elif payload.get("id") in (None, "litellm-mcp"):
request_id = request_data.get("litellm_call_id") or request_data.get("id")
if request_id:
payload["id"] = request_id
if "result" in normalized:
payload["result"] = normalized["result"]
if "error" in normalized:
payload["error"] = normalized["error"]
return payload
@staticmethod
def _hydrate_mcp_tool_context(request_data: Dict[str, Any]) -> None:
metadata = request_data.get("mcp_tool_call_metadata")
if metadata is None:
nested = request_data.get("metadata") or request_data.get(
"litellm_metadata"
)
if isinstance(nested, dict):
metadata = nested.get("mcp_tool_call_metadata")
if not isinstance(metadata, dict):
return
name = metadata.get("name")
arguments = metadata.get("arguments")
server_name = metadata.get("mcp_server_name")
if name:
request_data.setdefault("mcp_tool_name", name)
request_data.setdefault("tool_name", name)
request_data.setdefault("name", name)
if arguments is not None:
request_data.setdefault("mcp_arguments", arguments)
request_data.setdefault("arguments", arguments)
if server_name:
request_data.setdefault("mcp_server_name", server_name)
request_data.setdefault("server_name", server_name)
@staticmethod
def _normalize_mcp_response(response: object) -> Optional[Dict[str, Any]]:
"""Normalize an MCP tool response into a JSON-RPC envelope.
Handles JSON-RPC dicts, raw content lists, MCP SDK models, and
Pydantic-coerced ``[(field_name, value)]`` lists.
"""
if isinstance(response, dict):
if response.get("jsonrpc") == "2.0":
return dict(response)
if isinstance(response.get("result"), dict):
return {
"jsonrpc": "2.0",
"id": response.get("id") or "litellm-mcp",
"result": response["result"],
}
content = response.get("content")
if isinstance(content, list):
return {
"jsonrpc": "2.0",
"id": response.get("id") or "litellm-mcp",
"result": _CiscoAIDefenseMcpMixin._build_mcp_result(
content=content, source=response
),
}
if isinstance(response, list):
if response and all(
isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)
for item in response
):
response_fields = dict(response)
inner_content = response_fields.get("content")
if isinstance(inner_content, list):
return {
"jsonrpc": "2.0",
"id": "litellm-mcp",
"result": _CiscoAIDefenseMcpMixin._build_mcp_result(
content=inner_content, source=response_fields
),
}
else:
return None
return {
"jsonrpc": "2.0",
"id": "litellm-mcp",
"result": _CiscoAIDefenseMcpMixin._build_mcp_result(content=response),
}
model_dump = getattr(response, "model_dump", None)
if callable(model_dump):
try:
dumped = model_dump(exclude_none=True)
except TypeError:
dumped = model_dump()
if isinstance(dumped, dict):
return _CiscoAIDefenseMcpMixin._normalize_mcp_response(dumped)
content = getattr(response, "content", None)
if isinstance(content, list):
return {
"jsonrpc": "2.0",
"id": "litellm-mcp",
"result": _CiscoAIDefenseMcpMixin._build_mcp_result(
content=content, source=response
),
}
return None
@staticmethod
def _build_mcp_result(
content: List[Any],
source: object = None,
) -> Dict[str, Any]:
result: Dict[str, Any] = {
"content": [_serialize_mcp_content_item(item) for item in content]
}
for key in ("structuredContent", "isError"):
value = (
source.get(key)
if isinstance(source, dict)
else getattr(source, key, None)
)
if value is not None and (key != "isError" or isinstance(value, bool)):
result[key] = value
return result
# ------------------------------------------------------------------
# MCP redact (in-place rewrite of tool output)
# ------------------------------------------------------------------
@staticmethod
def _set_mcp_tool_response_text(response_obj: object, text: str) -> bool:
"""Replace text content in any supported MCP response shape."""
if response_obj is None:
return False
inner = getattr(response_obj, "mcp_tool_call_response", None)
if inner is not None:
return _CiscoAIDefenseMcpMixin._set_mcp_tool_response_text(inner, text)
content_list = _CiscoAIDefenseMcpMixin._coerce_to_content_list(response_obj)
replaced = False
if isinstance(content_list, list):
for item in content_list:
if isinstance(item, dict) and item.get("type") == "text":
item["text"] = text
replaced = True
elif hasattr(item, "type") and getattr(item, "type", None) == "text":
try:
setattr(item, "text", text)
replaced = True
except (AttributeError, TypeError, ValueError):
continue
replacement = {"result": text}
if (
isinstance(response_obj, list)
and response_obj
and all(
isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)
for item in response_obj
)
):
for index, item in enumerate(response_obj):
if item[0] == "structuredContent":
response_obj[index] = (item[0], replacement)
replaced = True
elif hasattr(response_obj, "structuredContent"):
try:
setattr(response_obj, "structuredContent", replacement)
replaced = True
except (AttributeError, TypeError, ValueError):
pass
elif isinstance(response_obj, dict):
result = response_obj.get("result")
target: Dict[Any, Any] = (
result if isinstance(result, dict) else response_obj
)
if "structuredContent" in target:
target["structuredContent"] = replacement
replaced = True
return replaced
@staticmethod
def _coerce_to_content_list(response_obj: object) -> Optional[List[Any]]:
"""Find the MCP content list inside supported response shapes."""
if response_obj is None:
return None
inner = getattr(response_obj, "mcp_tool_call_response", None)
if inner is not None:
return _CiscoAIDefenseMcpMixin._coerce_to_content_list(inner)
content = getattr(response_obj, "content", None)
if isinstance(content, list):
return content
if isinstance(response_obj, list):
if response_obj and all(
isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str)
for item in response_obj
):
inner_content = dict(response_obj).get("content")
if isinstance(inner_content, list):
return inner_content
return None
return response_obj
return None
# ------------------------------------------------------------------
# MCP-specific verdict extraction
# ------------------------------------------------------------------
@staticmethod
def _extract_sanitized_mcp_arguments(
inspect_response: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
"""Pull sanitized MCP tool-call arguments off the verdict.
Cisco can return them at the top level (``params.arguments``) or
under ``sanitized_payload`` / ``modified_payload``.
"""
containers = [inspect_response]
for container_key in ("result", "data"):
container = inspect_response.get(container_key)
if isinstance(container, dict):
containers.append(container)
for container in containers:
params = container.get("params")
if isinstance(params, dict):
args = params.get("arguments")
if isinstance(args, dict) and args:
return dict(args)
for key in (
"sanitized_payload",
"sanitizedPayload",
"modified_payload",
"modifiedPayload",
):
payload = container.get(key)
if isinstance(payload, dict):
inner_params = payload.get("params")
if isinstance(inner_params, dict):
args = inner_params.get("arguments")
if isinstance(args, dict) and args:
return dict(args)
direct = payload.get("arguments")
if isinstance(direct, dict) and direct:
return dict(direct)
return None

5
litellm/types/guardrails.py
View file

@ -47,6 +47,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.qohash import (
from litellm.types.proxy.guardrails.guardrail_hooks.vigil_guard import (
VigilGuardGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
CiscoAIDefenseGuardrailConfigModel,
)
"""
Pydantic object defining how to set guardrails on litellm proxy
@ -80,6 +83,7 @@ class SupportedGuardrailIntegrations(Enum):
PILLAR = "pillar"
GRAYSWAN = "grayswan"
PANW_PRISMA_AIRS = "panw_prisma_airs"
CISCO_AI_DEFENSE = "cisco_ai_defense"
AZURE_PROMPT_SHIELD = "azure/prompt_shield"
AZURE_TEXT_MODERATIONS = "azure/text_moderations"
MODEL_ARMOR = "model_armor"
@ -840,6 +844,7 @@ class Mode(BaseModel):
class LitellmParams(
CiscoAIDefenseGuardrailConfigModel,
PresidioConfigModel,
BedrockGuardrailConfigModel,
LakeraV2GuardrailConfigModel,

148
litellm/types/proxy/guardrails/guardrail_hooks/cisco_ai_defense.py Normal file
View file

@ -0,0 +1,148 @@
"""
Cisco AI Defense Guardrail Config Model
"""
from typing import List, Literal, Optional
from pydantic import BaseModel, ConfigDict, Field
from .base import GuardrailConfigModel
CISCO_AI_DEFENSE_RULE_NAMES = Literal[
"Code Detection",
"Harassment",
"Hate Speech",
"PCI",
"PHI",
"PII",
"Prompt Injection",
"Profanity",
"Sexual Content & Exploitation",
"Social Division & Polarization",
"Violence & Public Safety Threats",
]
# Inspection surfaces supported by Cisco AI Defense. The Cisco Inspection API
# exposes two separate endpoints — one for LLM chat conversations and one for
# MCP tool calls. The user picks exactly one surface to scan per guardrail
# instance; configure two guardrails if you need to scan both.
CISCO_AI_DEFENSE_INSPECTION_TYPE = Literal["chat", "mcp"]
class CiscoAIDefenseRule(BaseModel):
"""A single rule to enable for Cisco AI Defense inspection."""
rule_name: CISCO_AI_DEFENSE_RULE_NAMES = Field(
description="The canonical Cisco AI Defense rule name to evaluate.",
)
entity_types: Optional[List[str]] = Field(
default=None,
description=(
"Optional list of entity types for the rule (e.g. 'Email Address', "
"'Phone Number'). Applies to rules such as PII, PCI, and PHI."
),
)
class CiscoAIDefenseGuardrailConfigModelOptionalParams(BaseModel):
"""Optional parameters for the Cisco AI Defense guardrail."""
model_config = ConfigDict(extra="allow")
inspection_type: CISCO_AI_DEFENSE_INSPECTION_TYPE = Field(
default="chat",
description=(
"Which Cisco AI Defense inspection surface to use. "
"'chat' scans LLM model conversations via /api/v1/inspect/chat. "
"'mcp' scans MCP tool calls via /api/v1/inspect/mcp. "
"Each guardrail instance targets exactly one surface; configure "
"two guardrails to scan both chat and MCP traffic."
),
)
inspect_path: Optional[str] = Field(
default=None,
description=(
"Override for the inspection endpoint path. Defaults to "
"/api/v1/inspect/chat when inspection_type='chat' and "
"/api/v1/inspect/mcp when inspection_type='mcp'."
),
)
enabled_rules: Optional[List[CiscoAIDefenseRule]] = Field(
default=None,
description=(
"Explicit list of Cisco AI Defense rules to evaluate. If omitted, "
"the policies configured for the API key in the Cisco AI Defense "
"UI are used."
),
)
integration_profile_id: Optional[str] = Field(
default=None,
description="Integration profile id to apply (advanced).",
)
integration_profile_version: Optional[str] = Field(
default=None,
description="Integration profile version to apply (advanced).",
)
integration_tenant_id: Optional[str] = Field(
default=None,
description="Integration tenant id to apply (advanced).",
)
integration_type: Optional[str] = Field(
default=None,
description="Integration type to apply (advanced).",
)
on_flagged_action: Optional[str] = Field(
default="block",
description=(
"Action to take when Cisco AI Defense flags content. 'block' raises "
"an HTTPException; 'monitor' logs the detection and lets the "
"request continue."
),
)
fallback_on_error: Optional[Literal["allow", "block"]] = Field(
default="block",
description=(
"Behaviour when the Cisco AI Defense API is unavailable: 'allow' "
"proceeds without scanning (high availability), 'block' rejects "
"the request (maximum security)."
),
)
timeout: Optional[float] = Field(
default=10.0,
ge=1.0,
le=60.0,
description="Timeout (seconds) for Cisco AI Defense API calls (1-60).",
)
class CiscoAIDefenseGuardrailConfigModel(
GuardrailConfigModel[CiscoAIDefenseGuardrailConfigModelOptionalParams]
):
"""Configuration parameters for the Cisco AI Defense guardrail."""
api_key: Optional[str] = Field(
default=None,
description=(
"API key for the Cisco AI Defense inspection endpoint. If "
"not provided, the `CISCO_AI_DEFENSE_API_KEY` environment variable "
"is used. Sent in the `X-Cisco-AI-Defense-API-Key` header. "
"Both the chat and MCP endpoints use this key."
),
)
api_base: Optional[str] = Field(
default=None,
description=(
"Regional base URL for the Cisco AI Defense Inspection API. "
"Defaults to https://us.api.inspect.aidefense.security.cisco.com. "
"Supported regions: us (us-west-2), ap (ap-ne-1), eu "
"(eu-central-1). The environment variable "
"`CISCO_AI_DEFENSE_API_BASE` is consulted as a fallback. The "
"endpoint path is derived from inspection_type "
"(/api/v1/inspect/chat for 'chat', /api/v1/inspect/mcp for 'mcp')."
),
)
@staticmethod
def ui_friendly_name() -> str:
return "Cisco AI Defense"

6
tests/llm_translation/reasoning_effort_grid/grid_spec.py
View file

@ -141,6 +141,12 @@ ANTHROPIC_DIRECT_MODELS: Tuple[ModelEntry, ...] = (
mode="adaptive",
required_env=_ANTHROPIC_REQ,
caps=_CAPS_XHIGH_MAX,
fail_reason=(
"claude-fable-5 is not yet released on the Anthropic API for the CI "
"account; Anthropic returns not_found_error until the model is "
"available, so this cell stays loud in CI. Remove this fail_reason "
"once the model is available."
),
),
ModelEntry(
alias="claude-opus-4-8",

362
tests/test_litellm/proxy/guardrails/guardrail_hooks/_cisco_ai_defense_test_utils.py Normal file
View file

@ -0,0 +1,362 @@
import json
import os
import sys
from contextlib import contextmanager
from datetime import datetime
from types import SimpleNamespace
from typing import Any, Dict
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import HTTPException
from httpx import Request, Response
from litellm.types.utils import (
Choices,
Delta,
Message,
ModelResponse,
ModelResponseStream,
StreamingChoices,
TextChoices,
TextCompletionResponse,
)
def _make_text_completion_response(text: str) -> TextCompletionResponse:
return TextCompletionResponse(
choices=[{"text": text, "index": 0, "finish_reason": "stop"}]
)
def _make_model_response_with_content(content: str) -> ModelResponse:
return ModelResponse(
choices=[
Choices(
index=0,
finish_reason="stop",
message=Message(role="assistant", content=content),
)
]
)
sys.path.insert(0, os.path.abspath("../.."))
import litellm
from litellm import DualCache
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
CiscoAIDefenseGuardrail,
CiscoAIDefenseGuardrailMissingSecrets,
)
from litellm.proxy.guardrails.init_guardrails import init_guardrails_v2
CISCO_BASE = "https://us.api.inspect.aidefense.security.cisco.com"
CHAT_URL = f"{CISCO_BASE}/api/v1/inspect/chat"
MCP_URL = f"{CISCO_BASE}/api/v1/inspect/mcp"
@contextmanager
def _patch_inspection_post(g: CiscoAIDefenseGuardrail, post_mock: Any):
async def _send(request: Request, **kwargs: Any) -> Response:
return await post_mock(
url=str(request.url),
headers=request.headers,
json=json.loads(request.content.decode("utf-8")),
follow_redirects=kwargs.get("follow_redirects"),
)
with patch.object(g.async_handler.client, "send", new=_send):
yield post_mock
def _mock_inspect_response(
json_body: dict, *, status: int = 200, url: str = CHAT_URL
) -> Response:
return Response(
status_code=status,
json=json_body,
request=Request(method="POST", url=url),
)
def _safe_response(url: str = CHAT_URL) -> Response:
return _mock_inspect_response(
{
"is_safe": True,
"classifications": [],
"severity": "NONE_SEVERITY",
"rules": [],
"action": "allow",
},
url=url,
)
def _violation_response(url: str = CHAT_URL) -> Response:
return _mock_inspect_response(
{
"is_safe": False,
"classifications": ["SECURITY_VIOLATION", "PRIVACY_VIOLATION"],
"severity": "HIGH",
"rules": [
{"rule_name": "Prompt Injection"},
{"rule_name": "PII", "entity_types": ["Email Address"]},
],
"explanation": "Detected jailbreak attempt with PII exfiltration",
"event_id": "evt_123",
"action": "block",
},
url=url,
)
def _mcp_request(name="lookup", args=None, jsonrpc=False, **extra):
args = args if args is not None else {}
if jsonrpc:
return {
"jsonrpc": "2.0",
"id": "1",
"method": "tools/call",
"params": {"name": name, "arguments": args},
**extra,
}
return {"mcp_tool_name": name, "mcp_arguments": args, **extra}
def _mcp_response(content=None, response_cost=0.0):
if content is None:
content = [{"type": "text", "text": "ok"}]
return SimpleNamespace(
mcp_tool_call_response=content,
hidden_params=SimpleNamespace(response_cost=response_cost),
)
def _mcp_result_text(content) -> str:
if not content:
return ""
item = content[0] if isinstance(content, list) else content
return getattr(item, "text", None) or item.get("text", "")
def _chat_request_tool_call_args(arguments: str) -> dict:
return {
"messages": [
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "send_data",
"arguments": arguments,
},
}
],
}
]
}
def _chat_request_function_call_args(arguments: str) -> dict:
return {
"messages": [
{
"role": "assistant",
"content": None,
"function_call": {
"name": "exfil",
"arguments": arguments,
},
}
]
}
def _redact_response(
*,
sanitized_text=None,
sanitized_messages=None,
sanitized_mcp_arguments=None,
sanitized_payload=None,
classifications=("PRIVACY_VIOLATION",),
rules=({"rule_name": "PII"},),
severity="HIGH",
url=CHAT_URL,
):
body = {
"is_safe": False,
"classifications": list(classifications),
"severity": severity,
"rules": list(rules),
"action": "redact",
}
if sanitized_text is not None:
body["sanitized_text"] = sanitized_text
if sanitized_messages is not None:
body["sanitized_messages"] = sanitized_messages
if sanitized_mcp_arguments is not None:
body["sanitized_mcp_arguments"] = sanitized_mcp_arguments
if sanitized_payload is not None:
body["sanitized_payload"] = sanitized_payload
return _mock_inspect_response(body, url=url)
def _responses_api_response(text, role="assistant"):
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.responses.main import GenericResponseOutputItem, OutputText
return ResponsesAPIResponse(
id="resp_1",
created_at=0,
output=[
GenericResponseOutputItem(
type="message",
id="msg_1",
status="completed",
role=role,
content=[OutputText(type="output_text", text=text, annotations=[])],
)
],
parallel_tool_calls=False,
tool_choice=None,
tools=None,
top_p=None,
usage=None,
)
def _make_guardrail(
inspection_type="chat",
event_hook="pre_call",
*,
name="t",
api_key="x",
default_on=True,
**kwargs,
):
return CiscoAIDefenseGuardrail(
guardrail_name=name,
api_key=api_key,
inspection_type=inspection_type,
event_hook=event_hook,
default_on=default_on,
**kwargs,
)
def _find_callback(name):
from litellm.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
CiscoAIDefenseGuardrail,
)
for cb in litellm.callbacks:
if isinstance(cb, CiscoAIDefenseGuardrail) and cb.guardrail_name == name:
return cb
raise AssertionError(f"Cisco guardrail {name!r} not in litellm.callbacks")
def _make_streaming_chunks(parts):
chunks = []
for i, part in enumerate(parts):
chunks.append(
ModelResponseStream(
id="resp_1",
choices=[
StreamingChoices(
delta=Delta(content=part, role="assistant" if i == 0 else None),
finish_reason="stop" if i == len(parts) - 1 else None,
index=0,
)
],
created=1234567890,
model="gpt-4",
object="chat.completion.chunk",
)
)
return chunks
async def _aiter(items):
for item in items:
yield item
async def _streaming_setup(
g,
chunks,
cisco_response=None,
upstream=None,
request_data=None,
post_mock=None,
):
if post_mock is None:
post_mock = (
AsyncMock(return_value=cisco_response) if cisco_response else AsyncMock()
)
stream_source = upstream if upstream is not None else _aiter(chunks)
if request_data is None:
request_data = {"messages": [{"role": "user", "content": "hi"}]}
received: list = []
with _patch_inspection_post(g, post_mock):
async for chunk in g.async_post_call_streaming_iterator_hook(
user_api_key_dict=UserAPIKeyAuth(),
response=stream_source,
request_data=request_data,
):
received.append(chunk)
return received, post_mock
__all__ = [
"Any",
"AsyncMock",
"CHAT_URL",
"CISCO_BASE",
"Choices",
"CiscoAIDefenseGuardrail",
"CiscoAIDefenseGuardrailMissingSecrets",
"Delta",
"Dict",
"DualCache",
"HTTPException",
"MCP_URL",
"Message",
"ModelResponse",
"ModelResponseStream",
"Request",
"Response",
"SimpleNamespace",
"StreamingChoices",
"TextChoices",
"TextCompletionResponse",
"UserAPIKeyAuth",
"_aiter",
"_chat_request_function_call_args",
"_chat_request_tool_call_args",
"_find_callback",
"_make_guardrail",
"_make_model_response_with_content",
"_make_streaming_chunks",
"_make_text_completion_response",
"_mcp_request",
"_mcp_response",
"_mcp_result_text",
"_mock_inspect_response",
"_patch_inspection_post",
"_redact_response",
"_responses_api_response",
"_safe_response",
"_streaming_setup",
"_violation_response",
"contextmanager",
"datetime",
"init_guardrails_v2",
"json",
"litellm",
"os",
"patch",
"pytest",
"sys",
]

2842
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_cisco_ai_defense_chat.py Normal file
View file

File diff suppressed because it is too large Load diff

832
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_cisco_ai_defense_mcp.py Normal file
View file

@ -0,0 +1,832 @@
from tests.test_litellm.proxy.guardrails.guardrail_hooks._cisco_ai_defense_test_utils import (
Any,
AsyncMock,
CiscoAIDefenseGuardrail,
Dict,
DualCache,
HTTPException,
MCP_URL,
Response,
SimpleNamespace,
UserAPIKeyAuth,
_make_guardrail,
_make_model_response_with_content,
_mcp_request,
_mcp_response,
_mcp_result_text,
_mock_inspect_response,
_patch_inspection_post,
_redact_response,
_safe_response,
_violation_response,
datetime,
init_guardrails_v2,
json,
litellm,
pytest,
)
def test_cisco_ai_defense_config_via_init_v2_mcp(monkeypatch):
monkeypatch.setenv("CISCO_AI_DEFENSE_API_KEY", "test-key")
litellm.guardrail_name_config_map = {}
init_guardrails_v2(
all_guardrails=[
{
"guardrail_name": "cisco-mcp",
"litellm_params": {
"guardrail": "cisco_ai_defense",
"mode": "pre_mcp_call",
"default_on": True,
"optional_params": {"inspection_type": "mcp"},
},
}
],
config_file_path="",
)
class TestCiscoAIDefenseMCPMode:
@pytest.mark.asyncio
async def test_mcp_mode_inspects_mcp_request(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = _mcp_request(
name="send_email", args={"to": "x@y.com"}, litellm_call_id="call-1"
)
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert result == data
assert post_mock.call_args.kwargs["url"] == MCP_URL
assert post_mock.call_args.kwargs["follow_redirects"] is False
sent_payload = post_mock.call_args.kwargs["json"]
assert sent_payload["jsonrpc"] == "2.0"
assert sent_payload["method"] == "tools/call"
assert sent_payload["params"]["name"] == "send_email"
assert sent_payload["params"]["arguments"] == {"to": "x@y.com"}
assert "request" not in sent_payload
assert "metadata" not in sent_payload
assert "config" not in sent_payload
@pytest.mark.asyncio
async def test_mcp_mode_blocks_violation(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = _mcp_request(name="leak_secrets", args={"target": "evil"})
with _patch_inspection_post(
g, AsyncMock(return_value=_violation_response(url=MCP_URL))
):
with pytest.raises(HTTPException) as exc:
await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert exc.value.detail["surface"] == "mcp"
@pytest.mark.asyncio
async def test_mcp_mode_skips_chat_traffic(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = {"messages": [{"role": "user", "content": "hello"}]}
post_mock = AsyncMock()
with _patch_inspection_post(g, post_mock):
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="completion",
)
assert result == data
post_mock.assert_not_called()
@pytest.mark.asyncio
async def test_mcp_mode_inspects_jsonrpc_envelope(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = _mcp_request(name="do_thing", args={"x": 1}, jsonrpc=True, id="abc")
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
sent_payload = post_mock.call_args.kwargs["json"]
assert sent_payload["jsonrpc"] == "2.0"
assert sent_payload["id"] == "abc"
assert sent_payload["params"]["name"] == "do_thing"
assert sent_payload["params"]["arguments"] == {"x": 1}
@pytest.mark.parametrize(
"verdict_extra",
[
{"sanitized_payload": {"params": {"arguments": {"note": "ssn [REDACTED]"}}}},
{"sanitized_text": "ssn [REDACTED]"},
],
ids=["structured_arguments", "sanitized_text_fallback"],
)
@pytest.mark.asyncio
async def test_mcp_input_redaction_reaches_tool_call(self, verdict_extra):
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.utils import ProxyLogging
original_args = {"note": "ssn 123-45-6789"}
sanitized_args = {"note": "ssn [REDACTED]"}
g = _make_guardrail(
inspection_type="mcp",
event_hook="pre_mcp_call",
on_flagged_action="monitor",
)
data = _mcp_request(name="send_email", args=dict(original_args))
cisco_resp = _mock_inspect_response(
{
"is_safe": False,
"classifications": ["PRIVACY_VIOLATION"],
"severity": "HIGH",
"rules": [{"rule_name": "PII"}],
"action": "redact",
**verdict_extra,
},
url=MCP_URL,
)
with _patch_inspection_post(g, AsyncMock(return_value=cisco_resp)):
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
forwarded = ProxyLogging(
user_api_key_cache=UserApiKeyCache()
)._convert_mcp_hook_response_to_kwargs(
response_data=result, original_kwargs={"arguments": dict(original_args)}
)
assert forwarded["arguments"] == sanitized_args, (
"Sanitized MCP arguments did not reach the tool call. The proxy "
"bridge forwards redactions only via ``modified_arguments``, so a "
"redact verdict proceeded while the original unsanitized arguments "
f"still hit the MCP server. Got: {forwarded['arguments']!r}"
)
@pytest.mark.asyncio
async def test_mcp_response_hook_inspects_tool_output(self):
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
response_obj = _mcp_response(
SimpleNamespace(
content=[{"type": "text", "text": "Here is the secret API key abc123"}]
)
)
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
kwargs = {
"name": "lookup_secret",
"arguments": {"key": "production"},
"mcp_server_name": "vault",
"litellm_call_id": "call-42",
}
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs=kwargs,
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert result is None
assert post_mock.called
assert post_mock.call_args.kwargs["url"] == MCP_URL
sent_payload = post_mock.call_args.kwargs["json"]
assert sent_payload["jsonrpc"] == "2.0"
assert sent_payload["id"] == "call-42"
assert sent_payload["method"] == "tools/call"
assert sent_payload["params"] == {
"name": "lookup_secret",
"arguments": {"key": "production"},
}
assert sent_payload["result"]["content"][0]["text"] == (
"Here is the secret API key abc123"
)
assert "request" not in sent_payload
assert "metadata" not in sent_payload
@pytest.mark.asyncio
async def test_mcp_response_hook_blocks_violation(self):
from litellm.types.mcp import MCPPostCallResponseObject
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
response_obj = _mcp_response(
SimpleNamespace(content=[{"type": "text", "text": "leaked"}])
)
post_mock = AsyncMock(return_value=_violation_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs={"name": "leak", "arguments": {}},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert result is not None, (
"MCP response block was silently dropped — the litellm "
"dispatcher swallows raised exceptions, so the hook must "
"return a non-None MCPPostCallResponseObject to enforce a block."
)
assert isinstance(result, MCPPostCallResponseObject)
replacement = result.mcp_tool_call_response
assert len(replacement) == 1
text = _mcp_result_text(replacement)
assert "Blocked by Cisco AI Defense" in text
assert "evt_123" in text
assert "SECURITY_VIOLATION" in text
@pytest.mark.asyncio
async def test_mcp_response_hook_skipped_in_chat_mode(self):
g = _make_guardrail()
response_obj = _mcp_response(
SimpleNamespace(content=[{"type": "text", "text": "hi"}])
)
post_mock = AsyncMock()
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs={"name": "tool", "arguments": {}},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert result is None
post_mock.assert_not_called()
@pytest.mark.asyncio
async def test_post_call_skipped_for_mcp_mode_guardrail(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = {"messages": [{"role": "user", "content": "hi"}]}
response = _make_model_response_with_content("fine")
post_mock = AsyncMock()
with _patch_inspection_post(g, post_mock):
result = await g.async_post_call_success_hook(
data=data,
user_api_key_dict=UserAPIKeyAuth(),
response=response,
)
assert result is response
post_mock.assert_not_called()
@pytest.mark.asyncio
async def test_mcp_response_hook_runs_with_pre_mcp_call_only(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
response_obj = _mcp_response(
SimpleNamespace(
content=[{"type": "text", "text": "would have been scanned"}]
)
)
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
await g.async_post_mcp_tool_call_hook(
kwargs={"name": "lookup", "arguments": {}},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert post_mock.called, (
"MCP response scan was skipped when only ``pre_mcp_call`` "
"was configured. Per product decision, pre_mcp_call means "
"'guard the MCP call' — request AND response."
)
@pytest.mark.parametrize(
"cisco_response_kind,expected_block",
[("safe", False), ("violation", True)],
)
@pytest.mark.asyncio
async def test_mcp_response_hook_handles_raw_list_content(
self, cisco_response_kind, expected_block
):
from litellm.types.mcp import MCPPostCallResponseObject
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
text_content = (
"exfiltrated data: ..."
if cisco_response_kind == "violation"
else "Here is the secret API key abc123"
)
response_obj = _mcp_response([{"type": "text", "text": text_content}])
cisco_resp = (
_violation_response(url=MCP_URL)
if cisco_response_kind == "violation"
else _safe_response(url=MCP_URL)
)
post_mock = AsyncMock(return_value=cisco_resp)
kwargs = {
"name": "leak" if expected_block else "lookup_secret",
"arguments": {"key": "production"} if not expected_block else {},
"mcp_server_name": "vault",
"litellm_call_id": "call-raw-list",
}
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs=kwargs,
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert post_mock.called, (
"MCP response inspect was silently skipped for raw-list "
"shape — _normalize_mcp_response failed."
)
assert post_mock.call_args.kwargs["url"] == MCP_URL
if expected_block:
assert isinstance(result, MCPPostCallResponseObject)
replacement = result.mcp_tool_call_response
assert len(replacement) == 1
assert "Blocked by Cisco AI Defense" in _mcp_result_text(replacement)
else:
sent_payload = post_mock.call_args.kwargs["json"]
assert sent_payload["jsonrpc"] == "2.0"
assert sent_payload["id"] == "call-raw-list"
assert sent_payload["method"] == "tools/call"
assert sent_payload["params"] == {
"name": "lookup_secret",
"arguments": {"key": "production"},
}
assert sent_payload["result"]["content"][0]["text"] == text_content
assert result is None
@pytest.mark.asyncio
async def test_mcp_response_hook_through_real_logging_wrapper(self):
from mcp.types import CallToolResult, TextContent
from litellm.types.mcp import MCPPostCallResponseObject
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
real_result = CallToolResult(
content=[TextContent(type="text", text="leak 9045629876")],
structuredContent={"patient": {"ssn": "123-45-6789"}},
isError=False,
)
wrapped = MCPPostCallResponseObject(
mcp_tool_call_response=real_result,
hidden_params={},
)
assert isinstance(wrapped.mcp_tool_call_response, list)
assert all(
isinstance(item, tuple) and len(item) == 2
for item in wrapped.mcp_tool_call_response
), (
"Pydantic coercion shape changed — update the normalizer to "
"match the new wire format."
)
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs={
"name": "leak_tool",
"arguments": {},
"mcp_server_name": "vault",
"litellm_call_id": "real-wire-call",
},
response_obj=wrapped,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert post_mock.called, (
"Inspect API not called for real CallToolResult shape — "
"_normalize_mcp_response failed to handle Pydantic's "
"iterated-BaseModel coercion."
)
assert post_mock.call_args.kwargs["url"] == MCP_URL
sent_payload = post_mock.call_args.kwargs["json"]
content_items = sent_payload["result"]["content"]
assert len(content_items) == 1, (
f"expected exactly 1 content item from the real "
f"CallToolResult.content list, got {len(content_items)}: "
f"{content_items!r}"
)
assert content_items[0].get("text") == "leak 9045629876", (
f"Cisco wire payload missed the real tool text; got "
f"{content_items[0]!r}. This means the Pydantic-coerced "
f"(field_name, value) tuple shape was serialized as text "
f"content instead of being unwrapped to find the inner "
f"``content`` field."
)
assert content_items[0].get("type") == "text"
assert sent_payload["result"]["structuredContent"] == {
"patient": {"ssn": "123-45-6789"}
}
assert sent_payload["result"]["isError"] is False
assert sent_payload["id"] == "real-wire-call"
assert sent_payload["method"] == "tools/call"
assert sent_payload["params"] == {"name": "leak_tool", "arguments": {}}
assert result is None
@pytest.mark.asyncio
async def test_mcp_response_hook_uses_standard_logging_tool_metadata(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
response_obj = _mcp_response([{"type": "text", "text": "tool output"}])
post_mock = AsyncMock(return_value=_safe_response(url=MCP_URL))
with _patch_inspection_post(g, post_mock):
result = await g.async_post_mcp_tool_call_hook(
kwargs={
"litellm_call_id": "metadata-call",
"mcp_tool_call_metadata": {
"name": "lookup_secret",
"arguments": {"key": "production"},
"mcp_server_name": "vault",
},
},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert result is None
sent_payload = post_mock.call_args.kwargs["json"]
assert sent_payload["method"] == "tools/call"
assert sent_payload["params"] == {
"name": "lookup_secret",
"arguments": {"key": "production"},
}
assert sent_payload["result"]["content"][0]["text"] == "tool output"
class TestCiscoAIDefenseRedactListShape:
@staticmethod
def _violation_with_redact_response(text: str = "[REDACTED tool output]"):
return _mock_inspect_response(
{
"is_safe": False,
"classifications": ["PRIVACY_VIOLATION"],
"severity": "HIGH",
"rules": [{"rule_name": "PII", "entity_types": ["SSN"]}],
"explanation": "PII detected, redaction available",
"event_id": "evt_redact_1",
"action": "redact",
"sanitized_text": text,
},
url=MCP_URL,
)
@staticmethod
def _raw_list_factory():
original_content = [{"type": "text", "text": "Your SSN is 123-45-6789."}]
return original_content, lambda: original_content[0]["text"]
@staticmethod
def _pydantic_tuple_list_factory():
from mcp.types import TextContent
inner_content = [TextContent(type="text", text="SSN: 123-45-6789")]
tuples_list = [
("meta", None),
("content", inner_content),
("structuredContent", {"patient": {"ssn": "123-45-6789"}}),
("isError", False),
]
return tuples_list, lambda: inner_content[0].text
@pytest.mark.parametrize(
"factory_name",
["_raw_list_factory", "_pydantic_tuple_list_factory"],
)
@pytest.mark.asyncio
async def test_redact_rewrites_mcp_response_list_shape(self, factory_name):
from litellm.types.mcp import MCPPostCallResponseObject
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
content, get_text = getattr(self, factory_name)()
response_obj = _mcp_response(content)
with _patch_inspection_post(
g, AsyncMock(return_value=self._violation_with_redact_response())
):
result = await g.async_post_mcp_tool_call_hook(
kwargs={"name": "leak", "arguments": {}},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert result is None or not isinstance(result, MCPPostCallResponseObject), (
f"Redact silently fell through to block for {factory_name}. "
f"result={result!r}"
)
assert get_text() == "[REDACTED tool output]", (
f"Redact silently failed for {factory_name}; original text "
f"not rewritten."
)
if factory_name == "_pydantic_tuple_list_factory":
structured_content = dict(content)["structuredContent"]
assert structured_content == {"result": "[REDACTED tool output]"}
assert "123-45-6789" not in json.dumps(structured_content)
@pytest.mark.asyncio
async def test_redact_rewrites_client_visible_original_response(self):
from mcp.types import CallToolResult, TextContent
from litellm.types.llms.base import HiddenParams
from litellm.types.mcp import MCPPostCallResponseObject
original_response = CallToolResult(
content=[TextContent(type="text", text="SSN: 123-45-6789")],
structuredContent={"patient": {"ssn": "123-45-6789"}},
isError=False,
)
wrapper = MCPPostCallResponseObject(
mcp_tool_call_response=original_response,
hidden_params=HiddenParams(),
)
g = _make_guardrail(
inspection_type="mcp", event_hook=["pre_mcp_call", "during_mcp_call"]
)
with _patch_inspection_post(
g, AsyncMock(return_value=self._violation_with_redact_response())
):
await g.async_post_mcp_tool_call_hook(
kwargs={
"name": "leak",
"arguments": {},
"original_response": original_response,
},
response_obj=wrapper,
start_time=datetime.now(),
end_time=datetime.now(),
)
assert original_response.content[0].text == "[REDACTED tool output]"
assert "123-45-6789" not in json.dumps(original_response.structuredContent), (
"Redact verdict left the client-visible MCP tool output unchanged. "
"The post-call hook receives a wrapped MCPPostCallResponseObject but "
"the endpoint returns kwargs['original_response'], so the redaction "
"must rewrite that object too. structuredContent still leaks: "
f"{original_response.structuredContent!r}"
)
class TestCiscoAIDefenseMcpInputRedactionFallback:
"""``sanitized_text``-only redaction of structured MCP arguments."""
@pytest.mark.asyncio
async def test_single_string_arg_is_rewritten(self):
g = _make_guardrail(inspection_type="mcp", event_hook="pre_mcp_call")
data = _mcp_request(
name="search", args={"query": "my SSN is 123-45-6789", "limit": 10}
)
cisco = _redact_response(sanitized_text="my SSN is [REDACTED]", url=MCP_URL)
with _patch_inspection_post(g, AsyncMock(return_value=cisco)):
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert result == data
assert data["mcp_arguments"]["query"] == "my SSN is [REDACTED]"
assert data["mcp_arguments"]["limit"] == 10
@pytest.mark.asyncio
async def test_ambiguous_multi_string_args_block_instead_of_leaking(self):
g = _make_guardrail(
inspection_type="mcp",
event_hook="pre_mcp_call",
on_flagged_action="block",
)
original = {"query": "PII data", "filter": "sensitive term", "limit": 10}
data = _mcp_request(name="search", args=dict(original))
cisco = _redact_response(sanitized_text="[REDACTED]", url=MCP_URL)
with _patch_inspection_post(g, AsyncMock(return_value=cisco)):
with pytest.raises(HTTPException):
await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert data["mcp_arguments"] == original
@pytest.mark.asyncio
async def test_ambiguous_multi_string_args_not_partially_redacted_in_monitor(self):
g = _make_guardrail(
inspection_type="mcp",
event_hook="pre_mcp_call",
on_flagged_action="monitor",
)
original = {"query": "PII data", "filter": "sensitive term"}
data = _mcp_request(name="search", args=dict(original))
cisco = _redact_response(sanitized_text="[REDACTED]", url=MCP_URL)
with _patch_inspection_post(g, AsyncMock(return_value=cisco)):
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert result == data
assert data["mcp_arguments"] == original
class TestCiscoAIDefenseMCPBlockingContract:
@pytest.mark.asyncio
async def test_block_response_survives_dispatcher_contract(self):
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.types.mcp import MCPPostCallResponseObject
from mcp.types import CallToolResult, TextContent
g = _make_guardrail(
name="cisco-mcp",
inspection_type="mcp",
event_hook=["pre_mcp_call", "during_mcp_call"],
)
raw_response = CallToolResult(
content=[TextContent(type="text", text="exfiltrated")],
structuredContent={"result": "exfiltrated"},
isError=False,
)
response_obj = MCPPostCallResponseObject(
mcp_tool_call_response=raw_response,
hidden_params={},
)
post_mock = AsyncMock(return_value=_violation_response(url=MCP_URL))
captured: Dict[str, Any] = {}
with _patch_inspection_post(g, post_mock):
try:
captured["result"] = await g.async_post_mcp_tool_call_hook(
kwargs={
"name": "leak",
"arguments": {},
"original_response": raw_response,
},
response_obj=response_obj,
start_time=datetime.now(),
end_time=datetime.now(),
)
except Exception as e:
captured["swallowed"] = repr(e)
assert "swallowed" not in captured, (
f"async_post_mcp_tool_call_hook raised — the litellm "
f"dispatcher would swallow this and the block would be lost. "
f"Got: {captured.get('swallowed')}"
)
result = captured["result"]
assert isinstance(result, MCPPostCallResponseObject), (
"Hook must keep returning a MCPPostCallResponseObject for "
"dispatcher paths that do honor returned replacements."
)
assert raw_response.isError is True
assert "Blocked by Cisco AI Defense" in raw_response.content[0].text
assert raw_response.structuredContent is not None
assert "Blocked by Cisco AI Defense" in raw_response.structuredContent["result"]
assert "exfiltrated" not in raw_response.structuredContent["result"]
logging_stub = Logging.__new__(Logging)
logging_stub.model_call_details = {}
parsed = logging_stub._parse_post_mcp_call_hook_response(response=result)
assert parsed is not None
assert "Blocked by Cisco AI Defense" in _mcp_result_text(parsed)
class TestCiscoAIDefenseJsonRpcSuccessEnvelope:
@staticmethod
def _cisco_mcp_envelope(*, is_safe: bool, action: str = "Block") -> Response:
return _mock_inspect_response(
{
"jsonrpc": "2.0",
"id": 3,
"result": {
"is_safe": is_safe,
"action": action,
"classifications": [],
"rules": [
{
"rule_name": "PII",
"rule_id": 0,
"entity_types": [],
"classification": "NONE_VIOLATION",
}
],
"event_id": "645d9d22-b016-47e0-a12c-9d587fb11c57",
"detected_pii": [],
},
},
url=MCP_URL,
)
@pytest.mark.parametrize(
"is_safe,action,should_block",
[
(False, "Block", True),
(True, "Allow", False),
(False, "Allow", False),
(True, "Block", True),
],
)
@pytest.mark.asyncio
async def test_mcp_jsonrpc_envelope_respects_verdict(
self, is_safe, action, should_block
):
g = _make_guardrail(
name="cisco-mcp", inspection_type="mcp", event_hook="pre_mcp_call"
)
data = _mcp_request(
name="ask_question",
args={
"repoName": "facebook/react",
"question": "What is React Fiber 9045629876?",
},
)
with _patch_inspection_post(
g,
AsyncMock(
return_value=self._cisco_mcp_envelope(is_safe=is_safe, action=action)
),
):
if should_block:
with pytest.raises(HTTPException) as exc:
await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert exc.value.status_code == 400
assert exc.value.detail["surface"] == "mcp"
assert (
exc.value.detail["event_id"]
== "645d9d22-b016-47e0-a12c-9d587fb11c57"
)
else:
result = await g.async_pre_call_hook(
user_api_key_dict=UserAPIKeyAuth(),
cache=DualCache(),
data=data,
call_type="mcp_call",
)
assert result == data
@pytest.mark.parametrize(
"verdict,expected",
[
(
{
"is_safe": False,
"classifications": ["SECURITY_VIOLATION"],
"action": "block",
},
"passthrough",
),
(
{
"jsonrpc": "2.0",
"id": 1,
"result": {"is_safe": False, "action": "Block"},
},
{"is_safe": False, "action": "Block"},
),
],
)
def test_unwrap_verdict_envelope(self, verdict, expected):
unwrapped = CiscoAIDefenseGuardrail._unwrap_verdict_envelope(verdict)
if expected == "passthrough":
assert unwrapped is verdict
else:
assert unwrapped == expected

BIN
ui/litellm-dashboard/public/assets/logos/cisco.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 1.9 KiB

6
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_configs.ts
View file

@ -210,6 +210,12 @@ export const GUARDRAIL_PRESETS: Record<string, GuardrailPreset> = {
mode: "pre_call",
defaultOn: false,
},
cisco_ai_defense: {
provider: "CiscoAiDefense",
guardrailNameSuggestion: "Cisco AI Defense",
mode: "pre_call",
defaultOn: false,
},
noma: {
provider: "Noma",
guardrailNameSuggestion: "Noma Security",

10
ui/litellm-dashboard/src/components/guardrails/guardrail_garden_data.ts
View file

@ -307,6 +307,16 @@ export const PARTNER_GUARDRAIL_CARDS: GuardrailCardInfo[] = [
logo: `${ASSET_PREFIX}palo_alto_networks.jpeg`,
tags: ["Enterprise", "Security"],
},
{
id: "cisco_ai_defense",
name: "Cisco AI Defense",
description:
"Cisco AI Defense Inspection API for runtime protection: prompt injection, PII/PCI/PHI, harassment, hate speech, profanity, violence, and code detection.",
category: "partner",
logo: `${ASSET_PREFIX}cisco.png`,
tags: ["Enterprise", "Security", "Prompt Injection", "PII"],
providerKey: "CiscoAiDefense",
},
{
id: "noma",
name: "Noma Security",

1
ui/litellm-dashboard/src/components/guardrails/guardrail_info_helpers.tsx
View file

@ -123,6 +123,7 @@ export const guardrailLogoMap: Record<string, string> = {
"Azure Content Safety Text Moderation": `${asset_logos_folder}microsoft_azure.svg`,
"Aporia AI": `${asset_logos_folder}aporia.png`,
"PANW Prisma AIRS": `${asset_logos_folder}palo_alto_networks.jpeg`,
"Cisco AI Defense": `${asset_logos_folder}cisco.png`,
"Noma Security": `${asset_logos_folder}noma_security.png`,
"Javelin Guardrails": `${asset_logos_folder}javelin.png`,
"Pillar Guardrail": `${asset_logos_folder}pillar.jpeg`,