feat(context): track agent.messages token size

Author: lizradway

src/strands/telemetry/metrics.py

@@ -202,6 +202,20 @@ class EventLoopMetrics:
     accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
     accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
 
+    @property
+    def latest_context_tokens(self) -> int:
+        """Most recent context size from the last LLM call.
+
+        Returns the inputTokens from the most recent cycle of the most recent
+        invocation. This represents the current context size as reported by the model.
+
+        Returns:
+            The input token count from the most recent cycle, or 0 if no cycles exist.
+        """
+        if self.agent_invocations and self.agent_invocations[-1].cycles:
+            return self.agent_invocations[-1].cycles[-1].usage.get("inputTokens", 0)
+        return 0
+
     @property
     def _metrics_client(self) -> "MetricsClient":
         """Get the singleton MetricsClient instance."""

tests/strands/telemetry/test_metrics.py

@@ -566,3 +566,50 @@ def test_reset_usage_metrics(usage, event_loop_metrics, mock_get_meter_provider)
 
     # Verify accumulated_usage is NOT cleared
     assert event_loop_metrics.accumulated_usage["inputTokens"] == 11
+
+
+def test_latest_context_tokens_no_invocations(event_loop_metrics):
+    assert event_loop_metrics.latest_context_tokens == 0
+
+
+def test_latest_context_tokens_invocation_with_no_cycles(event_loop_metrics):
+    event_loop_metrics.reset_usage_metrics()
+    assert event_loop_metrics.latest_context_tokens == 0
+
+
+def test_latest_context_tokens_returns_last_cycle(event_loop_metrics, mock_get_meter_provider):
+    event_loop_metrics.reset_usage_metrics()
+    event_loop_metrics.start_cycle(attributes={"event_loop_cycle_id": "c1"})
+    event_loop_metrics.update_usage(Usage(inputTokens=100, outputTokens=50, totalTokens=150))
+
+    event_loop_metrics.start_cycle(attributes={"event_loop_cycle_id": "c2"})
+    event_loop_metrics.update_usage(Usage(inputTokens=250, outputTokens=80, totalTokens=330))
+
+    assert event_loop_metrics.latest_context_tokens == 250
+
+
+def test_latest_context_tokens_returns_from_latest_invocation(event_loop_metrics, mock_get_meter_provider):
+    # First invocation
+    event_loop_metrics.reset_usage_metrics()
+    event_loop_metrics.start_cycle(attributes={"event_loop_cycle_id": "c1"})
+    event_loop_metrics.update_usage(Usage(inputTokens=100, outputTokens=50, totalTokens=150))
+
+    # Second invocation
+    event_loop_metrics.reset_usage_metrics()
+    event_loop_metrics.start_cycle(attributes={"event_loop_cycle_id": "c2"})
+    event_loop_metrics.update_usage(Usage(inputTokens=500, outputTokens=80, totalTokens=580))
+
+    assert event_loop_metrics.latest_context_tokens == 500
+
+
+def test_latest_context_tokens_missing_input_tokens_key(event_loop_metrics):
+    """Returns 0 when usage dict is missing inputTokens (e.g. provider bug)."""
+    event_loop_metrics.reset_usage_metrics()
+    invocation = event_loop_metrics.agent_invocations[-1]
+    invocation.cycles.append(
+        strands.telemetry.metrics.EventLoopCycleMetric(
+            event_loop_cycle_id="c1",
+            usage={"outputTokens": 50, "totalTokens": 50},
+        )
+    )
+    assert event_loop_metrics.latest_context_tokens == 0