refactor(types): replace Any with proven types in 5 files (#43304)

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

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

* fix(types): keep enterprise email import inside try-except for unsafe-import check

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

* fix(types): keep email_logging_instance annotation as Any pending a guarded alias

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

* refactor(types): revert iterator override typing in proxy utils

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

---------

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

litellm/assistants/main.py

@@ -3,7 +3,7 @@
 import asyncio
 import contextvars
 import os
-from collections.abc import Coroutine, Iterable
+from collections.abc import Coroutine, Iterable, Mapping, Sequence
 from functools import partial
 from typing import Any, Final, Literal
 
@@ -233,8 +233,8 @@ def create_assistants(
     name: str | None = None,
     description: str | None = None,
     instructions: str | None = None,
-    tools: list[dict[str, Any]] | None = None,
-    tool_resources: dict[str, Any] | None = None,
+    tools: Sequence[Mapping[str, object]] | None = None,
+    tool_resources: Mapping[str, object] | None = None,
     metadata: dict[str, str] | None = None,
     temperature: float | None = None,
     top_p: float | None = None,
@@ -244,7 +244,7 @@ def create_assistants(
     api_base: str | None = None,
     api_version: str | None = None,
     **kwargs,
-) -> Assistant | Coroutine[Any, Any, Assistant]:
+) -> Assistant | Coroutine[None, None, Assistant]:
     async_create_assistants: Final[bool | None] = kwargs.pop("async_create_assistants", None)
     if async_create_assistants is not None and not isinstance(async_create_assistants, bool):
         raise ValueError("Invalid value passed in for async_create_assistants. Only bool or None allowed")
@@ -283,7 +283,7 @@ def create_assistants(
     # only send params that are not None
     create_assistant_data = {k: v for k, v in create_assistant_data.items() if v is not None}
 
-    response: Coroutine[Any, Any, Assistant] | Assistant | None = None
+    response: Coroutine[None, None, Assistant] | Assistant | None = None
     if custom_llm_provider == "openai":
         api_base = (
             optional_params.api_base  # for deepinfra/perplexity/anyscale/groq we check in get_llm_provider and pass in the api base from there
@@ -415,7 +415,7 @@ def delete_assistant(
     api_base: str | None = None,
     api_version: str | None = None,
     **kwargs,
-) -> AssistantDeleted | Coroutine[Any, Any, AssistantDeleted]:
+) -> AssistantDeleted | Coroutine[None, None, AssistantDeleted]:
     optional_params: Final = GenericLiteLLMParams(api_key=api_key, api_base=api_base, api_version=api_version, **kwargs)
 
     litellm_params_dict: Final = get_litellm_params(**kwargs)
@@ -440,7 +440,7 @@ def delete_assistant(
     elif timeout is None:
         timeout = 600.0
 
-    response: AssistantDeleted | Coroutine[Any, Any, AssistantDeleted] | None = None
+    response: AssistantDeleted | Coroutine[None, None, AssistantDeleted] | None = None
     if custom_llm_provider == "openai":
         api_base = (
             optional_params.api_base

litellm/litellm_core_utils/litellm_logging.py

@@ -640,8 +640,8 @@ class Logging(LiteLLMLoggingBaseClass):
         self._own_session_id: str = session_id_var.get()
 
         self.function_id = function_id
-        self.streaming_chunks: list[Any] = []  # for generating complete stream response
-        self.sync_streaming_chunks: list[Any] = []  # for generating complete stream response
+        self.streaming_chunks: list[object] = []  # for generating complete stream response
+        self.sync_streaming_chunks: list[object] = []  # for generating complete stream response
         self.log_raw_request_response = log_raw_request_response
         self.raw_request_only = raw_request_only
 
@@ -693,7 +693,7 @@ class Logging(LiteLLMLoggingBaseClass):
         self.response_timing_metrics: Mapping[str, float] = {}  # mutable-ok: kept deep-copyable
 
         # Passthrough endpoint guardrails config for field targeting
-        self.passthrough_guardrails_config: dict[str, Any] | None = None
+        self.passthrough_guardrails_config: dict[str, object] | None = None
 
         self.model_call_details: dict[str, Any] = {
             "litellm_trace_id": self.litellm_trace_id,
@@ -4479,7 +4479,7 @@ def set_callbacks(callback_list, function_id=None):
 def _init_custom_logger_compatible_class(
     logging_integration: _custom_logger_compatible_callbacks_literal,
     internal_usage_cache: DualCache | None,
-    llm_router: Any | None,  # expect litellm.Router, but typing errors due to circular import
+    llm_router: object,  # expect litellm.Router, but typing errors due to circular import
     custom_logger_init_args: dict | None = {},
 ) -> CustomLogger | None:
     """
@@ -6439,7 +6439,7 @@ def _autorouter_savings_for_payload(
 
 def get_standard_logging_object_payload(
     kwargs: dict | None,
-    init_response_obj: Any | BaseModel | dict,
+    init_response_obj: object,
     start_time: dt_object,
     end_time: dt_object,
     logging_obj: Logging,

litellm/llms/custom_httpx/llm_http_handler.py

@@ -6375,7 +6375,7 @@ class BaseLLMHTTPHandler:
         custom_llm_provider: str | None = None,
         first_message: str | None = None,
         request_defaults: ResponsesWebSocketRequestDefaults | None = None,
-        **kwargs: Any,
+        **kwargs: object,
     ) -> Exception | None:
         """
         Handles Responses API WebSocket mode.
@@ -10378,13 +10378,14 @@ class BaseLLMHTTPHandler:
         encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
         url: Final = f"{api_base}/{encoded_vector_store_id}"
 
-        request_body: Final[dict[str, Any]] = dict(vector_store_update_optional_params)
+        request_body: Final[dict[str, object]] = dict(vector_store_update_optional_params)
+        metadata: Final = vector_store_update_optional_params.get("metadata")
 
         # Clean metadata to only include string values (OpenAI requirement)
-        if "metadata" in request_body and request_body["metadata"] is not None:
+        if metadata is not None:
             from litellm.utils import add_openai_metadata
 
-            request_body["metadata"] = add_openai_metadata(request_body["metadata"])
+            request_body["metadata"] = add_openai_metadata(metadata)
 
         if extra_body:
             request_body.update(extra_body)
@@ -10456,13 +10457,14 @@ class BaseLLMHTTPHandler:
         encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
         url: Final = f"{api_base}/{encoded_vector_store_id}"
 
-        request_body: Final[dict[str, Any]] = dict(vector_store_update_optional_params)
+        request_body: Final[dict[str, object]] = dict(vector_store_update_optional_params)
+        metadata: Final = vector_store_update_optional_params.get("metadata")
 
         # Clean metadata to only include string values (OpenAI requirement)
-        if "metadata" in request_body and request_body["metadata"] is not None:
+        if metadata is not None:
             from litellm.utils import add_openai_metadata
 
-            request_body["metadata"] = add_openai_metadata(request_body["metadata"])
+            request_body["metadata"] = add_openai_metadata(metadata)
 
         if extra_body:
             request_body.update(extra_body)

litellm/proxy/common_request_processing.py

@@ -739,7 +739,7 @@ async def _parse_event_data_for_error(event_line: str | bytes) -> int | None:
         if not json_str or json_str == "[DONE]":  # handle empty data or [DONE] message
             return None
         try:
-            data: Final = orjson.loads(json_str)
+            data: Final[object] = orjson.loads(json_str)
             if isinstance(data, dict) and "error" in data and isinstance(data["error"], dict):
                 error_code_raw: Final = data["error"].get("code")
                 error_code: int | None = None
@@ -792,7 +792,7 @@ def _extract_error_from_sse_chunk(event_line: str | bytes) -> dict:
             return default_error
 
         try:
-            data: Final = orjson.loads(json_str)
+            data: Final[object] = orjson.loads(json_str)
             if isinstance(data, dict) and "error" in data:
                 error_obj: Final = data["error"]
                 if isinstance(error_obj, dict):
@@ -4131,7 +4131,9 @@ class ProxyBaseLLMRequestProcessing:
                 if stripped_ln.startswith("data:"):
                     json_part = stripped_ln.split("data:", 1)[1].strip()
                     if json_part and json_part != "[DONE]":
-                        obj = json.loads(json_part)
+                        obj: object = json.loads(json_part)
+                        if not isinstance(obj, dict):
+                            return None
                         maybe_modified = ProxyBaseLLMRequestProcessing._inject_cost_into_usage_dict(
                             obj, model_name, litellm_logging_obj
                         )

litellm/utils.py

@@ -1469,7 +1469,7 @@ async def async_pre_call_deployment_hook(kwargs: dict[str, Any], call_type: str)
 
 async def async_post_call_success_deployment_hook(
     request_data: dict, response: object, call_type: CallTypes | None
-) -> Any | None:
+) -> object:
     """
     Allow modifying / reviewing the response just after it's received from the deployment.
     """