fix: map OpenAI cached_tokens to cache_read_input_tokens on /v1/messages

litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py

@@ -230,6 +230,9 @@ def _process_event(self, event: Any) -> None:
             output_tokens = 0
             cache_creation_tokens = 0
             cache_read_tokens = 0
+            cached_tokens = 0
+            cache_write_tokens = 0
+            has_openai_cache_details = False
 
             if response_obj is not None:
                 status = getattr(response_obj, "status", None)
@@ -239,18 +242,32 @@ def _process_event(self, event: Any) -> None:
                 if usage is not None:
                     input_tokens = getattr(usage, "input_tokens", 0) or 0
                     output_tokens = getattr(usage, "output_tokens", 0) or 0
+                    # Prefer direct Anthropic-style cache fields if present
                     cache_creation_tokens = int(
                         getattr(usage, "cache_creation_input_tokens", 0) or 0
                     )
                     cache_read_tokens = int(
                         getattr(usage, "cache_read_input_tokens", 0) or 0
                     )
-                    if not cache_read_tokens:
-                        details = getattr(usage, "input_tokens_details", None)
-                        if details is not None:
-                            cache_read_tokens = int(
-                                getattr(details, "cached_tokens", 0) or 0
-                            )
+                    # Fall back to OpenAI-style input_tokens_details.
+                    # Reads: cached_tokens. Writes (GPT-5.6+): cache_write_tokens
+                    # (or cache_creation_tokens alias used elsewhere in LiteLLM).
+                    # Keep OpenAI native fields on the wire for proxy billing.
+                    details = getattr(usage, "input_tokens_details", None)
+                    if details is not None:
+                        has_openai_cache_details = True
+                        cached_tokens = int(
+                            getattr(details, "cached_tokens", 0) or 0
+                        )
+                        cache_write_tokens = int(
+                            getattr(details, "cache_write_tokens", 0)
+                            or getattr(details, "cache_creation_tokens", 0)
+                            or 0
+                        )
+                        if not cache_read_tokens:
+                            cache_read_tokens = cached_tokens
+                        if not cache_creation_tokens:
+                            cache_creation_tokens = cache_write_tokens
 
             # Check if tool_use was in the output to override stop_reason
             if response_obj is not None:
@@ -271,6 +288,9 @@ def _process_event(self, event: Any) -> None:
                 usage_delta["cache_creation_input_tokens"] = cache_creation_tokens
             if cache_read_tokens:
                 usage_delta["cache_read_input_tokens"] = cache_read_tokens
+            if has_openai_cache_details:
+                usage_delta["cached_tokens"] = cached_tokens
+                usage_delta["cache_write_tokens"] = cache_write_tokens
 
             self._chunk_queue.append(
                 {

litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py

@@ -468,10 +468,42 @@ def translate_response(
         input_tokens = int(getattr(raw_usage, "input_tokens", 0) or 0)
         output_tokens = int(getattr(raw_usage, "output_tokens", 0) or 0)
 
+        # Prefer direct Anthropic-style cache fields if present, then fall back
+        # to OpenAI-style input_tokens_details (same order as the streaming
+        # iterator). GPT-5.6+ reports writes as cache_write_tokens; older
+        # OpenAI models typically omit writes so creation stays 0.
+        # Also keep OpenAI native cached_tokens / cache_write_tokens on the
+        # wire so proxies can record usage before Anthropic client shaping.
+        cache_read_input_tokens = int(
+            getattr(raw_usage, "cache_read_input_tokens", 0) or 0
+        )
+        cache_creation_input_tokens = int(
+            getattr(raw_usage, "cache_creation_input_tokens", 0) or 0
+        )
+        cached_tokens = 0
+        cache_write_tokens = 0
+        details = getattr(raw_usage, "input_tokens_details", None) if raw_usage else None
+        if details is not None:
+            cached_tokens = int(getattr(details, "cached_tokens", 0) or 0)
+            cache_write_tokens = int(
+                getattr(details, "cache_write_tokens", 0)
+                or getattr(details, "cache_creation_tokens", 0)
+                or 0
+            )
+            if not cache_read_input_tokens:
+                cache_read_input_tokens = cached_tokens
+            if not cache_creation_input_tokens:
+                cache_creation_input_tokens = cache_write_tokens
+
         anthropic_usage = AnthropicUsage(
             input_tokens=input_tokens,
             output_tokens=output_tokens,
+            cache_read_input_tokens=cache_read_input_tokens,
+            cache_creation_input_tokens=cache_creation_input_tokens,
         )
+        if details is not None:
+            anthropic_usage["cached_tokens"] = cached_tokens
+            anthropic_usage["cache_write_tokens"] = cache_write_tokens
 
         cache_creation = int(getattr(raw_usage, "cache_creation_input_tokens", 0) or 0)
         if cache_creation:

litellm/types/llms/anthropic_messages/anthropic_response.py

@@ -71,6 +71,11 @@ class AnthropicUsage(TypedDict, total=False):
     cache_creation_input_tokens: int
     cache_read_input_tokens: int
 
+    # OpenAI-native cache fields preserved on the Messages wire so downstream
+    # proxies can bill cache write/read before returning Anthropic-shaped usage.
+    cached_tokens: int
+    cache_write_tokens: int
+
 
 class AnthropicMessagesResponse(TypedDict, total=False):
     """

tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py

@@ -130,3 +130,134 @@ def test_process_event_registered_item_id_does_not_synthesize_start(self):
             ("content_block_start", 0),
             ("content_block_delta", 0),
         ]
+
+
+class TestProcessEventCacheUsage:
+    """response.completed usage must map OpenAI cache details onto Anthropic
+    cache_read_input_tokens / cache_creation_input_tokens, preferring direct
+    Anthropic-style fields when set, while preserving OpenAI native
+    cached_tokens / cache_write_tokens on the wire.
+    """
+
+    def test_cache_read_from_input_tokens_details_cached_tokens(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 1846,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 0,
+                "input_tokens_details": type("Details", (), {"cached_tokens": 1664})(),
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert len(deltas) == 1
+        assert deltas[0]["usage"]["cache_read_input_tokens"] == 1664
+        assert deltas[0]["usage"]["cached_tokens"] == 1664
+        assert deltas[0]["usage"]["cache_write_tokens"] == 0
+
+    def test_cache_read_prefers_direct_cache_read_input_tokens(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 1846,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 900,
+                "cache_creation_input_tokens": 0,
+                "input_tokens_details": type("Details", (), {"cached_tokens": 1664})(),
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert deltas[0]["usage"]["cache_read_input_tokens"] == 900
+        assert deltas[0]["usage"]["cached_tokens"] == 1664
+
+    def test_cache_write_from_input_tokens_details_cache_write_tokens(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 2000,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 0,
+                "input_tokens_details": type(
+                    "Details",
+                    (),
+                    {"cached_tokens": 0, "cache_write_tokens": 1969},
+                )(),
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert deltas[0]["usage"]["cache_creation_input_tokens"] == 1969
+        assert deltas[0]["usage"]["cache_write_tokens"] == 1969
+        assert deltas[0]["usage"]["cached_tokens"] == 0
+
+    def test_cache_write_from_input_tokens_details_cache_creation_tokens_alias(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 2000,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 0,
+                "input_tokens_details": type(
+                    "Details",
+                    (),
+                    {"cached_tokens": 0, "cache_creation_tokens": 800},
+                )(),
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert deltas[0]["usage"]["cache_creation_input_tokens"] == 800
+        assert deltas[0]["usage"]["cache_write_tokens"] == 800
+
+    def test_cache_write_prefers_direct_cache_creation_input_tokens(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 2000,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 700,
+                "input_tokens_details": type(
+                    "Details",
+                    (),
+                    {"cached_tokens": 0, "cache_write_tokens": 1969},
+                )(),
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert deltas[0]["usage"]["cache_creation_input_tokens"] == 700
+        assert deltas[0]["usage"]["cache_write_tokens"] == 1969
+
+    def test_openai_cache_fields_absent_when_details_missing(self):
+        usage = type(
+            "Usage",
+            (),
+            {
+                "input_tokens": 100,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 0,
+                "cache_creation_input_tokens": 0,
+                "input_tokens_details": None,
+            },
+        )()
+        response = type("Response", (), {"status": "completed", "usage": usage, "output": []})()
+        chunks = _process_all([{"type": "response.completed", "response": response}])
+        deltas = [c for c in chunks if c["type"] == "message_delta"]
+        assert "cached_tokens" not in deltas[0]["usage"]
+        assert "cache_write_tokens" not in deltas[0]["usage"]

tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py

@@ -823,11 +823,30 @@ def _make_mock_response(
     model: str = "gpt-4o",
     input_tokens: int = 100,
     output_tokens: int = 50,
+    cache_read_input_tokens: int | None = None,
+    cache_creation_input_tokens: int | None = None,
+    cached_tokens: int | None = None,
+    cache_write_tokens: int | None = None,
+    cache_creation_tokens: int | None = None,
 ) -> MagicMock:
     """Build a minimal mock ResponsesAPIResponse."""
     usage = MagicMock()
     usage.input_tokens = input_tokens
     usage.output_tokens = output_tokens
+    usage.cache_read_input_tokens = cache_read_input_tokens
+    usage.cache_creation_input_tokens = cache_creation_input_tokens
+    if (
+        cached_tokens is None
+        and cache_write_tokens is None
+        and cache_creation_tokens is None
+    ):
+        usage.input_tokens_details = None
+    else:
+        details = MagicMock()
+        details.cached_tokens = cached_tokens
+        details.cache_write_tokens = cache_write_tokens
+        details.cache_creation_tokens = cache_creation_tokens
+        usage.input_tokens_details = details
 
     resp = MagicMock()
     resp.id = response_id
@@ -960,6 +979,81 @@ def test_usage_mapped_correctly(self):
         result: Any = _ADAPTER.translate_response(response)
         assert result["usage"]["input_tokens"] == 200
         assert result["usage"]["output_tokens"] == 75
+        assert result["usage"].get("cache_read_input_tokens", 0) == 0
+        assert result["usage"].get("cache_creation_input_tokens", 0) == 0
+
+    def test_cache_read_from_input_tokens_details_cached_tokens(self):
+        """OpenAI-style input_tokens_details.cached_tokens maps to cache_read_input_tokens."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cached_tokens=1664,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_read_input_tokens"] == 1664
+        assert result["usage"]["cached_tokens"] == 1664
+        assert result["usage"]["cache_write_tokens"] == 0
+
+    def test_cache_read_prefers_direct_cache_read_input_tokens(self):
+        """Direct Anthropic-style cache_read_input_tokens wins over details.cached_tokens."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cache_read_input_tokens=900,
+            cached_tokens=1664,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_read_input_tokens"] == 900
+        assert result["usage"]["cached_tokens"] == 1664
+
+    def test_cache_read_from_direct_field_when_details_absent(self):
+        """Direct cache_read_input_tokens is used when input_tokens_details is missing."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cache_read_input_tokens=512,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_read_input_tokens"] == 512
+        assert "cached_tokens" not in result["usage"]
+        assert "cache_write_tokens" not in result["usage"]
+
+    def test_cache_write_from_input_tokens_details_cache_write_tokens(self):
+        """OpenAI-style cache_write_tokens maps to Anthropic creation and is kept native."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cache_write_tokens=1969,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_creation_input_tokens"] == 1969
+        assert result["usage"]["cache_write_tokens"] == 1969
+        assert result["usage"]["cached_tokens"] == 0
+
+    def test_cache_write_from_input_tokens_details_cache_creation_tokens_alias(self):
+        """LiteLLM cache_creation_tokens alias also maps to cache_creation_input_tokens."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cache_creation_tokens=800,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_creation_input_tokens"] == 800
+        assert result["usage"]["cache_write_tokens"] == 800
+
+    def test_cache_write_prefers_direct_cache_creation_input_tokens(self):
+        """Direct Anthropic-style cache_creation_input_tokens wins over details writes."""
+        response = _make_mock_response(
+            output=[_make_output_message(["OK"])],
+            cache_creation_input_tokens=700,
+            cache_write_tokens=1969,
+        )
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_creation_input_tokens"] == 700
+        assert result["usage"]["cache_write_tokens"] == 1969
+
+    def test_cache_creation_field_always_populated(self):
+        """cache_creation_input_tokens is present even when there was no write."""
+        response = _make_mock_response(output=[_make_output_message(["OK"])])
+        result: Any = _ADAPTER.translate_response(response)
+        assert result["usage"]["cache_creation_input_tokens"] == 0
+        assert "cache_write_tokens" not in result["usage"]
+        assert "cached_tokens" not in result["usage"]
 
     def test_should_populate_cache_creation_input_tokens(self):
         """cache_creation_input_tokens extra on usage is mapped to AnthropicUsage."""