refactor(types): replace Any with proven types in 14 files (#48928)

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

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

* fix(types): hide input values in validation errors and drop secret-bearing seams

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

* fix(types): revert the Phoenix prompt seam that rejects string content

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

* fix(types): revert files that break mocked tests and Python 3.10 imports

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

* fix(types): revert pegasus response validation to main

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

* fix(types): revert azure search, milvus and onyx response validation to main

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

* fix(proxy): keep accepting numeric keys in /cache/delete

* fix(caching): answer 400 for a non-list /cache/delete keys and accept null Stability lists

Return pydantic errors without the caller's input on a bad /cache/delete body, let the Bedrock Stability
image edit schema take null finish_reasons and images like the old reads did, declare user_id and user_role
on the plugin session claim, wrap the retyped Vertex fields in ReadOnly, and drop unused Guardrails AI types

* refactor(types): leave provider reply parsers for a follow-up with live provider calls

Restores the Bedrock Stability, Bedrock KB, AgentCore, Replicate, Recraft, Aleph Alpha and Guardrails AI parsers to main, since no live provider call could prove the stricter schemas. Also types the cache delete route tests.

* refactor(types): keep /cache/delete on main's behavior

Restores the cache delete route and its test to main, so this types PR changes no response. Rejecting a non-list keys is left for its own change, since an object keys deletes its keys on main today.

* fix(types): keep text completion, batch ids and guardrail callback on main's code

* fix(types): keep policy suggestion and auto router parsing on main's code

---------

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

litellm/caching/qdrant_semantic_cache.py

@@ -13,7 +13,7 @@ import asyncio
 import json
 import os
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, Final, Protocol, cast
+from typing import TYPE_CHECKING, Any, Final, Protocol
 
 from pydantic import TypeAdapter
 
@@ -292,10 +292,7 @@ class QdrantSemanticCache(BaseCache):
         prompt: Final = get_semantic_cache_prompt_from_messages(messages)
 
         # create an embedding for prompt
-        embedding_response: Final = cast(
-            EmbeddingResponse,
-            self._get_embedding(prompt, metadata=kwargs.get("metadata")),
-        )
+        embedding_response: Final = self._get_embedding(prompt, metadata=kwargs.get("metadata"))
 
         # get the embedding
         embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])
@@ -331,10 +328,7 @@ class QdrantSemanticCache(BaseCache):
         prompt: Final = get_semantic_cache_prompt_from_messages(messages)
 
         # convert to embedding
-        embedding_response: Final = cast(
-            EmbeddingResponse,
-            self._get_embedding(prompt, metadata=kwargs.get("metadata")),
-        )
+        embedding_response: Final = self._get_embedding(prompt, metadata=kwargs.get("metadata"))
 
         # get the embedding
         embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])

litellm/caching/redis_semantic_cache.py

@@ -14,7 +14,9 @@ import asyncio
 import json
 import os
 from collections.abc import Callable, Mapping
-from typing import TYPE_CHECKING, Any, Final, cast
+from typing import TYPE_CHECKING, Final, cast
+
+from pydantic import ConfigDict, TypeAdapter
 
 import litellm
 from litellm._logging import print_verbose, verbose_logger
@@ -38,6 +40,11 @@ if TYPE_CHECKING:
     from litellm.router import Router
 
 
+_CACHE_RESPONSE_ADAPTER: Final[TypeAdapter[str | bytes | None]] = TypeAdapter(
+    str | bytes | None, config=ConfigDict(strict=True, hide_input_in_errors=True)
+)
+
+
 class RedisSemanticCache(BaseCache):
     """
     Redis-backed semantic cache for LLM responses.
@@ -380,7 +387,7 @@ class RedisSemanticCache(BaseCache):
             )
         return embedding_response["data"][0]["embedding"]
 
-    def _get_cache_logic(self, cached_response: Any) -> object:
+    def _get_cache_logic(self, cached_response: str | bytes | None) -> object:
         """
         Process the cached response to prepare it for use.
 
@@ -391,24 +398,23 @@ class RedisSemanticCache(BaseCache):
             The processed cache response, or None if input was None
         """
         if cached_response is None:
-            return cached_response
+            return None
 
         # Convert bytes to string if needed
-        if isinstance(cached_response, bytes):
-            cached_response = cached_response.decode("utf-8")
+        response_text: Final = (
+            cached_response.decode("utf-8") if isinstance(cached_response, bytes) else cached_response
+        )
 
         # Convert string representation to Python object
         try:
-            cached_response = json.loads(cached_response)
+            return json.loads(response_text)
         except json.JSONDecodeError:
             try:
-                cached_response = ast.literal_eval(cached_response)
+                return ast.literal_eval(response_text)
             except (ValueError, SyntaxError) as e:
                 print_verbose(f"Error parsing cached response: {e}")
                 return None
 
-        return cached_response
-
     def set_cache(self, key: str, value: object, **kwargs) -> None:
         """
         Store a value in the semantic cache.
@@ -494,7 +500,7 @@ class RedisSemanticCache(BaseCache):
             similarity: Final = 1 - vector_distance
 
             cached_prompt: Final = cache_hit["prompt"]
-            cached_response: Final = cache_hit["response"]
+            cached_response: Final = _CACHE_RESPONSE_ADAPTER.validate_python(cache_hit["response"])
 
             # update kwargs["metadata"] with similarity, don't rewrite the original metadata
             kwargs.setdefault("metadata", {})["semantic-similarity"] = similarity
@@ -645,7 +651,7 @@ class RedisSemanticCache(BaseCache):
             similarity: Final = 1 - vector_distance
 
             cached_prompt: Final = cache_hit["prompt"]
-            cached_response: Final = cache_hit["response"]
+            cached_response: Final = _CACHE_RESPONSE_ADAPTER.validate_python(cache_hit["response"])
 
             # update kwargs["metadata"] with similarity, don't rewrite the original metadata
             kwargs.setdefault("metadata", {})["semantic-similarity"] = similarity

litellm/integrations/custom_logger.py

@@ -513,7 +513,7 @@ class CustomLogger:  # https://docs.litellm.ai/docs/observability/custom_callbac
         data: dict,
         user_api_key_dict: UserAPIKeyAuth,
         call_type: CallTypesLiteral,
-    ) -> Any:
+    ) -> object:
         pass
 
     async def async_post_call_streaming_hook(

litellm/integrations/opentelemetry_utils/base_otel_llm_obs_attributes.py

@@ -15,7 +15,7 @@ class BaseLLMObsOTELAttributes(ABC):
         pass
 
 
-def cast_as_primitive_value_type(value) -> str | bool | int | float:
+def cast_as_primitive_value_type(value: object) -> str | bool | int | float:
     """
     Converts a value to an OTEL-supported primitive for Arize/Phoenix observability.
     """

litellm/proxy/health_check_utils/shared_health_check_manager.py

@@ -2,7 +2,10 @@ import asyncio
 import json
 import time
 from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Final
+from typing import TYPE_CHECKING, Final
+
+from pydantic import ConfigDict, TypeAdapter, with_config
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 from litellm._internal_context import with_service_target
 from litellm._logging import verbose_proxy_logger
@@ -19,6 +22,17 @@ if TYPE_CHECKING:
     from litellm.router import Router
 
 
+@with_config(ConfigDict(extra="allow", strict=True))
+class _CachedHealthCheckResults(TypedDict):
+    healthy_endpoints: NotRequired[ReadOnly[list[Mapping[str, object]]]]
+    unhealthy_endpoints: NotRequired[ReadOnly[list[Mapping[str, object]]]]
+    timestamp: NotRequired[ReadOnly[int | float]]
+    checked_by: NotRequired[ReadOnly[object]]
+
+
+_CACHED_HEALTH_CHECK_RESULTS: Final = TypeAdapter(_CachedHealthCheckResults)
+
+
 class SharedHealthCheckManager:
     """
     Manager for coordinating health checks across multiple pods using Redis.
@@ -109,7 +123,7 @@ class SharedHealthCheckManager:
             verbose_proxy_logger.error("Error releasing health check lock: %s", str(e))
 
     @with_service_target(HEALTH_CHECKS_TARGET)
-    async def get_cached_health_check_results(self) -> dict[str, Any] | None:
+    async def get_cached_health_check_results(self) -> _CachedHealthCheckResults | None:
         """
         Get cached health check results from Redis.
 
@@ -127,10 +141,9 @@ class SharedHealthCheckManager:
                 return None
 
             # Parse the cached data
-            if isinstance(cached_data, str):
-                cached_results = json.loads(cached_data)
-            else:
-                cached_results = cached_data
+            cached_results: Final = _CACHED_HEALTH_CHECK_RESULTS.validate_python(
+                json.loads(cached_data) if isinstance(cached_data, str) else cached_data
+            )
 
             # Check if the cache is still valid
             cache_timestamp: Final = cached_results.get("timestamp", 0)

litellm/proxy/mcp_server/oauth_issuer_stamp_backfill.py

@@ -43,6 +43,7 @@ from litellm._logging import verbose_proxy_logger
 from litellm.proxy.db.db_span import db_span
 from litellm.proxy.mcp_server.oauth_utils import canonicalize_url_identity
 from litellm.proxy.utils import PrismaClient
+from litellm.repositories.table_repositories import MCPServerRepository
 
 # The actor the removed discovery write-back stamped rows with.
 _DISCOVERY_ACTOR: Final = "mcp_oauth_discovery"
@@ -106,7 +107,7 @@ def _is_stamped_issuer_row(row: _MCPServerRow) -> bool:
 
 async def backfill_discovery_stamped_issuers(prisma_client: PrismaClient) -> int:
     """Clear gateway-written issuer stamps, returning the number of rows healed."""
-    candidate_rows: Final[list[_MCPServerRow]] = await prisma_client.db.litellm_mcpservertable.find_many(
+    candidate_rows: Final = await MCPServerRepository(prisma_client).table.find_many(
         where={
             "updated_by": _DISCOVERY_ACTOR,
             "auth_type": {"in": list(_AUTH_TYPES_WITH_ISSUER_ANCHORING)},

litellm/proxy/plugin_routes.py

@@ -31,6 +31,8 @@ from typing import Final
 
 from cryptography.fernet import Fernet, InvalidToken
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from pydantic import ConfigDict, TypeAdapter, with_config
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.proxy._types import PluginConfig, SpecialHeaders, UserAPIKeyAuth
@@ -39,6 +41,18 @@ from litellm.types.llms.custom_http import httpxSpecialProvider
 
 router: Final = APIRouter()
 
+
+@with_config(ConfigDict(extra="allow", strict=True))
+class PluginSessionClaim(TypedDict):
+    plugin: NotRequired[ReadOnly[object]]
+    user_id: NotRequired[ReadOnly[str]]
+    user_role: NotRequired[ReadOnly[str]]
+    exp: NotRequired[ReadOnly[int | float | str]]
+
+
+_PLUGIN_SESSION_CLAIM: Final = TypeAdapter(PluginSessionClaim)
+
+
 # Hop-by-hop headers (RFC 7230) and the litellm session cookie — never forwarded
 # to a plugin backend.  Credential headers are added on top per-request from the
 # canonical SpecialHeaders set so the plugin only ever authenticates via its own
@@ -155,7 +169,7 @@ def issue_plugin_session_claim(plugin_name: str, user_id: str | None, user_role:
     return _plugin_fernet(plugin_name).encrypt(json.dumps(claim).encode()).decode()
 
 
-def verify_plugin_session_claim(plugin_name: str, ciphertext: str) -> dict:
+def verify_plugin_session_claim(plugin_name: str, ciphertext: str) -> PluginSessionClaim:
     """Verify and decode a plugin session claim.
 
     Raises ValueError if the HMAC is invalid, the audience is wrong, or
@@ -163,7 +177,7 @@ def verify_plugin_session_claim(plugin_name: str, ciphertext: str) -> dict:
     """
     try:
         raw: Final = _plugin_fernet(plugin_name).decrypt(ciphertext.encode(), ttl=_CLAIM_TTL_SECONDS)
-        claim: Final = json.loads(raw)
+        claim: Final = _PLUGIN_SESSION_CLAIM.validate_python(json.loads(raw))
     except (InvalidToken, Exception) as exc:
         raise ValueError("Invalid, tampered, or expired plugin session claim") from exc
 

litellm/types/integrations/custom_logger.py

@@ -97,9 +97,9 @@ class AgenticLoopRequestPatch(LiteLLMBaseModel):
 
     model: str | None = None
     messages: list[dict[str, Any]] | None = None
-    tools: list[dict[str, Any]] | None = None
+    tools: list[dict[str, object]] | None = None
     max_tokens: int | None = None
-    optional_params: dict[str, Any] = Field(default_factory=dict)
+    optional_params: dict[str, object] = Field(default_factory=dict)
     kwargs: dict[str, Any] = Field(default_factory=dict)
 
 

litellm/types/integrations/datadog_llm_obs.py

@@ -5,7 +5,7 @@ API Reference: https://docs.datadoghq.com/llm_observability/setup/api/?tab=examp
 """
 
 from collections.abc import Sequence
-from typing import Any, Literal
+from typing import Literal
 
 from typing_extensions import ReadOnly, TypedDict
 
@@ -16,7 +16,7 @@ class ToolCall(TypedDict, total=False):
     """A tool call on a message, as LLM Obs names its fields."""
 
     name: ReadOnly[str]
-    arguments: ReadOnly[dict[str, Any] | str]  # parsed object, or the raw string when it will not parse to one
+    arguments: ReadOnly[dict[str, object] | str]  # parsed object, or the raw string when it will not parse to one
     tool_id: ReadOnly[str]
     type: ReadOnly[str]
 
@@ -35,7 +35,7 @@ class ToolDefinition(TypedDict, total=False):
 
     name: ReadOnly[str]
     description: ReadOnly[str]
-    schema: ReadOnly[dict[str, Any]]
+    schema: ReadOnly[dict[str, object]]
 
 
 class Message(TypedDict, total=False):
@@ -49,13 +49,15 @@ class Message(TypedDict, total=False):
 
 
 class InputMeta(TypedDict):
-    messages: Sequence[
-        Message | dict[str, Any]  # changed to fit with tool calls
+    messages: ReadOnly[
+        Sequence[
+            Message | dict[str, object]  # changed to fit with tool calls
+        ]
     ]  # Relevant Issue: https://github.com/BerriAI/litellm/issues/9494
 
 
 class OutputMeta(TypedDict):
-    messages: Sequence[Any]
+    messages: ReadOnly[Sequence[object]]
 
 
 class DDLLMObsError(TypedDict, total=False):
@@ -71,7 +73,7 @@ class Meta(TypedDict, total=False):
     kind: Literal["llm", "tool", "task", "embedding", "retrieval"]
     input: InputMeta  # The span's input information.
     output: OutputMeta  # The span's output information.
-    metadata: dict[str, Any]
+    metadata: ReadOnly[dict[str, object]]
     error: DDLLMObsError | None  # Error information on the span
     tool_definitions: ReadOnly[Sequence[ToolDefinition]]  # The tools offered to the model on this request
 

litellm/types/llms/openai.py

@@ -1600,7 +1600,7 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject):
 
     @field_validator("reasoning", mode="before")
     @classmethod
-    def validate_reasoning_to_dict(cls, value: Any) -> dict[str, Any] | None:
+    def validate_reasoning_to_dict(cls, value: Any) -> builtins.object:
         """Accept API reasoning dict (including effort 'none'/'xhigh'); always store as dict."""
         if value is None:
             return None

litellm/types/llms/vertex_ai.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 from enum import Enum
-from typing import Any, Final, Literal, Protocol
+from typing import Final, Literal, Protocol
 
 from typing_extensions import (
     ReadOnly,
@@ -140,7 +140,7 @@ class Schema(TypedDict, total=False):
     title: str
     description: str
     nullable: bool
-    default: Any
+    default: ReadOnly[object]
     items: "Schema"
     minItems: str
     maxItems: str
@@ -155,7 +155,7 @@ class Schema(TypedDict, total=False):
     minLength: str
     maxLength: str
     pattern: str
-    example: Any
+    example: ReadOnly[object]
     anyOf: list["Schema"]
 
 
@@ -794,9 +794,9 @@ class VertexVideoGenerationResponse(TypedDict, total=False):
 
     name: str
     done: bool
-    response: dict[str, Any]
-    metadata: dict[str, Any]
-    error: dict[str, Any]
+    response: ReadOnly[dict[str, object]]
+    metadata: ReadOnly[dict[str, object]]
+    error: ReadOnly[dict[str, object]]
 
 
 VERTEX_CREDENTIALS_TYPES = str | dict[str, str]

litellm/types/proxy/guardrails/guardrail_hooks/generic_guardrail_api.py

@@ -48,7 +48,7 @@ class GenericGuardrailAPIMetadata(TypedDict, total=False):
 class GenericGuardrailAPIOptionalParams(LiteLLMBaseModel):
     """Optional parameters for the Generic Guardrail API"""
 
-    additional_provider_specific_params: dict[str, Any] | None = Field(
+    additional_provider_specific_params: dict[str, object] | None = Field(
         default=None,
         description="Additional provider-specific parameters to send with the guardrail request",
     )
@@ -161,7 +161,7 @@ class GenericGuardrailAPIRequest(LiteLLMBaseModel):
         default=None,
         description="LiteLLM library version running this proxy.",
     )
-    additional_provider_specific_params: dict[str, Any] | None = None
+    additional_provider_specific_params: dict[str, object] | None = None
     tool_calls: list[ChatCompletionToolCallChunk] | list[ChatCompletionMessageToolCall] | None = None
     model: str | None = None  # the model being used for the LLM call
 

litellm/types/proxy/guardrails/guardrail_hooks/ibm/ibm_detector.py

@@ -1,7 +1,5 @@
-from typing import Any
-
 from pydantic import Field
-from typing_extensions import TypedDict
+from typing_extensions import ReadOnly, TypedDict
 
 from litellm.types.llms.base import LiteLLMBaseModel
 
@@ -14,41 +12,41 @@ from .base import IBMGuardrailsBaseConfigModel
 class IBMDetectorRequestBodyDetectorServer(TypedDict):
     """Request body for calling IBM Detector Server directly"""
 
-    contents: list[str]
-    detector_params: dict[str, Any]
+    contents: ReadOnly[list[str]]
+    detector_params: ReadOnly[dict[str, object]]
 
 
 class IBMDetectorRequestBodyOrchestrator(TypedDict):
     """Request body for calling IBM Detector via FMS Guardrails Orchestrator"""
 
-    content: str
-    detectors: dict[str, dict[str, Any]]
+    content: ReadOnly[str]
+    detectors: ReadOnly[dict[str, dict[str, object]]]
 
 
 class IBMDetectorDetection(TypedDict, total=False):
     """Individual detection from IBM Detector"""
 
-    start: int
-    end: int
-    text: str
-    detection: str
-    detection_type: str
-    score: float
-    evidences: list[Any]
-    metadata: dict[str, Any]
-    detector_id: str | None  # Only present in orchestrator response
+    start: ReadOnly[int]
+    end: ReadOnly[int]
+    text: ReadOnly[str]
+    detection: ReadOnly[str]
+    detection_type: ReadOnly[str]
+    score: ReadOnly[float]
+    evidences: ReadOnly[list[object]]
+    metadata: ReadOnly[dict[str, object]]
+    detector_id: ReadOnly[str | None]  # Only present in orchestrator response
 
 
 class IBMDetectorResponseDetectorServer(TypedDict):
     """Response from IBM Detector Server (returns list of lists)"""
 
-    detections: list[list[IBMDetectorDetection]]
+    detections: ReadOnly[list[list[IBMDetectorDetection]]]
 
 
 class IBMDetectorResponseOrchestrator(TypedDict):
     """Response from IBM FMS Guardrails Orchestrator"""
 
-    detections: list[IBMDetectorDetection]
+    detections: ReadOnly[list[IBMDetectorDetection]]
 
 
 # Pydantic Config Models
@@ -57,12 +55,12 @@ class IBMDetectorResponseOrchestrator(TypedDict):
 class IBMDetectorOptionalParams(LiteLLMBaseModel):
     """Optional parameters for IBM Detector guardrail"""
 
-    detector_params: dict[str, Any] | None = Field(
+    detector_params: dict[str, object] | None = Field(
         default_factory=lambda: {},
         description="Dictionary of arguments to pass to the detector.",
     )
 
-    extra_headers: dict[str, Any] | None = Field(
+    extra_headers: dict[str, object] | None = Field(
         default_factory=lambda: {},
         description="Dictionary of extra headers to pass to the detector.",
     )

litellm/types/proxy/management_endpoints/key_management_endpoints.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Final, Literal, TypeAlias
+from typing import Final, Literal, TypeAlias
 
 from pydantic import ConfigDict, model_validator
 from typing_extensions import ReadOnly, TypedDict
@@ -50,14 +50,14 @@ class SuccessfulKeyUpdate(LiteLLMBaseModel):
     """Successfully updated key with its updated information"""
 
     key: str
-    key_info: dict[str, Any]
+    key_info: dict[str, object]
 
 
 class FailedKeyUpdate(LiteLLMBaseModel):
     """Failed key update with reason"""
 
     key: str
-    key_info: dict[str, Any] | None = None
+    key_info: dict[str, object] | None = None
     failed_reason: str
 
 
@@ -78,14 +78,14 @@ class KeyUpdateFields(LiteLLMBaseModel):
     max_budget: float | None = None
     budget_id: str | None = None
     budget_duration: str | None = None
-    budget_limits: list[Any] | None = None
-    model_max_budget: dict[str, Any] | None = None
+    budget_limits: list[object] | None = None
+    model_max_budget: dict[str, object] | None = None
 
     # Rate limits
     tpm_limit: int | None = None
     rpm_limit: int | None = None
-    model_tpm_limit: dict[str, Any] | None = None
-    model_rpm_limit: dict[str, Any] | None = None
+    model_tpm_limit: dict[str, object] | None = None
+    model_rpm_limit: dict[str, object] | None = None
     max_parallel_requests: int | None = None
     rpm_limit_type: Literal["guaranteed_throughput", "best_effort_throughput", "dynamic"] | None = None
     tpm_limit_type: Literal["guaranteed_throughput", "best_effort_throughput", "dynamic"] | None = None
@@ -99,7 +99,7 @@ class KeyUpdateFields(LiteLLMBaseModel):
 
     # Operational metadata
     tags: list[str] | None = None
-    metadata: dict[str, Any] | None = None
+    metadata: dict[str, object] | None = None
 
     @model_validator(mode="after")
     def validate_temp_budget(self) -> "KeyUpdateFields":

tests/unit/litellm_core_utils/test_text_completion_conversion.py

@@ -1,10 +1,12 @@
 import asyncio
+from typing import Final
 from unittest.mock import MagicMock, patch
 
 import pytest
 
 import litellm
 from litellm import text_completion
+from litellm.llms.custom_llm import CustomLLM
 from litellm.types.utils import ModelResponse, TextCompletionResponse, Usage
 from litellm.utils import LiteLLMResponseObjectHandler
 
@@ -1298,3 +1300,30 @@ def mock_post(*args, **kwargs):
         "usage": {"prompt_tokens": 2, "total_tokens": 18, "completion_tokens": 16},
     }
     return mock_response
+
+
+def test_text_completion_reads_a_custom_provider_dict_reply(monkeypatch: pytest.MonkeyPatch) -> None:
+    class DictReplyLLM(CustomLLM):
+        def completion(self, *args: object, **kwargs: object) -> dict[str, object]:
+            return {
+                "id": "dict-reply",
+                "object": "chat.completion",
+                "created": 1,
+                "model": "x",
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "hi from a dict reply"},
+                        "finish_reason": "stop",
+                    }
+                ],
+            }
+
+    monkeypatch.setattr(litellm, "custom_provider_map", [{"provider": "dictreply", "custom_handler": DictReplyLLM()}])
+    monkeypatch.setattr(litellm, "provider_list", [*litellm.provider_list, "dictreply"])
+    monkeypatch.setattr(litellm, "_custom_providers", [*litellm._custom_providers, "dictreply"])
+
+    response: Final = text_completion(model="dictreply/x", prompt="hi")
+
+    assert response.choices[0].text == "hi from a dict reply"
+    assert response.choices[0].finish_reason == "stop"

tests/unit/proxy/management_endpoints/policy_endpoints/test_ai_policy_suggester.py

@@ -3,6 +3,7 @@ Tests for AiPolicySuggester class.
 """
 
 import json
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -214,6 +215,26 @@ class TestAiPolicySuggester:
         assert result["selected_templates"] == []
         assert "No templates" in result["explanation"]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("selected_templates", [{}, ""])
+    async def test_suggest_reads_an_empty_non_list_selection_as_no_templates(self, selected_templates: object) -> None:
+        mock_tool_call: Final = MagicMock()
+        mock_tool_call.function.arguments = json.dumps(
+            {"selected_templates": selected_templates, "explanation": "No template fits"}
+        )
+        mock_response: Final = MagicMock()
+        mock_response.choices = [MagicMock()]
+        mock_response.choices[0].message.tool_calls = [mock_tool_call]
+
+        with patch("litellm.acompletion", new_callable=AsyncMock, return_value=mock_response):
+            result: Final = await AiPolicySuggester().suggest(
+                templates=SAMPLE_TEMPLATES,
+                attack_examples=["Ignore all previous instructions"],
+                description="",
+            )
+
+        assert result == {"selected_templates": [], "explanation": "No template fits"}
+
     @pytest.mark.asyncio
     async def test_suggest_calls_litellm_with_correct_params(self):
         suggester = AiPolicySuggester()

tests/unit/router_strategy/test_auto_router.py

@@ -398,6 +398,19 @@ def _auto_router(routelayer: Any, litellm_router_instance: Any = None, **kwargs:
     return auto_router
 
 
+@pytest.mark.parametrize("routes", [{}, ""])
+def test_should_load_no_routes_from_an_empty_non_list_routes_value(routes: object) -> None:
+    auto_router: Final = AutoRouter(
+        model_name="my-auto-router",
+        auto_router_config=json.dumps({"routes": routes}),
+        default_model="fallback-model",
+        embedding_model="text-embedding-3-small",
+        litellm_router_instance=StubEmbeddingRouter(),
+    )
+
+    assert auto_router.loaded_routes == []
+
+
 class TestAutoRouterAlwaysResolvesARoutableModel:
     """The hook returns the alias's default model instead of failing or leaking the alias downstream."""