fix(guardrails): restore main's guardrails types and re-add tealtiger enum member

This commit is contained in:
AlphaRex-pixel 2026-10-11 21:57:05 +05:30
parent 195226e42e
commit e0e2e009ed

View file

@ -2,12 +2,14 @@ from collections.abc import Mapping
from datetime import datetime
from enum import Enum
from types import MappingProxyType
from typing import Final, Literal
from typing import Final, Literal, cast
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from pydantic import ConfigDict, Field, field_validator, model_validator
from typing_extensions import ReadOnly, Required, TypedDict
from litellm._logging import verbose_logger
from litellm.constants import BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import (
Agent365GuardrailConfigModel,
)
@ -23,6 +25,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import (
from litellm.types.proxy.guardrails.guardrail_hooks.compresr import (
CompresrGuardrailConfigModel,
)
from litellm.types.proxy.guardrails.guardrail_hooks.decision_model import (
DecisionModelCheck,
)
from litellm.types.proxy.guardrails.guardrail_hooks.enkryptai import (
EnkryptAIGuardrailConfigs,
)
@ -145,9 +150,10 @@ class SupportedGuardrailIntegrations(Enum):
STRAIKER = "straiker"
ALICE = "alice"
AGENT_365 = "agent_365"
LLM_SHIELD_PROXY = "llm_shield_proxy"
CONDUCT = "conduct"
TEALTIGER = "tealtiger"
DECISION_MODEL = "decision_model"
TEALTIGER = 'tealtiger'
class Role(Enum):
@ -167,7 +173,7 @@ class GuardrailItemSpec(TypedDict, total=False):
callback_args: dict[str, dict]
class GuardrailItem(BaseModel):
class GuardrailItem(LiteLLMBaseModel):
callbacks: list[str]
default_on: bool
logging_only: bool | None
@ -437,7 +443,7 @@ class GuardrailParamUITypes(str, Enum):
PERCENTAGE = "percentage"
class PresidioPresidioConfigModelUserInterface(BaseModel):
class PresidioPresidioConfigModelUserInterface(LiteLLMBaseModel):
"""Configuration parameters for the Presidio PII masking guardrail on LiteLLM UI"""
presidio_analyzer_api_base: str | None = Field(
@ -546,31 +552,31 @@ BedrockChecksSensitiveInformationEntity = Literal[
]
class BedrockChecksContentFilterCategoryItem(BaseModel):
class BedrockChecksContentFilterCategoryItem(LiteLLMBaseModel):
category: BedrockChecksContentFilterCategory
class BedrockChecksContentFilterModel(BaseModel):
class BedrockChecksContentFilterModel(LiteLLMBaseModel):
categories: list[BedrockChecksContentFilterCategoryItem]
class BedrockChecksPromptAttackCategoryItem(BaseModel):
class BedrockChecksPromptAttackCategoryItem(LiteLLMBaseModel):
category: BedrockChecksPromptAttackCategory
class BedrockChecksPromptAttackModel(BaseModel):
class BedrockChecksPromptAttackModel(LiteLLMBaseModel):
categories: list[BedrockChecksPromptAttackCategoryItem]
class BedrockChecksSensitiveInformationEntityItem(BaseModel):
class BedrockChecksSensitiveInformationEntityItem(LiteLLMBaseModel):
type: BedrockChecksSensitiveInformationEntity
class BedrockChecksSensitiveInformationModel(BaseModel):
class BedrockChecksSensitiveInformationModel(LiteLLMBaseModel):
entities: list[BedrockChecksSensitiveInformationEntityItem]
class BedrockChecksConfigModel(BaseModel):
class BedrockChecksConfigModel(LiteLLMBaseModel):
"""Inline `checks` config for the resource-less Bedrock InvokeGuardrailChecks API.
Include only the checks you want to run; at least one must be set.
@ -589,7 +595,7 @@ class BedrockChecksConfigModel(BaseModel):
return self
class BedrockGuardrailConfigModel(BaseModel):
class BedrockGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the AWS Bedrock guardrail"""
guardrailIdentifier: str | None = Field(default=None, description="The ID of your guardrail on Bedrock")
@ -664,7 +670,7 @@ class BedrockGuardrailConfigModel(BaseModel):
)
class BedrockGuardrailStreamingParams(BaseModel):
class BedrockGuardrailStreamingParams(LiteLLMBaseModel):
streaming_buffer_until_moderated: bool = Field(
default=True,
description="If True (default), withhold every streamed chunk until the end-of-stream "
@ -704,7 +710,7 @@ class BedrockGuardrailStreamingParams(BaseModel):
)
class LakeraV2GuardrailConfigModel(BaseModel):
class LakeraV2GuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Lakera AI v2 guardrail"""
api_key: str | None = Field(default=None, description="API key for the Lakera AI service")
@ -729,7 +735,7 @@ class LakeraV2GuardrailConfigModel(BaseModel):
)
class LassoGuardrailConfigModel(BaseModel):
class LassoGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Lasso guardrail"""
lasso_user_id: str | None = Field(default=None, description="User ID for the Lasso guardrail")
@ -737,7 +743,7 @@ class LassoGuardrailConfigModel(BaseModel):
mask: bool | None = Field(default=False, description="Enable content masking using Lasso classifix API")
class DeepKeepGuardrailConfigModel(BaseModel):
class DeepKeepGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the DeepKeep AI Firewall guardrail"""
deepkeep_firewall_id: str | None = Field(
@ -749,7 +755,7 @@ class DeepKeepGuardrailConfigModel(BaseModel):
)
class PillarGuardrailConfigModel(BaseModel):
class PillarGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Pillar Security guardrail"""
on_flagged_action: str | None = Field(
@ -774,7 +780,7 @@ class PillarGuardrailConfigModel(BaseModel):
)
class NomaGuardrailConfigModel(BaseModel):
class NomaGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Noma Security guardrail"""
use_v2: bool | None = Field(
@ -803,7 +809,7 @@ class NomaGuardrailConfigModel(BaseModel):
)
class ZscalerAIGuardConfigModel(BaseModel):
class ZscalerAIGuardConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Zscaler AI Guard guardrail"""
policy_id: int | None = Field(
@ -821,7 +827,7 @@ class ZscalerAIGuardConfigModel(BaseModel):
)
class JavelinGuardrailConfigModel(BaseModel):
class JavelinGuardrailConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the Javelin guardrail"""
guard_name: str | None = Field(default=None, description="Name of the Javelin guard to use")
@ -838,7 +844,7 @@ class ContentFilterAction(str, Enum):
MASK = "MASK"
class BlockedWord(BaseModel):
class BlockedWord(LiteLLMBaseModel):
"""Represents a blocked word with its action and optional description"""
keyword: str = Field(description="The keyword to block or mask")
@ -849,7 +855,7 @@ class BlockedWord(BaseModel):
)
class ContentFilterPattern(BaseModel):
class ContentFilterPattern(LiteLLMBaseModel):
"""Represents a content filter pattern (prebuilt or custom regex)"""
pattern_type: Literal["prebuilt", "regex"] = Field(
@ -870,7 +876,7 @@ class ContentFilterPattern(BaseModel):
action: ContentFilterAction = Field(description="Action to take when pattern matches (BLOCK or MASK)")
class ContentFilterConfigModel(BaseModel):
class ContentFilterConfigModel(LiteLLMBaseModel):
"""Configuration parameters for the content filter guardrail"""
patterns: list[ContentFilterPattern] | None = Field(
@ -901,6 +907,95 @@ class ContentFilterConfigModel(BaseModel):
MCP_SECURITY_ON_VIOLATION: Final = frozenset({"block", "alert"})
GuardrailStreamScope = Literal["streaming", "non_streaming", "both"]
DEFAULT_GUARDRAIL_STREAM_SCOPE: Final[GuardrailStreamScope] = "both"
class GuardrailEventHooks(str, Enum):
pre_call = "pre_call"
post_call = "post_call"
during_call = "during_call"
logging_only = "logging_only"
pre_mcp_call = "pre_mcp_call"
during_mcp_call = "during_mcp_call"
post_mcp_call = "post_mcp_call"
realtime_input_transcription = "realtime_input_transcription"
GUARDRAIL_EVENT_HOOK_VALUES: Final = frozenset(member.value for member in GuardrailEventHooks)
_GUARDRAIL_STREAM_SCOPES: Final[Mapping[str, GuardrailStreamScope]] = MappingProxyType(
{
"streaming": "streaming",
"non_streaming": "non_streaming",
"both": "both",
}
)
def _as_guardrail_stream_scope(value: object) -> GuardrailStreamScope:
if not isinstance(value, str):
raise ValueError(f"stream_scope values must be strings, got {type(value).__name__}")
scope: Final = _GUARDRAIL_STREAM_SCOPES.get(value.lower())
if scope is None:
raise ValueError(f"stream_scope must be one of both, streaming, non_streaming, got {value!r}")
return scope
def _validated_stream_scope_hook(key: object) -> str:
if not isinstance(key, str):
raise ValueError(f"stream_scope keys must be strings, got {type(key).__name__}")
hook: Final = key.lower()
if hook not in GUARDRAIL_EVENT_HOOK_VALUES:
raise ValueError(
f"stream_scope keys must be guardrail modes ({sorted(GUARDRAIL_EVENT_HOOK_VALUES)}), got {key!r}"
)
return hook
def coerce_stream_scope(value: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None:
if value is None:
return None
if isinstance(value, str):
return _as_guardrail_stream_scope(value)
if isinstance(value, Mapping):
scope_map: Final[Mapping[str, object]] = cast(Mapping[str, object], value) # cast-ok: keys validated below
return {
_validated_stream_scope_hook(key): _as_guardrail_stream_scope(scope) for key, scope in scope_map.items()
}
raise ValueError(f"stream_scope must be a string or mapping, got {type(value).__name__}")
def stored_stream_scope(value: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None:
try:
return coerce_stream_scope(value)
except ValueError:
verbose_logger.warning("Ignoring invalid stored stream_scope value of type %s", type(value).__name__)
return None
def with_tolerated_stream_scope(params: Mapping[str, object]) -> dict[str, object]:
if "stream_scope" not in params:
return dict(params)
return {
**params,
"stream_scope": stored_stream_scope(params["stream_scope"]),
}
def runtime_stream_scope(
stream_scope: object,
) -> tuple[GuardrailStreamScope, MappingProxyType[str, GuardrailStreamScope]]:
coerced: Final = coerce_stream_scope(stream_scope)
if coerced is None:
return DEFAULT_GUARDRAIL_STREAM_SCOPE, MappingProxyType({})
if isinstance(coerced, str):
return coerced, MappingProxyType({})
return DEFAULT_GUARDRAIL_STREAM_SCOPE, MappingProxyType(coerced)
LoggingOnlyScope = Literal["input", "output"]
class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch update guardrails
api_key: str | None = Field(default=None, description="API key for the guardrail service")
@ -1043,9 +1138,11 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
skip_unscannable_attachments: bool | None = Field(
default=False,
description=(
"Implemented by guardrail='model_armor'. When True, attachment references that carry no "
"inline bytes (file_id, gs://, or http(s) URLs) pass through unscanned instead of blocking, "
"while fail_on_error still governs real Model Armor API errors. Default False blocks them."
"Implemented by guardrail='model_armor' and guardrail='bedrock'. When True, attachments the "
"guardrail cannot scan pass through unscanned instead of blocking. For Model Armor these are "
"references with no inline bytes (file_id, gs://, or http(s) URLs), and fail_on_error still "
"governs real Model Armor API errors. For Bedrock these are documents, files, audio, video, "
"and images that are not inline PNG or JPEG up to 4 MB. Default False blocks them."
),
)
sanitize_error_detail: bool | None = Field(
@ -1066,10 +1163,23 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
description=(
"Behavior when a guardrail endpoint is unreachable due to network errors. "
"Implemented by guardrail='generic_guardrail_api', 'agent_365', 'akto', 'vigil_guard', 'repelloai', 'headroom', 'compresr', and 'typesafe'. "
"'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
"'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed. "
"Also implemented by guardrail='decision_model'."
),
)
max_input_chars: int | None = Field(
default=None,
gt=0,
description="Character budget for each text sent to the guardrail's model. Implemented by guardrail='decision_model'.",
)
max_concurrent_decision_calls: int | None = Field(
default=None,
gt=0,
description="Maximum decisions calls in flight at once on the guardrail in each proxy worker. Implemented by guardrail='decision_model'.",
)
extra_headers: list[str] | None = Field(
default=None,
description=(
@ -1142,6 +1252,39 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
),
)
stream_scope: GuardrailStreamScope | dict[str, GuardrailStreamScope] | None = Field(
default=None,
description=(
"Whether this guardrail runs on streaming requests, non-streaming requests, or both. "
"A string applies to every configured mode. A map overrides named modes "
"(pre_call, during_call, post_call, ...); omitted keys default to both. "
"Unset means both, matching historical behavior."
),
)
@field_validator("stream_scope", mode="before")
@classmethod
def normalize_stream_scope(cls, v: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None:
return coerce_stream_scope(v)
logging_only_scope: LoggingOnlyScope | None = Field(
default=None,
description=(
"which direction a logging_only scan observes: 'input' (request) or 'output' (response); "
"unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the "
"same guardrail keep blocking."
),
)
logging_only_continue_on_input_failure: bool | None = Field(
default=None,
description=(
"when True, a flagged or raising logging_only request scan is logged and the response is "
"still scanned, so both verdicts land. Only applies to mode logging_only and is ignored "
"when logging_only_scope is 'input' or 'output'."
),
)
@field_validator(
"mode",
"default_action",
@ -1169,7 +1312,7 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up
model_config = ConfigDict(extra="allow", protected_namespaces=())
class Mode(BaseModel):
class Mode(LiteLLMBaseModel):
tags: dict[str, str | list[str]] = Field(description="Tags for the guardrail mode")
default: str | list[str] | None = Field(default=None, description="Default mode when no tags match")
@ -1207,6 +1350,17 @@ class LitellmParams( # pyright: ignore[reportIncompatibleVariableOverride] # o
Agent365GuardrailConfigModel,
):
guardrail: str = Field(description="The type of guardrail integration to use")
decision_model: str | None = Field(
default=None,
description="For guardrail='decision_model': the Decisions-API model that scores each check",
)
checks: BedrockChecksConfigModel | tuple[DecisionModelCheck, ...] | None = Field( # pyright: ignore[reportIncompatibleVariableOverride] # widened to also accept decision-model checks
default=None,
description=(
"Inline Bedrock InvokeGuardrailChecks config for guardrail='bedrock', or the predicate "
"checks a decision_model guardrail scores for guardrail='decision_model'"
),
)
mode: str | list[str] | Mode = Field(
description="When to apply the guardrail (pre_call, post_call, during_call, logging_only)"
)
@ -1232,6 +1386,14 @@ class LitellmParams( # pyright: ignore[reportIncompatibleVariableOverride] # o
raise ValueError(f"on_violation={self.on_violation!r} is only supported by guardrail='mcp_security'")
return self
@model_validator(mode="after")
def validate_checks_list_for_guardrail(self) -> "LitellmParams":
if isinstance(self.checks, tuple) and self.guardrail != SupportedGuardrailIntegrations.DECISION_MODEL.value:
raise ValueError(
f"checks as a list is only supported by guardrail='decision_model', got guardrail={self.guardrail!r}"
)
return self
def __init__(self, **kwargs) -> None:
default_on: Final = kwargs.pop("default_on", None)
if default_on is not None:
@ -1268,17 +1430,6 @@ class guardrailConfig(TypedDict):
guardrails: list[Guardrail]
class GuardrailEventHooks(str, Enum):
pre_call = "pre_call"
post_call = "post_call"
during_call = "during_call"
logging_only = "logging_only"
pre_mcp_call = "pre_mcp_call"
during_mcp_call = "during_mcp_call"
post_mcp_call = "post_mcp_call"
realtime_input_transcription = "realtime_input_transcription"
class DynamicGuardrailParams(TypedDict):
extra_body: ReadOnly[dict[str, object]]
@ -1288,7 +1439,7 @@ class GUARDRAIL_DEFINITION_LOCATION(str, Enum):
CONFIG = "config"
class GuardrailInfoResponse(BaseModel):
class GuardrailInfoResponse(LiteLLMBaseModel):
guardrail_id: str | None = None
guardrail_name: str
litellm_params: BaseLitellmParams | None = None
@ -1301,20 +1452,22 @@ class GuardrailInfoResponse(BaseModel):
super().__init__(**kwargs)
class ListGuardrailsResponse(BaseModel):
class ListGuardrailsResponse(LiteLLMBaseModel):
guardrails: list[GuardrailInfoResponse]
class GuardrailUIAddGuardrailSettings(BaseModel):
class GuardrailUIAddGuardrailSettings(LiteLLMBaseModel):
supported_entities: list[str]
supported_actions: list[str]
supported_modes: list[str]
supported_modes_by_provider: dict[str, list[str]]
providers_without_directional_logging_only_scope: tuple[str, ...]
pii_entity_categories: list[PiiEntityCategoryMap]
content_filter_settings: dict[str, object] | None = None
decision_model_providers: tuple[str, ...] = Field(default_factory=tuple)
class PresidioPerRequestConfig(BaseModel):
class PresidioPerRequestConfig(LiteLLMBaseModel):
"""
presdio params that can be controlled per request, api key
"""
@ -1323,7 +1476,7 @@ class PresidioPerRequestConfig(BaseModel):
entities: list[PiiEntityType] | None = None
class ApplyGuardrailRequest(BaseModel):
class ApplyGuardrailRequest(LiteLLMBaseModel):
guardrail_name: str
text: str
language: str | None = None
@ -1333,11 +1486,11 @@ class ApplyGuardrailRequest(BaseModel):
metadata: dict[str, object] | None = None
class ApplyGuardrailResponse(BaseModel):
class ApplyGuardrailResponse(LiteLLMBaseModel):
response_text: str
class PatchGuardrailRequest(BaseModel):
class PatchGuardrailRequest(LiteLLMBaseModel):
guardrail_name: str | None = None
litellm_params: BaseLitellmParams | None = None
guardrail_info: dict[str, object] | None = None