[LLM Translation + Coding tools] Added litellm claude code count tokens support (#13261)

* Added litellm claude code count tokens support

* fix mypy

* create helper

* Revert construct

* revert construct

* fix return

* Add reutrn none

* change to factory approach

* refactor to BaseModelInfo

* enum fix

litellm/llms/anthropic/common_utils.py

@@ -2,7 +2,7 @@
 This file contains common utils for anthropic calls.
 """
 
-from typing import Dict, List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import httpx
 
@@ -229,6 +229,60 @@ class AnthropicModelInfo(BaseLLMModelInfo):
             litellm_model_names.append(litellm_model_name)
         return litellm_model_names
 
+    def get_token_counter(self) -> Optional["AnthropicTokenCounter"]:
+        """
+        Factory method to create an Anthropic token counter.
+        
+        Returns:
+            AnthropicTokenCounter instance for this provider.
+        """
+        return AnthropicTokenCounter()
+
+
+class AnthropicTokenCounter:
+    """Token counter implementation for Anthropic provider."""
+    
+    def supports_provider(
+        self, 
+        deployment: Optional[Dict[str, Any]] = None,
+        from_endpoint: bool = False
+    ) -> bool:
+        if not from_endpoint:
+            return False
+            
+        if deployment is None:
+            return False
+            
+        full_model = deployment.get("litellm_params", {}).get("model", "")
+        is_anthropic_provider = full_model.startswith("anthropic/") or "anthropic" in full_model.lower()
+        
+        return is_anthropic_provider
+    
+    async def count_tokens(
+        self,
+        model_to_use: str,
+        messages: Optional[List[Dict[str, Any]]],
+        deployment: Optional[Dict[str, Any]] = None,
+        request_model: str = "",
+    ) -> Optional[Dict[str, Any]]:
+        from litellm.proxy.utils import count_tokens_with_anthropic_api
+        
+        result = await count_tokens_with_anthropic_api(
+            model_to_use=model_to_use,
+            messages=messages,
+            deployment=deployment,
+        )
+        
+        if result is not None:
+            return {
+                "total_tokens": result["total_tokens"],
+                "request_model": request_model,
+                "model_used": model_to_use,
+                "tokenizer_type": result["tokenizer_used"],
+            }
+        
+        return None
+
 
 def process_anthropic_headers(headers: Union[httpx.Headers, dict]) -> dict:
     openai_headers = {}

litellm/llms/base_llm/base_utils.py

@@ -70,6 +70,16 @@ class BaseLLMModelInfo(ABC):
         """
         pass
 
+    def get_token_counter(self):
+        """
+        Factory method to create a token counter for this provider.
+        
+        Returns:
+            Optional TokenCounterInterface implementation for this provider,
+            or None if token counting is not supported.
+        """
+        return None
+
 
 def _convert_tool_response_to_message(
     tool_calls: List[ChatCompletionToolCallChunk],

litellm/proxy/anthropic_endpoints/endpoints.py

@@ -200,3 +200,80 @@ async def anthropic_response(  # noqa: PLR0915
             param=getattr(e, "param", "None"),
             code=getattr(e, "status_code", 500),
         )
+
+
+@router.post(
+    "/v1/messages/count_tokens",
+    tags=["[beta] Anthropic Messages Token Counting"],
+    dependencies=[Depends(user_api_key_auth)],
+)
+async def count_tokens(
+    request: Request,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),  # Used for auth
+):
+    """
+    Count tokens for Anthropic Messages API format.
+    
+    This endpoint follows the Anthropic Messages API token counting specification.
+    It accepts the same parameters as the /v1/messages endpoint but returns
+    token counts instead of generating a response.
+    
+    Example usage:
+    ```
+    curl -X POST "http://localhost:4000/v1/messages/count_tokens?beta=true" \
+      -H "Content-Type: application/json" \
+      -H "Authorization: Bearer your-key" \
+      -d '{
+        "model": "claude-3-sonnet-20240229",
+        "messages": [{"role": "user", "content": "Hello Claude!"}]
+      }'
+    ```
+    
+    Returns: {"input_tokens": <number>}
+    """
+    from litellm.proxy.proxy_server import token_counter as internal_token_counter
+    
+    try:
+        request_data = await _read_request_body(request=request)
+        data: dict = {**request_data}
+        
+        # Extract required fields
+        model_name = data.get("model")
+        messages = data.get("messages", [])
+        
+        if not model_name:
+            raise HTTPException(
+                status_code=400,
+                detail={"error": "model parameter is required"}
+            )
+        
+        if not messages:
+            raise HTTPException(
+                status_code=400,
+                detail={"error": "messages parameter is required"}
+            )
+        
+        # Create TokenCountRequest for the internal endpoint
+        from litellm.proxy._types import TokenCountRequest
+        
+        token_request = TokenCountRequest(
+            model=model_name,
+            messages=messages
+        )
+        
+        # Call the internal token counter function with direct request flag set to False
+        token_response = await internal_token_counter(token_request, is_direct_request=False)
+        
+        # Convert the internal response to Anthropic API format
+        return {"input_tokens": token_response.total_tokens}
+        
+    except HTTPException:
+        raise
+    except Exception as e:
+        verbose_proxy_logger.exception(
+            "litellm.proxy.anthropic_endpoints.count_tokens(): Exception occurred - {}".format(str(e))
+        )
+        raise HTTPException(
+            status_code=500,
+            detail={"error": f"Internal server error: {str(e)}"}
+        )

litellm/proxy/proxy_server.py

@@ -679,6 +679,7 @@ app = FastAPI(
 )
 
 
+
 ### CUSTOM API DOCS [ENTERPRISE FEATURE] ###
 # Custom OpenAPI schema generator to include only selected routes
 from fastapi.routing import APIWebSocketRoute
@@ -5603,13 +5604,57 @@ async def run_thread(
 # async def get_available_routes(user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth)):
 
 
+def _get_provider_token_counter(deployment: dict, model_to_use: str):
+    """
+    Auto-route to the correct provider's token counter based on model/deployment.
+    Uses the existing get_provider_model_info infrastructure with switch-case pattern.
+    """
+    if deployment is None:
+        return None
+        
+    from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+    
+    full_model = deployment.get("litellm_params", {}).get("model", "")
+    
+    try:
+        # Use existing LiteLLM logic to determine provider
+        model, provider, dynamic_api_key, api_base = get_llm_provider(
+            model=full_model,
+            custom_llm_provider=deployment.get("litellm_params", {}).get("custom_llm_provider"),
+            api_base=deployment.get("litellm_params", {}).get("api_base"),
+            api_key=deployment.get("litellm_params", {}).get("api_key")
+        )
+        
+        # Switch case pattern using existing get_provider_model_info
+        from litellm.utils import ProviderConfigManager
+        from litellm.types.utils import LlmProviders
+        
+        # Convert string provider to LlmProviders enum
+        llm_provider_enum = LlmProviders(provider)
+        # Add more provider mappings as needed
+        
+        if llm_provider_enum:
+            provider_model_info = ProviderConfigManager.get_provider_model_info(model=full_model, provider=llm_provider_enum)
+            if provider_model_info is not None:
+                return provider_model_info.get_token_counter()
+            
+    except Exception:
+        # If provider detection fails, fall back to manual checks
+        if full_model.startswith("anthropic/") or "anthropic" in full_model.lower():
+            from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+            anthropic_model_info = AnthropicModelInfo()
+            return anthropic_model_info.get_token_counter()
+    
+    return None
+
+
 @router.post(
     "/utils/token_counter",
     tags=["llm utils"],
     dependencies=[Depends(user_api_key_auth)],
     response_model=TokenCountResponse,
 )
-async def token_counter(request: TokenCountRequest):
+async def token_counter(request: TokenCountRequest, is_direct_request: bool = True):
     """ """
     from litellm import token_counter
 
@@ -5634,7 +5679,7 @@ async def token_counter(request: TokenCountRequest):
                 break
     if deployment is not None:
         litellm_model_name = deployment.get("litellm_params", {}).get("model")
-        # remove the custom_llm_provider_prefix in the litellm_model_name
+        # remove the custom_llm_provider_prefix in the litellm_model_name  
         if "/" in litellm_model_name:
             litellm_model_name = litellm_model_name.split("/", 1)[1]
 
@@ -5642,6 +5687,29 @@ async def token_counter(request: TokenCountRequest):
         litellm_model_name or request.model
     )  # use litellm model name, if it's not avalable then fallback to request.model
 
+    # Try provider-specific token counting first - only for non-direct requests (from provider endpoints)
+    provider_counter = None
+    if deployment is not None and not is_direct_request:
+        # Auto-route to the correct provider based on model
+        provider_counter = _get_provider_token_counter(deployment, model_to_use)
+    
+    if provider_counter is not None and provider_counter.supports_provider(deployment=deployment, from_endpoint=not is_direct_request):
+        result = await provider_counter.count_tokens(
+            model_to_use=model_to_use,
+            messages=messages, # type: ignore
+            deployment=deployment,
+            request_model=request.model,
+        )
+        
+        if result is not None:
+            return TokenCountResponse(
+                total_tokens=result["total_tokens"],
+                request_model=result["request_model"],
+                model_used=result["model_used"],
+                tokenizer_type=result["tokenizer_type"],
+            )
+
+    # Default LiteLLM token counting
     custom_tokenizer: Optional[CustomHuggingfaceTokenizer] = None
     if model_info is not None:
         custom_tokenizer = cast(

litellm/proxy/utils.py

@@ -3802,7 +3802,6 @@ def is_valid_api_key(key: str) -> bool:
 def construct_database_url_from_env_vars() -> Optional[str]:
     """
     Construct a DATABASE_URL from individual environment variables.
-
     Returns:
         Optional[str]: The constructed DATABASE_URL or None if required variables are missing
     """
@@ -3829,9 +3828,65 @@ def construct_database_url_from_env_vars() -> Optional[str]:
             database_url = f"postgresql://{database_username_enc}@{database_host}/{database_name_enc}"
 
         return database_url
-
+    
     return None
 
+async def count_tokens_with_anthropic_api(
+    model_to_use: str,
+    messages: Optional[List[Dict[str, Any]]],
+    deployment: Optional[Dict[str, Any]] = None,
+) -> Optional[Dict[str, Any]]:
+    """
+    Helper function to count tokens using Anthropic API directly.
+
+    Args:
+        model_to_use: The model name to use for token counting
+        messages: The messages to count tokens for
+        deployment: Optional deployment configuration containing API key
+
+    Returns:
+        Optional dict with token count and tokenizer info, or None if failed
+    """
+    if not messages:
+        return None
+
+    try:
+        import anthropic
+        import os
+
+        # Get Anthropic API key from deployment config
+        anthropic_api_key = None
+        if deployment is not None:
+            anthropic_api_key = deployment.get("litellm_params", {}).get("api_key")
+
+        # Fallback to environment variable
+        if not anthropic_api_key:
+            anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
+
+        if anthropic_api_key and messages:
+            # Call Anthropic API directly for more accurate token counting
+            client = anthropic.Anthropic(api_key=anthropic_api_key)
+
+            # Call with explicit parameters to satisfy type checking
+            # Type ignore for now since messages come from generic dict input
+            response = client.beta.messages.count_tokens(
+                model=model_to_use,
+                messages=messages,  # type: ignore
+                betas=["token-counting-2024-11-01"]
+            )
+            total_tokens = response.input_tokens
+            tokenizer_used = "anthropic_api"
+
+            return {
+                "total_tokens": total_tokens,
+                "tokenizer_used": tokenizer_used,
+            }
+
+    except ImportError:
+        verbose_proxy_logger.warning("Anthropic library not available, falling back to LiteLLM tokenizer")
+    except Exception as e:
+        verbose_proxy_logger.warning(f"Error calling Anthropic API: {e}, falling back to LiteLLM tokenizer")
+    return None
 
 async def get_available_models_for_user(
     user_api_key_dict: "UserAPIKeyAuth",
@@ -4003,4 +4058,4 @@ def validate_model_access(
         raise HTTPException(
             status_code=404,
             detail="The model `{}` does not exist or is not accessible".format(model_id)
-        )
+        )

tests/proxy_unit_tests/test_proxy_token_counter.py

@@ -136,3 +136,427 @@ async def test_gpt_token_counting():
         response.tokenizer_type == "openai_tokenizer"
     )  # SHOULD use the OpenAI tokenizer
     assert response.request_model == "gpt-4"
+
+
+@pytest.mark.asyncio
+async def test_anthropic_messages_count_tokens_endpoint():
+    """
+    Test /v1/messages/count_tokens endpoint with Anthropic model
+    - Should return response in Anthropic format: {"input_tokens": <count>}
+    - Should work as wrapper around internal token_counter function
+    """
+    from litellm.proxy.anthropic_endpoints.endpoints import count_tokens
+    from fastapi import Request
+    from unittest.mock import AsyncMock, MagicMock
+    
+    # Mock request object
+    mock_request = MagicMock(spec=Request)
+    mock_request_data = {
+        "model": "claude-3-sonnet-20240229",
+        "messages": [{"role": "user", "content": "Hello Claude!"}]
+    }
+    
+    # Mock the _read_request_body function
+    async def mock_read_request_body(request):
+        return mock_request_data
+    
+    # Mock UserAPIKeyAuth
+    mock_user_api_key_dict = MagicMock()
+    
+    # Patch the _read_request_body function
+    import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints
+    original_read_request_body = anthropic_endpoints._read_request_body
+    anthropic_endpoints._read_request_body = mock_read_request_body
+    
+    # Mock the internal token_counter function to return a controlled response
+    async def mock_token_counter(request, is_direct_request=True):
+        assert is_direct_request == False, "Should be called with is_direct_request=False for Anthropic endpoint"
+        assert request.model == "claude-3-sonnet-20240229"
+        assert request.messages == [{"role": "user", "content": "Hello Claude!"}]
+        
+        from litellm.proxy._types import TokenCountResponse
+        return TokenCountResponse(
+            total_tokens=15,
+            request_model="claude-3-sonnet-20240229",
+            model_used="claude-3-sonnet-20240229",
+            tokenizer_type="openai_tokenizer"
+        )
+    
+    # Patch the imported token_counter function from proxy_server
+    import litellm.proxy.proxy_server as proxy_server
+    original_token_counter = proxy_server.token_counter
+    proxy_server.token_counter = mock_token_counter
+    
+    try:
+        # Call the endpoint
+        response = await count_tokens(mock_request, mock_user_api_key_dict)
+        
+        # Verify response format matches Anthropic spec
+        assert isinstance(response, dict)
+        assert "input_tokens" in response
+        assert response["input_tokens"] == 15
+        assert len(response) == 1  # Should only contain input_tokens
+        
+        print("✅ Anthropic endpoint test passed!")
+        
+    finally:
+        # Restore original functions
+        anthropic_endpoints._read_request_body = original_read_request_body
+        proxy_server.token_counter = original_token_counter
+
+
+@pytest.mark.asyncio 
+async def test_anthropic_messages_count_tokens_with_non_anthropic_model():
+    """
+    Test /v1/messages/count_tokens endpoint with non-Anthropic model (GPT-4)
+    - Should still work and return Anthropic format
+    - Should call internal token_counter with from_anthropic_endpoint=True
+    """
+    from litellm.proxy.anthropic_endpoints.endpoints import count_tokens
+    from fastapi import Request
+    from unittest.mock import AsyncMock, MagicMock
+    
+    # Mock request object
+    mock_request = MagicMock(spec=Request)
+    mock_request_data = {
+        "model": "gpt-4",
+        "messages": [{"role": "user", "content": "Hello GPT!"}]
+    }
+    
+    # Mock the _read_request_body function
+    async def mock_read_request_body(request):
+        return mock_request_data
+    
+    # Mock UserAPIKeyAuth
+    mock_user_api_key_dict = MagicMock()
+    
+    # Patch the _read_request_body function
+    import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints
+    original_read_request_body = anthropic_endpoints._read_request_body
+    anthropic_endpoints._read_request_body = mock_read_request_body
+    
+    # Mock the internal token_counter function to return a controlled response
+    async def mock_token_counter(request, is_direct_request=True):
+        assert is_direct_request == False, "Should be called with is_direct_request=False for Anthropic endpoint"
+        assert request.model == "gpt-4"
+        assert request.messages == [{"role": "user", "content": "Hello GPT!"}]
+        
+        from litellm.proxy._types import TokenCountResponse
+        return TokenCountResponse(
+            total_tokens=12,
+            request_model="gpt-4", 
+            model_used="gpt-4",
+            tokenizer_type="openai_tokenizer"
+        )
+    
+    # Patch the imported token_counter function from proxy_server
+    import litellm.proxy.proxy_server as proxy_server
+    original_token_counter = proxy_server.token_counter
+    proxy_server.token_counter = mock_token_counter
+    
+    try:
+        # Call the endpoint
+        response = await count_tokens(mock_request, mock_user_api_key_dict)
+        
+        # Verify response format matches Anthropic spec
+        assert isinstance(response, dict)
+        assert "input_tokens" in response
+        assert response["input_tokens"] == 12
+        assert len(response) == 1  # Should only contain input_tokens
+        
+        print("✅ Non-Anthropic model test passed!")
+        
+    finally:
+        # Restore original functions
+        anthropic_endpoints._read_request_body = original_read_request_body
+        proxy_server.token_counter = original_token_counter
+
+
+@pytest.mark.asyncio
+async def test_internal_token_counter_anthropic_provider_detection():
+    """
+    Test that the internal token_counter correctly detects Anthropic providers
+    and handles the from_anthropic_endpoint flag appropriately
+    """
+    
+    # Test with Anthropic provider
+    llm_router = Router(
+        model_list=[
+            {
+                "model_name": "claude-test",
+                "litellm_params": {
+                    "model": "anthropic/claude-3-sonnet-20240229",
+                    "api_key": "test-key"
+                },
+            }
+        ]
+    )
+    
+    setattr(litellm.proxy.proxy_server, "llm_router", llm_router)
+    
+    # Test with is_direct_request=False (simulating call from Anthropic endpoint)
+    response = await token_counter(
+        request=TokenCountRequest(
+            model="claude-test",
+            messages=[{"role": "user", "content": "hello"}],
+        ),
+        is_direct_request=False
+    )
+    
+    print("Anthropic provider test response:", response)
+    
+    # Verify response structure
+    assert response.request_model == "claude-test"
+    assert response.model_used == "claude-3-sonnet-20240229"
+    assert response.total_tokens > 0
+    
+    # Test with non-Anthropic provider
+    llm_router = Router(
+        model_list=[
+            {
+                "model_name": "gpt-test",
+                "litellm_params": {
+                    "model": "gpt-4",
+                },
+            }
+        ]
+    )
+    
+    setattr(litellm.proxy.proxy_server, "llm_router", llm_router)
+    
+    # Test with is_direct_request=False but non-Anthropic provider
+    response = await token_counter(
+        request=TokenCountRequest(
+            model="gpt-test",
+            messages=[{"role": "user", "content": "hello"}],
+        ),
+        is_direct_request=False
+    )
+    
+    print("Non-Anthropic provider test response:", response)
+    
+    # Verify response structure 
+    assert response.request_model == "gpt-test"
+    assert response.model_used == "gpt-4"
+    assert response.total_tokens > 0
+    assert response.tokenizer_type == "openai_tokenizer"  # Should use LiteLLM tokenizer
+
+
+@pytest.mark.asyncio
+async def test_anthropic_endpoint_error_handling():
+    """
+    Test error handling in the /v1/messages/count_tokens endpoint
+    """
+    from litellm.proxy.anthropic_endpoints.endpoints import count_tokens
+    from fastapi import Request, HTTPException
+    from unittest.mock import MagicMock
+    
+    # Mock request object
+    mock_request = MagicMock(spec=Request)
+    mock_user_api_key_dict = MagicMock()
+    
+    # Test missing model parameter
+    mock_request_data = {
+        "messages": [{"role": "user", "content": "Hello!"}]
+        # Missing "model" key
+    }
+    
+    async def mock_read_request_body(request):
+        return mock_request_data
+    
+    import litellm.proxy.anthropic_endpoints.endpoints as anthropic_endpoints
+    original_read_request_body = anthropic_endpoints._read_request_body
+    anthropic_endpoints._read_request_body = mock_read_request_body
+    
+    try:
+        # Should raise HTTPException for missing model
+        with pytest.raises(HTTPException) as exc_info:
+            await count_tokens(mock_request, mock_user_api_key_dict)
+        
+        assert exc_info.value.status_code == 400
+        assert "model parameter is required" in str(exc_info.value.detail)
+        
+        print("✅ Error handling test passed!")
+        
+    finally:
+        anthropic_endpoints._read_request_body = original_read_request_body
+
+
+@pytest.mark.asyncio
+async def test_factory_anthropic_endpoint_calls_anthropic_counter():
+    """Test that /v1/messages/count_tokens with Anthropic model uses Anthropic counter."""
+    from unittest.mock import patch
+    from fastapi.testclient import TestClient
+    from litellm.proxy.proxy_server import app
+    
+    # Mock the anthropic token counting function
+    with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count:
+        mock_anthropic_count.return_value = {
+            "total_tokens": 42,
+            "tokenizer_used": "anthropic"
+        }
+        
+        # Mock router to return Anthropic deployment
+        with patch('litellm.proxy.proxy_server.llm_router') as mock_router:
+            mock_router.model_list = [{
+                "model_name": "claude-3-5-sonnet",
+                "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"},
+                "model_info": {}
+            }]
+            
+            client = TestClient(app)
+            
+            response = client.post(
+                "/v1/messages/count_tokens",
+                json={
+                    "model": "claude-3-5-sonnet",
+                    "messages": [{"role": "user", "content": "Hello"}]
+                },
+                headers={"Authorization": "Bearer test-key"}
+            )
+            
+            assert response.status_code == 200
+            data = response.json()
+            assert data["input_tokens"] == 42
+            
+            # Verify that Anthropic API was called
+            mock_anthropic_count.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_factory_gpt4_endpoint_does_not_call_anthropic_counter():
+    """Test that /v1/messages/count_tokens with GPT-4 does NOT use Anthropic counter."""
+    from unittest.mock import patch
+    from fastapi.testclient import TestClient
+    from litellm.proxy.proxy_server import app
+    
+    # Mock the anthropic token counting function
+    with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count:
+        # Mock litellm token counter
+        with patch('litellm.token_counter') as mock_litellm_counter:
+            mock_litellm_counter.return_value = 50
+            
+            # Mock router to return GPT-4 deployment
+            with patch('litellm.proxy.proxy_server.llm_router') as mock_router:
+                mock_router.model_list = [{
+                    "model_name": "gpt-4",
+                    "litellm_params": {"model": "openai/gpt-4"},
+                    "model_info": {}
+                }]
+                
+                client = TestClient(app)
+                
+                response = client.post(
+                    "/v1/messages/count_tokens",
+                    json={
+                        "model": "gpt-4",
+                        "messages": [{"role": "user", "content": "Hello"}]
+                    },
+                    headers={"Authorization": "Bearer test-key"}
+                )
+                
+                assert response.status_code == 200
+                data = response.json()
+                assert data["input_tokens"] == 50
+                
+                # Verify that Anthropic API was NOT called
+                mock_anthropic_count.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_factory_normal_token_counter_endpoint_does_not_call_anthropic():
+    """Test that /utils/token_counter does NOT use Anthropic counter even with Anthropic model."""
+    from unittest.mock import patch
+    from fastapi.testclient import TestClient
+    from litellm.proxy.proxy_server import app
+    
+    # Mock the anthropic token counting function
+    with patch('litellm.proxy.utils.count_tokens_with_anthropic_api') as mock_anthropic_count:
+        # Mock litellm token counter
+        with patch('litellm.token_counter') as mock_litellm_counter:
+            mock_litellm_counter.return_value = 35
+            
+            # Mock router to return Anthropic deployment
+            with patch('litellm.proxy.proxy_server.llm_router') as mock_router:
+                mock_router.model_list = [{
+                    "model_name": "claude-3-5-sonnet",
+                    "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"},
+                    "model_info": {}
+                }]
+                
+                client = TestClient(app)
+                
+                response = client.post(
+                    "/utils/token_counter",
+                    json={
+                        "model": "claude-3-5-sonnet",
+                        "messages": [{"role": "user", "content": "Hello"}]
+                    },
+                    headers={"Authorization": "Bearer test-key"}
+                )
+                
+                assert response.status_code == 200
+                data = response.json()
+                assert data["total_tokens"] == 35
+                
+                # Verify that Anthropic API was NOT called (since is_direct_request=True)
+                mock_anthropic_count.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_factory_registration():
+    """Test that the new factory pattern correctly provides counters."""
+    from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+    
+    # Test Anthropic ModelInfo provides token counter
+    anthropic_model_info = AnthropicModelInfo()
+    counter = anthropic_model_info.get_token_counter()
+    assert counter is not None
+    
+    # Create test deployments
+    anthropic_deployment = {
+        "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"}
+    }
+    
+    non_anthropic_deployment = {
+        "litellm_params": {"model": "openai/gpt-4"}
+    }
+    
+    # Test Anthropic counter supports provider
+    assert counter.supports_provider(anthropic_deployment, from_endpoint=True)
+    assert not counter.supports_provider(anthropic_deployment, from_endpoint=False)
+    
+    # Test non-Anthropic provider
+    assert not counter.supports_provider(non_anthropic_deployment, from_endpoint=True)
+    assert not counter.supports_provider(non_anthropic_deployment, from_endpoint=False)
+    
+    # Test None deployment
+    assert not counter.supports_provider(None, from_endpoint=True)
+    assert not counter.supports_provider(None, from_endpoint=False)
+
+
+@pytest.mark.asyncio 
+async def test_factory_anthropic_counter_supports_provider():
+    """Test AnthropicTokenCounter provider detection logic."""
+    from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+    
+    anthropic_model_info = AnthropicModelInfo()
+    counter = anthropic_model_info.get_token_counter()
+    
+    # Test Anthropic provider detection
+    anthropic_deployment = {
+        "litellm_params": {"model": "anthropic/claude-3-5-sonnet-20241022"}
+    }
+    assert counter.supports_provider(anthropic_deployment, from_endpoint=True)
+    assert not counter.supports_provider(anthropic_deployment, from_endpoint=False)
+    
+    # Test non-Anthropic provider
+    openai_deployment = {
+        "litellm_params": {"model": "openai/gpt-4"}
+    }
+    assert not counter.supports_provider(openai_deployment, from_endpoint=True)
+    assert not counter.supports_provider(openai_deployment, from_endpoint=False)
+    
+    # Test None deployment
+    assert not counter.supports_provider(None, from_endpoint=True)
+    assert not counter.supports_provider(None, from_endpoint=False)