fix(cache): handle string content in is_cached_message (#17853)

Fixes #17821

The `is_cached_message` function crashed with TypeError when message
content was a string instead of a list of content blocks.

Changes:
- Add explicit `isinstance(content, list)` check before iteration
- Add `isinstance(content_item, dict)` check inside loop to skip non-dict items
- Use `.get()` for safer nested dict access
- Follow same pattern as `extract_ttl_from_cached_messages` (same module)

Tests:
- Add TestIsCachedMessage class with 9 test cases covering:
  - String content (the reported bug)
  - None content
  - Missing content key
  - Empty list content
  - List with/without cache_control
  - Mixed content types (strings + dicts)
  - Wrong cache_control type

Author: peterkc

litellm/utils.py

@@ -793,10 +793,8 @@ def function_setup(  # noqa: PLR0915
             or call_type == CallTypes.transcription.value
         ):
             _file_obj: FileTypes = args[1] if len(args) > 1 else kwargs["file"]
-            file_checksum = (
-                litellm.litellm_core_utils.audio_utils.utils.get_audio_file_content_hash(
-                    file_obj=_file_obj
-                )
+            file_checksum = litellm.litellm_core_utils.audio_utils.utils.get_audio_file_content_hash(
+                file_obj=_file_obj
             )
             if "metadata" in kwargs:
                 kwargs["metadata"]["file_checksum"] = file_checksum
@@ -2903,7 +2901,7 @@ def _check_valid_arg(supported_params: Optional[list]):
             non_default_params=non_default_params,
             optional_params={},
             model=model,
-            drop_params=drop_params if drop_params is not None else False
+            drop_params=drop_params if drop_params is not None else False,
         )
     elif custom_llm_provider == "infinity":
         supported_params = get_supported_openai_params(
@@ -5063,7 +5061,9 @@ def _get_model_info_helper(  # noqa: PLR0915
                     "output_cost_per_video_per_second", None
                 ),
                 output_cost_per_image=_model_info.get("output_cost_per_image", None),
-                output_cost_per_image_token=_model_info.get("output_cost_per_image_token", None),
+                output_cost_per_image_token=_model_info.get(
+                    "output_cost_per_image_token", None
+                ),
                 output_vector_size=_model_info.get("output_vector_size", None),
                 citation_cost_per_token=_model_info.get(
                     "citation_cost_per_token", None
@@ -6719,7 +6719,9 @@ def _get_base_model_from_metadata(model_call_details=None):
             return _base_model
         metadata = litellm_params.get("metadata", {})
 
-        base_model_from_metadata = _get_base_model_from_litellm_call_metadata(metadata=metadata)
+        base_model_from_metadata = _get_base_model_from_litellm_call_metadata(
+            metadata=metadata
+        )
         if base_model_from_metadata is not None:
             return base_model_from_metadata
 
@@ -6808,14 +6810,22 @@ def is_cached_message(message: AllMessageValues) -> bool:
     """
     if "content" not in message:
         return False
-    if message["content"] is None or isinstance(message["content"], str):
+
+    content = message["content"]
+
+    # Handle non-list content types (None, str, etc.)
+    if not isinstance(content, list):
         return False
 
-    for content in message["content"]:
+    for content_item in content:
+        # Ensure content_item is a dictionary before accessing keys
+        if not isinstance(content_item, dict):
+            continue
+
         if (
-            content["type"] == "text"
-            and content.get("cache_control") is not None
-            and content["cache_control"]["type"] == "ephemeral"  # type: ignore
+            content_item.get("type") == "text"
+            and content_item.get("cache_control") is not None
+            and content_item.get("cache_control", {}).get("type") == "ephemeral"
         ):
             return True
 
@@ -7475,8 +7485,11 @@ def get_provider_responses_api_config(
             # Note: GPT models (gpt-3.5, gpt-4, gpt-5, etc.) support temperature parameter
             # O-series models (o1, o3) do not contain "gpt" and have different parameter restrictions
             is_gpt_model = model and "gpt" in model.lower()
-            is_o_series = model and ("o_series" in model.lower() or (supports_reasoning(model) and not is_gpt_model))
-            
+            is_o_series = model and (
+                "o_series" in model.lower()
+                or (supports_reasoning(model) and not is_gpt_model)
+            )
+
             if is_o_series:
                 return litellm.AzureOpenAIOSeriesResponsesAPIConfig()
             else:
@@ -7495,10 +7508,10 @@ def get_provider_skills_api_config(
     ) -> Optional["BaseSkillsAPIConfig"]:
         """
         Get provider-specific Skills API configuration
-        
+
         Args:
             provider: The LLM provider
-            
+
         Returns:
             Provider-specific Skills API config or None
         """
@@ -8197,7 +8210,9 @@ def get_non_default_transcription_params(kwargs: dict) -> dict:
     return non_default_params
 
 
-def add_openai_metadata(metadata: Optional[Mapping[str, Any]]) -> Optional[Dict[str, str]]:
+def add_openai_metadata(
+    metadata: Optional[Mapping[str, Any]],
+) -> Optional[Dict[str, str]]:
     """
     Add metadata to openai optional parameters, excluding hidden params.
 
@@ -8231,6 +8246,7 @@ def add_openai_metadata(metadata: Optional[Mapping[str, Any]]) -> Optional[Dict[
 
     return visible_metadata.copy()
 
+
 def get_requester_metadata(metadata: dict):
     if not metadata:
         return None
@@ -8247,6 +8263,7 @@ def get_requester_metadata(metadata: dict):
 
     return None
 
+
 def return_raw_request(endpoint: CallTypes, kwargs: dict) -> RawRequestTypedDict:
     """
     Return the json str of the request

tests/test_litellm/test_utils.py

@@ -23,6 +23,7 @@
     TextCompletionStreamWrapper,
     get_llm_provider,
     get_optional_params_image_gen,
+    is_cached_message,
 )
 
 # Adds the parent directory to the system path
@@ -846,6 +847,7 @@ def test_check_provider_match():
     model_info = {"litellm_provider": "bedrock"}
     assert litellm.utils._check_provider_match(model_info, "openai") is False
 
+
 def test_get_provider_rerank_config():
     """
     Test the get_provider_rerank_config function for various providers
@@ -854,9 +856,12 @@ def test_get_provider_rerank_config():
     from litellm.utils import LlmProviders, ProviderConfigManager
 
     # Test for hosted_vllm provider
-    config = ProviderConfigManager.get_provider_rerank_config("my_model", LlmProviders.HOSTED_VLLM, 'http://localhost', [])
+    config = ProviderConfigManager.get_provider_rerank_config(
+        "my_model", LlmProviders.HOSTED_VLLM, "http://localhost", []
+    )
     assert isinstance(config, HostedVLLMRerankConfig)
 
+
 # Models that should be skipped during testing
 OLD_PROVIDERS = ["aleph_alpha", "palm"]
 SKIP_MODELS = [
@@ -2513,3 +2518,87 @@ def test_get_valid_models_with_cli_pattern(self):
             assert "headers" in call_kwargs
             headers = call_kwargs["headers"]
             assert headers.get("Authorization") == "Bearer sk-test-cli-key-123"
+
+
+class TestIsCachedMessage:
+    """Test is_cached_message function for context caching detection.
+
+    Fixes GitHub issue #17821 - TypeError when content is string instead of list.
+    """
+
+    def test_string_content_returns_false(self):
+        """String content should return False without crashing."""
+        message = {"role": "user", "content": "Hello world"}
+        assert is_cached_message(message) is False
+
+    def test_none_content_returns_false(self):
+        """None content should return False."""
+        message = {"role": "user", "content": None}
+        assert is_cached_message(message) is False
+
+    def test_missing_content_returns_false(self):
+        """Message without content key should return False."""
+        message = {"role": "user"}
+        assert is_cached_message(message) is False
+
+    def test_list_content_without_cache_control_returns_false(self):
+        """List content without cache_control should return False."""
+        message = {"role": "user", "content": [{"type": "text", "text": "Hello"}]}
+        assert is_cached_message(message) is False
+
+    def test_list_content_with_cache_control_returns_true(self):
+        """List content with cache_control ephemeral should return True."""
+        message = {
+            "role": "user",
+            "content": [
+                {
+                    "type": "text",
+                    "text": "Hello",
+                    "cache_control": {"type": "ephemeral"},
+                }
+            ],
+        }
+        assert is_cached_message(message) is True
+
+    def test_list_with_non_dict_items_skips_them(self):
+        """List content with non-dict items should skip them gracefully."""
+        message = {
+            "role": "user",
+            "content": ["string_item", 123, {"type": "text", "text": "Hello"}],
+        }
+        assert is_cached_message(message) is False
+
+    def test_list_with_mixed_items_finds_cached(self):
+        """Mixed content list should find cached item."""
+        message = {
+            "role": "user",
+            "content": [
+                "string_item",
+                {"type": "image", "url": "..."},
+                {
+                    "type": "text",
+                    "text": "cached",
+                    "cache_control": {"type": "ephemeral"},
+                },
+            ],
+        }
+        assert is_cached_message(message) is True
+
+    def test_wrong_cache_control_type_returns_false(self):
+        """Non-ephemeral cache_control type should return False."""
+        message = {
+            "role": "user",
+            "content": [
+                {
+                    "type": "text",
+                    "text": "Hello",
+                    "cache_control": {"type": "permanent"},
+                }
+            ],
+        }
+        assert is_cached_message(message) is False
+
+    def test_empty_list_content_returns_false(self):
+        """Empty list content should return False."""
+        message = {"role": "user", "content": []}
+        assert is_cached_message(message) is False