""" Aetheel Configuration ===================== Loads configuration from ~/.aetheel/config.json with .env fallback for secrets. Config hierarchy (highest priority wins): 1. CLI arguments (--model, --claude, etc.) 2. Environment variables 3. ~/.aetheel/config.json 4. Defaults Secrets (tokens, passwords) should stay in .env — they are never written to the config file. Everything else goes in config.json. Usage: from config import load_config, AetheelConfig cfg = load_config() print(cfg.runtime.model) print(cfg.slack.bot_token) # from .env """ import json import logging import os from dataclasses import dataclass, field from pathlib import Path logger = logging.getLogger("aetheel.config") CONFIG_DIR = os.path.expanduser("~/.aetheel") CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json") # --------------------------------------------------------------------------- # Config Dataclasses # --------------------------------------------------------------------------- @dataclass class RuntimeConfig: """AI runtime configuration.""" engine: str = "opencode" # "opencode" or "claude" mode: str = "cli" # "cli" or "sdk" model: str | None = None # e.g. "anthropic/claude-sonnet-4-20250514" provider: str | None = None timeout_seconds: int = 120 server_url: str = "http://localhost:4096" workspace: str | None = None format: str = "json" @dataclass class ClaudeConfig: """Claude Code runtime configuration.""" model: str | None = None timeout_seconds: int = 120 max_turns: int = 3 no_tools: bool = False allowed_tools: list[str] = field(default_factory=lambda: [ "Bash", "Read", "Write", "Edit", "Glob", "Grep", "WebSearch", "WebFetch", "Task", "TaskOutput", "TaskStop", "Skill", "TeamCreate", "TeamDelete", "SendMessage", ]) @dataclass class SlackConfig: """Slack adapter configuration. Tokens come from .env.""" enabled: bool = True # auto-enabled when tokens present bot_token: str = "" # from .env: SLACK_BOT_TOKEN app_token: str = "" # from .env: SLACK_APP_TOKEN @dataclass class TelegramConfig: """Telegram adapter configuration.""" enabled: bool = False # enable in config or set token bot_token: str = "" # from .env: TELEGRAM_BOT_TOKEN @dataclass class DiscordConfig: """Discord adapter configuration.""" enabled: bool = False # enable in config or set token bot_token: str = "" # from .env: DISCORD_BOT_TOKEN listen_channels: list[str] = field(default_factory=list) @dataclass class MemoryConfig: """Memory system configuration.""" workspace: str = "~/.aetheel/workspace" db_path: str = "~/.aetheel/memory.db" @dataclass class SchedulerConfig: """Scheduler configuration.""" db_path: str = "~/.aetheel/scheduler.db" @dataclass class MCPServerConfig: """Single MCP server entry.""" command: str = "" args: list[str] = field(default_factory=list) env: dict[str, str] = field(default_factory=dict) @dataclass class MCPConfig: """MCP server configuration.""" servers: dict[str, MCPServerConfig] = field(default_factory=dict) @dataclass class HeartbeatConfig: """Heartbeat / proactive system configuration.""" enabled: bool = True default_channel: str = "slack" default_channel_id: str = "" silent: bool = False @dataclass class WebChatConfig: """WebChat adapter configuration.""" enabled: bool = False port: int = 8080 host: str = "127.0.0.1" @dataclass class HooksConfig: """Lifecycle hooks configuration.""" enabled: bool = True @dataclass class WebhookConfig: """Webhook receiver configuration.""" enabled: bool = False port: int = 8090 host: str = "127.0.0.1" token: str = "" # Bearer token for auth @dataclass class AetheelConfig: """Top-level configuration for Aetheel.""" log_level: str = "INFO" runtime: RuntimeConfig = field(default_factory=RuntimeConfig) claude: ClaudeConfig = field(default_factory=ClaudeConfig) slack: SlackConfig = field(default_factory=SlackConfig) telegram: TelegramConfig = field(default_factory=TelegramConfig) discord: DiscordConfig = field(default_factory=DiscordConfig) memory: MemoryConfig = field(default_factory=MemoryConfig) scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) heartbeat: HeartbeatConfig = field(default_factory=HeartbeatConfig) webchat: WebChatConfig = field(default_factory=WebChatConfig) mcp: MCPConfig = field(default_factory=MCPConfig) hooks: HooksConfig = field(default_factory=HooksConfig) webhooks: WebhookConfig = field(default_factory=WebhookConfig) # --------------------------------------------------------------------------- # Loader # --------------------------------------------------------------------------- def _deep_get(data: dict, *keys, default=None): """Safely traverse nested dicts.""" for key in keys: if not isinstance(data, dict): return default data = data.get(key, default) return data def load_config() -> AetheelConfig: """ Load configuration from config.json + environment variables. Config file provides non-secret settings. Environment variables (from .env or shell) provide secrets and can override any setting. """ cfg = AetheelConfig() file_data: dict = {} # 1. Load config.json if it exists if os.path.isfile(CONFIG_PATH): try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: file_data = json.load(f) logger.info(f"Config loaded from {CONFIG_PATH}") except (json.JSONDecodeError, OSError) as e: logger.warning(f"Failed to load {CONFIG_PATH}: {e}") # 2. Apply config.json values cfg.log_level = _deep_get(file_data, "log_level", default=cfg.log_level) # Runtime rt = file_data.get("runtime", {}) cfg.runtime.engine = rt.get("engine", cfg.runtime.engine) cfg.runtime.mode = rt.get("mode", cfg.runtime.mode) cfg.runtime.model = rt.get("model", cfg.runtime.model) cfg.runtime.provider = rt.get("provider", cfg.runtime.provider) cfg.runtime.timeout_seconds = rt.get("timeout_seconds", cfg.runtime.timeout_seconds) cfg.runtime.server_url = rt.get("server_url", cfg.runtime.server_url) cfg.runtime.workspace = rt.get("workspace", cfg.runtime.workspace) cfg.runtime.format = rt.get("format", cfg.runtime.format) # Claude cl = file_data.get("claude", {}) cfg.claude.model = cl.get("model", cfg.claude.model) cfg.claude.timeout_seconds = cl.get("timeout_seconds", cfg.claude.timeout_seconds) cfg.claude.max_turns = cl.get("max_turns", cfg.claude.max_turns) cfg.claude.no_tools = cl.get("no_tools", cfg.claude.no_tools) # Discord (non-secret settings from config) dc = file_data.get("discord", {}) cfg.discord.enabled = dc.get("enabled", cfg.discord.enabled) cfg.discord.listen_channels = dc.get("listen_channels", cfg.discord.listen_channels) # Slack (enabled flag from config) sl = file_data.get("slack", {}) cfg.slack.enabled = sl.get("enabled", cfg.slack.enabled) # Telegram (enabled flag from config) tg = file_data.get("telegram", {}) cfg.telegram.enabled = tg.get("enabled", cfg.telegram.enabled) # Memory mem = file_data.get("memory", {}) cfg.memory.workspace = mem.get("workspace", cfg.memory.workspace) cfg.memory.db_path = mem.get("db_path", cfg.memory.db_path) # Scheduler sched = file_data.get("scheduler", {}) cfg.scheduler.db_path = sched.get("db_path", cfg.scheduler.db_path) # Claude allowed_tools cl_allowed = cl.get("allowed_tools") if cl_allowed is not None: cfg.claude.allowed_tools = cl_allowed # Heartbeat hb = file_data.get("heartbeat", {}) cfg.heartbeat.enabled = hb.get("enabled", cfg.heartbeat.enabled) cfg.heartbeat.default_channel = hb.get("default_channel", cfg.heartbeat.default_channel) cfg.heartbeat.default_channel_id = hb.get("default_channel_id", cfg.heartbeat.default_channel_id) cfg.heartbeat.silent = hb.get("silent", cfg.heartbeat.silent) # WebChat wc = file_data.get("webchat", {}) cfg.webchat.enabled = wc.get("enabled", cfg.webchat.enabled) cfg.webchat.port = wc.get("port", cfg.webchat.port) cfg.webchat.host = wc.get("host", cfg.webchat.host) # MCP mcp_data = file_data.get("mcp", {}) servers_data = mcp_data.get("servers", {}) for name, srv in servers_data.items(): cfg.mcp.servers[name] = MCPServerConfig( command=srv.get("command", ""), args=srv.get("args", []), env=srv.get("env", {}), ) # Hooks hk = file_data.get("hooks", {}) cfg.hooks.enabled = hk.get("enabled", cfg.hooks.enabled) # Webhooks wh = file_data.get("webhooks", {}) cfg.webhooks.enabled = wh.get("enabled", cfg.webhooks.enabled) cfg.webhooks.port = wh.get("port", cfg.webhooks.port) cfg.webhooks.host = wh.get("host", cfg.webhooks.host) cfg.webhooks.token = wh.get("token", cfg.webhooks.token) # 3. Environment variables override everything (secrets + overrides) cfg.log_level = os.environ.get("LOG_LEVEL", cfg.log_level) cfg.runtime.engine = os.environ.get("AETHEEL_ENGINE", cfg.runtime.engine) cfg.runtime.mode = os.environ.get("OPENCODE_MODE", cfg.runtime.mode) cfg.runtime.model = os.environ.get("OPENCODE_MODEL") or cfg.runtime.model cfg.runtime.provider = os.environ.get("OPENCODE_PROVIDER") or cfg.runtime.provider cfg.runtime.timeout_seconds = int( os.environ.get("OPENCODE_TIMEOUT", str(cfg.runtime.timeout_seconds)) ) cfg.runtime.server_url = os.environ.get("OPENCODE_SERVER_URL", cfg.runtime.server_url) cfg.runtime.workspace = ( os.environ.get("OPENCODE_WORKSPACE") or os.environ.get("AETHEEL_WORKSPACE") or cfg.runtime.workspace ) cfg.claude.model = os.environ.get("CLAUDE_MODEL") or cfg.claude.model cfg.claude.timeout_seconds = int( os.environ.get("CLAUDE_TIMEOUT", str(cfg.claude.timeout_seconds)) ) cfg.claude.max_turns = int( os.environ.get("CLAUDE_MAX_TURNS", str(cfg.claude.max_turns)) ) cfg.claude.no_tools = os.environ.get( "CLAUDE_NO_TOOLS", str(cfg.claude.no_tools) ).lower() == "true" # Secrets from .env only cfg.slack.bot_token = os.environ.get("SLACK_BOT_TOKEN", cfg.slack.bot_token) cfg.slack.app_token = os.environ.get("SLACK_APP_TOKEN", cfg.slack.app_token) cfg.telegram.bot_token = os.environ.get("TELEGRAM_BOT_TOKEN", cfg.telegram.bot_token) cfg.discord.bot_token = os.environ.get("DISCORD_BOT_TOKEN", cfg.discord.bot_token) # Discord listen channels: env overrides config env_channels = os.environ.get("DISCORD_LISTEN_CHANNELS", "").strip() if env_channels: cfg.discord.listen_channels = [ch.strip() for ch in env_channels.split(",") if ch.strip()] cfg.memory.workspace = os.environ.get("AETHEEL_WORKSPACE", cfg.memory.workspace) cfg.memory.db_path = os.environ.get("AETHEEL_MEMORY_DB", cfg.memory.db_path) return cfg def save_default_config() -> str: """ Write a default config.json if one doesn't exist. Returns the path to the config file. """ os.makedirs(CONFIG_DIR, exist_ok=True) if os.path.isfile(CONFIG_PATH): return CONFIG_PATH default = { "$schema": "Aetheel configuration — edit this file, keep secrets in .env", "log_level": "INFO", "runtime": { "engine": "opencode", "mode": "cli", "model": None, "timeout_seconds": 120, "server_url": "http://localhost:4096", "format": "json", }, "claude": { "model": None, "timeout_seconds": 120, "max_turns": 3, "no_tools": False, "allowed_tools": [ "Bash", "Read", "Write", "Edit", "Glob", "Grep", "WebSearch", "WebFetch", "Task", "TaskOutput", "TaskStop", "Skill", "TeamCreate", "TeamDelete", "SendMessage", ], }, "slack": { "enabled": True, }, "telegram": { "enabled": False, }, "discord": { "enabled": False, "listen_channels": [], }, "memory": { "workspace": "~/.aetheel/workspace", "db_path": "~/.aetheel/memory.db", }, "scheduler": { "db_path": "~/.aetheel/scheduler.db", }, "heartbeat": { "enabled": True, "default_channel": "slack", "default_channel_id": "", "silent": False, }, "webchat": { "enabled": False, "port": 8080, "host": "127.0.0.1", }, "mcp": { "servers": {}, }, "hooks": { "enabled": True, }, "webhooks": { "enabled": False, "port": 8090, "host": "127.0.0.1", "token": "", }, } with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(default, f, indent=2) logger.info(f"Default config written to {CONFIG_PATH}") return CONFIG_PATH def write_mcp_config(mcp_config: MCPConfig, workspace_dir: str, use_claude: bool) -> None: """Write MCP server config to the appropriate file for the runtime. Produces ``.mcp.json`` (Claude Code) or ``opencode.json`` (OpenCode) in *workspace_dir*. Skips writing when no servers are configured. """ if not mcp_config.servers: return config_data = { "mcpServers" if use_claude else "mcp": { name: {"command": s.command, "args": s.args, "env": s.env} for name, s in mcp_config.servers.items() } } filename = ".mcp.json" if use_claude else "opencode.json" path = os.path.join(os.path.expanduser(workspace_dir), filename) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(config_data, f, indent=2) logger.info(f"MCP config written to {path}")