feat: add Claude subscription OAuth support

README.md

@@ -232,6 +232,7 @@ Assistant: Hi! I'm ClawCodex, a Python reimplementation...
 clawcodex                       # Interactive Ink TUI (default)
 clawcodex tui                   # Interactive Ink TUI (explicit)
 clawcodex login                 # Configure API keys (interactive)
+clawcodex logout anthropic      # Remove Claude Pro/Max OAuth credentials
 clawcodex config                # Show ~/.clawcodex/config.json-backed settings
 clawcodex --version             # Version string
 
@@ -250,6 +251,23 @@ clawcodex --dangerously-skip-permissions -p "ls"       # bypass all permission c
 clawcodex --allow-dangerously-skip-permissions         # allow /permission-mode bypass later
 ```
 
+### Claude Pro/Max subscription login
+
+Run `clawcodex login`, choose `anthropic`, then choose `subscription`. The CLI
+opens Claude's PKCE authorization page; paste the returned authorization code
+back into the prompt. ClawCodex stores the OAuth tokens in
+`~/.clawcodex/anthropic-oauth.json` with user-only permissions, refreshes them
+automatically, and uses the subscription whenever no Anthropic API key is set.
+An explicit `ANTHROPIC_API_KEY` or configured Anthropic API key takes precedence.
+
+> **Important:** Anthropic does not officially support using Claude Pro/Max
+> subscriptions from third-party clients. This integration may stop working
+> and remains subject to Anthropic's terms. The login flow requires explicit
+> confirmation before opening the authorization page.
+
+Use `clawcodex config` to check connection status and
+`clawcodex logout anthropic` to delete the stored OAuth credentials.
+
 > **`--dangerously-skip-permissions`** disables every tool permission check
 > for the session. Recommended only inside sandboxed containers/VMs with no
 > internet access. The flag is refused when the process is running as

src/auth/anthropic_subscription.py

@@ -0,0 +1,186 @@
+"""Claude Pro/Max OAuth credentials for the Anthropic provider.
+
+This implements the authorization-code + PKCE flow formerly bundled by
+OpenCode's ``opencode-anthropic-auth`` plugin.  Claude subscription use from
+third-party clients is not officially supported by Anthropic; callers must
+surface that fact before starting login.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+import secrets
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
+AUTHORIZE_URL = "https://claude.ai/oauth/authorize"
+TOKEN_URL = "https://console.anthropic.com/v1/oauth/token"
+REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback"
+SCOPES = "org:create_api_key user:profile user:inference"
+OAUTH_BETAS = ("oauth-2025-04-20", "interleaved-thinking-2025-05-14")
+_refresh_lock = threading.Lock()
+
+
+def credentials_path() -> Path:
+    root = Path(os.environ.get("CLAWCODEX_CONFIG_DIR", Path.home() / ".clawcodex"))
+    return root / "anthropic-oauth.json"
+
+
+@dataclass
+class SubscriptionCredentials:
+    access_token: str
+    refresh_token: str
+    expires_at: float
+    scope: str = SCOPES
+
+    @property
+    def needs_refresh(self) -> bool:
+        return self.expires_at <= time.time() + 60
+
+
+def load_credentials() -> SubscriptionCredentials | None:
+    path = credentials_path()
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+        return SubscriptionCredentials(
+            access_token=str(value["access_token"]),
+            refresh_token=str(value["refresh_token"]),
+            expires_at=float(value["expires_at"]),
+            scope=str(value.get("scope", SCOPES)),
+        )
+    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+        return None
+
+
+def save_credentials(credentials: SubscriptionCredentials) -> None:
+    path = credentials_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(f".tmp-{os.getpid()}-{secrets.token_hex(4)}")
+    try:
+        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        with os.fdopen(fd, "w", encoding="utf-8") as stream:
+            json.dump(asdict(credentials), stream, indent=2)
+            stream.write("\n")
+        os.replace(tmp, path)
+        if os.name != "nt":
+            os.chmod(path, 0o600)
+    finally:
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+
+
+def remove_credentials() -> bool:
+    try:
+        credentials_path().unlink()
+        return True
+    except FileNotFoundError:
+        return False
+
+
+def _post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(payload).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=30) as response:
+            result = json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", "replace")
+        raise RuntimeError(f"Claude OAuth request failed ({exc.code}): {detail}") from exc
+    if not isinstance(result, dict):
+        raise RuntimeError("Claude OAuth endpoint returned an invalid response")
+    return result
+
+
+def begin_login() -> tuple[str, str]:
+    verifier = secrets.token_urlsafe(64)[:128]
+    digest = hashlib.sha256(verifier.encode("ascii")).digest()
+    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+    query = urllib.parse.urlencode({
+        "code": "true",
+        "client_id": CLIENT_ID,
+        "response_type": "code",
+        "redirect_uri": REDIRECT_URI,
+        "scope": SCOPES,
+        "code_challenge": challenge,
+        "code_challenge_method": "S256",
+        # Claude returns this value after the '#' in its copy/paste code.
+        "state": verifier,
+    })
+    return f"{AUTHORIZE_URL}?{query}", verifier
+
+
+def complete_login(code: str, verifier: str) -> SubscriptionCredentials:
+    authorization_code, separator, returned_state = code.strip().partition("#")
+    if not authorization_code:
+        raise ValueError("Authorization code is empty")
+    payload: dict[str, Any] = {
+        "code": authorization_code,
+        "grant_type": "authorization_code",
+        "client_id": CLIENT_ID,
+        "redirect_uri": REDIRECT_URI,
+        "code_verifier": verifier,
+    }
+    if separator:
+        payload["state"] = returned_state
+    result = _post_json(TOKEN_URL, payload)
+    credentials = _credentials_from_response(result)
+    save_credentials(credentials)
+    return credentials
+
+
+def _credentials_from_response(
+    result: dict[str, Any], *, old_refresh_token: str = ""
+) -> SubscriptionCredentials:
+    try:
+        access = str(result["access_token"])
+        refresh = str(result.get("refresh_token") or old_refresh_token)
+        expires = time.time() + float(result["expires_in"])
+    except (KeyError, TypeError, ValueError) as exc:
+        raise RuntimeError("Claude OAuth response omitted required token fields") from exc
+    if not access or not refresh:
+        raise RuntimeError("Claude OAuth response omitted required token fields")
+    return SubscriptionCredentials(access, refresh, expires, str(result.get("scope", SCOPES)))
+
+
+def get_valid_credentials() -> SubscriptionCredentials | None:
+    credentials = load_credentials()
+    if credentials is None or not credentials.needs_refresh:
+        return credentials
+    with _refresh_lock:
+        credentials = load_credentials()
+        if credentials is None or not credentials.needs_refresh:
+            return credentials
+        result = _post_json(TOKEN_URL, {
+            "grant_type": "refresh_token",
+            "refresh_token": credentials.refresh_token,
+            "client_id": CLIENT_ID,
+        })
+        refreshed = _credentials_from_response(
+            result, old_refresh_token=credentials.refresh_token
+        )
+        save_credentials(refreshed)
+        return refreshed
+
+
+def subscription_headers(existing: dict[str, str] | None = None) -> dict[str, str]:
+    headers = dict(existing or {})
+    present = [part.strip() for part in headers.get("anthropic-beta", "").split(",") if part.strip()]
+    headers["anthropic-beta"] = ",".join(dict.fromkeys([*OAUTH_BETAS, *present]))
+    headers["user-agent"] = "claude-cli/2.1.2 (external, cli)"
+    return headers

src/cli.py

@@ -102,6 +102,8 @@ def main():
         rest = argv[1:]
         if token == 'login':
             return handle_login()
+        if token == 'logout':
+            return handle_logout(rest)
         if token == 'config':
             return show_config()
         if token == 'mcp':
@@ -714,6 +716,15 @@ def handle_login():
 
     info = PROVIDER_INFO[provider]
 
+    if provider == "anthropic":
+        auth_method = Prompt.ask(
+            "Authentication method",
+            choices=["subscription", "api-key"],
+            default="subscription",
+        )
+        if auth_method == "subscription":
+            return _handle_anthropic_subscription_login(console, Prompt)
+
     api_key = Prompt.ask(
         f"Enter {provider.upper()} API Key",
         password=True
@@ -746,6 +757,60 @@ def handle_login():
     return 0
 
 
+def _handle_anthropic_subscription_login(console, Prompt):
+    """Run Claude Pro/Max's copy/paste OAuth authorization flow."""
+    import webbrowser
+
+    from src.auth.anthropic_subscription import begin_login, complete_login
+    from src.config import set_api_key, set_default_provider
+    from src.providers import PROVIDER_INFO
+
+    console.print(
+        "\n[yellow]Important:[/yellow] Anthropic does not officially support "
+        "using Claude Pro/Max subscriptions from third-party clients. "
+        "This flow may stop working and is subject to Anthropic's terms."
+    )
+    consent = Prompt.ask("Continue?", choices=["yes", "no"], default="no")
+    if consent != "yes":
+        console.print("Login cancelled.")
+        return 1
+    url, verifier = begin_login()
+    console.print("\nOpening Claude authorization in your browser...")
+    if not webbrowser.open(url):
+        console.print(f"Open this URL manually:\n{url}")
+    code = Prompt.ask("Paste the authorization code")
+    try:
+        complete_login(code, verifier)
+    except Exception as exc:
+        console.print(f"\n[red]Claude subscription login failed:[/red] {exc}")
+        return 1
+    info = PROVIDER_INFO["anthropic"]
+    set_api_key(
+        "anthropic", api_key="", base_url=info["default_base_url"],
+        default_model=info["default_model"],
+    )
+    set_default_provider("anthropic")
+    console.print("\n[green]✓ Claude Pro/Max subscription connected.[/green]")
+    return 0
+
+
+def handle_logout(args: list[str]) -> int:
+    """Remove locally stored provider OAuth credentials."""
+    from rich.console import Console
+
+    console = Console()
+    provider = args[0] if args else "anthropic"
+    if provider != "anthropic":
+        console.print(f"[red]OAuth logout is not supported for provider '{provider}'.[/red]")
+        return 2
+    from src.auth.anthropic_subscription import remove_credentials
+    if remove_credentials():
+        console.print("[green]✓ Claude Pro/Max subscription disconnected.[/green]")
+    else:
+        console.print("No Claude Pro/Max subscription login was stored.")
+    return 0
+
+
 def show_config():
     """Show current configuration."""
     from rich.console import Console
@@ -773,6 +838,12 @@ def show_config():
             console.print(f"    Base URL: {provider_config.get('base_url', 'Not set')}")
             console.print(f"    Default Model: {provider_config.get('default_model', 'Not set')}")
 
+        from src.auth.anthropic_subscription import load_credentials
+        console.print(
+            "\n[cyan]Claude Pro/Max:[/cyan] "
+            + ("Connected" if load_credentials() is not None else "Not connected")
+        )
+
         # Stored API keys (config "env" block, e.g. TAVILY_API_KEY). Values are
         # masked — only the name and a hint are shown.
         from src.secret_store import get_secret, list_secret_names

src/cost_tracker.py

@@ -45,7 +45,7 @@ def record_api_usage(model: str, usage: Any) -> float:
     """
     if not isinstance(usage, dict):
         usage = {}
-    cost = compute_cost(model, usage)
+    cost = 0.0 if usage.get("billing_mode") == "subscription" else compute_cost(model, usage)
     bootstrap_usage = ModelUsage(
         input_tokens=int(usage.get("input_tokens", 0) or 0),
         output_tokens=int(usage.get("output_tokens", 0) or 0),

src/entrypoints/provider_validation.py

@@ -50,6 +50,10 @@ def get_provider_validation_error(provider_name: str | None) -> str | None:
     # ``DEEPSEEK_API_KEY``). Local providers (Ollama / vLLM / SGLang) need
     # no key. Same check the headless path ran inline pre-ENTRY-2.
     api_key = resolve_api_key(name, provider_cfg)
+    if name == "anthropic" and not api_key:
+        from src.auth.anthropic_subscription import load_credentials
+        if load_credentials() is not None:
+            return None
     if not api_key and provider_requires_api_key(name):
         return (
             f"error: API key for provider '{name}' is not configured. "