feat(config): add global user configuration for the tool

- langfuse
- default driver
- and api keys
This commit is contained in:
2025-03-11 12:47:18 -06:00
parent d42af870ff
commit dab783b01d
4 changed files with 468 additions and 13 deletions

View File

@@ -7,15 +7,19 @@ from rich.table import Table
from .config import ConfigManager
from .container import ContainerManager
from .models import SessionStatus
from .user_config import UserConfigManager
app = typer.Typer(help="Monadical Container Tool")
session_app = typer.Typer(help="Manage MC sessions")
driver_app = typer.Typer(help="Manage MC drivers", no_args_is_help=True)
config_app = typer.Typer(help="Manage MC configuration")
app.add_typer(session_app, name="session", no_args_is_help=True)
app.add_typer(driver_app, name="driver", no_args_is_help=True)
app.add_typer(config_app, name="config", no_args_is_help=True)
console = Console()
config_manager = ConfigManager()
user_config = UserConfigManager()
container_manager = ContainerManager(config_manager)
@@ -110,12 +114,16 @@ def create_session(
),
) -> None:
"""Create a new MC session"""
# Use default driver if not specified
# Use default driver from user configuration
if not driver:
driver = config_manager.config.defaults.get("driver", "goose")
driver = user_config.get(
"defaults.driver", config_manager.config.defaults.get("driver", "goose")
)
# Parse environment variables
environment = {}
# Start with environment variables from user configuration
environment = user_config.get_environment_variables()
# Override with environment variables from command line
for var in env:
if "=" in var:
key, value = var.split("=", 1)
@@ -131,7 +139,7 @@ def create_session(
project=project,
environment=environment,
session_name=name,
mount_local=not no_mount,
mount_local=not no_mount and user_config.get("defaults.mount_local", True),
)
if session:
@@ -144,8 +152,9 @@ def create_session(
for container_port, host_port in session.ports.items():
console.print(f" {container_port} -> {host_port}")
# Auto-connect unless --no-connect flag is provided
if not no_connect:
# Auto-connect based on user config, unless overridden by --no-connect flag
auto_connect = user_config.get("defaults.connect", True)
if not no_connect and auto_connect:
container_manager.connect_session(session.id)
else:
console.print(
@@ -280,6 +289,10 @@ def quick_create(
),
) -> None:
"""Create a new MC session with a project repository"""
# Use user config for defaults if not specified
if not driver:
driver = user_config.get("defaults.driver")
create_session(
driver=driver,
project=project,
@@ -398,5 +411,100 @@ def driver_info(
console.print(f.read())
# Configuration commands
@config_app.command("list")
def list_config() -> None:
"""List all configuration values"""
# Create table
table = Table(show_header=True, header_style="bold")
table.add_column("Configuration", style="cyan")
table.add_column("Value")
# Add rows from flattened config
for key, value in user_config.list_config():
table.add_row(key, str(value))
console.print(table)
@config_app.command("get")
def get_config(
key: str = typer.Argument(
..., help="Configuration key to get (e.g., langfuse.url)"
),
) -> None:
"""Get a configuration value"""
value = user_config.get(key)
if value is None:
console.print(f"[yellow]Configuration key '{key}' not found[/yellow]")
return
# Mask sensitive values
if (
any(substr in key.lower() for substr in ["key", "token", "secret", "password"])
and value
):
display_value = "*****"
else:
display_value = value
console.print(f"{key} = {display_value}")
@config_app.command("set")
def set_config(
key: str = typer.Argument(
..., help="Configuration key to set (e.g., langfuse.url)"
),
value: str = typer.Argument(..., help="Value to set"),
) -> None:
"""Set a configuration value"""
try:
# Convert string value to appropriate type
if value.lower() == "true":
typed_value = True
elif value.lower() == "false":
typed_value = False
elif value.isdigit():
typed_value = int(value)
else:
typed_value = value
user_config.set(key, typed_value)
# Mask sensitive values in output
if (
any(
substr in key.lower()
for substr in ["key", "token", "secret", "password"]
)
and value
):
display_value = "*****"
else:
display_value = typed_value
console.print(f"[green]Configuration updated: {key} = {display_value}[/green]")
except Exception as e:
console.print(f"[red]Error setting configuration: {e}[/red]")
@config_app.command("reset")
def reset_config(
confirm: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt"),
) -> None:
"""Reset configuration to defaults"""
if not confirm:
should_reset = typer.confirm(
"Are you sure you want to reset all configuration to defaults?"
)
if not should_reset:
console.print("Reset canceled")
return
user_config.reset()
console.print("[green]Configuration reset to defaults[/green]")
if __name__ == "__main__":
app()

220
mcontainer/user_config.py Normal file
View File

@@ -0,0 +1,220 @@
"""
User configuration manager for Monadical Container Tool.
"""
import os
import yaml
from pathlib import Path
from typing import Any, Dict, Optional, List, Tuple
# Define the environment variable mappings
ENV_MAPPINGS = {
"services.langfuse.url": "LANGFUSE_URL",
"services.langfuse.public_key": "LANGFUSE_INIT_PROJECT_PUBLIC_KEY",
"services.langfuse.secret_key": "LANGFUSE_INIT_PROJECT_SECRET_KEY",
"services.openai.api_key": "OPENAI_API_KEY",
"services.anthropic.api_key": "ANTHROPIC_API_KEY",
"services.openrouter.api_key": "OPENROUTER_API_KEY",
}
class UserConfigManager:
"""Manager for user-specific configuration."""
def __init__(self, config_path: Optional[str] = None):
"""Initialize the user configuration manager.
Args:
config_path: Optional path to the configuration file.
Defaults to ~/.config/mc/config.yaml.
"""
# Default to ~/.config/mc/config.yaml
self.config_path = Path(
config_path or os.path.expanduser("~/.config/mc/config.yaml")
)
self.config = self._load_config()
def _load_config(self) -> Dict[str, Any]:
"""Load configuration from file or create with defaults if it doesn't exist."""
if not self.config_path.exists():
# Create directory if it doesn't exist
self.config_path.parent.mkdir(parents=True, exist_ok=True)
# Create default config
default_config = self._get_default_config()
# Save to file
with open(self.config_path, "w") as f:
yaml.safe_dump(default_config, f)
# Set secure permissions
os.chmod(self.config_path, 0o600)
return default_config
# Load existing config
with open(self.config_path, "r") as f:
config = yaml.safe_load(f) or {}
# Merge with defaults for any missing fields
return self._merge_with_defaults(config)
def _get_default_config(self) -> Dict[str, Any]:
"""Get the default configuration."""
return {
"defaults": {
"driver": "goose",
"connect": True,
"mount_local": True,
},
"services": {
"langfuse": {},
"openai": {},
"anthropic": {},
"openrouter": {},
},
"docker": {
"network": "mc-network",
},
"ui": {
"colors": True,
"verbose": False,
},
}
def _merge_with_defaults(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Merge user config with defaults for missing values."""
defaults = self._get_default_config()
# Deep merge of config with defaults
def _deep_merge(source, destination):
for key, value in source.items():
if key not in destination:
destination[key] = value
elif isinstance(value, dict) and isinstance(destination[key], dict):
_deep_merge(value, destination[key])
return destination
return _deep_merge(defaults, config)
def get(self, key_path: str, default: Any = None) -> Any:
"""Get a configuration value by dot-notation path.
Args:
key_path: The configuration path (e.g., "defaults.driver")
default: The default value to return if not found
Returns:
The configuration value or default if not found
"""
# Handle shorthand service paths (e.g., "langfuse.url")
if (
"." in key_path
and not key_path.startswith("services.")
and not any(
key_path.startswith(section + ".")
for section in ["defaults", "docker", "remote", "ui"]
)
):
service, setting = key_path.split(".", 1)
key_path = f"services.{service}.{setting}"
parts = key_path.split(".")
result = self.config
for part in parts:
if part not in result:
return default
result = result[part]
return result
def set(self, key_path: str, value: Any) -> None:
"""Set a configuration value by dot-notation path.
Args:
key_path: The configuration path (e.g., "defaults.driver")
value: The value to set
"""
# Handle shorthand service paths (e.g., "langfuse.url")
if (
"." in key_path
and not key_path.startswith("services.")
and not any(
key_path.startswith(section + ".")
for section in ["defaults", "docker", "remote", "ui"]
)
):
service, setting = key_path.split(".", 1)
key_path = f"services.{service}.{setting}"
parts = key_path.split(".")
config = self.config
# Navigate to the containing dictionary
for part in parts[:-1]:
if part not in config:
config[part] = {}
config = config[part]
# Set the value
config[parts[-1]] = value
self.save()
def save(self) -> None:
"""Save the configuration to file."""
with open(self.config_path, "w") as f:
yaml.safe_dump(self.config, f)
def reset(self) -> None:
"""Reset the configuration to defaults."""
self.config = self._get_default_config()
self.save()
def get_environment_variables(self) -> Dict[str, str]:
"""Get environment variables from the configuration.
Returns:
A dictionary of environment variables to set in the container.
"""
env_vars = {}
# Process the service configurations and map to environment variables
for config_path, env_var in ENV_MAPPINGS.items():
value = self.get(config_path)
if value:
# Handle environment variable references
if (
isinstance(value, str)
and value.startswith("${")
and value.endswith("}")
):
env_var_name = value[2:-1]
value = os.environ.get(env_var_name, "")
env_vars[env_var] = str(value)
return env_vars
def list_config(self) -> List[Tuple[str, Any]]:
"""List all configuration values as flattened key-value pairs.
Returns:
A list of (key, value) tuples with flattened key paths.
"""
result = []
def _flatten_dict(d, prefix=""):
for key, value in d.items():
full_key = f"{prefix}.{key}" if prefix else key
if isinstance(value, dict):
_flatten_dict(value, full_key)
else:
# Mask sensitive values
if any(
substr in full_key.lower()
for substr in ["key", "token", "secret", "password"]
):
displayed_value = "*****" if value else value
else:
displayed_value = value
result.append((full_key, displayed_value))
_flatten_dict(self.config)
return sorted(result)