feat(anthropic): mint OAuth2 client_credentials bearer for anthropic chat and /v1/messages

An anthropic/ model with the same oauth_* litellm_params now resolves a
client_credentials token: the chat path overrides api_key and sets
use_bearer_for_custom_base so the header builder emits Authorization:
Bearer for the gateway base, and the native /v1/messages path swaps
x-api-key for the minted bearer. The five oauth_* params are registered
in all_litellm_params so they are treated as LiteLLM-level config and
never swept into extra_body toward the provider

litellm/llms/anthropic/experimental_pass_through/messages/transformation.py

@@ -21,6 +21,7 @@ from litellm.types.llms.anthropic import (
 from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
 )
+from litellm.llms.openai_like.oauth_authenticator import resolve_client_credentials_token
 from litellm.types.llms.anthropic_tool_search import get_tool_search_beta_header
 from litellm.types.router import GenericLiteLLMParams
 
@@ -307,6 +308,11 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
         # Check for Anthropic OAuth token in Authorization header
         headers, api_key = optionally_handle_anthropic_oauth(headers=headers, api_key=api_key)
 
+        oauth_token: Final = resolve_client_credentials_token(litellm_params)
+        if oauth_token is not None:
+            headers.pop("x-api-key", None)
+            headers["authorization"] = f"Bearer {oauth_token}"
+
         if "x-api-key" not in headers and "authorization" not in headers:
             auth_header: Final = AnthropicModelInfo.get_auth_header(api_key)
             if auth_header is not None:

litellm/main.py

@@ -2755,6 +2755,18 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
     timeout: Final = ctx.timeout
 
     api_key = api_key or litellm.anthropic_key or litellm.api_key or os.environ.get("ANTHROPIC_API_KEY")
+
+    from litellm.llms.openai_like.oauth_authenticator import (
+        resolve_client_credentials_token,
+    )
+
+    oauth_api_key = resolve_client_credentials_token(cast("dict[str, object]", litellm_params))
+    if oauth_api_key is not None:
+        api_key = oauth_api_key
+    request_litellm_params: Final = (
+        {**litellm_params, "use_bearer_for_custom_base": True} if oauth_api_key is not None else litellm_params
+    )
+
     custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
     # call /messages
     # default route for all anthropic models
@@ -2783,7 +2795,7 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
         model_response=model_response,
         print_verbose=print_verbose,
         optional_params=optional_params,
-        litellm_params=litellm_params,
+        litellm_params=request_litellm_params,
         logger_fn=logger_fn,
         encoding=_get_encoding(),  # for calculating input/output tokens
         api_key=api_key,

litellm/types/utils.py

@@ -3594,6 +3594,11 @@ all_litellm_params = (
         "enable_tag_filtering",
         "enable_json_schema_validation",
         "use_xai_oauth",
+        "oauth_client_credentials",
+        "oauth_token_url",
+        "oauth_client_id",
+        "oauth_client_secret",
+        "oauth_scope",
         "auto_router_config_path",
         "auto_router_config",
         "auto_router_default_model",

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_messages_transformation.py

@@ -0,0 +1,99 @@
+"""OAuth2 client_credentials bearer injection on the native /v1/messages path.
+
+A deployment with oauth_client_credentials in litellm_params fronts the
+Anthropic Messages API with an OAuth2 gateway; the minted bearer must replace
+x-api-key auth.
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
+    AnthropicMessagesConfig,
+)
+from litellm.llms.openai_like.oauth_authenticator import _token_cache
+
+OAUTH_LITELLM_PARAMS = {
+    "oauth_client_credentials": True,
+    "oauth_token_url": "https://idp.test/token",
+    "oauth_client_id": "cid-messages",
+    "oauth_client_secret": "secret-messages",
+}
+
+
+@pytest.fixture(autouse=True)
+def _flush_token_cache():
+    _token_cache.flush_cache()
+    yield
+    _token_cache.flush_cache()
+
+
+def _token_client(access_token="tok-messages", expires_in=3600):
+    client = MagicMock()
+    response = MagicMock()
+    response.raise_for_status.return_value = None
+    response.json.return_value = {"access_token": access_token, "expires_in": expires_in}
+    client.post.return_value = response
+    return client
+
+
+class TestValidateAnthropicMessagesEnvironmentOAuth:
+    def test_flag_swaps_x_api_key_for_minted_bearer(self):
+        token_client = _token_client(access_token="tok-messages")
+        with patch(
+            "litellm.llms.openai_like.oauth_authenticator._get_httpx_client",
+            return_value=token_client,
+        ):
+            headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment(
+                headers={"x-api-key": "sk-config"},
+                model="claude-sonnet-5",
+                messages=[{"role": "user", "content": "hi"}],
+                optional_params={},
+                litellm_params=dict(OAUTH_LITELLM_PARAMS),
+                api_key="sk-config",
+                api_base="https://gateway.test",
+            )
+
+        token_client.post.assert_called_once()
+        assert "x-api-key" not in headers
+        assert headers["authorization"] == "Bearer tok-messages"
+
+    def test_flag_sets_bearer_when_no_prior_auth_header(self):
+        token_client = _token_client(access_token="tok-fresh")
+        with patch(
+            "litellm.llms.openai_like.oauth_authenticator._get_httpx_client",
+            return_value=token_client,
+        ):
+            headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment(
+                headers={},
+                model="claude-sonnet-5",
+                messages=[{"role": "user", "content": "hi"}],
+                optional_params={},
+                litellm_params=dict(OAUTH_LITELLM_PARAMS),
+                api_key="sk-config",
+                api_base="https://gateway.test",
+            )
+
+        assert headers["authorization"] == "Bearer tok-fresh"
+        assert "x-api-key" not in headers
+
+    def test_no_flag_keeps_x_api_key_and_fetches_nothing(self):
+        token_client = _token_client(access_token="should-not-be-used")
+        with patch(
+            "litellm.llms.openai_like.oauth_authenticator._get_httpx_client",
+            return_value=token_client,
+        ):
+            headers, _ = AnthropicMessagesConfig().validate_anthropic_messages_environment(
+                headers={"x-api-key": "sk-config"},
+                model="claude-sonnet-5",
+                messages=[{"role": "user", "content": "hi"}],
+                optional_params={},
+                litellm_params={"oauth_token_url": "https://idp.test/token"},
+                api_key="sk-config",
+                api_base="https://gateway.test",
+            )
+
+        token_client.post.assert_not_called()
+        assert headers["x-api-key"] == "sk-config"
+        assert "authorization" not in headers

tests/test_litellm/llms/openai_like/test_oauth_authenticator.py

@@ -390,3 +390,92 @@ class TestCompletionInjection:
 
         token_client.post.assert_not_called()
         assert chat.call_args.kwargs["api_key"] == "sk-static"
+
+
+class TestAnthropicCompletionInjection:
+    def test_flag_injects_minted_token_and_bearer_mode(self):
+        # End-to-end through litellm.completion on an anthropic/ model: the flag
+        # plus oauth_* in litellm_params mint a bearer that overrides the
+        # configured api_key, and use_bearer_for_custom_base is set so the
+        # anthropic header builder emits Authorization: Bearer for the custom
+        # gateway base instead of x-api-key.
+        import litellm
+        import litellm.main as main_mod
+        from litellm.types.utils import ModelResponse
+
+        token_client = _token_client(access_token="tok-anthropic")
+        fake = ModelResponse()
+        with (
+            patch(
+                "litellm.llms.openai_like.oauth_authenticator._get_httpx_client",
+                return_value=token_client,
+            ),
+            patch.object(
+                main_mod.anthropic_chat_completions, "completion", return_value=fake
+            ) as chat,
+        ):
+            litellm.completion(
+                model="anthropic/claude-sonnet-5",
+                messages=[{"role": "user", "content": "hi"}],
+                api_base="https://gateway.test",
+                api_key="sk-should-be-overridden",
+                oauth_client_credentials=True,
+                oauth_token_url="https://idp.test/token",
+                oauth_client_id="cid-anthropic",
+                oauth_client_secret="secret-anthropic",
+                oauth_scope="scope-anthropic",
+            )
+
+        token_client.post.assert_called_once()
+        assert chat.call_args.kwargs["api_key"] == "tok-anthropic"
+        assert chat.call_args.kwargs["litellm_params"]["use_bearer_for_custom_base"] is True
+
+    def test_no_flag_keeps_configured_api_key(self):
+        import litellm
+        import litellm.main as main_mod
+        from litellm.types.utils import ModelResponse
+
+        token_client = _token_client(access_token="should-not-be-used")
+        fake = ModelResponse()
+        with (
+            patch(
+                "litellm.llms.openai_like.oauth_authenticator._get_httpx_client",
+                return_value=token_client,
+            ),
+            patch.object(
+                main_mod.anthropic_chat_completions, "completion", return_value=fake
+            ) as chat,
+        ):
+            litellm.completion(
+                model="anthropic/claude-sonnet-5",
+                messages=[{"role": "user", "content": "hi"}],
+                api_base="https://gateway.test",
+                api_key="sk-static",
+                oauth_token_url="https://idp.test/token",
+                oauth_client_id="cid",
+                oauth_client_secret="secret",
+            )
+
+        token_client.post.assert_not_called()
+        assert chat.call_args.kwargs["api_key"] == "sk-static"
+        assert "use_bearer_for_custom_base" not in chat.call_args.kwargs["litellm_params"]
+
+
+class TestOAuthParamsExcludedFromProviderBody:
+    def test_oauth_fields_are_litellm_params_not_provider_params(self):
+        # Regression: without the all_litellm_params registration, unrecognized
+        # top-level kwargs are swept into extra_body and sent to the provider,
+        # leaking oauth_client_secret to the upstream.
+        from litellm.utils import get_non_default_completion_params
+
+        non_default = get_non_default_completion_params(
+            kwargs={
+                "oauth_client_credentials": True,
+                "oauth_token_url": "https://idp.test/token",
+                "oauth_client_id": "cid",
+                "oauth_client_secret": "secret",
+                "oauth_scope": "scope",
+                "some_provider_param": "stays",
+            }
+        )
+        assert non_default == {"some_provider_param": "stays"}