Reconcile unreconciled Anthropic cache tokens in UsageMetrics.

LiteLLM and flow event paths can pass raw Anthropic usage where input_tokens excludes cache counters; fold them into prompt and total tokens without double-counting native provider payloads.

Co-authored-by: Cursor <cursoragent@cursor.com>

lib/crewai/src/crewai/types/usage_metrics.py

@@ -108,6 +108,40 @@ class UsageMetrics(BaseModel):
             ),
         )
 
+    @staticmethod
+    def _has_unreconciled_anthropic_cache_keys(usage_data: dict[str, Any]) -> bool:
+        """Detect raw Anthropic usage that still splits cache from ``input_tokens``.
+
+        The native ``AnthropicCompletion`` provider folds cache read/creation
+        counters into ``input_tokens`` before usage reaches this normalizer.
+        LiteLLM and flow-level event aggregation can still deliver the raw
+        Anthropic API shape, where ``input_tokens`` is only the uncached
+        portion and cache counters arrive as separate keys. Without
+        reconciling here, ``prompt_tokens`` and ``total_tokens`` undercount
+        billed usage on cached Anthropic workloads.
+        """
+        return (
+            "input_tokens" in usage_data
+            and (
+                "cache_read_input_tokens" in usage_data
+                or "cache_creation_input_tokens" in usage_data
+            )
+        )
+
+    @staticmethod
+    def _resolve_billed_prompt_tokens(usage_data: dict[str, Any]) -> int:
+        """Return the full billed prompt/input token count for a usage dict."""
+        if UsageMetrics._has_unreconciled_anthropic_cache_keys(usage_data):
+            return (
+                _coerce_int(usage_data.get("input_tokens"))
+                + _coerce_int(usage_data.get("cache_read_input_tokens"))
+                + _coerce_int(usage_data.get("cache_creation_input_tokens"))
+            )
+
+        return _first_int(
+            usage_data, "prompt_tokens", "prompt_token_count", "input_tokens"
+        )
+
     @classmethod
     def from_provider_dict(cls, usage_data: dict[str, Any] | None) -> Self | None:
         """Normalize a provider's raw usage dict into a ``UsageMetrics``.
@@ -125,9 +159,7 @@ class UsageMetrics(BaseModel):
         if not usage_data:
             return None
 
-        prompt_tokens = _first_int(
-            usage_data, "prompt_tokens", "prompt_token_count", "input_tokens"
-        )
+        prompt_tokens = cls._resolve_billed_prompt_tokens(usage_data)
         completion_tokens = _first_int(
             usage_data,
             "completion_tokens",
@@ -151,6 +183,5 @@ class UsageMetrics(BaseModel):
             completion_tokens=completion_tokens,
             cached_prompt_tokens=cached_prompt_tokens,
             reasoning_tokens=_coerce_int(usage_data.get("reasoning_tokens")),
-            cache_creation_tokens=_coerce_int(usage_data.get("cache_creation_tokens")),
             successful_requests=1,
         )

lib/crewai/tests/events/test_llm_usage_event.py

@@ -297,3 +297,123 @@ class TestUsageMetricsNewFields:
         dumped = metrics.model_dump()
         assert dumped["reasoning_tokens"] == 10
         assert dumped["cache_creation_tokens"] == 5
+
+
+class TestFromProviderDictAnthropicCacheTokens:
+    def test_cache_read_tokens_included_in_prompt_and_total(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 3,
+                "output_tokens": 44,
+                "cache_read_input_tokens": 2061,
+            }
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 2064
+        assert metrics.completion_tokens == 44
+        assert metrics.total_tokens == 2108
+        assert metrics.cached_prompt_tokens == 2061
+
+    def test_cache_creation_tokens_included_in_prompt_and_total(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 100,
+                "output_tokens": 50,
+                "cache_creation_input_tokens": 20,
+            }
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 120
+        assert metrics.total_tokens == 170
+
+    def test_cache_read_and_creation_tokens_both_included(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 100,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 30,
+                "cache_creation_input_tokens": 20,
+            }
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 150
+        assert metrics.total_tokens == 200
+        assert metrics.cached_prompt_tokens == 30
+
+    def test_missing_cache_fields_preserve_non_cached_totals(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {"input_tokens": 100, "output_tokens": 50}
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 100
+        assert metrics.total_tokens == 150
+        assert metrics.cached_prompt_tokens == 0
+
+    def test_reconciled_native_dict_is_not_double_counted(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 150,
+                "output_tokens": 50,
+                "cached_prompt_tokens": 30,
+            }
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 150
+        assert metrics.total_tokens == 200
+
+    def test_openai_cached_prompt_tokens_are_not_added_twice(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        metrics = UsageMetrics.from_provider_dict(
+            {
+                "prompt_tokens": 100,
+                "completion_tokens": 50,
+                "prompt_tokens_details": {"cached_tokens": 30},
+            }
+        )
+
+        assert metrics is not None
+        assert metrics.prompt_tokens == 100
+        assert metrics.total_tokens == 150
+        assert metrics.cached_prompt_tokens == 30
+
+    def test_cumulative_usage_via_add_usage_metrics(self):
+        from crewai.types.usage_metrics import UsageMetrics
+
+        first = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 100,
+                "output_tokens": 50,
+                "cache_read_input_tokens": 30,
+            }
+        )
+        second = UsageMetrics.from_provider_dict(
+            {
+                "input_tokens": 40,
+                "output_tokens": 20,
+            }
+        )
+
+        assert first is not None and second is not None
+        first.add_usage_metrics(second)
+
+        assert first.prompt_tokens == 170
+        assert first.completion_tokens == 70
+        assert first.total_tokens == 240
+        assert first.cached_prompt_tokens == 30
+        assert first.successful_requests == 2

lib/crewai/tests/test_flow_usage_metrics.py

@@ -130,7 +130,6 @@ class TestUsageDictToMetrics:
                     "total_tokens": 180,
                     "cached_prompt_tokens": 40,
                     "reasoning_tokens": 25,
-                    "cache_creation_tokens": 10,
                 },
                 UsageMetrics(
                     prompt_tokens=100,
@@ -138,7 +137,6 @@ class TestUsageDictToMetrics:
                     total_tokens=180,
                     cached_prompt_tokens=40,
                     reasoning_tokens=25,
-                    cache_creation_tokens=10,
                     successful_requests=1,
                 ),
             ),