refactor(types): replace Any with proven types in 3 files (#45565)

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

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

* refactor(types): drop Azure AI Search and OpenRouter image edit seams that reject previously accepted payloads

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/litellm_core_utils/llm_response_utils/convert_dict_to_response.py

@@ -6,6 +6,9 @@ import traceback
 from collections.abc import Mapping, Sequence
 from typing import Final, Literal, cast
 
+from pydantic import ConfigDict, TypeAdapter, with_config
+from typing_extensions import ReadOnly, TypedDict
+
 import litellm
 from litellm._logging import verbose_logger
 from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
@@ -373,6 +376,20 @@ def convert_to_streaming_response(
 from collections import defaultdict
 
 
+@with_config(ConfigDict(extra="allow", strict=True))
+class _FakeToolUse(TypedDict):
+    parameters: ReadOnly[object]
+    recipient_name: ReadOnly[str]
+
+
+@with_config(ConfigDict(extra="allow", strict=True))
+class _ParallelToolUseArgs(TypedDict):
+    tool_uses: ReadOnly[Sequence[_FakeToolUse]]
+
+
+_PARALLEL_TOOL_USE_ARGS: Final = TypeAdapter(_ParallelToolUseArgs)
+
+
 def handle_invalid_parallel_tool_calls(
     tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall],
 ) -> list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None:
@@ -393,7 +410,9 @@ def handle_invalid_parallel_tool_calls(
             function_args = json.loads(tool_call.function.arguments)
             if current_function == "multi_tool_use.parallel":
                 verbose_logger.debug("OpenAI did a weird pseudo-multi-tool-use call, fixing call structure..")
-                for _fake_i, _fake_tool_use in enumerate(function_args["tool_uses"]):
+                for _fake_i, _fake_tool_use in enumerate(
+                    _PARALLEL_TOOL_USE_ARGS.validate_python(function_args)["tool_uses"]
+                ):
                     _function_args = _fake_tool_use["parameters"]
                     _current_function = _fake_tool_use["recipient_name"]
                     _current_function = _current_function.removeprefix("functions.")

litellm/llms/bedrock/files/transformation.py

@@ -84,6 +84,13 @@ LIST_FILES_PURPOSE_PARAM: Final = "_s3_list_files_purpose"
 LIST_FILES_LOCATION_PARAM: Final = "_s3_list_files_location"
 
 
+def _header_or_empty(headers: Headers, name: str) -> str:
+    try:
+        return headers[name]
+    except KeyError:
+        return ""
+
+
 def _is_empty_s3_object_range_error(raw_response: Response) -> bool:
     if raw_response.status_code != 416:
         return False
@@ -101,7 +108,7 @@ def _retrieved_s3_file_size(raw_response: Response) -> int:
     if _is_empty_s3_object_range_error(raw_response):
         return 0
     if status_code == 206:
-        content_range: Final = raw_response.headers.get("Content-Range", "")
+        content_range: Final = _header_or_empty(raw_response.headers, "Content-Range")
         range_parts: Final = content_range.removeprefix("bytes 0-0/")
         if content_range.startswith("bytes 0-0/") and range_parts.isdigit():
             return int(range_parts)
@@ -112,7 +119,7 @@ def _retrieved_s3_file_size(raw_response: Response) -> int:
             response=raw_response,
         )
     if status_code == 200:
-        content_length: Final = raw_response.headers.get("Content-Length", "")
+        content_length: Final = _header_or_empty(raw_response.headers, "Content-Length")
         if content_length.isdigit():
             return int(content_length)
         raise BedrockError(
@@ -1400,7 +1407,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
 
         file_size: Final = _retrieved_s3_file_size(raw_response)
 
-        last_modified: Final = raw_response.headers.get("Last-Modified", "")
+        last_modified: Final = _header_or_empty(raw_response.headers, "Last-Modified")
         created_at: Final = int(parsedate_to_datetime(last_modified).timestamp()) if last_modified else 0
         return OpenAIFileObject(
             id=file_id,

litellm/llms/black_forest_labs/image_edit/transformation.py

@@ -14,6 +14,8 @@ from typing import TYPE_CHECKING, Any, Final
 
 import httpx
 from httpx._types import RequestFiles
+from pydantic import ConfigDict, TypeAdapter, with_config
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 import litellm
 from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
@@ -38,6 +40,19 @@ else:
     LiteLLMLoggingObj = Any
 
 
+@with_config(ConfigDict(extra="allow", strict=True))
+class _BflSample(TypedDict):
+    sample: NotRequired[ReadOnly[str | None]]
+
+
+@with_config(ConfigDict(extra="allow", strict=True))
+class _BflPolledResult(TypedDict):
+    result: NotRequired[ReadOnly[_BflSample]]
+
+
+_BFL_POLLED_RESULT_ADAPTER: Final = TypeAdapter(_BflPolledResult)
+
+
 _BFL_REQUEST_PARAMS: Final = (
     "seed",
     "output_format",
@@ -302,13 +317,15 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
         The response contains: {"status": "Ready", "result": {"sample": "https://..."}}
         """
         try:
-            response_data: Final = raw_response.json()
+            raw_response_data: Final = raw_response.json()
         except Exception as e:
             raise BlackForestLabsError(
                 status_code=raw_response.status_code,
                 message=f"Error parsing BFL response: {e}",
             )
 
+        response_data: Final = _BFL_POLLED_RESULT_ADAPTER.validate_python(raw_response_data)
+
         # Get image URL from result
         image_url: Final = response_data.get("result", {}).get("sample")
         if not image_url: