fix(models): take the final input and cache counts from Anthropic message_delta

src/google/adk/models/anthropic_llm.py

@@ -776,6 +776,14 @@ def _extract_cached_token_count(usage: Any) -> int | None:
   return cached if isinstance(cached, int) else None
 
 
+def _extract_uncached_input_token_count(
+    usage: Any, default: int | None = 0
+) -> int | None:
+  """Returns ``input_tokens``, the part of the prompt outside the cache."""
+  value = getattr(usage, "input_tokens", None)
+  return value if isinstance(value, int) else default
+
+
 def _extract_prompt_token_count(usage: anthropic_types.Usage) -> int:
   """Returns every input token billed for the turn.
 
@@ -1343,7 +1351,7 @@ async def _generate_content_streaming(
     tool_use_blocks: dict[int, _ToolUseAccumulator] = {}
     thinking_blocks: dict[int, _ThinkingAccumulator] = {}
     redacted_thinking_blocks: dict[int, str] = {}
-    input_tokens = 0
+    uncached_input_tokens = 0
     output_tokens = 0
     thinking_tokens: int | None = None
     cached_input_tokens: int | None = None
@@ -1353,7 +1361,9 @@ async def _generate_content_streaming(
 
     async for event in raw_stream:
       if event.type == "message_start":
-        input_tokens = _extract_prompt_token_count(event.message.usage)
+        uncached_input_tokens = _extract_uncached_input_token_count(
+            event.message.usage
+        )
         output_tokens = event.message.usage.output_tokens
         thinking_tokens = _extract_thinking_token_count(event.message.usage)
         cached_input_tokens = _extract_cached_token_count(event.message.usage)
@@ -1491,6 +1501,17 @@ async def _generate_content_streaming(
         # thinking detail is refreshed alongside the total it is nested in.
         output_tokens = event.usage.output_tokens
         thinking_tokens = _extract_thinking_token_count(event.usage)
+        # The input and cache counts are cumulative here too, and are final
+        # where ``message_start`` could only report what it knew up front.
+        final_uncached = _extract_uncached_input_token_count(event.usage, None)
+        if final_uncached is not None:
+          uncached_input_tokens = final_uncached
+        final_cached = _extract_cached_token_count(event.usage)
+        if final_cached is not None:
+          cached_input_tokens = final_cached
+        final_creation = _extract_cache_creation_token_count(event.usage)
+        if final_creation is not None:
+          cache_creation_tokens = final_creation
         if event.delta and event.delta.stop_reason:
           stop_reason = event.delta.stop_reason
 
@@ -1536,6 +1557,11 @@ async def _generate_content_streaming(
         function_call.id = tool_acc.id
         all_parts.append(part)
 
+    input_tokens = (
+        uncached_input_tokens
+        + (cached_input_tokens or 0)
+        + (cache_creation_tokens or 0)
+    )
     usage_metadata = types.GenerateContentResponseUsageMetadata(
         prompt_token_count=input_tokens,
         candidates_token_count=output_tokens - (thinking_tokens or 0),

tests/unittests/models/test_anthropic_llm.py

@@ -2103,6 +2103,72 @@ async def test_streaming_text_yields_partial_and_final():
   assert responses[2].finish_reason == "STOP"
 
 
+@pytest.mark.asyncio
+async def test_streaming_takes_final_input_and_cache_counts_from_message_delta():
+  """message_delta carries the cumulative input and cache counts."""
+  llm = AnthropicLlm(model="claude-sonnet-4-20250514")
+
+  events = [
+      MagicMock(
+          type="message_start",
+          message=MagicMock(
+              model="claude-sonnet-4-20250514",
+              usage=anthropic_types.Usage(
+                  input_tokens=3,
+                  output_tokens=1,
+                  cache_read_input_tokens=0,
+                  cache_creation_input_tokens=0,
+              ),
+          ),
+      ),
+      MagicMock(
+          type="content_block_start",
+          index=0,
+          content_block=anthropic_types.TextBlock(text="", type="text"),
+      ),
+      MagicMock(
+          type="content_block_delta",
+          index=0,
+          delta=anthropic_types.TextDelta(text="Hi", type="text_delta"),
+      ),
+      MagicMock(type="content_block_stop", index=0),
+      MagicMock(
+          type="message_delta",
+          delta=MagicMock(stop_reason="end_turn"),
+          usage=anthropic_types.MessageDeltaUsage(
+              output_tokens=7,
+              input_tokens=120,
+              cache_read_input_tokens=30,
+              cache_creation_input_tokens=10,
+          ),
+      ),
+      MagicMock(type="message_stop"),
+  ]
+
+  mock_client = MagicMock()
+  mock_client.messages.create = AsyncMock(
+      return_value=_make_mock_stream_events(events)
+  )
+
+  llm_request = LlmRequest(
+      model="claude-sonnet-4-20250514",
+      contents=[Content(role="user", parts=[Part.from_text(text="Hi")])],
+      config=types.GenerateContentConfig(system_instruction="You are helpful"),
+  )
+
+  with mock.patch.object(llm, "_anthropic_client", mock_client):
+    responses = [
+        r async for r in llm.generate_content_async(llm_request, stream=True)
+    ]
+
+  usage = responses[-1].usage_metadata
+  assert usage.prompt_token_count == 160
+  assert usage.cached_content_token_count == 30
+  assert usage.candidates_token_count == 7
+  assert usage.total_token_count == 167
+  assert usage.cache_creation_input_tokens == 10
+
+
 @pytest.mark.asyncio
 async def test_streaming_tool_use_yields_function_call():
   """Streaming tool_use should accumulate args and yield in final."""