fix(guardrails): type hook call_type with CallTypesLiteral

This commit is contained in:
Alexis 2026-10-16 23:46:38 +02:00
parent ac6300f380
commit cd3415558f
7 changed files with 24 additions and 149 deletions

View file

@ -11,7 +11,7 @@ import sys
sys.path.insert(0, os.path.abspath("../..")) # Adds the parent directory to the system path
import json
import sys
from typing import TYPE_CHECKING, Final, Literal
from typing import TYPE_CHECKING, Final
from fastapi import HTTPException
@ -29,6 +29,7 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypesLiteral
GUARDRAIL_NAME: Final = "aporia"
@ -172,16 +173,7 @@ class AporiaGuardrail(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
):
from litellm.proxy.common_utils.callback_utils import (
add_guardrail_to_applied_guardrails_header,

View file

@ -22,7 +22,7 @@ import os
from collections.abc import AsyncIterator, Mapping, Sequence
from dataclasses import dataclass, replace
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, Literal
from typing import TYPE_CHECKING, Any, Final
import httpx
from fastapi import HTTPException
@ -45,6 +45,7 @@ from litellm.proxy.common_utils.callback_utils import (
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import (
CallTypesLiteral,
Choices,
LLMResponseTypes,
ModelResponse,
@ -299,18 +300,7 @@ class CiscoAIDefenseGuardrail(CiscoAIDefenseMcpMixin, CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
# Trust proxy call_type, not caller-controlled request shape.
is_mcp: Final = self._is_mcp_call_type(call_type)
@ -350,16 +340,7 @@ class CiscoAIDefenseGuardrail(CiscoAIDefenseMcpMixin, CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
is_mcp: Final = self._is_mcp_call_type(call_type)

View file

@ -1,4 +1,4 @@
from typing import Final, Literal
from typing import Final
import litellm
from litellm._logging import verbose_proxy_logger
@ -8,6 +8,7 @@ from litellm.integrations.custom_guardrail import (
log_guardrail_information,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.utils import CallTypesLiteral
class myCustomGuardrail(CustomGuardrail):
@ -26,18 +27,7 @@ class myCustomGuardrail(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
"""
Runs before the LLM API call
@ -64,16 +54,7 @@ class myCustomGuardrail(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
):
"""
Runs in parallel to LLM API call

View file

@ -22,6 +22,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypesLiteral
if TYPE_CHECKING:
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
@ -167,17 +168,7 @@ class GuardrailsAI(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: litellm.DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
],
call_type: CallTypesLiteral,
) -> (
Exception | str | dict | None
): # raise exception if invalid, return a str for the user to receive - if rejected, or return a modified dictionary for passing into litellm

View file

@ -36,6 +36,7 @@ from litellm.types.guardrails import (
Role,
default_roles,
)
from litellm.types.utils import CallTypesLiteral
GUARDRAIL_NAME: Final = "lakera_prompt_injection"
@ -116,19 +117,7 @@ class lakeraAI_Moderation(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
):
if (
await should_proceed_based_on_metadata(
@ -290,18 +279,7 @@ class lakeraAI_Moderation(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: litellm.DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
from litellm.types.guardrails import GuardrailEventHooks
@ -321,16 +299,7 @@ class lakeraAI_Moderation(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
):
if self.event_hook is None:
if self.moderation_check == "pre_call":

View file

@ -49,6 +49,7 @@ from litellm.proxy.guardrails._content_utils import (
has_non_string_content,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypesLiteral
class LassoResponse(TypedDict):
@ -177,18 +178,7 @@ class LassoGuardrail(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache, # Deprecated, use global_cache instead (kept to align with CustomGuardrail interface)
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
"""
Runs before the LLM API call to validate and potentially modify input.
@ -210,16 +200,7 @@ class LassoGuardrail(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
cache: DualCache,
):
"""

View file

@ -8,7 +8,7 @@
# Standard library imports
import json
import os
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol
from typing import TYPE_CHECKING, Any, Final, Protocol
from urllib.parse import quote
import httpx
@ -37,7 +37,7 @@ from litellm.proxy.common_utils.callback_utils import (
get_metadata_variable_name_from_kwargs,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import LLMResponseTypes
from litellm.types.utils import CallTypesLiteral, LLMResponseTypes
if TYPE_CHECKING:
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
@ -315,18 +315,7 @@ class PillarGuardrail(CustomGuardrail):
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
"""
Pre-call hook to scan the request for security threats before sending to LLM.
@ -361,16 +350,7 @@ class PillarGuardrail(CustomGuardrail):
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Exception | str | dict | None:
"""
During-call hook to scan the request in parallel with LLM processing.