refactor(types): replace Any with proven types in 6 files

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/otel/logger.py

@@ -5,7 +5,7 @@ from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import contextmanager
 from datetime import datetime
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Final, cast
+from typing import TYPE_CHECKING, Final, cast
 
 from opentelemetry.context import Context, attach, get_current
 from opentelemetry.sdk._logs import LoggerProvider
@@ -21,6 +21,7 @@ from opentelemetry.trace import (
     use_span,
 )
 from opentelemetry.trace import TracerProvider as ApiTracerProvider
+from typing_extensions import TypedDict, Unpack
 
 import litellm
 from litellm._logging import verbose_logger
@@ -140,6 +141,10 @@ def _request_trace_links(context: Context | None) -> tuple[Link, ...] | None:
     return (Link(anchor),) if anchor.is_valid else None
 
 
+class _CustomLoggerOptions(TypedDict, total=False, extra_items=object):
+    """Keyword arguments forwarded untouched to ``CustomLogger`` and ``OpenTelemetryV2Config``."""
+
+
 class _LLMCallSpan:
     """The state carried from the ``pre_call`` boundary to span close.
 
@@ -179,7 +184,7 @@ class OpenTelemetryV2(CustomLogger):
         tracer_provider: TracerProvider | None = None,
         logger_provider: LoggerProvider | None = None,
         meter_provider: "MeterProvider | None" = None,
-        **kwargs: Any,
+        **kwargs: Unpack[_CustomLoggerOptions],
     ) -> None:
         super().__init__(**kwargs)
         self.config: OpenTelemetryV2Config = config or OpenTelemetryV2Config(**kwargs)

litellm/llms/bedrock/chat/agentcore/transformation.py

@@ -6,7 +6,7 @@ https://docs.aws.amazon.com/bedrock/latest/APIReference/API_agentcore_InvokeAgen
 
 import json
 from collections.abc import AsyncGenerator
-from typing import TYPE_CHECKING, Any, Final, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Final, Optional, Union
 from urllib.parse import quote
 
 import httpx
@@ -31,6 +31,7 @@ from litellm.types.llms.openai import AllMessageValues
 from litellm.types.utils import (
     Choices,
     Delta,
+    LlmProviders,
     Message,
     ModelResponse,
     ModelResponseStream,
@@ -872,7 +873,7 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM):
         )
 
         if client is None or not isinstance(client, AsyncHTTPHandler):
-            client = get_async_httpx_client(llm_provider=cast(Any, "bedrock"), params={})
+            client = get_async_httpx_client(llm_provider=LlmProviders.BEDROCK, params={})
 
         verbose_logger.debug("Making async streaming request to: %s", api_base)
 

litellm/proxy/guardrails/guardrail_hooks/alice/alice.py

@@ -11,14 +11,13 @@ from collections.abc import Mapping
 from itertools import islice
 from typing import (
     TYPE_CHECKING,
-    Any,  # noqa: TID251  # **kwargs forwards verbatim to CustomGuardrail.__init__; see ruff-strict.toml
     Final,
     Literal,
     Optional,
 )
 
 import httpx
-from typing_extensions import NotRequired, ReadOnly, TypedDict
+from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
 
 from litellm._logging import verbose_proxy_logger
 from litellm.exceptions import GuardrailRaisedException, Timeout
@@ -92,6 +91,10 @@ class AliceVerdict(TypedDict):
     replacements: ReadOnly[NotRequired["tuple[AliceReplacement, ...]"]]
 
 
+class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
+    """Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
+
+
 class AliceGuardrailMissingSecrets(Exception):
     """Raised when the Alice API key is not configured."""
 
@@ -144,7 +147,9 @@ class AliceGuardrail(CustomGuardrail):
         api_key: str | None = None,
         api_base: str | None = None,
         unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
-        **kwargs: Any,  # kwargs-ok: forwarded verbatim to CustomGuardrail.__init__, whose param list is wide and evolving
+        **kwargs: Unpack[  # kwargs-ok: forwarded verbatim to CustomGuardrail.__init__, whose param list is wide and evolving
+            _CustomGuardrailOptions
+        ],
     ) -> None:
         self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
 

litellm/proxy/guardrails/guardrail_hooks/grayswan/grayswan.py

@@ -2,10 +2,10 @@
 
 import os
 import time
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol
+from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol
 
 from fastapi import HTTPException
-from typing_extensions import NotRequired, ReadOnly, TypedDict
+from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
 
 from litellm._logging import verbose_proxy_logger
 from litellm.integrations.custom_guardrail import (
@@ -38,6 +38,10 @@ class _GraySwanMonitorResponse(TypedDict):
     ipi: ReadOnly[NotRequired[bool | None]]
 
 
+class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
+    """Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
+
+
 class _GraySwanMonitorHTTPResponse(Protocol):
     def raise_for_status(self) -> object: ...
 
@@ -103,7 +107,7 @@ class GraySwanGuardrail(CustomGuardrail):
         streaming_sampling_rate: int = 5,
         fail_open: bool | None = True,
         guardrail_timeout: float | None = 30.0,
-        **kwargs: Any,
+        **kwargs: Unpack[_CustomGuardrailOptions],
     ) -> None:
         self.async_handler: _GraySwanMonitorHTTPClient = get_async_httpx_client(
             llm_provider=httpxSpecialProvider.GuardrailCallback

litellm/proxy/guardrails/guardrail_hooks/promptguard/promptguard.py

@@ -7,9 +7,10 @@ before and after LLM calls.
 """
 
 import os
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict
+from typing import TYPE_CHECKING, Final, Literal, Optional, TypedDict
 
-from typing_extensions import ReadOnly
+from typing_extensions import ReadOnly, Unpack
+from typing_extensions import TypedDict as ExtraItemsTypedDict
 
 from litellm._logging import verbose_proxy_logger
 from litellm.exceptions import GuardrailRaisedException
@@ -53,6 +54,12 @@ class PromptGuardHTTPView(TypedDict):
     guard_response: ReadOnly[PromptGuardGuardAPIResponse]
 
 
+class _CustomGuardrailOptions(ExtraItemsTypedDict, total=False, extra_items=object):
+    """Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
+
+    supported_event_hooks: ReadOnly[list[GuardrailEventHooks] | None]
+
+
 class PromptGuardMissingCredentials(Exception):
     pass
 
@@ -63,7 +70,7 @@ class PromptGuardGuardrail(CustomGuardrail):
         api_key: str | None = None,
         api_base: str | None = None,
         block_on_error: bool | None = None,
-        **kwargs: Any,
+        **kwargs: Unpack[_CustomGuardrailOptions],
     ) -> None:
         self.api_key = api_key or os.environ.get(
             "PROMPTGUARD_API_KEY",
@@ -92,9 +99,12 @@ class PromptGuardGuardrail(CustomGuardrail):
             llm_provider=httpxSpecialProvider.GuardrailCallback,
         )
 
-        kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
+        options: Final[_CustomGuardrailOptions] = {
+            "supported_event_hooks": list(self.get_supported_event_hooks()),
+            **kwargs,
+        }
 
-        super().__init__(**kwargs)
+        super().__init__(**options)
 
     @staticmethod
     def get_config_model() -> type["GuardrailConfigModel"] | None:

litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

@@ -24,6 +24,7 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.llms.openai import AllMessageValues
 from litellm.types.proxy.guardrails.guardrail_hooks.base import (
     GuardrailConfigModel,
 )
@@ -40,7 +41,7 @@ from litellm.types.utils import CallTypes, GenericGuardrailAPIInputs
 _DEFAULT_API_BASE: Final = "http://localhost:8003"
 _GUARD_ENDPOINT: Final = "/api/v1/ai-gateway/litellm-v2"
 _DEFAULT_TIMEOUT: Final = 30.0
-_EMPTY_MAPPING: Final[Mapping[str, Any]] = MappingProxyType({})
+_EMPTY_MAPPING: Final[Mapping[str, Mapping[str, object]]] = MappingProxyType({})
 _MCP_MODEL_PREFIX: Final = "MCP:"
 
 
@@ -159,7 +160,7 @@ class SingulrGuardrail(CustomGuardrail):
         return {key: value for key, value in resolved if value}  # mutable-ok: short-lived JSON payload dict
 
     @staticmethod
-    def _build_user_message(text: str) -> Mapping[str, Any]:
+    def _build_user_message(text: str) -> Mapping[str, str]:
         return {"role": "user", "content": text}  # mutable-ok: short-lived JSON payload dict
 
     def _build_headers(self) -> Mapping[str, str]:
@@ -224,7 +225,7 @@ class SingulrGuardrail(CustomGuardrail):
         self,
         inputs: GenericGuardrailAPIInputs,
         texts: Sequence[str],
-        structured_messages: Sequence[Any],
+        structured_messages: Sequence[AllMessageValues],
         request_data: Mapping[str, Any],
     ) -> GenericGuardrailAPIInputs:
         messages: Final = (
@@ -271,12 +272,12 @@ class SingulrGuardrail(CustomGuardrail):
         return request_data.get("mcp_tool_name") or request_data.get("name")
 
     @staticmethod
-    def _mcp_arguments(request_data: Mapping[str, Any]) -> object:
+    def _mcp_arguments(request_data: Mapping[str, object]) -> object:
         arguments: Final = request_data.get("mcp_arguments")
         return arguments if arguments is not None else request_data.get("arguments")
 
     @staticmethod
-    def _is_mcp_call(request_data: Mapping[str, Any], logging_obj: LiteLLMLoggingObj | None) -> bool:
+    def _is_mcp_call(request_data: Mapping[str, object], logging_obj: LiteLLMLoggingObj | None) -> bool:
         call_type: Final = logging_obj.call_type if logging_obj is not None else request_data.get("call_type")
         if call_type is not None:
             return call_type == CallTypes.call_mcp_tool.value