From 28975f6ee0640dbcf063500f2a76d196fe187cdf Mon Sep 17 00:00:00 2001 From: Krrish Dholakia Date: Tue, 14 Jul 2026 09:52:35 -0700 Subject: [PATCH] feat(cli): add `lite up`/`lite down` to ambiently route Claude Code through the proxy Patches ~/.claude/settings.json in place (env.ANTHROPIC_BASE_URL + apiKeyHelper via `lite auth print-token`) so any `claude` session started afterward, from any terminal, routes through the local LiteLLM proxy with no wrapper command needed, unlike the existing `lite claude` subprocess-exec approach. Backs up the original file first and restores it on Ctrl-C/SIGTERM, or via `lite down` after an unclean exit. Cursor is not supported: no equivalent file-based config to patch. --- litellm/proxy/client/cli/README.md | 18 ++ litellm/proxy/client/cli/commands/agents.py | 5 +- litellm/proxy/client/cli/commands/up.py | 231 ++++++++++++++++ litellm/proxy/client/cli/main.py | 4 + .../proxy/client/cli/test_up_commands.py | 253 ++++++++++++++++++ 5 files changed, 509 insertions(+), 2 deletions(-) create mode 100644 litellm/proxy/client/cli/commands/up.py create mode 100644 tests/test_litellm/proxy/client/cli/test_up_commands.py diff --git a/litellm/proxy/client/cli/README.md b/litellm/proxy/client/cli/README.md index ce13a906a36..e8b024dc31d 100644 --- a/litellm/proxy/client/cli/README.md +++ b/litellm/proxy/client/cli/README.md @@ -471,6 +471,24 @@ The token minted by `lite login` is a short-lived, per-session agent credential, The credential is short-lived by design (default 24h, configurable via `LITELLM_CLI_JWT_EXPIRATION_HOURS`); run `lite login` again to refresh it, which also re-reads your latest team and user settings. It does not appear in the Keys UI and cannot be rotated or revoked mid-session. `lite auth print-token` (usable as Claude Code's `apiKeyHelper`) prints it while it's still fresh and fails once it expires -- there is no silent renewal, so a long-running session needs a fresh `lite login` once a day. `lite claude`, `lite codex`, and `lite opencode` work with it on a default deployment; `EXPERIMENTAL_UI_LOGIN` is not required. If you need a long-lived, rotatable key that shows up in the Keys UI, create a dedicated virtual key in the dashboard and pass it via `--api-key` or `LITELLM_PROXY_API_KEY` instead. +### Route Every Claude Code Session Through the Proxy + +`lite claude` wraps a single invocation, but `lite up` goes further: it patches `~/.claude/settings.json`, Claude Code's own config file, so that every Claude Code session started afterward -- from any terminal, launched normally with just `claude`, no wrapper needed -- routes through your LiteLLM proxy. It sets `env.ANTHROPIC_BASE_URL` to the proxy URL and `apiKeyHelper` to a `lite auth print-token` invocation, drops any stray static `ANTHROPIC_API_KEY` so the helper-issued token wins, and leaves every other setting in the file untouched. It backs up the original file before patching it. + +Two things need to already be true: you've run `lite login`, since the apiKeyHelper depends on that stored token, and the proxy is already reachable, since `lite up` does not start one for you. + +```bash +lite login +litellm --config litellm/proxy/dev_config.yaml & +lite up +``` + +`lite up` runs in the foreground and blocks. Press Ctrl-C to stop it, which restores the original settings file and exits. If the process is ever killed uncleanly instead -- `kill -9`, a crash -- the settings file is left patched, and `lite down` is the manual recovery path: run it at any later point to restore from the same backup. + +This is a one-time file patch and restore, not a live traffic interceptor. A Claude Code session already running before `lite up` started keeps whatever `ANTHROPIC_BASE_URL` and token it loaded at its own startup, and a session still running when `lite up` stops keeps routing through the proxy until it exits; only sessions *started* while the patch is in effect are affected, and only *new* sessions after a restore go back to Anthropic directly. + +Cursor is not supported: it has no equivalent file-based config to hot-patch this way, since its model routing lives in its own app storage and is configured through its GUI. + ## Environment Variables The CLI respects the following environment variables: diff --git a/litellm/proxy/client/cli/commands/agents.py b/litellm/proxy/client/cli/commands/agents.py index dfcedb70686..6b6252d8ecb 100644 --- a/litellm/proxy/client/cli/commands/agents.py +++ b/litellm/proxy/client/cli/commands/agents.py @@ -212,7 +212,7 @@ def _is_interactive() -> bool: return sys.stdin.isatty() -def _resolve_api_key(ctx: click.Context) -> str: +def resolve_api_key(ctx: click.Context) -> str: base_url = ctx.obj["base_url"] api_key = ctx.obj.get("api_key") if api_key: @@ -238,7 +238,7 @@ _SKIP_VERIFY_HELP = "Skip the pre-launch key check against the proxy." def _launch(ctx: click.Context, binary: str, args: Sequence[str], *, skip_verify: bool) -> None: base_url = ctx.obj["base_url"] started_interactive = _is_interactive() - api_key = _resolve_api_key(ctx) + api_key = resolve_api_key(ctx) display_name, _ = agent_profile(binary) click.echo(f"litellm: routing {display_name} through proxy at {base_url.rstrip('/')}") @@ -288,5 +288,6 @@ __all__ = [ "agent_launch_args", "verify_proxy_key", "agent_profile", + "resolve_api_key", "AgentRunError", ] diff --git a/litellm/proxy/client/cli/commands/up.py b/litellm/proxy/client/cli/commands/up.py new file mode 100644 index 00000000000..fb9fb2b3fa4 --- /dev/null +++ b/litellm/proxy/client/cli/commands/up.py @@ -0,0 +1,231 @@ +import atexit +import json +import os +import shlex +import shutil +import signal +import sys +import threading +from dataclasses import dataclass +from pathlib import Path +from types import FrameType +from typing import Dict, Mapping, Optional + +import click +from pydantic import JsonValue, TypeAdapter + +from litellm.litellm_core_utils.cli_token_utils import is_cli_token_fresh + +from .agents import AgentRunError, resolve_api_key, verify_proxy_key +from .auth import load_token, login + +ENV_KEY = "env" +API_KEY_HELPER_KEY = "apiKeyHelper" +ANTHROPIC_BASE_URL_KEY = "ANTHROPIC_BASE_URL" +ANTHROPIC_API_KEY_KEY = "ANTHROPIC_API_KEY" + +CLAUDE_SETTINGS_PATH = Path.home() / ".claude" / "settings.json" +BACKUP_PATH = Path.home() / ".litellm" / "claude_settings_backup.json" + + +class UpError(Exception): + """Raised for any user-actionable failure while starting/stopping interception.""" + + +@dataclass(frozen=True, slots=True) +class BackupRecord: + """Snapshot of ~/.claude/settings.json taken right before `lite up` patches it.""" + + existed: bool + content: Optional[Dict[str, JsonValue]] + + +_SETTINGS_ADAPTER = TypeAdapter(Dict[str, JsonValue]) +_BACKUP_RECORD_ADAPTER = TypeAdapter(BackupRecord) + + +def load_json_or_empty(path: Path) -> Dict[str, JsonValue]: + if not path.exists(): + return {} + with open(path, "r") as f: + return _SETTINGS_ADAPTER.validate_json(f.read()) + + +def merge_claude_settings( + settings: Mapping[str, JsonValue], base_url: str, api_key_helper: str +) -> Dict[str, JsonValue]: + """Return a new settings dict wired to route Claude Code through the proxy. + + Only env.ANTHROPIC_BASE_URL and the top-level apiKeyHelper are overridden; a + stray env.ANTHROPIC_API_KEY is dropped so it cannot outrank the helper-issued + token (same reasoning as build_agent_env in agents.py). Every other key is + preserved untouched. + """ + raw_env = settings.get(ENV_KEY, {}) + base_env = raw_env if isinstance(raw_env, dict) else {} + env = {**base_env, ANTHROPIC_BASE_URL_KEY: base_url.rstrip("/")} + env.pop(ANTHROPIC_API_KEY_KEY, None) + return {**settings, ENV_KEY: env, API_KEY_HELPER_KEY: api_key_helper} + + +def write_backup(record: BackupRecord) -> None: + BACKUP_PATH.parent.mkdir(exist_ok=True) + with open(BACKUP_PATH, "w") as f: + json.dump({"existed": record.existed, "content": record.content}, f, indent=2) + os.chmod(BACKUP_PATH, 0o600) + + +def read_backup() -> Optional[BackupRecord]: + if not BACKUP_PATH.exists(): + return None + with open(BACKUP_PATH, "r") as f: + return _BACKUP_RECORD_ADAPTER.validate_json(f.read()) + + +def restore_claude_settings() -> Optional[BackupRecord]: + """Restore ~/.claude/settings.json from the backup, then delete the backup. + + Returns the restored record, or None if there was nothing to restore. + """ + record = read_backup() + if record is None: + return None + if record.existed and record.content is not None: + with open(CLAUDE_SETTINGS_PATH, "w") as f: + json.dump(record.content, f, indent=2) + elif CLAUDE_SETTINGS_PATH.exists(): + CLAUDE_SETTINGS_PATH.unlink() + BACKUP_PATH.unlink() + return record + + +def resolve_api_key_helper() -> str: + """Build the shell command Claude Code should run for its apiKeyHelper. + + Resolves `lite` to an absolute path so the helper works regardless of the + PATH visible to whatever subprocess Claude Code spawns it from. + """ + lite_path = shutil.which("lite") + if lite_path is None: + raise UpError( + "Could not find `lite` on your PATH. Claude Code's apiKeyHelper needs " + "an absolute path to it, so `lite up` cannot continue." + ) + return f"{shlex.quote(lite_path)} auth print-token" + + +def _ensure_fresh_login(ctx: click.Context) -> None: + token_data = load_token() + if token_data and is_cli_token_fresh(token_data): + return + + if not sys.stdin.isatty(): + raise UpError( + "No fresh LiteLLM login found. Run `lite login` first (apiKeyHelper " + "reads this token on every Claude Code request)." + ) + + click.echo("No fresh LiteLLM login found; starting login...") + ctx.invoke(login) + token_data = load_token() + if not token_data or not is_cli_token_fresh(token_data): + raise UpError("Login did not produce a usable token; cannot start `lite up`.") + + +def _restore_and_report() -> None: + record = restore_claude_settings() + if record is None: + click.echo("Nothing to restore.") + return + if record.existed: + click.echo(f"Restored {CLAUDE_SETTINGS_PATH} to its original contents.") + else: + click.echo(f"Removed {CLAUDE_SETTINGS_PATH} (it did not exist before `lite up`).") + + +@click.command(name="up") +@click.pass_context +def up(ctx: click.Context) -> None: + """Route every Claude Code session through your LiteLLM proxy until stopped. + + Patches ~/.claude/settings.json so Claude Code picks up the proxy on its own + next startup, from any terminal -- no need to launch it through `lite`. + Press Ctrl-C to stop and restore your original settings. Assumes the proxy + is already running (this does not start one for you). Cursor is not + supported: it has no equivalent file-based config to patch. + """ + base_url = ctx.obj["base_url"] + + try: + _ensure_fresh_login(ctx) + api_key = resolve_api_key(ctx) + verify_proxy_key(base_url, api_key) + + if BACKUP_PATH.exists(): + raise UpError( + f"{BACKUP_PATH} already exists -- `lite up` looks like it's already " + "running (or crashed without cleanup). Run `lite down` first." + ) + + api_key_helper = resolve_api_key_helper() + original_existed = CLAUDE_SETTINGS_PATH.exists() + original_settings = load_json_or_empty(CLAUDE_SETTINGS_PATH) + write_backup( + BackupRecord( + existed=original_existed, + content=original_settings if original_existed else None, + ) + ) + + CLAUDE_SETTINGS_PATH.parent.mkdir(exist_ok=True) + merged = merge_claude_settings(original_settings, base_url, api_key_helper) + with open(CLAUDE_SETTINGS_PATH, "w") as f: + json.dump(merged, f, indent=2) + except (AgentRunError, UpError) as e: + raise click.ClickException(str(e)) + + click.echo(f"litellm: routing Claude Code through proxy at {base_url.rstrip('/')}") + click.echo("Press Ctrl-C to stop and restore your original settings.") + + stop_event = threading.Event() + restored = threading.Lock() + + def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None: + stop_event.set() + + def _restore_once() -> None: + if restored.acquire(blocking=False): + _restore_and_report() + + signal.signal(signal.SIGINT, _handle_signal) + signal.signal(signal.SIGTERM, _handle_signal) + atexit.register(_restore_once) + + stop_event.wait() + _restore_once() + + +@click.command(name="down") +def down() -> None: + """Restore ~/.claude/settings.json if a `lite up` session left it patched. + + Use this after a `lite up` process was killed uncleanly (e.g. `kill -9`) + instead of stopped with Ctrl-C. + """ + _restore_and_report() + + +__all__ = [ + "up", + "down", + "BackupRecord", + "load_json_or_empty", + "merge_claude_settings", + "write_backup", + "read_backup", + "restore_claude_settings", + "resolve_api_key_helper", + "UpError", + "CLAUDE_SETTINGS_PATH", + "BACKUP_PATH", +] diff --git a/litellm/proxy/client/cli/main.py b/litellm/proxy/client/cli/main.py index 4de3ff5fc87..322ce2b287a 100644 --- a/litellm/proxy/client/cli/main.py +++ b/litellm/proxy/client/cli/main.py @@ -18,6 +18,7 @@ from .commands.keys import keys # local imports from .commands.models import models from .commands.teams import teams +from .commands.up import down, up from .commands.users import users from .interface import interactive_shell @@ -131,6 +132,9 @@ cli.add_command(users) # Add a top-level command per coding agent (claude, codex, opencode, ...) for agent_command in agent_commands(): cli.add_command(agent_command) +# Add the up/down commands (route Claude Code through the local LiteLLM proxy) +cli.add_command(up) +cli.add_command(down) if __name__ == "__main__": diff --git a/tests/test_litellm/proxy/client/cli/test_up_commands.py b/tests/test_litellm/proxy/client/cli/test_up_commands.py new file mode 100644 index 00000000000..7595707ac67 --- /dev/null +++ b/tests/test_litellm/proxy/client/cli/test_up_commands.py @@ -0,0 +1,253 @@ +import json +import shutil +import sys +from unittest.mock import patch + +import pytest +from click.testing import CliRunner + +from litellm.proxy.client.cli.commands import up as up_module +from litellm.proxy.client.cli.commands.agents import AgentRunError +from litellm.proxy.client.cli.commands.up import ( + BackupRecord, + UpError, + down, + merge_claude_settings, + read_backup, + resolve_api_key_helper, + restore_claude_settings, + up, + write_backup, +) + +UP_MODULE = "litellm.proxy.client.cli.commands.up" + + +def _patch_paths(monkeypatch, tmp_path): + settings_path = tmp_path / "claude_settings.json" + backup_path = tmp_path / "backup.json" + monkeypatch.setattr(up_module, "CLAUDE_SETTINGS_PATH", settings_path) + monkeypatch.setattr(up_module, "BACKUP_PATH", backup_path) + return settings_path, backup_path + + +class TestMergeClaudeSettings: + def test_preserves_unrelated_top_level_keys(self): + merged = merge_claude_settings({"theme": "dark"}, "http://localhost:4000", "helper") + assert merged["theme"] == "dark" + + def test_preserves_unrelated_env_keys(self): + settings = {"env": {"SOME_OTHER_VAR": "value"}} + merged = merge_claude_settings(settings, "http://localhost:4000", "helper") + assert merged["env"]["SOME_OTHER_VAR"] == "value" + + def test_overrides_base_url_and_helper(self): + settings = { + "env": {"ANTHROPIC_BASE_URL": "https://old.example.com"}, + "apiKeyHelper": "old-helper", + } + merged = merge_claude_settings(settings, "http://localhost:4000/", "new-helper") + assert merged["env"]["ANTHROPIC_BASE_URL"] == "http://localhost:4000" + assert merged["apiKeyHelper"] == "new-helper" + + def test_drops_stray_api_key(self): + settings = {"env": {"ANTHROPIC_API_KEY": "leaked-key"}} + merged = merge_claude_settings(settings, "http://localhost:4000", "helper") + assert "ANTHROPIC_API_KEY" not in merged["env"] + + def test_works_from_empty_settings(self): + merged = merge_claude_settings({}, "http://localhost:4000", "helper") + assert merged["env"] == {"ANTHROPIC_BASE_URL": "http://localhost:4000"} + assert merged["apiKeyHelper"] == "helper" + + def test_does_not_mutate_input(self): + settings = {"env": {"FOO": "bar"}} + merge_claude_settings(settings, "http://localhost:4000", "helper") + assert settings == {"env": {"FOO": "bar"}} + + +class TestBackupRoundTrip: + def test_restores_original_content_when_file_existed(self, monkeypatch, tmp_path): + settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + original = {"apiKeyHelper": "old-helper", "theme": "dark"} + settings_path.write_text(json.dumps(original)) + + write_backup(BackupRecord(existed=True, content=original)) + settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"})) + + restored = restore_claude_settings() + + assert restored is not None + assert restored.existed is True + assert json.loads(settings_path.read_text()) == original + assert not backup_path.exists() + + def test_deletes_settings_file_when_it_did_not_exist_before(self, monkeypatch, tmp_path): + settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + write_backup(BackupRecord(existed=False, content=None)) + settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"})) + + restored = restore_claude_settings() + + assert restored is not None + assert restored.existed is False + assert not settings_path.exists() + assert not backup_path.exists() + + def test_no_backup_is_a_no_op_returning_none(self, monkeypatch, tmp_path): + settings_path, _backup_path = _patch_paths(monkeypatch, tmp_path) + assert restore_claude_settings() is None + assert not settings_path.exists() + + def test_read_backup_round_trips_write_backup(self, monkeypatch, tmp_path): + _patch_paths(monkeypatch, tmp_path) + write_backup(BackupRecord(existed=True, content={"a": 1})) + assert read_backup() == BackupRecord(existed=True, content={"a": 1}) + + def test_read_backup_missing_file_returns_none(self, monkeypatch, tmp_path): + _patch_paths(monkeypatch, tmp_path) + assert read_backup() is None + + def test_backup_file_always_removed_after_restore(self, monkeypatch, tmp_path): + _settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + write_backup(BackupRecord(existed=False, content=None)) + assert backup_path.exists() + + restore_claude_settings() + + assert not backup_path.exists() + + +class TestResolveApiKeyHelper: + def test_returns_helper_command_when_lite_found(self, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda name: "/usr/local/bin/lite") + helper = resolve_api_key_helper() + assert helper == "/usr/local/bin/lite auth print-token" + + def test_raises_when_lite_not_on_path(self, monkeypatch): + monkeypatch.setattr(shutil, "which", lambda name: None) + with pytest.raises(UpError, match="Could not find `lite`"): + resolve_api_key_helper() + + +class TestUpCommand: + def setup_method(self): + self.runner = CliRunner() + + def test_refuses_double_start_without_touching_settings_file(self, monkeypatch, tmp_path): + settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + existing_backup = {"existed": False, "content": None} + backup_path.write_text(json.dumps(existing_backup)) + + with ( + patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}), + patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True), + patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"), + patch(f"{UP_MODULE}.verify_proxy_key"), + ): + result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"}) + + assert result.exit_code != 0 + assert "already" in result.output + assert "lite down" in result.output + assert not settings_path.exists() + assert json.loads(backup_path.read_text()) == existing_backup + + def test_no_fresh_login_non_interactive_fails_cleanly(self, monkeypatch, tmp_path): + _patch_paths(monkeypatch, tmp_path) + monkeypatch.setattr(sys.stdin, "isatty", lambda: False) + + with patch(f"{UP_MODULE}.load_token", return_value=None): + result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"}) + + assert result.exit_code != 0 + assert "lite login" in result.output + + def test_unreachable_proxy_fails_cleanly(self, monkeypatch, tmp_path): + _patch_paths(monkeypatch, tmp_path) + + with ( + patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}), + patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True), + patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"), + patch( + f"{UP_MODULE}.verify_proxy_key", + side_effect=AgentRunError("Could not reach the LiteLLM proxy at http://localhost:4000"), + ), + ): + result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"}) + + assert result.exit_code != 0 + assert "Could not reach the LiteLLM proxy" in result.output + + def test_happy_path_writes_settings_and_backup_then_restores_on_stop(self, monkeypatch, tmp_path): + settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + original = {"theme": "dark"} + settings_path.write_text(json.dumps(original)) + + captured = {} + + def fake_wait(self, timeout=None): + captured["settings"] = json.loads(settings_path.read_text()) + captured["backup_existed"] = backup_path.exists() + return True + + with ( + patch(f"{UP_MODULE}.load_token", return_value={"key": "sk-fresh"}), + patch(f"{UP_MODULE}.is_cli_token_fresh", return_value=True), + patch(f"{UP_MODULE}.resolve_api_key", return_value="sk-fresh"), + patch(f"{UP_MODULE}.verify_proxy_key"), + patch( + f"{UP_MODULE}.resolve_api_key_helper", + return_value="/usr/local/bin/lite auth print-token", + ), + patch(f"{UP_MODULE}.signal.signal"), + patch(f"{UP_MODULE}.atexit.register"), + patch("threading.Event.wait", new=fake_wait), + ): + result = self.runner.invoke(up, obj={"base_url": "http://localhost:4000"}) + + assert result.exit_code == 0, result.output + assert captured["backup_existed"] is True + assert captured["settings"]["theme"] == "dark" + assert captured["settings"]["env"]["ANTHROPIC_BASE_URL"] == "http://localhost:4000" + assert captured["settings"]["apiKeyHelper"] == "/usr/local/bin/lite auth print-token" + assert json.loads(settings_path.read_text()) == original + assert not backup_path.exists() + + +class TestDownCommand: + def setup_method(self): + self.runner = CliRunner() + + def test_restores_when_backup_exists(self, monkeypatch, tmp_path): + settings_path, backup_path = _patch_paths(monkeypatch, tmp_path) + original = {"apiKeyHelper": "old-helper"} + write_backup(BackupRecord(existed=True, content=original)) + settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"})) + + result = self.runner.invoke(down) + + assert result.exit_code == 0, result.output + assert "Restored" in result.output + assert json.loads(settings_path.read_text()) == original + assert not backup_path.exists() + + def test_removes_settings_file_when_it_did_not_exist_before(self, monkeypatch, tmp_path): + settings_path, _backup_path = _patch_paths(monkeypatch, tmp_path) + write_backup(BackupRecord(existed=False, content=None)) + settings_path.write_text(json.dumps({"apiKeyHelper": "lite-helper"})) + + result = self.runner.invoke(down) + + assert result.exit_code == 0, result.output + assert "Removed" in result.output + assert not settings_path.exists() + + def test_prints_nothing_to_restore_when_no_backup(self, monkeypatch, tmp_path): + _patch_paths(monkeypatch, tmp_path) + + result = self.runner.invoke(down) + + assert result.exit_code == 0, result.output + assert "Nothing to restore." in result.output