diff --git a/litellm/proxy/client/cli/commands/autoroute/commands.py b/litellm/proxy/client/cli/commands/autoroute/commands.py index aca2f395615..cbc07456ac2 100644 --- a/litellm/proxy/client/cli/commands/autoroute/commands.py +++ b/litellm/proxy/client/cli/commands/autoroute/commands.py @@ -4,15 +4,13 @@ import secrets import signal import threading from types import FrameType -from typing import Dict, Optional import click import yaml from pydantic import JsonValue, TypeAdapter -from ..up import CLAUDE_SETTINGS_PATH +from ..up import CLAUDE_SETTINGS_PATH, load_json_or_empty, restore_claude_settings, write_backup from ..up import BackupRecord as ClaudeBackupRecord -from ..up import load_json_or_empty, restore_claude_settings, write_backup from .process import ( AUTOROUTE_DIR, CONFIG_PATH, @@ -34,7 +32,7 @@ from .wizard import run_configure_wizard AUTOROUTE_BACKUP_PATH = AUTOROUTE_DIR / "claude_settings_backup.json" -_GENERATED_CONFIG_ADAPTER = TypeAdapter(Dict[str, JsonValue]) +_GENERATED_CONFIG_ADAPTER = TypeAdapter(dict[str, JsonValue]) def _mint_and_embed_master_key() -> str: @@ -49,11 +47,11 @@ def _mint_and_embed_master_key() -> str: with open(CONFIG_PATH, "r") as f: generated = _GENERATED_CONFIG_ADAPTER.validate_python(yaml.safe_load(f)) general_settings = generated.get("general_settings") - updated_settings: Dict[str, JsonValue] = { + updated_settings: dict[str, JsonValue] = { **(general_settings if isinstance(general_settings, dict) else {}), "master_key": master_key, } - updated: Dict[str, JsonValue] = {**generated, "general_settings": updated_settings} + updated: dict[str, JsonValue] = {**generated, "general_settings": updated_settings} with open(CONFIG_PATH, "w") as f: yaml.safe_dump(updated, f, sort_keys=False) CONFIG_PATH.chmod(0o600) @@ -122,7 +120,7 @@ def up() -> None: restore_claude_settings(CLAUDE_SETTINGS_PATH, AUTOROUTE_BACKUP_PATH) click.echo("\nStopped ephemeral proxy and restored Claude Code settings.") - def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None: + def _handle_signal(_signum: int, _frame: FrameType | None) -> None: stop_event.set() signal.signal(signal.SIGINT, _handle_signal) @@ -139,7 +137,7 @@ def up() -> None: @autoroute_group.command("down") def down() -> None: """Restore Claude Code settings and stop a leftover ephemeral proxy, if any""" - record: Optional[PidRecord] = read_pid_record() + record: PidRecord | None = read_pid_record() if record is not None and is_running(record.pid): terminate(record.pid) click.echo(f"Stopped leftover ephemeral proxy (pid {record.pid}).") diff --git a/litellm/proxy/client/cli/commands/autoroute/config.py b/litellm/proxy/client/cli/commands/autoroute/config.py index f23ebff2330..1d18f497324 100644 --- a/litellm/proxy/client/cli/commands/autoroute/config.py +++ b/litellm/proxy/client/cli/commands/autoroute/config.py @@ -1,8 +1,8 @@ -from typing import Dict, FrozenSet, List, Literal, Tuple, Union +from typing import Literal, Union from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter -TIER_NAMES: Tuple[str, ...] = ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING") +TIER_NAMES: tuple[str, ...] = ("SIMPLE", "MEDIUM", "COMPLEX", "REASONING") AUTOROUTER_MODEL_NAME = "autorouter" @@ -32,10 +32,10 @@ class _RawModelGroup(BaseModel): output_cost_per_token: float | None = None -_RAW_MODEL_GROUPS_ADAPTER = TypeAdapter(List[_RawModelGroup]) +_RAW_MODEL_GROUPS_ADAPTER = TypeAdapter(list[_RawModelGroup]) -def parse_discovered_models(raw: List[JsonValue]) -> Tuple[DiscoveredModel, ...]: +def parse_discovered_models(raw: list[JsonValue]) -> tuple[DiscoveredModel, ...]: """Validate a raw `/model_group/info` response into typed models.""" parsed = _RAW_MODEL_GROUPS_ADAPTER.validate_python(raw) return tuple( @@ -52,11 +52,11 @@ def parse_discovered_models(raw: List[JsonValue]) -> Tuple[DiscoveredModel, ...] ) -def chat_models(models: Tuple[DiscoveredModel, ...]) -> Tuple[DiscoveredModel, ...]: +def chat_models(models: tuple[DiscoveredModel, ...]) -> tuple[DiscoveredModel, ...]: return tuple(m for m in models if m.mode == "chat") -def embedding_models(models: Tuple[DiscoveredModel, ...]) -> Tuple[DiscoveredModel, ...]: +def embedding_models(models: tuple[DiscoveredModel, ...]) -> tuple[DiscoveredModel, ...]: return tuple(m for m in models if m.mode == "embedding") @@ -91,7 +91,7 @@ SemanticMatchingChoice = Union[NoSemanticMatching, SemanticMatching] # Satisfies complexity_router's "semantic matching requires non-empty keyword_tier_rules" # invariant with a sane starting point; the generated config.yaml can be hand-edited afterward. -_DEFAULT_KEYWORD_TIER_RULES: Tuple[Dict[str, JsonValue], ...] = ( +_DEFAULT_KEYWORD_TIER_RULES: tuple[dict[str, JsonValue], ...] = ( {"keywords": ["hi", "hello", "thanks"], "tier": "SIMPLE"}, {"keywords": ["explain", "how does"], "tier": "MEDIUM"}, {"keywords": ["refactor", "implement", "debug"], "tier": "COMPLEX"}, @@ -106,17 +106,17 @@ class AutorouteConfig(BaseModel): api_key: str # Each tier maps to a pool of one or more models; complexity_router picks randomly among # them per request (or, in adaptive mode, learns which to prefer within the pool). - tiers: Dict[str, Tuple[str, ...]] + tiers: dict[str, tuple[str, ...]] default_model: str classifier: ClassifierChoice = Field(default_factory=HeuristicClassifier) semantic_matching: SemanticMatchingChoice = Field(default_factory=NoSemanticMatching) adaptive: bool = False -def validate_config(config: AutorouteConfig, discovered: Tuple[DiscoveredModel, ...]) -> None: +def validate_config(config: AutorouteConfig, discovered: tuple[DiscoveredModel, ...]) -> None: """Raise ConfigGenerationError if config references a model discovery didn't return.""" - chat_names: FrozenSet[str] = frozenset(m.name for m in chat_models(discovered)) - embedding_names: FrozenSet[str] = frozenset(m.name for m in embedding_models(discovered)) + chat_names: frozenset[str] = frozenset(m.name for m in chat_models(discovered)) + embedding_names: frozenset[str] = frozenset(m.name for m in embedding_models(discovered)) for tier, models in config.tiers.items(): for model in models: @@ -138,7 +138,7 @@ def validate_config(config: AutorouteConfig, discovered: Tuple[DiscoveredModel, ) -def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> Dict[str, JsonValue]: +def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> dict[str, JsonValue]: return { "model_name": name, "litellm_params": { @@ -149,7 +149,7 @@ def _litellm_proxy_deployment(name: str, base_url: str, api_key: str) -> Dict[st } -def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]: +def build_generated_model_list(config: AutorouteConfig) -> list[JsonValue]: """Build the model_list for the ephemeral proxy's config.yaml. Every real model referenced anywhere (tier targets, classifier, embedding) is deduplicated @@ -167,7 +167,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]: _litellm_proxy_deployment(name, config.base_url, config.api_key) for name in sorted(referenced_names) ] - complexity_router_config: Dict[str, JsonValue] = { + complexity_router_config: dict[str, JsonValue] = { "tiers": {tier: list(models) for tier, models in config.tiers.items()}, "default_model": config.default_model, } @@ -185,7 +185,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]: if config.adaptive: complexity_router_config["adaptive"] = True - auto_router_litellm_params: Dict[str, JsonValue] = { + auto_router_litellm_params: dict[str, JsonValue] = { "model": "auto_router/complexity_router", "complexity_router_config": complexity_router_config, } @@ -202,7 +202,7 @@ def build_generated_model_list(config: AutorouteConfig) -> List[JsonValue]: ] -def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> Dict[str, JsonValue]: +def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> dict[str, JsonValue]: """Full config.yaml content for the ephemeral proxy, including its own auth key. master_key must live under general_settings, not litellm_settings -- the proxy server @@ -217,20 +217,20 @@ def build_generated_proxy_config(config: AutorouteConfig, master_key: str) -> Di __all__ = [ - "TIER_NAMES", "AUTOROUTER_MODEL_NAME", + "TIER_NAMES", + "AutorouteConfig", + "ClassifierChoice", "ConfigGenerationError", "DiscoveredModel", - "parse_discovered_models", - "chat_models", - "embedding_models", "HeuristicClassifier", "LLMClassifier", - "ClassifierChoice", "NoSemanticMatching", "SemanticMatching", "SemanticMatchingChoice", - "AutorouteConfig", - "validate_config", "build_generated_model_list", + "chat_models", + "embedding_models", + "parse_discovered_models", + "validate_config", ] diff --git a/litellm/proxy/client/cli/commands/autoroute/process.py b/litellm/proxy/client/cli/commands/autoroute/process.py index ce146e95eec..40d0585cfed 100644 --- a/litellm/proxy/client/cli/commands/autoroute/process.py +++ b/litellm/proxy/client/cli/commands/autoroute/process.py @@ -9,7 +9,6 @@ import threading import time from dataclasses import dataclass from pathlib import Path -from typing import Optional import click import requests @@ -76,7 +75,7 @@ def poll_liveliness(base_url: str, log_path: Path, process: "subprocess.Popen[by ) -def write_pid_record(record: PidRecord, path: Optional[Path] = None) -> None: +def write_pid_record(record: PidRecord, path: Path | None = None) -> None: resolved_path = path if path is not None else PID_RECORD_PATH resolved_path.parent.mkdir(parents=True, exist_ok=True) with open(resolved_path, "w") as f: @@ -87,7 +86,7 @@ def write_pid_record(record: PidRecord, path: Optional[Path] = None) -> None: ) -def read_pid_record(path: Optional[Path] = None) -> Optional[PidRecord]: +def read_pid_record(path: Path | None = None) -> PidRecord | None: resolved_path = path if path is not None else PID_RECORD_PATH if not resolved_path.exists(): return None @@ -95,7 +94,7 @@ def read_pid_record(path: Optional[Path] = None) -> Optional[PidRecord]: return _PID_RECORD_ADAPTER.validate_json(f.read()) -def clear_pid_record(path: Optional[Path] = None) -> None: +def clear_pid_record(path: Path | None = None) -> None: resolved_path = path if path is not None else PID_RECORD_PATH resolved_path.unlink(missing_ok=True) @@ -144,15 +143,15 @@ __all__ = [ "CONFIG_PATH", "LOG_PATH", "PID_RECORD_PATH", - "ProcessLaunchError", "PidRecord", + "ProcessLaunchError", "allocate_free_port", - "launch_proxy", - "poll_liveliness", - "write_pid_record", - "read_pid_record", "clear_pid_record", "is_running", - "terminate", + "launch_proxy", + "poll_liveliness", + "read_pid_record", "stream_log", + "terminate", + "write_pid_record", ] diff --git a/litellm/proxy/client/cli/commands/autoroute/settings.py b/litellm/proxy/client/cli/commands/autoroute/settings.py index 811d131f433..4bed184eb34 100644 --- a/litellm/proxy/client/cli/commands/autoroute/settings.py +++ b/litellm/proxy/client/cli/commands/autoroute/settings.py @@ -1,5 +1,3 @@ -from typing import Dict - from pydantic import JsonValue from .config import AUTOROUTER_MODEL_NAME @@ -22,8 +20,8 @@ ANTHROPIC_DEFAULT_MODEL_ENV_KEYS = ( def merge_claude_settings_static_token( - settings: Dict[str, JsonValue], base_url: str, auth_token: str -) -> Dict[str, JsonValue]: + settings: dict[str, JsonValue], base_url: str, auth_token: str +) -> dict[str, JsonValue]: """Return a new settings dict wired to a local ephemeral proxy with a static token. Unlike up.py's merge_claude_settings (which sets apiKeyHelper for a long-lived, real @@ -33,14 +31,14 @@ def merge_claude_settings_static_token( """ raw_env = settings.get(ENV_KEY, {}) base_env = raw_env if isinstance(raw_env, dict) else {} - env: Dict[str, JsonValue] = { + env: dict[str, JsonValue] = { **base_env, ANTHROPIC_BASE_URL_KEY: base_url.rstrip("/"), ANTHROPIC_AUTH_TOKEN_KEY: auth_token, **{key: AUTOROUTER_MODEL_NAME for key in ANTHROPIC_DEFAULT_MODEL_ENV_KEYS}, } env.pop(ANTHROPIC_API_KEY_KEY, None) - merged: Dict[str, JsonValue] = {**settings, ENV_KEY: env} + merged: dict[str, JsonValue] = {**settings, ENV_KEY: env} merged.pop(API_KEY_HELPER_KEY, None) return merged diff --git a/litellm/proxy/client/cli/commands/autoroute/wizard.py b/litellm/proxy/client/cli/commands/autoroute/wizard.py index 1aaa37a2b7c..06a6ea7d2b1 100644 --- a/litellm/proxy/client/cli/commands/autoroute/wizard.py +++ b/litellm/proxy/client/cli/commands/autoroute/wizard.py @@ -1,6 +1,5 @@ import sys from pathlib import Path -from typing import List, Tuple import click import yaml @@ -30,7 +29,7 @@ def _is_interactive() -> bool: return sys.stdin.isatty() -def _fuzzy_pick(models: Tuple[DiscoveredModel, ...], prompt_label: str, multiselect: bool) -> List[str]: +def _fuzzy_pick(models: tuple[DiscoveredModel, ...], prompt_label: str, multiselect: bool) -> list[str]: """Type-to-filter picker over a (possibly huge) model pool, using InquirerPy's fzf-style fuzzy prompt. A plain numbered table + typed index does not scale past a handful of models -- proxies with @@ -56,11 +55,11 @@ def _fuzzy_pick(models: Tuple[DiscoveredModel, ...], prompt_label: str, multisel click.echo("Select at least one model.") -def _render_and_prompt_for_model(models: Tuple[DiscoveredModel, ...], prompt_label: str) -> str: +def _render_and_prompt_for_model(models: tuple[DiscoveredModel, ...], prompt_label: str) -> str: return _fuzzy_pick(models, prompt_label, multiselect=False)[0] -def _render_and_prompt_for_models(models: Tuple[DiscoveredModel, ...], prompt_label: str) -> Tuple[str, ...]: +def _render_and_prompt_for_models(models: tuple[DiscoveredModel, ...], prompt_label: str) -> tuple[str, ...]: return tuple(_fuzzy_pick(models, prompt_label, multiselect=True)) diff --git a/litellm/proxy/client/cli/commands/up.py b/litellm/proxy/client/cli/commands/up.py index a4ed8a6cc5f..23cce90992c 100644 --- a/litellm/proxy/client/cli/commands/up.py +++ b/litellm/proxy/client/cli/commands/up.py @@ -9,7 +9,7 @@ import threading from dataclasses import dataclass from pathlib import Path from types import FrameType -from typing import Dict, Mapping, Optional +from typing import Mapping import click from pydantic import JsonValue, TypeAdapter @@ -37,14 +37,14 @@ class BackupRecord: """Snapshot of ~/.claude/settings.json taken right before `lite up` patches it.""" existed: bool - content: Optional[Dict[str, JsonValue]] + content: dict[str, JsonValue] | None -_SETTINGS_ADAPTER = TypeAdapter(Dict[str, JsonValue]) +_SETTINGS_ADAPTER = TypeAdapter(dict[str, JsonValue]) _BACKUP_RECORD_ADAPTER = TypeAdapter(BackupRecord) -def load_json_or_empty(path: Path) -> Dict[str, JsonValue]: +def load_json_or_empty(path: Path) -> dict[str, JsonValue]: if not path.exists(): return {} with open(path, "r") as f: @@ -56,7 +56,7 @@ def load_json_or_empty(path: Path) -> Dict[str, JsonValue]: def merge_claude_settings( settings: Mapping[str, JsonValue], base_url: str, api_key_helper: str -) -> Dict[str, JsonValue]: +) -> 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 @@ -71,7 +71,7 @@ def merge_claude_settings( return {**settings, ENV_KEY: env, API_KEY_HELPER_KEY: api_key_helper} -def write_backup(record: BackupRecord, backup_path: Optional[Path] = None) -> None: +def write_backup(record: BackupRecord, backup_path: Path | None = None) -> None: path = backup_path if backup_path is not None else BACKUP_PATH path.parent.mkdir(exist_ok=True) with open(path, "w") as f: @@ -79,7 +79,7 @@ def write_backup(record: BackupRecord, backup_path: Optional[Path] = None) -> No os.chmod(path, 0o600) -def read_backup(backup_path: Optional[Path] = None) -> Optional[BackupRecord]: +def read_backup(backup_path: Path | None = None) -> BackupRecord | None: path = backup_path if backup_path is not None else BACKUP_PATH if not path.exists(): return None @@ -87,9 +87,7 @@ def read_backup(backup_path: Optional[Path] = None) -> Optional[BackupRecord]: return _BACKUP_RECORD_ADAPTER.validate_json(f.read()) -def restore_claude_settings( - settings_path: Optional[Path] = None, backup_path: Optional[Path] = None -) -> Optional[BackupRecord]: +def restore_claude_settings(settings_path: Path | None = None, backup_path: Path | None = None) -> BackupRecord | None: """Restore settings_path from the backup at backup_path, then delete the backup. Returns the restored record, or None if there was nothing to restore. @@ -199,7 +197,7 @@ def up(ctx: click.Context) -> None: stop_event = threading.Event() restored = threading.Lock() - def _handle_signal(_signum: int, _frame: Optional[FrameType]) -> None: + def _handle_signal(_signum: int, _frame: FrameType | None) -> None: stop_event.set() def _restore_once() -> None: @@ -225,16 +223,16 @@ def down() -> None: __all__ = [ - "up", - "down", + "BACKUP_PATH", + "CLAUDE_SETTINGS_PATH", "BackupRecord", + "UpError", + "down", "load_json_or_empty", "merge_claude_settings", - "write_backup", "read_backup", - "restore_claude_settings", "resolve_api_key_helper", - "UpError", - "CLAUDE_SETTINGS_PATH", - "BACKUP_PATH", + "restore_claude_settings", + "up", + "write_backup", ]