Add Claude Max server-managed OAuth provider

docs/my-website/docs/providers/server_managed_oauth_subscriptions.md

@@ -0,0 +1,60 @@
+# Server-managed OAuth subscription providers
+
+## Overview
+
+| Property | Details |
+|-------|-------|
+| Description | Server-side refresh-token adapters for subscription-backed inference providers. Clients call LiteLLM normally and LiteLLM refreshes provider OAuth tokens on the server. |
+| Provider Routes on LiteLLM | `claude_max/` |
+| Supported Operations | `/chat/completions`, including streaming when the upstream provider supports it |
+
+These providers are intended for private gateways where the LiteLLM server owns the subscription login and clients authenticate only to LiteLLM. Do not expose the server token files to client machines.
+
+## Token storage
+
+Each adapter reads a JSON token file and refreshes it when the access token is near expiry. You can mount files into the default directory or point each provider at an explicit file.
+
+```bash showLineNumbers title="Environment Variables"
+export SERVER_OAUTH_TOKEN_DIR="$HOME/.config/litellm/server_oauth"
+export CLAUDE_MAX_OAUTH_FILE="$SERVER_OAUTH_TOKEN_DIR/claude-max.json"
+```
+
+For secret managers that inject environment variables instead of files, set `CLAUDE_MAX_OAUTH_JSON_B64` to a base64-encoded JSON token payload. LiteLLM restores it to the provider token file with owner-only permissions on first use.
+
+## Claude Max
+
+The `claude_max/` provider uses Anthropic-compatible chat completions with a server-managed Claude OAuth refresh token.
+
+```json showLineNumbers title="claude-max.json"
+{
+  "access_token": "sk-ant-oat...",
+  "refresh_token": "...",
+  "expires_at": 1790000000
+}
+```
+
+```yaml showLineNumbers title="config.yaml"
+model_list:
+  - model_name: frontier
+    litellm_params:
+      model: claude_max/claude-opus-4-8
+  - model_name: opus-4-8
+    litellm_params:
+      model: claude_max/claude-opus-4-8
+```
+
+Optional overrides:
+
+```bash showLineNumbers title="Claude Max overrides"
+export CLAUDE_MAX_TOKEN_URL="https://platform.claude.com/v1/oauth/token"
+export CLAUDE_MAX_CLIENT_ID="9d1c250a-e61b-44d9-88ed-5944d1962f5e"
+export CLAUDE_MAX_API_BASE="https://api.anthropic.com/v1/messages"
+```
+
+## Antigravity 2.0
+
+Antigravity 2.0 is not implemented by this provider page yet. The deprecated Cloud Code Assist / Gemini CLI `cloudcode-pa.googleapis.com/v1internal` gateway is intentionally not documented here because it is not the Antigravity 2.0 SDK/CLI runtime contract.
+
+## LiteLLM virtual keys
+
+Put these providers behind normal LiteLLM model groups and virtual keys. Client applications should send only the LiteLLM virtual key and the model alias, such as `frontier` or `opus-4-8`; provider OAuth secrets remain server-side.

litellm/__init__.py

@@ -1904,6 +1904,7 @@ if TYPE_CHECKING:
         GithubCopilotEmbeddingConfig as GithubCopilotEmbeddingConfig,
     )
     from .llms.chatgpt.chat.transformation import ChatGPTConfig as ChatGPTConfig
+    from .llms.claude_max.chat.transformation import ClaudeMaxConfig as ClaudeMaxConfig
     from .llms.chatgpt.responses.transformation import (
         ChatGPTResponsesAPIConfig as ChatGPTResponsesAPIConfig,
     )

litellm/_lazy_imports_registry.py

@@ -131,6 +131,7 @@ LLM_CONFIG_NAMES = (
     "OpenrouterConfig",
     "DataRobotConfig",
     "AnthropicConfig",
+    "ClaudeMaxConfig",
     "BedrockClaudePlatformConfig",
     "AnthropicTextConfig",
     "GroqSTTConfig",
@@ -178,6 +179,7 @@ LLM_CONFIG_NAMES = (
     "NLPCloudConfig",
     "VertexGeminiConfig",
     "GoogleAIStudioGeminiConfig",
+    "AntigravityConfig",
     "VertexAIAnthropicConfig",
     "VertexAILlama3Config",
     "VertexAIAi21Config",
@@ -612,6 +614,7 @@ _LLM_CONFIGS_IMPORT_MAP = {
     "OpenrouterConfig": (".llms.openrouter.chat.transformation", "OpenrouterConfig"),
     "DataRobotConfig": (".llms.datarobot.chat.transformation", "DataRobotConfig"),
     "AnthropicConfig": (".llms.anthropic.chat.transformation", "AnthropicConfig"),
+    "ClaudeMaxConfig": (".llms.claude_max.chat.transformation", "ClaudeMaxConfig"),
     "BedrockClaudePlatformConfig": (
         ".llms.bedrock.claude_platform.transformation",
         "BedrockClaudePlatformConfig",

litellm/llms/claude_max/authenticator.py

@@ -0,0 +1,29 @@
+import os
+import time
+from typing import Any, Dict
+
+from litellm.llms.server_oauth_base import RefreshTokenOAuthAuthenticator
+
+
+class ClaudeMaxAuthenticator(RefreshTokenOAuthAuthenticator):
+    provider = "claude-max"
+    env_prefix = "CLAUDE_MAX"
+    token_url = os.getenv("CLAUDE_MAX_TOKEN_URL", "https://platform.claude.com/v1/oauth/token")
+    client_id = os.getenv("CLAUDE_MAX_CLIENT_ID", "9d1c250a-e61b-44d9-88ed-5944d1962f5e")
+
+    def refresh_tokens(self, data: Dict[str, Any], refresh_token: str) -> Dict[str, Any]:
+        response = self._post_refresh_json(
+            {
+                "grant_type": "refresh_token",
+                "refresh_token": refresh_token,
+                "client_id": self.client_id,
+            },
+            headers={"content-type": "application/json"},
+        )
+        return {
+            "access_token": response.get("access_token"),
+            "refresh_token": response.get("refresh_token") or refresh_token,
+            "expires_at": int(time.time()) + int(response.get("expires_in", 3600)),
+            "id_token": response.get("id_token") or data.get("id_token"),
+            "scope": response.get("scope") or data.get("scope"),
+        }

litellm/llms/claude_max/chat/transformation.py

@@ -0,0 +1,26 @@
+import os
+from typing import Optional
+
+from litellm.llms.anthropic.chat.transformation import AnthropicConfig
+
+from ..authenticator import ClaudeMaxAuthenticator
+
+
+class ClaudeMaxConfig(AnthropicConfig):
+    @property
+    def custom_llm_provider(self) -> Optional[str]:
+        return "claude_max"
+
+    def get_access_token(self) -> str:
+        return ClaudeMaxAuthenticator().get_access_token()
+
+    def get_api_base(self, api_base: Optional[str] = None) -> str:
+        return api_base or os.getenv("CLAUDE_MAX_API_BASE") or "https://api.anthropic.com/v1/messages"
+
+    def get_default_headers(self) -> dict:
+        return {
+            "User-Agent": os.getenv("CLAUDE_MAX_USER_AGENT", "claude-cli/2.1.123 (external, sdk-cli)"),
+            "x-app": "cli",
+            "X-Stainless-Lang": "js",
+            "X-Stainless-Runtime": "node",
+        }

litellm/llms/server_oauth_base.py

@@ -0,0 +1,119 @@
+import base64
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import httpx
+
+from litellm._logging import verbose_logger
+from litellm.llms.custom_httpx.http_handler import _get_httpx_client
+
+TOKEN_EXPIRY_SKEW_SECONDS = 300
+DEFAULT_TOKEN_DIR = Path(os.getenv("SERVER_OAUTH_TOKEN_DIR", os.path.expanduser("~/.config/litellm/server_oauth")))
+
+
+class ServerOAuthError(Exception):
+    pass
+
+
+class JsonOAuthTokenStore:
+    def __init__(self, provider: str, env_prefix: str) -> None:
+        self.provider = provider
+        self.env_prefix = env_prefix
+        self.path = Path(os.getenv(f"{env_prefix}_OAUTH_FILE", str(DEFAULT_TOKEN_DIR / f"{provider}.json")))
+
+    def load(self) -> Dict[str, Any]:
+        self._restore_from_env_if_needed()
+        try:
+            with open(self.path, "r") as f:
+                data = json.load(f)
+        except IOError as exc:
+            raise ServerOAuthError(
+                f"{self.provider} OAuth token file missing at {self.path}. Set {self.env_prefix}_OAUTH_JSON_B64 or mount a token file."
+            ) from exc
+        except json.JSONDecodeError as exc:
+            raise ServerOAuthError(f"Invalid {self.provider} OAuth token file at {self.path}: {exc}") from exc
+        return data
+
+    def save(self, data: Dict[str, Any]) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(data, f)
+        try:
+            self.path.chmod(0o600)
+        except OSError:
+            pass
+
+    def _restore_from_env_if_needed(self) -> None:
+        encoded = os.getenv(f"{self.env_prefix}_OAUTH_JSON_B64")
+        if not encoded or self.path.exists():
+            return
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.path.write_bytes(base64.b64decode(encoded))
+        try:
+            self.path.chmod(0o600)
+        except OSError:
+            pass
+
+
+def normalize_expires_at(raw: Dict[str, Any]) -> int:
+    expires_at = raw.get("expires_at") or raw.get("expiresAt") or raw.get("expiry_date")
+    if expires_at is None and isinstance(raw.get("expires_in"), (int, float)):
+        expires_at = int(time.time()) + int(raw["expires_in"])
+    if expires_at is None:
+        return 0
+    expires_at = int(expires_at)
+    if expires_at > 10_000_000_000:
+        expires_at //= 1000
+    return expires_at
+
+
+class RefreshTokenOAuthAuthenticator:
+    provider: str
+    env_prefix: str
+    token_url: str
+
+    def __init__(self) -> None:
+        self.store = JsonOAuthTokenStore(self.provider, self.env_prefix)
+
+    def get_access_token(self) -> str:
+        data = self.store.load()
+        access_token = data.get("access_token") or data.get("accessToken")
+        if access_token and not self._is_expired(data):
+            return access_token
+        refresh_token = data.get("refresh_token") or data.get("refreshToken")
+        if not refresh_token:
+            if access_token:
+                return access_token
+            raise ServerOAuthError(f"{self.provider} OAuth token file has no access_token or refresh_token")
+        refreshed = self.refresh_tokens(data, refresh_token)
+        self.store.save(refreshed)
+        return refreshed["access_token"]
+
+    def _is_expired(self, data: Dict[str, Any]) -> bool:
+        return normalize_expires_at(data) <= int(time.time()) + TOKEN_EXPIRY_SKEW_SECONDS
+
+    def refresh_tokens(self, data: Dict[str, Any], refresh_token: str) -> Dict[str, Any]:
+        raise NotImplementedError
+
+    def _post_refresh_json(self, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
+        client = _get_httpx_client()
+        resp = client.post(self.token_url, json=payload, headers=headers or {})
+        try:
+            resp.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            verbose_logger.warning("%s OAuth refresh failed: %s", self.provider, exc)
+            raise ServerOAuthError(f"{self.provider} OAuth refresh failed: {exc}") from exc
+        return resp.json()
+
+    def _post_refresh_form(self, payload: Dict[str, Any], headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
+        client = _get_httpx_client()
+        resp = client.post(self.token_url, data=payload, headers=headers or {})
+        try:
+            resp.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+            verbose_logger.warning("%s OAuth refresh failed: %s", self.provider, exc)
+            raise ServerOAuthError(f"{self.provider} OAuth refresh failed: {exc}") from exc
+        return resp.json()

litellm/main.py

@@ -2911,23 +2911,30 @@ def completion(  # type: ignore # noqa: PLR0915
                 api_key=api_key,
                 logging_obj=logging,  # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
             )
-        elif custom_llm_provider == "anthropic":
-            api_key = (
-                api_key
-                or litellm.anthropic_key
-                or litellm.api_key
-                or os.environ.get("ANTHROPIC_API_KEY")
-            )
-            custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
-            # call /messages
-            # default route for all anthropic models
-            api_base = (
-                api_base
-                or litellm.api_base
-                or get_secret("ANTHROPIC_API_BASE")
-                or get_secret("ANTHROPIC_BASE_URL")
-                or "https://api.anthropic.com/v1/messages"
-            )
+        elif custom_llm_provider == "anthropic" or custom_llm_provider == "claude_max":
+            if custom_llm_provider == "claude_max":
+                from litellm.llms.claude_max.chat.transformation import ClaudeMaxConfig
+
+                claude_max_config = ClaudeMaxConfig()
+                api_key = api_key or claude_max_config.get_access_token()
+                api_base = claude_max_config.get_api_base(api_base)
+                headers = {**claude_max_config.get_default_headers(), **headers}
+                custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
+            else:
+                api_key = (
+                    api_key
+                    or litellm.anthropic_key
+                    or litellm.api_key
+                    or os.environ.get("ANTHROPIC_API_KEY")
+                )
+                custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
+                api_base = (
+                    api_base
+                    or litellm.api_base
+                    or get_secret("ANTHROPIC_API_BASE")
+                    or get_secret("ANTHROPIC_BASE_URL")
+                    or "https://api.anthropic.com/v1/messages"
+                )
 
             # Check if we should disable automatic URL suffix appending
             disable_url_suffix = get_secret_bool("LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX")
@@ -3487,7 +3494,7 @@ def completion(  # type: ignore # noqa: PLR0915
             raise ValueError(
                 "Palm was decommisioned on October 2024. Please use the `gemini/` route for Gemini Google AI Studio Models. Announcement: https://ai.google.dev/palm_docs/palm?hl=en"
             )
-        elif custom_llm_provider == "vertex_ai_beta" or custom_llm_provider == "gemini":
+        elif custom_llm_provider in ("vertex_ai_beta", "gemini"):
             vertex_ai_project = (
                 optional_params.pop("vertex_project", None)
                 or optional_params.pop("vertex_ai_project", None)
@@ -3512,8 +3519,8 @@ def completion(  # type: ignore # noqa: PLR0915
                 or get_secret("PALM_API_KEY")  # older palm api key should also work
                 or litellm.api_key
             )
-
             api_base = api_base or litellm.api_base or get_secret("GEMINI_API_BASE")
+
             new_params = safe_deep_copy(optional_params or {})
             response = vertex_chat_completion.completion(  # type: ignore
                 model=model,

litellm/types/utils.py

@@ -3234,6 +3234,7 @@ GenericBudgetConfigType = Dict[str, BudgetConfig]
 class LlmProviders(str, Enum):
     OPENAI = "openai"
     CHATGPT = "chatgpt"
+    CLAUDE_MAX = "claude_max"
     OPENAI_LIKE = "openai_like"  # embedding only
     JINA_AI = "jina_ai"
     XAI = "xai"

litellm/utils.py

@@ -8260,6 +8260,7 @@ class ProviderConfigManager:
             LlmProviders.COMPACTIFAI: (lambda: litellm.CompactifAIChatConfig(), False),
             LlmProviders.GITHUB_COPILOT: (lambda: litellm.GithubCopilotConfig(), False),
             LlmProviders.CHATGPT: (lambda: litellm.ChatGPTConfig(), False),
+            LlmProviders.CLAUDE_MAX: (lambda: litellm.ClaudeMaxConfig(), False),
             LlmProviders.GIGACHAT: (lambda: litellm.GigaChatConfig(), False),
             LlmProviders.RAGFLOW: (lambda: litellm.RAGFlowConfig(), False),
             LlmProviders.CUSTOM: (lambda: litellm.OpenAILikeChatConfig(), False),

tests/test_litellm/llms/server_oauth/test_server_oauth.py

@@ -0,0 +1,54 @@
+import base64
+import json
+import time
+from unittest.mock import MagicMock, patch
+
+from litellm.llms.claude_max.authenticator import ClaudeMaxAuthenticator
+from litellm.llms.claude_max.chat.transformation import ClaudeMaxConfig
+from litellm.llms.server_oauth_base import JsonOAuthTokenStore
+from litellm.types.utils import LlmProviders
+from litellm.utils import ProviderConfigManager
+
+
+def test_json_store_restores_base64_env(tmp_path, monkeypatch):
+    token_path = tmp_path / "claude-max.json"
+    payload = {"access_token": "token", "expires_at": time.time() + 3600}
+    monkeypatch.setenv("CLAUDE_MAX_OAUTH_FILE", str(token_path))
+    monkeypatch.setenv("CLAUDE_MAX_OAUTH_JSON_B64", base64.b64encode(json.dumps(payload).encode()).decode())
+
+    store = JsonOAuthTokenStore("claude-max", "CLAUDE_MAX")
+
+    assert store.load()["access_token"] == "token"
+    assert token_path.exists()
+
+
+def test_claude_max_refresh_uses_anthropic_oauth_contract(tmp_path, monkeypatch):
+    token_path = tmp_path / "claude-max.json"
+    monkeypatch.setenv("CLAUDE_MAX_OAUTH_FILE", str(token_path))
+    token_path.write_text(json.dumps({"access_token": "old", "refresh_token": "refresh", "expires_at": 1}))
+    response = MagicMock()
+    response.json.return_value = {"access_token": "new", "expires_in": 3600}
+    response.raise_for_status.return_value = None
+    client = MagicMock()
+    client.post.return_value = response
+
+    with patch("litellm.llms.server_oauth_base._get_httpx_client", return_value=client):
+        assert ClaudeMaxAuthenticator().get_access_token() == "new"
+
+    client.post.assert_called_once()
+    _, kwargs = client.post.call_args
+    assert kwargs["json"]["grant_type"] == "refresh_token"
+    assert kwargs["json"]["client_id"] == "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
+
+
+def test_provider_config_registration():
+    claude_config = ProviderConfigManager.get_provider_chat_config("claude-opus-4-8", LlmProviders.CLAUDE_MAX)
+
+    assert isinstance(claude_config, ClaudeMaxConfig)
+
+
+def test_claude_max_default_headers():
+    headers = ClaudeMaxConfig().get_default_headers()
+
+    assert headers["x-app"] == "cli"
+    assert headers["X-Stainless-Lang"] == "js"