fix(anthropic): token counter honors ANTHROPIC_AUTH_TOKEN and degrades on a failed federation mint

This commit is contained in:
mateo-berri 2026-08-30 12:26:01 -07:00
parent 068c1d1885
commit c6f585e0ae
2 changed files with 84 additions and 17 deletions

View file

@ -6,6 +6,7 @@ import os
from typing import Any, Final
from litellm._logging import verbose_logger
from litellm.exceptions import AuthenticationError
from litellm.llms.anthropic.count_tokens.handler import AnthropicCountTokensHandler
from litellm.llms.base_llm.base_utils import BaseTokenCounter
from litellm.types.utils import LlmProviders, TokenCountResponse
@ -46,34 +47,32 @@ class AnthropicTokenCounter(BaseTokenCounter):
Returns:
TokenCountResponse with token count, or None if counting fails
"""
from litellm.llms.anthropic.common_utils import AnthropicError
from litellm.llms.anthropic.common_utils import AnthropicError, AnthropicModelInfo
from litellm.llms.anthropic.wif import aget_anthropic_wif_token
if not messages:
return None
deployment = deployment or {}
litellm_params: Final = deployment.get("litellm_params", {})
from litellm.llms.anthropic.wif import aget_anthropic_wif_token
static_key: Final = litellm_params.get("api_key") or os.getenv("ANTHROPIC_API_KEY")
# A federated deployment holds no static key by design. Without a minted one this returns
# None and the caller silently falls back to the local tokenizer, so a workload identity
# deployment would never reach Anthropic's authoritative count. The minted token is an
# sk-ant-oat, which get_required_headers already sends as a Bearer rather than x-api-key.
api_base: Final = litellm_params.get("api_base")
api_key: Final = static_key or await aget_anthropic_wif_token(litellm_params, api_base, model_to_use)
if not api_key:
verbose_logger.warning("No Anthropic credential found for token counting")
return None
static_key: Final = litellm_params.get("api_key") or os.getenv("ANTHROPIC_API_KEY")
auth_token_configured: Final = AnthropicModelInfo.get_auth_token() is not None
try:
api_key: Final = (
static_key
if static_key or auth_token_configured
else await aget_anthropic_wif_token(litellm_params, api_base, model_to_use)
)
if not api_key:
verbose_logger.warning("No Anthropic credential found for token counting")
return None
result: Final = await anthropic_count_tokens_handler.handle_count_tokens_request(
model=model_to_use,
messages=messages,
api_key=api_key,
# The token is minted for this base, so the count has to be asked of the same host.
api_base=api_base,
tools=tools,
system=system,
@ -87,8 +86,8 @@ class AnthropicTokenCounter(BaseTokenCounter):
tokenizer_type="anthropic_api",
original_response=result,
)
except AnthropicError as e:
verbose_logger.warning("Anthropic CountTokens API error: status=%s, message=%s", e.status_code, e.message)
except (AnthropicError, AuthenticationError) as e:
verbose_logger.warning("Anthropic CountTokens error: status=%s, message=%s", e.status_code, e.message)
return TokenCountResponse(
total_tokens=0,
request_model=request_model,

View file

@ -130,3 +130,71 @@ class TestCountTokensUsesWorkloadIdentity:
assert result is not None
assert result.total_tokens == 42
assert seen["api_key"] == minted
@pytest.mark.asyncio
async def test_an_auth_token_deployment_never_mints(self, monkeypatch):
from litellm.llms.anthropic.count_tokens import token_counter as token_counter_module
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", "bearer-token-for-testing")
mint_calls: list[str] = []
async def fake_mint(_params, _api_base, model):
mint_calls.append(model)
return "sk-ant-oat01-should-not-be-minted"
monkeypatch.setattr("litellm.llms.anthropic.wif.aget_anthropic_wif_token", fake_mint, raising=False)
result = await token_counter_module.AnthropicTokenCounter().count_tokens(
model_to_use="claude-sonnet-4-5",
messages=[{"role": "user", "content": "hi"}],
contents=None,
deployment={
"litellm_params": {
"model": "anthropic/claude-sonnet-4-5",
"anthropic_federation_rule_id": "fdrl_x",
"anthropic_organization_id": "org-x",
}
},
request_model="claude-sonnet-4-5",
)
assert result is None
assert mint_calls == []
@pytest.mark.asyncio
async def test_a_failed_mint_degrades_like_an_anthropic_error(self, monkeypatch):
import litellm
from litellm.llms.anthropic.count_tokens import token_counter as token_counter_module
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("ANTHROPIC_AUTH_TOKEN", raising=False)
async def failing_mint(_params, _api_base, model):
raise litellm.AuthenticationError(
message="federation_rule_id is not a well-formed fdrl_ tagged ID",
llm_provider="anthropic",
model=model,
)
monkeypatch.setattr("litellm.llms.anthropic.wif.aget_anthropic_wif_token", failing_mint, raising=False)
result = await token_counter_module.AnthropicTokenCounter().count_tokens(
model_to_use="claude-sonnet-4-5",
messages=[{"role": "user", "content": "hi"}],
contents=None,
deployment={
"litellm_params": {
"model": "anthropic/claude-sonnet-4-5",
"anthropic_federation_rule_id": "not-a-rule",
"anthropic_organization_id": "org-x",
}
},
request_model="claude-sonnet-4-5",
)
assert result is not None
assert result.error is True
assert result.status_code == 401
assert result.total_tokens == 0
assert "fdrl_" in (result.error_message or "")