feat: add cached token metrics support for Amazon Bedrock

- Add optional cacheReadInputTokens and cacheWriteInputTokens fields to Usage TypedDict
- Update EventLoopMetrics to accumulate cached token metrics
- Add OpenTelemetry instrumentation for cached token telemetry
- Enhance metrics summary display to show cached token information
- Maintain 100% backward compatibility with existing Usage objects
- Add comprehensive test coverage for cached token functionality

Resolves #529

Author: oaltagar-aws

src/strands/telemetry/metrics.py

@@ -264,6 +264,23 @@ def update_usage(self, usage: Usage) -> None:
         self.accumulated_usage["outputTokens"] += usage["outputTokens"]
         self.accumulated_usage["totalTokens"] += usage["totalTokens"]
 
+        # Handle optional cached token metrics
+        if "cacheReadInputTokens" in usage and usage["cacheReadInputTokens"] is not None:
+            cache_read_tokens = usage["cacheReadInputTokens"]
+            self._metrics_client.event_loop_cache_read_tokens.record(cache_read_tokens)
+            if "cacheReadInputTokens" not in self.accumulated_usage:
+                self.accumulated_usage["cacheReadInputTokens"] = 0
+            current_cache_read = self.accumulated_usage.get("cacheReadInputTokens", 0) or 0
+            self.accumulated_usage["cacheReadInputTokens"] = current_cache_read + cache_read_tokens
+
+        if "cacheWriteInputTokens" in usage and usage["cacheWriteInputTokens"] is not None:
+            cache_write_tokens = usage["cacheWriteInputTokens"]
+            self._metrics_client.event_loop_cache_write_tokens.record(cache_write_tokens)
+            if "cacheWriteInputTokens" not in self.accumulated_usage:
+                self.accumulated_usage["cacheWriteInputTokens"] = 0
+            current_cache_write = self.accumulated_usage.get("cacheWriteInputTokens", 0) or 0
+            self.accumulated_usage["cacheWriteInputTokens"] = current_cache_write + cache_write_tokens
+
     def update_metrics(self, metrics: Metrics) -> None:
         """Update the accumulated performance metrics with new metrics data.
 
@@ -325,11 +342,21 @@ def _metrics_summary_to_lines(event_loop_metrics: EventLoopMetrics, allowed_name
         f"├─ Cycles: total={summary['total_cycles']}, avg_time={summary['average_cycle_time']:.3f}s, "
         f"total_time={summary['total_duration']:.3f}s"
     )
-    yield (
-        f"├─ Tokens: in={summary['accumulated_usage']['inputTokens']}, "
-        f"out={summary['accumulated_usage']['outputTokens']}, "
-        f"total={summary['accumulated_usage']['totalTokens']}"
-    )
+
+    # Build token display with optional cached tokens
+    token_parts = [
+        f"in={summary['accumulated_usage']['inputTokens']}",
+        f"out={summary['accumulated_usage']['outputTokens']}",
+        f"total={summary['accumulated_usage']['totalTokens']}",
+    ]
+
+    # Add cached token info if present
+    if summary["accumulated_usage"].get("cacheReadInputTokens"):
+        token_parts.append(f"cache_read={summary['accumulated_usage']['cacheReadInputTokens']}")
+    if summary["accumulated_usage"].get("cacheWriteInputTokens"):
+        token_parts.append(f"cache_write={summary['accumulated_usage']['cacheWriteInputTokens']}")
+
+    yield f"├─ Tokens: {', '.join(token_parts)}"
     yield f"├─ Bedrock Latency: {summary['accumulated_metrics']['latencyMs']}ms"
 
     yield "├─ Tool Usage:"
@@ -421,6 +448,8 @@ class MetricsClient:
     event_loop_latency: Histogram
     event_loop_input_tokens: Histogram
     event_loop_output_tokens: Histogram
+    event_loop_cache_read_tokens: Histogram
+    event_loop_cache_write_tokens: Histogram
 
     tool_call_count: Counter
     tool_success_count: Counter
@@ -474,3 +503,9 @@ def create_instruments(self) -> None:
         self.event_loop_output_tokens = self.meter.create_histogram(
             name=constants.STRANDS_EVENT_LOOP_OUTPUT_TOKENS, unit="token"
         )
+        self.event_loop_cache_read_tokens = self.meter.create_histogram(
+            name=constants.STRANDS_EVENT_LOOP_CACHE_READ_TOKENS, unit="token"
+        )
+        self.event_loop_cache_write_tokens = self.meter.create_histogram(
+            name=constants.STRANDS_EVENT_LOOP_CACHE_WRITE_TOKENS, unit="token"
+        )

src/strands/telemetry/metrics_constants.py

@@ -13,3 +13,5 @@
 STRANDS_EVENT_LOOP_CYCLE_DURATION = "strands.event_loop.cycle_duration"
 STRANDS_EVENT_LOOP_INPUT_TOKENS = "strands.event_loop.input.tokens"
 STRANDS_EVENT_LOOP_OUTPUT_TOKENS = "strands.event_loop.output.tokens"
+STRANDS_EVENT_LOOP_CACHE_READ_TOKENS = "strands.event_loop.cache.read.tokens"
+STRANDS_EVENT_LOOP_CACHE_WRITE_TOKENS = "strands.event_loop.cache.write.tokens"

src/strands/types/event_loop.py

@@ -1,22 +1,26 @@
 """Event loop-related type definitions for the SDK."""
 
-from typing import Literal
+from typing import Literal, Optional
 
-from typing_extensions import TypedDict
+from typing_extensions import Required, TypedDict
 
 
-class Usage(TypedDict):
+class Usage(TypedDict, total=False):
     """Token usage information for model interactions.
 
     Attributes:
-        inputTokens: Number of tokens sent in the request to the model..
+        inputTokens: Number of tokens sent in the request to the model.
         outputTokens: Number of tokens that the model generated for the request.
         totalTokens: Total number of tokens (input + output).
+        cacheReadInputTokens: Number of tokens read from cache (optional).
+        cacheWriteInputTokens: Number of tokens written to cache (optional).
     """
 
-    inputTokens: int
-    outputTokens: int
-    totalTokens: int
+    inputTokens: Required[int]
+    outputTokens: Required[int]
+    totalTokens: Required[int]
+    cacheReadInputTokens: Optional[int]
+    cacheWriteInputTokens: Optional[int]
 
 
 class Metrics(TypedDict):