fix(anthropic): price and surface response service_tier in cost tracking (#30558)

litellm/cost_calculator.py

@@ -94,6 +94,7 @@ from litellm.types.utils import (
     LlmProviders,
     LlmProvidersSet,
     ModelInfo,
+    ServiceTier,
     StandardBuiltInToolsParams,
     TranscriptionUsageDurationObject,
     TranscriptionUsageTokensObject,
@@ -614,7 +615,9 @@ def cost_per_token(
                 service_tier=service_tier,
             )
     elif custom_llm_provider == "anthropic":
-        return anthropic_cost_per_token(model=model, usage=usage_block)
+        return anthropic_cost_per_token(
+            model=model, usage=usage_block, service_tier=service_tier
+        )
     elif custom_llm_provider == "bedrock":
         return bedrock_cost_per_token(
             model=model, usage=usage_block, service_tier=service_tier
@@ -1224,6 +1227,12 @@ def completion_cost(
         if service_tier is None and optional_params is not None:
             service_tier = optional_params.get("service_tier")
 
+        # "auto" is a routing preference, not a billable tier: the provider picks
+        # the tier and reports the one actually served on the response/usage, so
+        # defer to that instead of pricing the request-level "auto" as standard
+        if service_tier is not None and service_tier.lower() == ServiceTier.AUTO.value:
+            service_tier = None
+
         # Extract service_tier from completion_response if not provided
         if service_tier is None and completion_response is not None:
             if isinstance(completion_response, BaseModel):

litellm/llms/anthropic/chat/transformation.py

@@ -2213,6 +2213,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
         inference_geo: Optional[str] = None
         if "inference_geo" in _usage and _usage["inference_geo"] is not None:
             inference_geo = _usage["inference_geo"]
+        service_tier = cast(
+            str | None,
+            _usage.get("service_tier"),  # any-ok: untyped usage dict
+        )
 
         iterations: Optional[List[Any]] = _usage.get("iterations")
         if iterations:
@@ -2324,6 +2328,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             ),
             inference_geo=inference_geo,
             speed=speed,
+            service_tier=service_tier,
         )
         return usage
 

litellm/llms/anthropic/cost_calculation.py

@@ -18,7 +18,9 @@ if TYPE_CHECKING:
 import litellm
 
 
-def _compute_cache_only_cost(model_info: "ModelInfo", usage: "Usage") -> float:
+def _compute_cache_only_cost(
+    model_info: "ModelInfo", usage: "Usage", service_tier: str | None = None
+) -> float:
     """
     Return only the cache-related portion of the prompt cost (cache read + cache write).
 
@@ -36,7 +38,9 @@ def _compute_cache_only_cost(model_info: "ModelInfo", usage: "Usage") -> float:
         cache_creation_cost,
         cache_creation_cost_above_1hr,
         cache_read_cost,
-    ) = _get_token_base_cost(model_info=model_info, usage=usage)
+    ) = _get_token_base_cost(
+        model_info=model_info, usage=usage, service_tier=service_tier
+    )
 
     cache_cost = float(prompt_tokens_details["cache_hit_tokens"]) * cache_read_cost
 
@@ -56,19 +60,26 @@ def _compute_cache_only_cost(model_info: "ModelInfo", usage: "Usage") -> float:
     return cache_cost
 
 
-def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
+def cost_per_token(
+    model: str, usage: "Usage", service_tier: str | None = None
+) -> Tuple[float, float]:
     """
     Calculates the cost per token for a given model, prompt tokens, and completion tokens.
 
     Input:
         - model: str, the model name without provider prefix
         - usage: LiteLLM Usage block, containing anthropic caching information
+        - service_tier: the service tier the request was served at (e.g. "priority"),
+          read from the Anthropic response usage and used to select tier-specific pricing
 
     Returns:
         Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
     """
     prompt_cost, completion_cost = generic_cost_per_token(
-        model=model, usage=usage, custom_llm_provider="anthropic"
+        model=model,
+        usage=usage,
+        custom_llm_provider="anthropic",
+        service_tier=service_tier,
     )
 
     # Apply provider_specific_entry multipliers for geo/speed routing
@@ -89,7 +100,9 @@ def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
             multiplier *= provider_specific_entry.get("fast", 1.0)
 
         if multiplier != 1.0:
-            cache_cost = _compute_cache_only_cost(model_info=model_info, usage=usage)
+            cache_cost = _compute_cache_only_cost(
+                model_info=model_info, usage=usage, service_tier=service_tier
+            )
             prompt_cost = (prompt_cost - cache_cost) * multiplier + cache_cost
             completion_cost *= multiplier
     except Exception:

litellm/types/utils.py

@@ -3669,6 +3669,7 @@ class SpecialEnums(Enum):
 class ServiceTier(Enum):
     """Enum for service tier types used in cost calculations."""
 
+    AUTO = "auto"
     FLEX = "flex"
     PRIORITY = "priority"
 

tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py

@@ -3702,6 +3702,39 @@ def test_fast_mode_with_inference_geo():
         assert abs(completion_cost - base_completion * expected_multiplier) < 1e-10
 
 
+def test_calculate_usage_captures_service_tier():
+    """
+    Anthropic returns the assigned service tier on the response usage object
+    (e.g. ``"priority"``). It must be surfaced on the Usage object so it is
+    visible in logs and used to select tier-specific pricing.
+    """
+    config = AnthropicConfig()
+
+    usage_object = {
+        "input_tokens": 410,
+        "cache_creation_input_tokens": 0,
+        "cache_read_input_tokens": 0,
+        "output_tokens": 585,
+        "service_tier": "priority",
+    }
+
+    usage = config.calculate_usage(usage_object=usage_object, reasoning_content=None)
+
+    assert usage.service_tier == "priority"
+
+
+def test_calculate_usage_service_tier_defaults_to_none():
+    """A response without a service tier must not invent one."""
+    config = AnthropicConfig()
+
+    usage = config.calculate_usage(
+        usage_object={"input_tokens": 10, "output_tokens": 5},
+        reasoning_content=None,
+    )
+
+    assert usage.service_tier is None
+
+
 def test_fast_mode_parameter_in_supported_params():
     """
     Test that 'speed' is in the list of supported OpenAI params.

tests/test_litellm/proxy/spend_tracking/test_spend_management_endpoints.py

@@ -359,6 +359,7 @@ ignored_keys = [
     "metadata.additional_usage_values.cache_read_input_tokens",
     "metadata.additional_usage_values.inference_geo",
     "metadata.additional_usage_values.speed",
+    "metadata.additional_usage_values.service_tier",
     "metadata.additional_usage_values.iterations",
     "metadata.litellm_overhead_time_ms",
     "metadata.cost_breakdown",

tests/test_litellm/test_cost_calculator.py

@@ -2127,6 +2127,169 @@ def test_completion_cost_service_tier_for_bedrock():
     assert priority_cost > default_cost > flex_cost > 0
 
 
+def test_completion_cost_service_tier_for_anthropic():
+    """
+    Anthropic priority-tier requests must be priced at the priority rate.
+
+    Regression for LIT-3771: the Anthropic cost route dropped ``service_tier``,
+    so priority requests (whose tier is reported on the response usage) were
+    always billed at the standard rate. The tier is captured by the
+    transformation and must flow through to ``generic_cost_per_token``.
+    """
+    from litellm import completion_cost
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
+    litellm.model_cost = litellm.get_model_cost_map(url="")
+
+    model = "claude-test-service-tier-cost-model"
+    litellm.register_model(
+        model_cost={
+            model: {
+                "input_cost_per_token": 3e-6,
+                "output_cost_per_token": 15e-6,
+                "input_cost_per_token_priority": 6e-6,
+                "output_cost_per_token_priority": 30e-6,
+                "litellm_provider": "anthropic",
+                "max_tokens": 8192,
+            }
+        }
+    )
+
+    def _cost_for_tier(service_tier):
+        usage = AnthropicConfig().calculate_usage(
+            usage_object={
+                "input_tokens": 1000,
+                "output_tokens": 500,
+                "service_tier": service_tier,
+            },
+            reasoning_content=None,
+        )
+        response = ModelResponse(usage=usage, model=model)
+        return completion_cost(
+            completion_response=response,
+            model=model,
+            custom_llm_provider="anthropic",
+        )
+
+    standard_cost = _cost_for_tier("standard")
+    priority_cost = _cost_for_tier("priority")
+
+    expected_standard = 1000 * 3e-6 + 500 * 15e-6
+    assert standard_cost == pytest.approx(expected_standard)
+    # priority rates are exactly 2x standard for both input and output
+    assert priority_cost == pytest.approx(2 * standard_cost)
+
+
+def test_completion_cost_anthropic_auto_tier_uses_served_priority_rate():
+    """
+    Proxy billing path regression for LIT-3771.
+
+    Priority is opted into with ``service_tier="auto"``; Anthropic then serves
+    "priority" and reports it on the response usage. The proxy forwards the
+    request-level "auto" into ``completion_cost`` (via ``_response_cost_calculator``),
+    and that preference must not shadow the served tier, otherwise priority
+    requests are silently billed at the standard rate.
+    """
+    from litellm import completion_cost
+    from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+    os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
+    litellm.model_cost = litellm.get_model_cost_map(url="")
+
+    model = "claude-test-auto-tier-cost-model"
+    litellm.register_model(
+        model_cost={
+            model: {
+                "input_cost_per_token": 3e-6,
+                "output_cost_per_token": 15e-6,
+                "input_cost_per_token_priority": 6e-6,
+                "output_cost_per_token_priority": 30e-6,
+                "litellm_provider": "anthropic",
+                "max_tokens": 8192,
+            }
+        }
+    )
+
+    usage = AnthropicConfig().calculate_usage(
+        usage_object={
+            "input_tokens": 1000,
+            "output_tokens": 500,
+            "service_tier": "priority",
+        },
+        reasoning_content=None,
+    )
+    response = ModelResponse(usage=usage, model=model)
+
+    cost = completion_cost(
+        completion_response=response,
+        model=model,
+        custom_llm_provider="anthropic",
+        service_tier="auto",
+        optional_params={"service_tier": "auto"},
+    )
+
+    expected_priority = 1000 * 6e-6 + 500 * 30e-6
+    assert cost == pytest.approx(expected_priority)
+
+
+def test_anthropic_cost_per_token_prices_cache_at_served_tier_with_multiplier():
+    """
+    Regression for the cache/tier interaction in the Anthropic geo/speed path.
+
+    When a request is served at "priority" and also carries a geo/speed
+    multiplier (here ``speed="fast"``), the cache portion is held out of the
+    multiplier so it is not scaled. That held-out cache cost must use the
+    served tier's cache rate; pricing it at the standard rate while the cache
+    embedded in ``prompt_cost`` is priced at the priority rate leaves a
+    ``(cache_priority - cache_standard)(multiplier - 1)`` billing error.
+    """
+    from litellm.llms.anthropic.cost_calculation import (
+        cost_per_token as anthropic_cost_per_token,
+    )
+    from litellm.types.utils import PromptTokensDetailsWrapper, Usage
+
+    os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
+    litellm.model_cost = litellm.get_model_cost_map(url="")
+
+    model = "claude-test-priority-cache-fast-model"
+    litellm.register_model(
+        model_cost={
+            model: {
+                "input_cost_per_token": 3e-6,
+                "output_cost_per_token": 15e-6,
+                "cache_read_input_token_cost": 0.3e-6,
+                "input_cost_per_token_priority": 6e-6,
+                "output_cost_per_token_priority": 30e-6,
+                "cache_read_input_token_cost_priority": 0.6e-6,
+                "litellm_provider": "anthropic",
+                "max_tokens": 8192,
+                "provider_specific_entry": {"fast": 2.0},
+            }
+        }
+    )
+
+    usage = Usage(
+        prompt_tokens=1000,
+        completion_tokens=500,
+        total_tokens=1500,
+        prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=200),
+    )
+    usage.speed = "fast"
+
+    prompt_cost, completion_cost = anthropic_cost_per_token(
+        model=model, usage=usage, service_tier="priority"
+    )
+
+    # non-cache input priced at the priority rate and scaled by the fast
+    # multiplier; the 200 cache-hit tokens priced at the priority cache rate
+    # and held out of the multiplier
+    expected_prompt = (1000 - 200) * 6e-6 * 2 + 200 * 0.6e-6
+    expected_completion = 500 * 30e-6 * 2
+    assert prompt_cost == pytest.approx(expected_prompt)
+    assert completion_cost == pytest.approx(expected_completion)
+
+
 def test_gemini_cache_tokens_details_no_negative_values():
     """
     Test for Issue #18750: Negative text_tokens with Gemini caching