fix(anthropic): capture server-side tool usage metrics

Flatten Anthropic usage.server_tool_use into Braintrust span metrics so
server-side tool invocations are preserved for tracing and cost analysis.

Add regression tests using a red/green workflow to cover dict-backed span
logging and object-backed usage extraction.

Fixes #171

Author: AbhiPrasad

py/src/braintrust/integrations/anthropic/_utils.py

@@ -1,5 +1,6 @@
 """Shared utilities for Anthropic API wrappers."""
 
+from collections.abc import Iterator
 from typing import Any
 
 
@@ -13,6 +14,41 @@ def __getattr__(self, name: str) -> Any:
         return getattr(self.__wrapped, name)
 
 
+def _get_usage_value(usage: Any, key: str) -> Any:
+    if isinstance(usage, dict):
+        return usage.get(key)
+    return getattr(usage, key, None)
+
+
+def _iter_usage_items(usage: Any) -> Iterator[tuple[str, Any]]:
+    if isinstance(usage, dict):
+        yield from usage.items()
+        return
+
+    try:
+        yield from vars(usage).items()
+        return
+    except TypeError:
+        pass
+
+    for key in dir(usage):
+        if key.startswith("_"):
+            continue
+        value = getattr(usage, key, None)
+        if callable(value):
+            continue
+        yield key, value
+
+
+def _maybe_add_metric(metrics: dict[str, float], key: str, value: Any) -> None:
+    if value is None or isinstance(value, bool):
+        return
+    try:
+        metrics[key] = float(value)
+    except (ValueError, TypeError):
+        pass
+
+
 def extract_anthropic_usage(usage: Any) -> dict[str, float]:
     """Extract and normalize usage metrics from Anthropic usage object or dict.
 
@@ -28,44 +64,26 @@ def extract_anthropic_usage(usage: Any) -> dict[str, float]:
         - completion_tokens (from output_tokens)
         - prompt_cached_tokens (from cache_read_input_tokens)
         - prompt_cache_creation_tokens (from cache_creation_input_tokens)
+        - server_tool_use_* (flattened from server_tool_use)
     """
     metrics: dict[str, float] = {}
 
     if not usage:
         return metrics
 
-    def get_value(key: str) -> Any:
-        if isinstance(usage, dict):
-            return usage.get(key)
-        return getattr(usage, key, None)
-
-    input_tokens = get_value("input_tokens")
-    if input_tokens is not None:
-        try:
-            metrics["prompt_tokens"] = float(input_tokens)
-        except (ValueError, TypeError):
-            pass
-
-    output_tokens = get_value("output_tokens")
-    if output_tokens is not None:
-        try:
-            metrics["completion_tokens"] = float(output_tokens)
-        except (ValueError, TypeError):
-            pass
-
-    cache_read_tokens = get_value("cache_read_input_tokens")
-    if cache_read_tokens is not None:
-        try:
-            metrics["prompt_cached_tokens"] = float(cache_read_tokens)
-        except (ValueError, TypeError):
-            pass
-
-    cache_creation_tokens = get_value("cache_creation_input_tokens")
-    if cache_creation_tokens is not None:
-        try:
-            metrics["prompt_cache_creation_tokens"] = float(cache_creation_tokens)
-        except (ValueError, TypeError):
-            pass
+    _maybe_add_metric(metrics, "prompt_tokens", _get_usage_value(usage, "input_tokens"))
+    _maybe_add_metric(metrics, "completion_tokens", _get_usage_value(usage, "output_tokens"))
+    _maybe_add_metric(metrics, "prompt_cached_tokens", _get_usage_value(usage, "cache_read_input_tokens"))
+    _maybe_add_metric(
+        metrics,
+        "prompt_cache_creation_tokens",
+        _get_usage_value(usage, "cache_creation_input_tokens"),
+    )
+
+    server_tool_use = _get_usage_value(usage, "server_tool_use")
+    if server_tool_use:
+        for key, value in _iter_usage_items(server_tool_use):
+            _maybe_add_metric(metrics, f"server_tool_use_{key}", value)
 
     return metrics
 

py/src/braintrust/integrations/anthropic/test_anthropic.py

@@ -11,6 +11,7 @@
 import pytest
 from braintrust import logger
 from braintrust.integrations.anthropic import AnthropicIntegration, wrap_anthropic
+from braintrust.integrations.anthropic._utils import extract_anthropic_usage
 from braintrust.integrations.anthropic.tracing import _log_message_to_span
 from braintrust.test_helpers import init_test_logger
 
@@ -52,6 +53,10 @@ def test_log_message_to_span_includes_stop_reason_and_stop_sequence():
             "output_tokens": 7,
             "cache_read_input_tokens": 0,
             "cache_creation_input_tokens": 0,
+            "server_tool_use": {
+                "web_search_requests": 2,
+                "web_fetch_requests": 1,
+            },
         },
     )
 
@@ -70,12 +75,38 @@ def test_log_message_to_span_includes_stop_reason_and_stop_sequence():
             "completion_tokens": 7.0,
             "prompt_cached_tokens": 0.0,
             "prompt_cache_creation_tokens": 0.0,
+            "server_tool_use_web_search_requests": 2.0,
+            "server_tool_use_web_fetch_requests": 1.0,
             "tokens": 18.0,
             "time_to_first_token": 0.123,
         },
     )
 
 
+def test_extract_anthropic_usage_includes_server_tool_use_metrics_from_objects():
+    usage = SimpleNamespace(
+        input_tokens=11,
+        output_tokens=7,
+        cache_read_input_tokens=3,
+        cache_creation_input_tokens=2,
+        server_tool_use=SimpleNamespace(
+            web_search_requests=2,
+            web_fetch_requests=1,
+            code_execution_requests=4,
+        ),
+    )
+
+    assert extract_anthropic_usage(usage) == {
+        "prompt_tokens": 11.0,
+        "completion_tokens": 7.0,
+        "prompt_cached_tokens": 3.0,
+        "prompt_cache_creation_tokens": 2.0,
+        "server_tool_use_web_search_requests": 2.0,
+        "server_tool_use_web_fetch_requests": 1.0,
+        "server_tool_use_code_execution_requests": 4.0,
+    }
+
+
 @pytest.mark.vcr
 def test_anthropic_messages_create_stream_true(memory_logger):
     assert not memory_logger.pop()