"""Centralized configuration using Pydantic Settings.""" import os from collections.abc import Mapping from dataclasses import dataclass from functools import lru_cache from pathlib import Path from typing import Any from dotenv import dotenv_values from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict from .constants import HTTP_CONNECT_TIMEOUT_DEFAULT from .nim import NimSettings from .provider_ids import SUPPORTED_PROVIDER_IDS @dataclass(frozen=True, slots=True) class ConfiguredChatModelRef: """A unique configured chat model reference and the env keys that set it.""" model_ref: str provider_id: str model_id: str sources: tuple[str, ...] def _env_files() -> tuple[Path, ...]: """Return env file paths in priority order (later overrides earlier).""" files: list[Path] = [ Path.home() / ".config" / "free-claude-code" / ".env", Path(".env"), ] if explicit := os.environ.get("FCC_ENV_FILE"): files.append(Path(explicit)) return tuple(files) def _configured_env_files(model_config: Mapping[str, Any]) -> tuple[Path, ...]: """Return the currently configured env files for Settings.""" configured = model_config.get("env_file") if configured is None: return () if isinstance(configured, (str, Path)): return (Path(configured),) return tuple(Path(item) for item in configured) def _env_file_contains_key(path: Path, key: str) -> bool: """Check whether a dotenv-style file defines the given key.""" return _env_file_value(path, key) is not None def _env_file_value(path: Path, key: str) -> str | None: """Return a dotenv value when the file explicitly defines the key.""" if not path.is_file(): return None try: values = dotenv_values(path) except OSError: return None if key not in values: return None value = values[key] return "" if value is None else value def _env_file_override(model_config: Mapping[str, Any], key: str) -> str | None: """Return the last configured dotenv value that explicitly defines a key.""" configured_value: str | None = None for env_file in _configured_env_files(model_config): value = _env_file_value(env_file, key) if value is not None: configured_value = value return configured_value def _removed_env_var_message(model_config: Mapping[str, Any]) -> str | None: """Return a migration error for removed env vars, if present.""" removed_keys = ("NIM_ENABLE_THINKING", "ENABLE_THINKING") replacement = ( "ENABLE_MODEL_THINKING, ENABLE_OPUS_THINKING, " "ENABLE_SONNET_THINKING, or ENABLE_HAIKU_THINKING" ) for removed_key in removed_keys: if removed_key in os.environ: return ( f"{removed_key} has been removed in this release. " f"Rename it to {replacement}." ) for env_file in _configured_env_files(model_config): if _env_file_contains_key(env_file, removed_key): return ( f"{removed_key} has been removed in this release. " f"Rename it to {replacement}. Found in {env_file}." ) return None class Settings(BaseSettings): """Application settings loaded from environment variables.""" # ==================== Messaging Platform Selection ==================== # Valid: "telegram" | "discord" | "none" messaging_platform: str = Field( default="discord", validation_alias="MESSAGING_PLATFORM" ) messaging_rate_limit: int = Field( default=1, validation_alias="MESSAGING_RATE_LIMIT" ) messaging_rate_window: float = Field( default=1.0, validation_alias="MESSAGING_RATE_WINDOW" ) # ==================== NVIDIA NIM Config ==================== nvidia_nim_api_key_qwen: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_QWEN" ) nvidia_nim_api_key_glm: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_GLM" ) nvidia_nim_api_key_stepfun: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_STEPFUN" ) nvidia_nim_api_key_seed_oss: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_SEED_OSS" ) nvidia_nim_api_key_dracarys: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_DRACARYS" ) nvidia_nim_api_key_nemotron: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_NEMOTRON" ) nvidia_nim_api_key_mistral_large: str = Field( default="", validation_alias="NVIDIA_NIM_API_KEY_MISTRAL_LARGE" ) # ==================== Zen/OpenCode Config ==================== zen_api_key: str = Field(default="", validation_alias="ZEN_API_KEY") session_retention_minutes: int = Field( default=30, validation_alias="SESSION_RETENTION_MINUTES" ) # ==================== Cerebras Config ==================== cerebras_api_key: str = Field(default="", validation_alias="CEREBRAS_API_KEY") # ==================== Silicon Flow Config ==================== silicon_api_key: str = Field(default="", validation_alias="SILICON_API_KEY") # ==================== Groq Config ==================== groq_api_key: str = Field(default="", validation_alias="GROQ_API_KEY") zen_base_url: str = Field( default="https://opencode.ai/zen", validation_alias="ZEN_BASE_URL" ) # Comma-separated list of fallback models (provider-prefixed or bare model ids) # AUTO_MODEL_PRIORITY="nvidia_nim/qwen/qwen3-coder-480b-a35b-instruct,nvidia_nim/z-ai/glm4.7,nvidia_nim/stepfun-ai/step-3.5-flash,nvidia_nim/bytedance/seed-oss-36b-instruct" nvidia_nim_fallback_models: str = Field( default="", validation_alias="NVIDIA_NIM_FALLBACK_MODELS" ) # ==================== Model ==================== # All Claude model requests are mapped to this single model (fallback) # Format: provider_type/model/name model: str = Field(default="nvidia_nim/z-ai/glm4.7", validation_alias="MODEL") # Per-model overrides (optional, falls back to MODEL) model_opus: str | None = Field(default=None, validation_alias="MODEL_OPUS") model_sonnet: str | None = Field(default=None, validation_alias="MODEL_SONNET") model_haiku: str | None = Field(default=None, validation_alias="MODEL_HAIKU") # Optional CSV list of preferred provider/model refs used by the virtual # `auto` model. Format: provider/model/name entries separated by commas. # Example: "nvidia_nim/z-ai/glm4.7,nvidia_nim/stepfun-ai/step-3.5-flash" auto_model_order: str = Field(default="", validation_alias="AUTO_MODEL_PRIORITY") # ==================== Per-Provider Proxy ==================== nvidia_nim_proxy: str = Field(default="", validation_alias="NVIDIA_NIM_PROXY") # ==================== Provider Rate Limiting ==================== provider_rate_limit: int = Field(default=40, validation_alias="PROVIDER_RATE_LIMIT") provider_rate_window: int = Field( default=60, validation_alias="PROVIDER_RATE_WINDOW" ) provider_max_concurrency: int = Field( default=5, validation_alias="PROVIDER_MAX_CONCURRENCY" ) # NIM-specific throughput tuning (leaves headroom before upstream limits) nim_rate_limit: int = Field(default=100, validation_alias="NIM_RATE_LIMIT") nim_max_concurrency: int = Field(default=40, validation_alias="NIM_MAX_CONCURRENCY") enable_model_thinking: bool = Field( default=True, validation_alias="ENABLE_MODEL_THINKING" ) enable_opus_thinking: bool | None = Field( default=None, validation_alias="ENABLE_OPUS_THINKING" ) enable_sonnet_thinking: bool | None = Field( default=None, validation_alias="ENABLE_SONNET_THINKING" ) enable_haiku_thinking: bool | None = Field( default=None, validation_alias="ENABLE_HAIKU_THINKING" ) # ==================== HTTP Client Timeouts ==================== http_read_timeout: float = Field( default=120.0, validation_alias="HTTP_READ_TIMEOUT" ) http_write_timeout: float = Field( default=10.0, validation_alias="HTTP_WRITE_TIMEOUT" ) http_connect_timeout: float = Field( default=HTTP_CONNECT_TIMEOUT_DEFAULT, validation_alias="HTTP_CONNECT_TIMEOUT", ) # ==================== Fast Prefix Detection ==================== fast_prefix_detection: bool = True # ==================== Optimizations ==================== enable_network_probe_mock: bool = True enable_title_generation_skip: bool = True enable_suggestion_mode_skip: bool = True enable_filepath_extraction_mock: bool = True # ==================== Local web server tools (web_search / web_fetch) ==================== # Off by default: these tools perform outbound HTTP from the proxy (SSRF risk). enable_web_server_tools: bool = Field( default=False, validation_alias="ENABLE_WEB_SERVER_TOOLS" ) # Comma-separated URL schemes allowed for web_fetch (default: http,https). web_fetch_allowed_schemes: str = Field( default="http,https", validation_alias="WEB_FETCH_ALLOWED_SCHEMES" ) # When true, skip private/loopback/link-local IP blocking for web_fetch (lab only). web_fetch_allow_private_networks: bool = Field( default=False, validation_alias="WEB_FETCH_ALLOW_PRIVATE_NETWORKS" ) # ==================== Debug / diagnostic logging (avoid sensitive content) ==================== # When false (default), API and SSE helpers log only metadata (counts, lengths, ids). log_raw_api_payloads: bool = Field( default=False, validation_alias="LOG_RAW_API_PAYLOADS" ) log_raw_sse_events: bool = Field( default=False, validation_alias="LOG_RAW_SSE_EVENTS" ) # When false (default), unhandled exceptions log only type + route metadata (no message/traceback). log_api_error_tracebacks: bool = Field( default=False, validation_alias="LOG_API_ERROR_TRACEBACKS" ) # When false (default), messaging logs omit text/transcription previews (metadata only). log_raw_messaging_content: bool = Field( default=False, validation_alias="LOG_RAW_MESSAGING_CONTENT" ) # When true, log full Claude CLI stderr, non-JSON lines, and parser error text. log_raw_cli_diagnostics: bool = Field( default=False, validation_alias="LOG_RAW_CLI_DIAGNOSTICS" ) # When true, log exception text / CLI error strings in messaging (may leak user content). log_messaging_error_details: bool = Field( default=False, validation_alias="LOG_MESSAGING_ERROR_DETAILS" ) debug_platform_edits: bool = Field( default=False, validation_alias="DEBUG_PLATFORM_EDITS" ) debug_subagent_stack: bool = Field( default=False, validation_alias="DEBUG_SUBAGENT_STACK" ) # ==================== NIM Settings ==================== nim: NimSettings = Field(default_factory=NimSettings) # ==================== Voice Note Transcription ==================== voice_note_enabled: bool = Field( default=True, validation_alias="VOICE_NOTE_ENABLED" ) # Device: "cpu" | "cuda" | "nvidia_nim" # - "cpu"/"cuda": local Whisper (requires voice_local extra: uv sync --extra voice_local) # - "nvidia_nim": NVIDIA NIM Whisper API (requires voice extra: uv sync --extra voice) whisper_device: str = Field(default="cpu", validation_alias="WHISPER_DEVICE") # Whisper model ID or short name (for local Whisper) or NVIDIA NIM model (for nvidia_nim) # Local Whisper: "tiny", "base", "small", "medium", "large-v2", "large-v3", "large-v3-turbo" # NVIDIA NIM: "nvidia/parakeet-ctc-1.1b-asr", "openai/whisper-large-v3", etc. whisper_model: str = Field(default="base", validation_alias="WHISPER_MODEL") # Hugging Face token for faster model downloads (optional, for local Whisper) hf_token: str = Field(default="", validation_alias="HF_TOKEN") # ==================== Bot Wrapper Config ==================== telegram_bot_token: str | None = None allowed_telegram_user_id: str | None = None discord_bot_token: str | None = Field( default=None, validation_alias="DISCORD_BOT_TOKEN" ) allowed_discord_channels: str | None = Field( default=None, validation_alias="ALLOWED_DISCORD_CHANNELS" ) claude_workspace: str = "./agent_workspace" allowed_dir: str = "" claude_cli_bin: str = Field(default="claude", validation_alias="CLAUDE_CLI_BIN") max_message_log_entries_per_chat: int | None = Field( default=None, validation_alias="MAX_MESSAGE_LOG_ENTRIES_PER_CHAT" ) # ==================== Server ==================== host: str = "0.0.0.0" port: int = 8082 log_file: str = "server.log" # Optional server API key to protect endpoints (Anthropic-style) # Set via env `ANTHROPIC_AUTH_TOKEN`. When empty, no auth is required. anthropic_auth_token: str = Field( default="", validation_alias="ANTHROPIC_AUTH_TOKEN" ) # When true, only advertise models explicitly configured via MODEL / MODEL_OPUS / MODEL_SONNET / MODEL_HAIKU # Useful if your provider exposes hundreds of models and you want the picker to show only your selected ones. advertise_only_configured_models: bool = Field( default=False, validation_alias="ADVERTISE_ONLY_CONFIGURED_MODELS" ) @model_validator(mode="before") @classmethod def reject_removed_env_vars(cls, data: Any) -> Any: """Fail fast when removed environment variables are still configured.""" if message := _removed_env_var_message(cls.model_config): raise ValueError(message) return data # Handle empty strings for optional string fields @field_validator( "telegram_bot_token", "allowed_telegram_user_id", "discord_bot_token", "allowed_discord_channels", "model_opus", "model_sonnet", "model_haiku", "enable_opus_thinking", "enable_sonnet_thinking", "enable_haiku_thinking", mode="before", ) @classmethod def parse_optional_str(cls, v: Any) -> Any: if v == "": return None return v @field_validator("max_message_log_entries_per_chat", mode="before") @classmethod def parse_optional_log_cap(cls, v: Any) -> Any: if v == "" or v is None: return None return v @field_validator("whisper_device") @classmethod def validate_whisper_device(cls, v: str) -> str: if v not in ("cpu", "cuda", "nvidia_nim"): raise ValueError( f"whisper_device must be 'cpu', 'cuda', or 'nvidia_nim', got {v!r}" ) return v @field_validator("messaging_platform") @classmethod def validate_messaging_platform(cls, v: str) -> str: if v not in ("telegram", "discord", "none"): raise ValueError( f"messaging_platform must be 'telegram', 'discord', or 'none', got {v!r}" ) return v @field_validator("messaging_rate_limit") @classmethod def validate_messaging_rate_limit(cls, v: int) -> int: if v <= 0: raise ValueError("messaging_rate_limit must be > 0") return v @field_validator("messaging_rate_window") @classmethod def validate_messaging_rate_window(cls, v: float) -> float: if v <= 0: raise ValueError("messaging_rate_window must be > 0") return float(v) @field_validator("web_fetch_allowed_schemes") @classmethod def validate_web_fetch_allowed_schemes(cls, v: str) -> str: schemes = [part.strip().lower() for part in v.split(",") if part.strip()] if not schemes: raise ValueError("web_fetch_allowed_schemes must list at least one scheme") for scheme in schemes: if not scheme.isascii() or not scheme.isalpha(): raise ValueError( f"Invalid URL scheme in web_fetch_allowed_schemes: {scheme!r}" ) return ",".join(schemes) @field_validator("model", "model_opus", "model_sonnet", "model_haiku") @classmethod def validate_model_format(cls, v: str | None) -> str | None: if v is None: return None if "/" not in v: raise ValueError( f"Model must be prefixed with provider type. " f"Valid providers: {', '.join(SUPPORTED_PROVIDER_IDS)}. " f"Format: provider_type/model/name" ) provider = v.split("/", 1)[0] if provider not in SUPPORTED_PROVIDER_IDS: supported = ", ".join(f"'{p}'" for p in SUPPORTED_PROVIDER_IDS) raise ValueError(f"Invalid provider: '{provider}'. Supported: {supported}") return v @model_validator(mode="after") def check_nvidia_nim_api_key(self) -> "Settings": if ( self.voice_note_enabled and self.whisper_device == "nvidia_nim" and not self.nvidia_nim_api_key_qwen.strip() ): raise ValueError( "NVIDIA_NIM_API_KEY_QWEN is required when WHISPER_DEVICE is 'nvidia_nim'. " "Set it in your .env file." ) return self @model_validator(mode="after") def prefer_dotenv_anthropic_auth_token(self) -> "Settings": """Let explicit .env auth config override stale shell/client tokens.""" dotenv_value = _env_file_override(self.model_config, "ANTHROPIC_AUTH_TOKEN") if dotenv_value is not None: self.anthropic_auth_token = dotenv_value return self def uses_process_anthropic_auth_token(self) -> bool: """Return whether proxy auth came from process env, not dotenv config.""" if _env_file_override(self.model_config, "ANTHROPIC_AUTH_TOKEN") is not None: return False return bool(os.environ.get("ANTHROPIC_AUTH_TOKEN")) @property def provider_type(self) -> str: """Extract provider type from the default model string.""" return Settings.parse_provider_type(self.model) @property def model_name(self) -> str: """Extract the actual model name from the default model string.""" return Settings.parse_model_name(self.model) def resolve_model(self, claude_model_name: str) -> str: """Resolve a Claude model name to the configured provider/model string. Classifies the incoming Claude model (opus/sonnet/haiku) and returns the model-specific override if configured, otherwise the fallback MODEL. """ name_lower = claude_model_name.lower() if "opus" in name_lower and self.model_opus is not None: return self.model_opus if "haiku" in name_lower and self.model_haiku is not None: return self.model_haiku if "sonnet" in name_lower and self.model_sonnet is not None: return self.model_sonnet return self.model def configured_chat_model_refs(self) -> tuple[ConfiguredChatModelRef, ...]: """Return unique configured chat provider/model refs with source env keys.""" model_refs = [m.strip() for m in (self.model or "").split(",") if m.strip()] candidates = [("MODEL", m) for m in model_refs] candidates.extend( [ ("MODEL_OPUS", self.model_opus), ("MODEL_SONNET", self.model_sonnet), ("MODEL_HAIKU", self.model_haiku), ] ) sources_by_ref: dict[str, list[str]] = {} for source, model_ref in candidates: if model_ref is None: continue sources_by_ref.setdefault(model_ref, []).append(source) return tuple( ConfiguredChatModelRef( model_ref=model_ref, provider_id=Settings.parse_provider_type(model_ref), model_id=Settings.parse_model_name(model_ref), sources=tuple(sources), ) for model_ref, sources in sources_by_ref.items() ) def resolve_thinking(self, claude_model_name: str) -> bool: """Resolve whether thinking is enabled for an incoming Claude model name.""" name_lower = claude_model_name.lower() if "opus" in name_lower and self.enable_opus_thinking is not None: return self.enable_opus_thinking if "haiku" in name_lower and self.enable_haiku_thinking is not None: return self.enable_haiku_thinking if "sonnet" in name_lower and self.enable_sonnet_thinking is not None: return self.enable_sonnet_thinking return self.enable_model_thinking def web_fetch_allowed_scheme_set(self) -> frozenset[str]: """Return normalized schemes allowed for web_fetch.""" return frozenset( part.strip().lower() for part in self.web_fetch_allowed_schemes.split(",") if part.strip() ) @staticmethod def parse_provider_type(model_string: str) -> str: """Extract provider type from any 'provider/model' string.""" return model_string.split("/", 1)[0] @staticmethod def parse_model_name(model_string: str) -> str: """Extract model name from any 'provider/model' string.""" return model_string.split("/", 1)[1] def provider_is_configured(self, provider_id: str) -> bool: """Return whether a given provider appears configured in settings. This is a heuristic check used by the `auto` resolver to prefer providers that have credentials or base URLs present. """ if provider_id == "nvidia_nim": return bool( self.nvidia_nim_api_key_qwen.strip() or self.nvidia_nim_api_key_glm.strip() or self.nvidia_nim_api_key_stepfun.strip() or self.nvidia_nim_api_key_seed_oss.strip() or self.nvidia_nim_api_key_dracarys.strip() or self.nvidia_nim_api_key_nemotron.strip() or self.nvidia_nim_api_key_mistral_large.strip() ) if provider_id == "zen": return bool(self.zen_api_key.strip()) if provider_id == "cerebras": return bool(self.cerebras_api_key.strip()) if provider_id == "silicon": return bool(self.silicon_api_key.strip()) if provider_id == "groq": return bool(self.groq_api_key.strip()) # conservative default: assume not configured return False def nvidia_nim_api_key_for_model(self, model_name: str) -> str: """Return the NVIDIA API key that should be used for a specific model id.""" model_name = model_name.strip().lower() if model_name.startswith("z-ai/glm"): return ( self.nvidia_nim_api_key_glm.strip() or self.nvidia_nim_api_key_qwen.strip() ) if model_name.startswith("stepfun-ai/step-"): return ( self.nvidia_nim_api_key_stepfun.strip() or self.nvidia_nim_api_key_qwen.strip() ) if model_name.startswith("bytedance/seed-oss"): return ( self.nvidia_nim_api_key_seed_oss.strip() or self.nvidia_nim_api_key_qwen.strip() ) if model_name.startswith("abacusai/dracarys"): return ( self.nvidia_nim_api_key_dracarys.strip() or self.nvidia_nim_api_key_qwen.strip() ) if model_name.startswith("mistralai/mistral-nemotron"): return ( self.nvidia_nim_api_key_nemotron.strip() or self.nvidia_nim_api_key_qwen.strip() ) if model_name.startswith("mistralai/mistral-large"): return ( self.nvidia_nim_api_key_mistral_large.strip() or self.nvidia_nim_api_key_qwen.strip() ) return self.nvidia_nim_api_key_qwen.strip() def resolve_api_key_for_model(self, provider_id: str, model_id: str) -> str: """Return the API key for a given provider and model.""" if provider_id == "nvidia_nim": return self.nvidia_nim_api_key_for_model(model_id) if provider_id == "zen": return self.zen_api_key.strip() return "" model_config = SettingsConfigDict( env_file=_env_files(), env_file_encoding="utf-8", extra="ignore", ) @lru_cache def get_settings() -> Settings: """Get cached settings instance.""" return Settings()