fix(bedrock): price 1h Converse cache writes at the 1h rate

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

litellm/llms/bedrock/chat/converse_transformation.py

@@ -57,6 +57,7 @@ from litellm.types.llms.openai import (
     OpenAIMessageContentListBlock,
 )
 from litellm.types.utils import (
+    CacheCreationTokenDetails,
     ChatCompletionMessageToolCall,
     CompletionTokensDetailsWrapper,
     Function,
@@ -1770,6 +1771,30 @@ class AmazonConverseConfig(BaseConfig):
                 thinking_blocks_list.append(_redacted_block)
         return thinking_blocks_list
 
+    @staticmethod
+    def _transform_cache_creation_token_details(
+        usage: ConverseTokenUsageBlock,
+        cache_creation_input_tokens: int,
+    ) -> CacheCreationTokenDetails | None:
+        """Split Bedrock's ``cacheDetails`` per-TTL breakdown into 5m and 1h cache write buckets.
+
+        Returns ``None`` when Bedrock reports no breakdown or when the known TTLs don't account for
+        every cache write token, so pricing falls back to the aggregate cache write rate.
+        """
+        cache_details: Final = usage.get("cacheDetails")
+        if not cache_details:
+            return None
+        tokens_by_ttl: Final = {
+            ttl: sum(detail["inputTokens"] for detail in cache_details if detail["ttl"] == ttl)
+            for ttl in ("5m", "1h")
+        }
+        if sum(tokens_by_ttl.values()) != cache_creation_input_tokens:
+            return None
+        return CacheCreationTokenDetails(
+            ephemeral_5m_input_tokens=tokens_by_ttl["5m"],
+            ephemeral_1h_input_tokens=tokens_by_ttl["1h"],
+        )
+
     def _transform_usage(
         self,
         usage: ConverseTokenUsageBlock,
@@ -1792,6 +1817,9 @@ class AmazonConverseConfig(BaseConfig):
         prompt_tokens_details: Final = PromptTokensDetailsWrapper(
             cached_tokens=cache_read_input_tokens,
             cache_creation_tokens=cache_creation_input_tokens,
+            cache_creation_token_details=self._transform_cache_creation_token_details(
+                usage, cache_creation_input_tokens
+            ),
             text_tokens=raw_input_tokens,
         )
         reasoning_tokens = token_counter(text=reasoning_content, count_response_tokens=True) if reasoning_content else 0

litellm/types/llms/bedrock.py

@@ -1,8 +1,9 @@
 import json
+from collections.abc import Sequence
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Final, Literal
 
-from typing_extensions import Required, TypedDict, override
+from typing_extensions import NotRequired, ReadOnly, Required, TypedDict, override
 
 from .openai import ChatCompletionToolCallChunk
 
@@ -216,6 +217,11 @@ class ConverseResponseOutputBlock(TypedDict):
     message: MessageBlock | None
 
 
+class ConverseCacheDetailBlock(TypedDict):
+    ttl: ReadOnly[str]
+    inputTokens: ReadOnly[int]
+
+
 class ConverseTokenUsageBlock(TypedDict):
     inputTokens: int
     outputTokens: int
@@ -224,6 +230,7 @@ class ConverseTokenUsageBlock(TypedDict):
     cacheReadInputTokens: int
     cacheWriteInputTokenCount: int
     cacheWriteInputTokens: int
+    cacheDetails: NotRequired[ReadOnly[Sequence[ConverseCacheDetailBlock]]]
 
 
 class ServiceTierBlock(TypedDict):

tests/test_litellm/llms/bedrock/chat/test_converse_transformation.py

@@ -51,6 +51,69 @@ def test_transform_usage():
     assert openai_usage.completion_tokens_details.text_tokens == usage["outputTokens"]
 
 
+@pytest.mark.parametrize(
+    "cache_details, expected_5m, expected_1h",
+    [
+        ([{"ttl": "1h", "inputTokens": 11632}], 0, 11632),
+        ([{"ttl": "5m", "inputTokens": 11632}], 11632, 0),
+        (
+            [{"ttl": "5m", "inputTokens": 1632}, {"ttl": "1h", "inputTokens": 10000}],
+            1632,
+            10000,
+        ),
+        (None, None, None),
+        ([{"ttl": "3h", "inputTokens": 11632}], None, None),
+    ],
+)
+def test_transform_usage_splits_cache_details_by_ttl(cache_details, expected_5m, expected_1h):
+    """Bedrock reports cache write TTLs in `cacheDetails`; without it 1h writes are billed at the 5m rate."""
+    usage = ConverseTokenUsageBlock(
+        **{
+            "inputTokens": 16,
+            "outputTokens": 4,
+            "totalTokens": 11652,
+            "cacheReadInputTokens": 0,
+            "cacheWriteInputTokens": 11632,
+            **({"cacheDetails": cache_details} if cache_details is not None else {}),
+        }
+    )
+    openai_usage = AmazonConverseConfig()._transform_usage(usage)
+    assert openai_usage._cache_creation_input_tokens == 11632
+    details = getattr(openai_usage.prompt_tokens_details, "cache_creation_token_details", None)
+    if expected_5m is None:
+        assert details is None
+        return
+    assert details is not None
+    assert details.ephemeral_5m_input_tokens == expected_5m
+    assert details.ephemeral_1h_input_tokens == expected_1h
+
+
+def test_bedrock_converse_1h_cache_write_cost_uses_1h_rate():
+    """Regression for 1h Bedrock Converse cache writes being priced at the 5m rate."""
+    usage = ConverseTokenUsageBlock(
+        **{
+            "inputTokens": 16,
+            "outputTokens": 4,
+            "totalTokens": 11652,
+            "cacheReadInputTokens": 0,
+            "cacheWriteInputTokens": 11632,
+            "cacheDetails": [{"ttl": "1h", "inputTokens": 11632}],
+        }
+    )
+    openai_usage = AmazonConverseConfig()._transform_usage(usage)
+    model = "bedrock/converse/global.anthropic.claude-opus-4-8"
+    prompt_cost, completion_cost = litellm.cost_calculator.cost_per_token(model=model, usage_object=openai_usage)
+    model_info = litellm.get_model_info(model=model)
+    expected_prompt_cost = (
+        16 * model_info["input_cost_per_token"] + 11632 * model_info["cache_creation_input_token_cost_above_1hr"]
+    )
+    assert prompt_cost == pytest.approx(expected_prompt_cost)
+    assert prompt_cost > 16 * model_info["input_cost_per_token"] + 11632 * model_info[
+        "cache_creation_input_token_cost"
+    ]
+    assert completion_cost == pytest.approx(4 * model_info["output_cost_per_token"])
+
+
 def test_transform_usage_with_reasoning_content():
     """Test that completion_tokens_details correctly tracks reasoning vs text tokens."""
     usage = ConverseTokenUsageBlock(