selfhostyourtech/cubbi
2
0
Fork 0
mirror of https://github.com/Monadical-SAS/cubbi.git synced 2025-12-20 12:19:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: comprehensive configuration system and environment variable forwarding (#29)

Browse Source 
* feat: migrate container configuration from env vars to YAML config files

- Replace environment variable-based configuration with structured YAML config files
- Add Pydantic models for type-safe configuration management in cubbi_init.py
- Update container.py to generate /cubbi/config.yaml and mount into containers
- Simplify goose plugin to extract provider from default model format
- Remove complex environment variable handling in favor of direct config access
- Maintain backward compatibility while enabling cleaner plugin architecture

* feat: optimize goose plugin to only pass required API key for selected model

- Update goose plugin to set only the API key for the provider of the selected model
- Add selective API key configuration for anthropic, openai, google, and openrouter
- Update README.md with comprehensive automated testing documentation
- Add litellm/gpt-oss:120b to test.sh model matrix (now 5 images × 4 models = 20 tests)
- Include single prompt command syntax for each tool in the documentation

* feat: add comprehensive integration tests with pytest parametrization

- Create tests/test_integration.py with parametrized tests for 5 images × 4 models (20 combinations)
- Add pytest configuration to exclude integration tests by default
- Add integration marker for selective test running
- Include help command tests and image availability tests
- Document test usage in tests/README_integration.md

Integration tests cover:
- goose, aider, claudecode, opencode, crush images
- anthropic/claude-sonnet-4-20250514, openai/gpt-4o, openrouter/openai/gpt-4o, litellm/gpt-oss:120b models
- Proper command syntax for each tool
- Success validation with exit codes and completion markers

Usage:
- pytest (regular tests only)
- pytest -m integration (integration tests only)
- pytest -m integration -k "goose" (specific image)

* feat: update OpenCode plugin with perfect multi-provider configuration

- Add global STANDARD_PROVIDERS constant for maintainability
- Support custom providers (with baseURL) vs standard providers
- Custom providers: include npm package, name, baseURL, apiKey, models
- Standard providers: include only apiKey and empty models
- Use direct API key values from cubbi config instead of env vars
- Only add default model to the provider that matches the default model
- Use @ai-sdk/openai-compatible for OpenAI-compatible providers
- Preserve model names without transformation
- All providers get required empty models{} section per OpenCode spec

This ensures OpenCode can properly recognize and use both native
providers (anthropic, openai, google, openrouter) and custom
providers (litellm, etc.) with correct configuration format.

* refactor: model is now a combination of provider/model

* feat: add separate integration test for Claude Code without model config

Claude Code is Anthropic-specific and doesn't require model selection like other tools.
Created dedicated test that verifies basic functionality without model preselection.

* feat: update Claude Code and Crush plugins to use new config system

- Claude Code plugin now uses cubbi_config.providers to get Anthropic API key
- Crush plugin updated to use cubbi_config.providers for provider configuration
- Both plugins maintain backwards compatibility with environment variables
- Consistent plugin structure across all cubbi images

* feat: add environments_to_forward support for images

- Add environments_to_forward field to ImageConfig and Image models
- Update container creation logic to forward specified environment variables from host
- Add environments_to_forward to claudecode cubbi_image.yaml to ensure Anthropic API key is always available
- Claude Code now gets required environment variables regardless of model selection
- This ensures Claude Code works properly even when other models are specified

Fixes the issue where Claude Code couldn't access Anthropic API key when using different model configurations.

* refactor: remove unused environment field from cubbi_image.yaml files

The 'environment' field was loaded but never processed at runtime.
Only 'environments_to_forward' is actually used to pass environment
variables from host to container.

Cleaned up configuration files by removing:
- 72 lines from aider/cubbi_image.yaml
- 42 lines from claudecode/cubbi_image.yaml
- 28 lines from crush/cubbi_image.yaml
- 16 lines from goose/cubbi_image.yaml
- Empty environment: [] from opencode/cubbi_image.yaml

This makes the configuration files cleaner and only contains
fields that are actually used by the system.

* feat: implement environment variable forwarding for aider

Updates aider to automatically receive all relevant environment variables
from the host, similar to how opencode works.

Changes:
- Added environments_to_forward field to aider/cubbi_image.yaml with
  comprehensive list of API keys, configuration, and proxy variables
- Updated aider_plugin.py to use cubbi_config system for provider/model setup
- Environment variables now forwarded automatically during container creation
- Maintains backward compatibility with legacy environment variables

Environment variables forwarded:
- API Keys: OPENAI_API_KEY, ANTHROPIC_API_KEY, DEEPSEEK_API_KEY, etc.
- Configuration: AIDER_MODEL, GIT_* variables, HTTP_PROXY, etc.
- Timezone: TZ for proper log timestamps

Tested: All aider tests pass, environment variables confirmed forwarded.

* refactor: remove unused volumes and init fields from cubbi_image.yaml files

Both 'volumes' and 'init' fields were loaded but never processed at runtime.
These were incomplete implementations that didn't affect container behavior.

Removed from all 5 images:
- volumes: List with mountPath: /app (incomplete, missing host paths)
- init: pre_command and command fields (unused during container creation)

The cubbi_image.yaml files now only contain fields that are actually used:
- Basic metadata (name, description, version, maintainer, image)
- persistent_configs (working functionality)
- environments_to_forward (working functionality where present)

This makes the configuration files cleaner and eliminates confusion
about what functionality is actually implemented.

* refactor: remove unused ImageInit and VolumeMount models

These models were only referenced in the Image model definition but
never used at runtime since we removed all init: and volumes: fields
from cubbi_image.yaml files.

Removed:
- VolumeMount class (mountPath, description fields)
- ImageInit class (pre_command, command fields)
- init: Optional[ImageInit] field from Image model
- volumes: List[VolumeMount] field from Image model

The Image model now only contains fields that are actually used:
- Basic metadata (name, description, version, maintainer, image)
- environment (loaded but unused - kept for future cleanup)
- persistent_configs (working functionality)
- environments_to_forward (working functionality)

This makes the data model cleaner and eliminates dead code.

* feat: add interactive configuration command

Adds `cubbi configure` command for interactive setup of LLM providers
and models through a user-friendly questionnaire interface.

New features:
- Interactive provider configuration (OpenAI, Anthropic, OpenRouter, etc.)
- API key management with environment variable references
- Model selection with provider/model format validation
- Default settings configuration (image, ports, volumes, etc.)
- Added questionary dependency for interactive prompts

Changes:
- Added cubbi/configure.py with full interactive configuration logic
- Added configure command to cubbi/cli.py
- Updated uv.lock with questionary and prompt-toolkit dependencies

Usage: `cubbi configure`

* refactor: update integration tests for current functionality

Updates integration tests to reflect current cubbi functionality:

test_integration.py:
- Simplified image list (removed crush temporarily)
- Updated model list with current supported models
- Removed outdated help command tests that were timing out
- Simplified claudecode test to basic functionality test
- Updated command templates for current tool versions

test_integration_docker.py:
- Cleaned up container management tests
- Fixed formatting and improved readability
- Updated assertion formatting for better error messages

These changes align the tests with the current state of the codebase
and remove tests that were causing timeouts or failures.

* fix: fix temporary file chmod
This commit is contained in:
Mathieu Virbel
2025-08-06 21:27:26 -06:00
committed by GitHub
parent e4c64a54ed
commit bae951cf7c
23 changed files with 2741 additions and 826 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
README.md
View File

@@ -144,13 +144,37 @@ Cubbi includes an image management system that allows you to build, manage, and
**Supported Images**
| Image Name | Langtrace Support |
|------------|-------------------|
| goose | yes |
| opencode | no |
| claudecode | no |
| aider | no |
| crush | no |
| Image Name | Langtrace Support | Single Prompt Command |
|------------|-------------------|----------------------|
| goose | yes | `goose run -t 'prompt' --no-session --quiet` |
| opencode | no | `opencode run -m MODEL 'prompt'` |
| claudecode | no | `claude -p 'prompt'` |
| aider | no | `aider --message 'prompt' --yes-always --no-fancy-input` |
| crush | no | `crush run 'prompt'` |
**Automated Testing:**
Each image can be tested with single prompt commands using different models:
```bash
# Test a single image with a specific model
cubbix -i goose -m anthropic/claude-sonnet-4-20250514 --no-connect --no-shell --run "goose run -t 'What is 2+2?' --no-session --quiet"
# Test aider with non-interactive flags
cubbix -i aider -m openai/gpt-4o --no-connect --no-shell --run "aider --message 'What is 2+2?' --yes-always --no-fancy-input --no-check-update"
# Test claude-code (note: binary name is 'claude', not 'claude-code')
cubbix -i claudecode -m anthropic/claude-sonnet-4-20250514 --no-connect --no-shell --run "claude -p 'What is 2+2?'"
# Test opencode with model specification
cubbix -i opencode -m anthropic/claude-sonnet-4-20250514 --no-connect --no-shell --run "opencode run -m anthropic/claude-sonnet-4-20250514 'What is 2+2?'"
# Test crush
cubbix -i crush -m anthropic/claude-sonnet-4-20250514 --no-connect --no-shell --run "crush run 'What is 2+2?'"
# Run comprehensive test suite (requires test.sh script)
./test.sh # Tests all images with multiple models: anthropic/claude-sonnet-4-20250514, openai/gpt-4o, openrouter/openai/gpt-4o, litellm/gpt-oss:120b
```
```bash
# List available images

23
cubbi/cli.py
View File

@@ -14,6 +14,7 @@ from rich.console import Console
from rich.table import Table
from .config import ConfigManager
from .configure import run_interactive_config
from .container import ContainerManager
from .mcp import MCPManager
from .models import SessionStatus
@@ -60,6 +61,12 @@ def main(
logging.getLogger().setLevel(logging.INFO)
@app.command()
def configure() -> None:
"""Interactive configuration of LLM providers and models"""
run_interactive_config()
@app.command()
def version() -> None:
"""Show Cubbi version information"""
@@ -173,9 +180,11 @@ def create_session(
gid: Optional[int] = typer.Option(
None, "--gid", help="Group ID to run the container as (defaults to host user)"
),
model: Optional[str] = typer.Option(None, "--model", help="Model to use"),
provider: Optional[str] = typer.Option(
None, "--provider", "-p", help="Provider to use"
model: Optional[str] = typer.Option(
None,
"--model",
"-m",
help="Model to use in 'provider/model' format (e.g., 'anthropic/claude-3-5-sonnet')",
),
ssh: bool = typer.Option(False, "--ssh", help="Start SSH server in the container"),
config: List[str] = typer.Option(
@@ -387,15 +396,10 @@ def create_session(
"[yellow]Warning: --no-shell is ignored without --run[/yellow]"
)
# Use model and provider from config overrides if not explicitly provided
# Use model from config overrides if not explicitly provided
final_model = (
model if model is not None else temp_user_config.get("defaults.model")
)
final_provider = (
provider
if provider is not None
else temp_user_config.get("defaults.provider")
)
session = container_manager.create_session(
image_name=image_name,
@@ -414,7 +418,6 @@ def create_session(
gid=target_gid,
ssh=ssh,
model=final_model,
provider=final_provider,
domains=all_domains,
)

908
cubbi/configure.py Normal file
View File

@@ -0,0 +1,908 @@
"""
Interactive configuration tool for Cubbi providers and models.
"""
import os
import docker
import questionary
from rich.console import Console
from .user_config import UserConfigManager
console = Console()
class ProviderConfigurator:
"""Interactive configuration for LLM providers."""
def __init__(self, user_config: UserConfigManager):
self.user_config = user_config
# Initialize Docker client for network autocomplete
try:
self.docker_client = docker.from_env()
self.docker_client.ping() # Test connection
except Exception:
self.docker_client = None
def run(self) -> None:
"""Run the interactive configuration tool."""
console.print("\nCubbi Configuration\n")
while True:
# Get current default model for display
current_default = self.user_config.get("defaults.model", "Not set")
choice = questionary.select(
"What would you like to configure?",
choices=[
"Configure providers",
f"Set default model ({current_default})",
"Configure MCP servers",
"Configure networks",
"Configure volumes",
"Configure ports",
"View current configuration",
"Exit",
],
).ask()
if choice == "Configure providers":
self._configure_providers()
elif choice and choice.startswith("Set default model"):
self._set_default_model()
elif choice == "Configure MCP servers":
self._configure_mcps()
elif choice == "Configure networks":
self._configure_networks()
elif choice == "Configure volumes":
self._configure_volumes()
elif choice == "Configure ports":
self._configure_ports()
elif choice == "View current configuration":
self._show_current_config()
elif choice == "Exit" or choice is None:
console.print("\n[green]Configuration complete![/green]")
break
def _configure_providers(self) -> None:
"""Configure LLM providers."""
while True:
providers = self.user_config.list_providers()
choices = []
# Add existing providers
for name, config in providers.items():
provider_type = config.get("type", "unknown")
choices.append(f"{name} ({provider_type})")
# Add separator and options
if choices:
choices.append("---")
choices.extend(["Add new provider", "Back to main menu"])
choice = questionary.select(
"Select a provider to configure:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu":
break
elif choice == "Add new provider":
self._add_new_provider()
else:
# Extract provider name from the choice
# Format: "provider_name (provider_type)"
provider_name = choice.split(" (")[0]
self._edit_provider(provider_name)
def _add_new_provider(self) -> None:
"""Add a new provider configuration."""
# Ask for provider type
provider_type = questionary.select(
"Select provider type:",
choices=[
"Anthropic",
"OpenAI",
"Google",
"OpenRouter",
"OpenAI-compatible (custom)",
],
).ask()
if provider_type is None:
return
# Map display names to internal types
type_mapping = {
"Anthropic": "anthropic",
"OpenAI": "openai",
"Google": "google",
"OpenRouter": "openrouter",
"Other (openai compatible)": "openai",
}
internal_type = type_mapping[provider_type]
# Ask for provider name
if provider_type == "OpenAI-compatible (custom)":
provider_name = questionary.text(
"Enter a name for this provider (e.g., 'litellm', 'local-llm'):",
validate=lambda name: len(name.strip()) > 0
or "Please enter a provider name",
).ask()
else:
# Use standard name but allow customization
standard_name = internal_type
provider_name = questionary.text(
"Provider name:",
default=standard_name,
validate=lambda name: len(name.strip()) > 0
or "Please enter a provider name",
).ask()
if provider_name is None:
return
provider_name = provider_name.strip()
# Check if provider already exists
if self.user_config.get_provider(provider_name):
console.print(
f"[yellow]Provider '{provider_name}' already exists![/yellow]"
)
return
# Ask for API key configuration
api_key_choice = questionary.select(
"How would you like to provide the API key?",
choices=[
"Enter API key directly (saved in config)",
"Reference environment variable (recommended)",
],
).ask()
if api_key_choice is None:
return
if "environment variable" in api_key_choice:
env_var = questionary.text(
"Environment variable name:",
default=f"{provider_name.upper().replace('-', '_')}_API_KEY",
validate=lambda var: len(var.strip()) > 0
or "Please enter a variable name",
).ask()
if env_var is None:
return
api_key = f"${{{env_var.strip()}}}"
# Check if the environment variable exists
if not os.environ.get(env_var.strip()):
console.print(
f"[yellow]Warning: Environment variable '{env_var}' is not currently set[/yellow]"
)
else:
api_key = questionary.password(
"Enter API key:",
validate=lambda key: len(key.strip()) > 0 or "Please enter an API key",
).ask()
if api_key is None:
return
base_url = None
if internal_type == "openai" and provider_type == "OpenAI-compatible (custom)":
base_url = questionary.text(
"Base URL for API calls:",
validate=lambda url: url.startswith("http")
or "Please enter a valid URL starting with http",
).ask()
if base_url is None:
return
# Add the provider
self.user_config.add_provider(
name=provider_name,
provider_type=internal_type,
api_key=api_key,
base_url=base_url,
)
console.print(f"[green]Added provider '{provider_name}'[/green]")
def _edit_provider(self, provider_name: str) -> None:
"""Edit an existing provider."""
provider_config = self.user_config.get_provider(provider_name)
if not provider_config:
console.print(f"[red]Provider '{provider_name}' not found![/red]")
return
choices = ["View configuration", "Remove provider", "---", "Back"]
choice = questionary.select(
f"What would you like to do with '{provider_name}'?",
choices=choices,
).ask()
if choice == "View configuration":
console.print(f"\n[bold]Configuration for '{provider_name}':[/bold]")
for key, value in provider_config.items():
if key == "api_key" and not value.startswith("${"):
# Mask direct API keys
display_value = (
f"{'*' * (len(value) - 4)}{value[-4:]}"
if len(value) > 4
else "****"
)
else:
display_value = value
console.print(f" {key}: {display_value}")
console.print()
elif choice == "Remove provider":
confirm = questionary.confirm(
f"Are you sure you want to remove provider '{provider_name}'?"
).ask()
if confirm:
self.user_config.remove_provider(provider_name)
console.print(f"[green]Removed provider '{provider_name}'[/green]")
def _set_default_model(self) -> None:
"""Set the default model."""
providers = self.user_config.list_providers()
if not providers:
console.print(
"[yellow]No providers configured. Please add providers first.[/yellow]"
)
return
# Create choices in provider/model format
choices = []
for provider_name, provider_config in providers.items():
provider_type = provider_config.get("type", "unknown")
has_key = bool(provider_config.get("api_key"))
if has_key:
choices.append(f"{provider_name} ({provider_type})")
if not choices:
console.print("[yellow]No providers with API keys configured.[/yellow]")
return
# Add separator and cancel option
choices.append("---")
choices.append("Back to main menu")
choice = questionary.select(
"Select a provider for the default model:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu" or choice == "---":
return
# Extract provider name
provider_name = choice.split(" (")[0]
# Ask for model name
model_name = questionary.text(
f"Enter model name for {provider_name} (e.g., 'claude-3-5-sonnet', 'gpt-4', 'llama3:70b'):",
validate=lambda name: len(name.strip()) > 0 or "Please enter a model name",
).ask()
if model_name is None:
return
# Set the default model in provider/model format
default_model = f"{provider_name}/{model_name.strip()}"
self.user_config.set("defaults.model", default_model)
console.print(f"[green]Set default model to '{default_model}'[/green]")
def _show_current_config(self) -> None:
"""Show current configuration."""
console.print()
# Show default model
default_model = self.user_config.get("defaults.model", "Not set")
console.print(f"Default model: [cyan]{default_model}[/cyan]")
# Show providers
console.print("\n[bold]Providers[/bold]")
providers = self.user_config.list_providers()
if providers:
for name in providers.keys():
console.print(f" - {name}")
else:
console.print(" (no providers configured)")
# Show MCP servers
console.print("\n[bold]MCP Servers[/bold]")
mcp_configs = self.user_config.list_mcp_configurations()
default_mcps = self.user_config.list_mcps()
if mcp_configs:
for mcp_config in mcp_configs:
name = mcp_config.get("name", "unknown")
is_default = " (default)" if name in default_mcps else ""
console.print(f" - {name}{is_default}")
else:
console.print(" (no MCP servers configured)")
# Show networks
console.print("\n[bold]Networks[/bold]")
networks = self.user_config.list_networks()
if networks:
for network in networks:
console.print(f" - {network}")
else:
console.print(" (no networks configured)")
# Show volumes
console.print("\n[bold]Volumes[/bold]")
volumes = self.user_config.list_volumes()
if volumes:
for volume in volumes:
console.print(f" - {volume}")
else:
console.print(" (no volumes configured)")
# Show ports
console.print("\n[bold]Ports[/bold]")
ports = self.user_config.list_ports()
if ports:
for port in sorted(ports):
console.print(f" - {port}")
else:
console.print(" (no ports configured)")
console.print()
def _get_docker_networks(self):
"""Get list of existing Docker networks for autocomplete."""
if not self.docker_client:
return []
try:
networks = self.docker_client.networks.list()
return [network.name for network in networks if network.name != "none"]
except Exception:
return []
def _configure_mcps(self) -> None:
"""Configure MCP servers."""
while True:
mcp_configs = self.user_config.list_mcp_configurations()
default_mcps = self.user_config.list_mcps()
choices = []
if mcp_configs:
for mcp_config in mcp_configs:
name = mcp_config.get("name", "unknown")
mcp_type = mcp_config.get("type", "unknown")
is_default = " ⭐" if name in default_mcps else ""
choices.append(f"{name} ({mcp_type}){is_default}")
choices.append("---")
choices.extend(["Add MCP server", "---", "Back to main menu"])
choice = questionary.select(
"Select an MCP server to configure:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu" or choice == "---":
break
elif choice == "Add MCP server":
self._add_mcp_server()
else:
# Extract MCP name from choice (format: "name (type)⭐")
mcp_name = choice.split(" (")[0]
self._edit_mcp_server(mcp_name)
def _add_mcp_server(self) -> None:
"""Add a new MCP server."""
# Ask for MCP type first
mcp_type = questionary.select(
"Select MCP server type:",
choices=[
"Remote MCP (URL-based)",
"Docker MCP (containerized)",
"Proxy MCP (proxy + base image)",
],
).ask()
if mcp_type is None:
return
if "Remote MCP" in mcp_type:
self._add_remote_mcp()
elif "Docker MCP" in mcp_type:
self._add_docker_mcp()
elif "Proxy MCP" in mcp_type:
self._add_proxy_mcp()
def _add_remote_mcp(self) -> None:
"""Add a remote MCP server."""
name = questionary.text(
"Enter MCP server name:",
validate=lambda n: len(n.strip()) > 0 or "Please enter a name",
).ask()
if name is None:
return
url = questionary.text(
"Enter server URL:",
validate=lambda u: u.startswith("http")
or "Please enter a valid URL starting with http",
).ask()
if url is None:
return
# Ask for optional headers
add_headers = questionary.confirm("Add custom headers?").ask()
headers = {}
if add_headers:
while True:
header_name = questionary.text("Header name (empty to finish):").ask()
if not header_name or not header_name.strip():
break
header_value = questionary.text(f"Value for {header_name}:").ask()
if header_value:
headers[header_name.strip()] = header_value.strip()
mcp_config = {
"name": name.strip(),
"type": "remote",
"url": url.strip(),
"headers": headers,
}
self.user_config.add_mcp_configuration(mcp_config)
# Ask if it should be a default
make_default = questionary.confirm(f"Add '{name}' to default MCPs?").ask()
if make_default:
self.user_config.add_mcp(name.strip())
console.print(f"[green]Added remote MCP server '{name}'[/green]")
def _add_docker_mcp(self) -> None:
"""Add a Docker MCP server."""
name = questionary.text(
"Enter MCP server name:",
validate=lambda n: len(n.strip()) > 0 or "Please enter a name",
).ask()
if name is None:
return
image = questionary.text(
"Enter Docker image:",
validate=lambda i: len(i.strip()) > 0 or "Please enter an image",
).ask()
if image is None:
return
command = questionary.text(
"Enter command to run (optional):",
).ask()
# Ask for environment variables
add_env = questionary.confirm("Add environment variables?").ask()
env = {}
if add_env:
while True:
env_name = questionary.text(
"Environment variable name (empty to finish):"
).ask()
if not env_name or not env_name.strip():
break
env_value = questionary.text(f"Value for {env_name}:").ask()
if env_value:
env[env_name.strip()] = env_value.strip()
mcp_config = {
"name": name.strip(),
"type": "docker",
"image": image.strip(),
"command": command.strip() if command else "",
"env": env,
}
self.user_config.add_mcp_configuration(mcp_config)
# Ask if it should be a default
make_default = questionary.confirm(f"Add '{name}' to default MCPs?").ask()
if make_default:
self.user_config.add_mcp(name.strip())
console.print(f"[green]Added Docker MCP server '{name}'[/green]")
def _add_proxy_mcp(self) -> None:
"""Add a Proxy MCP server."""
name = questionary.text(
"Enter MCP server name:",
validate=lambda n: len(n.strip()) > 0 or "Please enter a name",
).ask()
if name is None:
return
base_image = questionary.text(
"Enter base Docker image (the actual MCP server):",
validate=lambda i: len(i.strip()) > 0 or "Please enter a base image",
).ask()
if base_image is None:
return
proxy_image = questionary.text(
"Enter proxy Docker image:",
default="mcp-proxy",
).ask()
if proxy_image is None:
return
command = questionary.text(
"Enter command to run in base image (optional):",
).ask()
host_port = questionary.text(
"Enter host port (optional, will auto-assign if empty):",
validate=lambda p: not p.strip()
or (p.strip().isdigit() and 1 <= int(p.strip()) <= 65535)
or "Please enter a valid port number (1-65535) or leave empty",
).ask()
# Ask for environment variables
add_env = questionary.confirm("Add environment variables?").ask()
env = {}
if add_env:
while True:
env_name = questionary.text(
"Environment variable name (empty to finish):"
).ask()
if not env_name or not env_name.strip():
break
env_value = questionary.text(f"Value for {env_name}:").ask()
if env_value:
env[env_name.strip()] = env_value.strip()
mcp_config = {
"name": name.strip(),
"type": "proxy",
"base_image": base_image.strip(),
"proxy_image": proxy_image.strip(),
"command": command.strip() if command else "",
"proxy_options": {
"sse_port": 8080,
"sse_host": "0.0.0.0",
"allow_origin": "*",
},
"env": env,
}
if host_port and host_port.strip():
mcp_config["host_port"] = int(host_port.strip())
self.user_config.add_mcp_configuration(mcp_config)
# Ask if it should be a default
make_default = questionary.confirm(f"Add '{name}' to default MCPs?").ask()
if make_default:
self.user_config.add_mcp(name.strip())
console.print(f"[green]Added Proxy MCP server '{name}'[/green]")
def _edit_mcp_server(self, server_name: str) -> None:
"""Edit an existing MCP server."""
mcp_config = self.user_config.get_mcp_configuration(server_name)
if not mcp_config:
console.print(f"[red]MCP server '{server_name}' not found![/red]")
return
is_default = server_name in self.user_config.list_mcps()
choices = [
"View configuration",
f"{'Remove from' if is_default else 'Add to'} defaults",
"Remove server",
"---",
"Back",
]
choice = questionary.select(
f"What would you like to do with MCP server '{server_name}'?",
choices=choices,
).ask()
if choice == "View configuration":
console.print("\n[bold]MCP server configuration:[/bold]")
for key, value in mcp_config.items():
if isinstance(value, dict) and value:
console.print(f" {key}:")
for sub_key, sub_value in value.items():
console.print(f" {sub_key}: {sub_value}")
elif isinstance(value, list) and value:
console.print(f" {key}: {', '.join(map(str, value))}")
else:
console.print(f" {key}: {value}")
console.print()
elif "defaults" in choice:
if is_default:
self.user_config.remove_mcp(server_name)
console.print(
f"[green]Removed '{server_name}' from default MCPs[/green]"
)
else:
self.user_config.add_mcp(server_name)
console.print(f"[green]Added '{server_name}' to default MCPs[/green]")
elif choice == "Remove server":
confirm = questionary.confirm(
f"Are you sure you want to remove MCP server '{server_name}'?"
).ask()
if confirm:
if self.user_config.remove_mcp_configuration(server_name):
console.print(f"[green]Removed MCP server '{server_name}'[/green]")
else:
console.print(
f"[red]Failed to remove MCP server '{server_name}'[/red]"
)
def _configure_networks(self) -> None:
"""Configure default networks."""
while True:
networks = self.user_config.list_networks()
choices = []
if networks:
for network in networks:
choices.append(f"{network}")
choices.append("---")
choices.extend(["Add network", "---", "Back to main menu"])
choice = questionary.select(
"Select a network to configure:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu" or choice == "---":
break
elif choice == "Add network":
self._add_network()
else:
# Edit network
self._edit_network(choice)
def _add_network(self) -> None:
"""Add a new network."""
# Get existing Docker networks for autocomplete
docker_networks = self._get_docker_networks()
if docker_networks:
network_name = questionary.autocomplete(
"Enter Docker network name:",
choices=docker_networks,
validate=lambda name: len(name.strip()) > 0
or "Please enter a network name",
).ask()
else:
# Fallback to text input if Docker is not available
network_name = questionary.text(
"Enter Docker network name:",
validate=lambda name: len(name.strip()) > 0
or "Please enter a network name",
).ask()
if network_name is None:
return
network_name = network_name.strip()
self.user_config.add_network(network_name)
console.print(f"[green]Added network '{network_name}'[/green]")
def _edit_network(self, network_name: str) -> None:
"""Edit an existing network."""
choices = ["View configuration", "Remove network", "---", "Back"]
choice = questionary.select(
f"What would you like to do with network '{network_name}'?",
choices=choices,
).ask()
if choice == "View configuration":
console.print("\n[bold]Network configuration:[/bold]")
console.print(f" Name: {network_name}")
console.print()
elif choice == "Remove network":
confirm = questionary.confirm(
f"Are you sure you want to remove network '{network_name}'?"
).ask()
if confirm:
self.user_config.remove_network(network_name)
console.print(f"[green]Removed network '{network_name}'[/green]")
def _configure_volumes(self) -> None:
"""Configure default volume mappings."""
while True:
volumes = self.user_config.list_volumes()
choices = []
if volumes:
for volume in volumes:
choices.append(f"{volume}")
choices.append("---")
choices.extend(["Add volume mapping", "---", "Back to main menu"])
choice = questionary.select(
"Select a volume to configure:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu" or choice == "---":
break
elif choice == "Add volume mapping":
self._add_volume()
else:
# Edit volume
self._edit_volume(choice)
def _add_volume(self) -> None:
"""Add a new volume mapping."""
# Ask for source directory
source = questionary.path(
"Enter source directory path:",
validate=lambda path: len(path.strip()) > 0 or "Please enter a source path",
).ask()
if source is None:
return
# Ask for destination directory
destination = questionary.text(
"Enter destination path in container:",
validate=lambda path: len(path.strip()) > 0
or "Please enter a destination path",
).ask()
if destination is None:
return
# Create the volume mapping
volume_mapping = f"{source.strip()}:{destination.strip()}"
self.user_config.add_volume(volume_mapping)
console.print(f"[green]Added volume mapping '{volume_mapping}'[/green]")
def _edit_volume(self, volume_mapping: str) -> None:
"""Edit an existing volume mapping."""
choices = ["View configuration", "Remove volume", "---", "Back"]
choice = questionary.select(
f"What would you like to do with volume '{volume_mapping}'?",
choices=choices,
).ask()
if choice == "View configuration":
console.print("\n[bold]Volume mapping configuration:[/bold]")
if ":" in volume_mapping:
source, destination = volume_mapping.split(":", 1)
console.print(f" Source: {source}")
console.print(f" Destination: {destination}")
else:
console.print(f" Mapping: {volume_mapping}")
console.print()
elif choice == "Remove volume":
confirm = questionary.confirm(
f"Are you sure you want to remove volume mapping '{volume_mapping}'?"
).ask()
if confirm:
self.user_config.remove_volume(volume_mapping)
console.print(
f"[green]Removed volume mapping '{volume_mapping}'[/green]"
)
def _configure_ports(self) -> None:
"""Configure default port forwards."""
while True:
ports = self.user_config.list_ports()
choices = []
if ports:
for port in sorted(ports):
choices.append(f"{port}")
choices.append("---")
choices.extend(["Add port", "---", "Back to main menu"])
choice = questionary.select(
"Select a port to configure:",
choices=choices,
).ask()
if choice is None or choice == "Back to main menu" or choice == "---":
break
elif choice == "Add port":
self._add_port()
else:
# Edit port
try:
port_num = int(choice)
self._edit_port(port_num)
except ValueError:
pass
def _add_port(self) -> None:
"""Add a new port forward."""
def validate_port(value: str) -> bool:
try:
port = int(value.strip())
return 1 <= port <= 65535
except ValueError:
return False
port_str = questionary.text(
"Enter port number (1-65535):",
validate=lambda p: validate_port(p)
or "Please enter a valid port number (1-65535)",
).ask()
if port_str is None:
return
port_num = int(port_str.strip())
self.user_config.add_port(port_num)
console.print(f"[green]Added port {port_num}[/green]")
def _edit_port(self, port_num: int) -> None:
"""Edit an existing port forward."""
choices = ["Remove port", "---", "Back"]
choice = questionary.select(
f"What would you like to do with port {port_num}?",
choices=choices,
).ask()
if choice == "Remove port":
confirm = questionary.confirm(
f"Are you sure you want to remove port {port_num}?"
).ask()
if confirm:
self.user_config.remove_port(port_num)
console.print(f"[green]Removed port {port_num}[/green]")
def run_interactive_config() -> None:
"""Entry point for the interactive configuration tool."""
user_config = UserConfigManager()
configurator = ProviderConfigurator(user_config)
configurator.run()
if __name__ == "__main__":
run_interactive_config()

269
cubbi/container.py
View File

@@ -4,10 +4,13 @@ import logging
import os
import pathlib
import sys
import tempfile
import uuid
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import docker
import yaml
from docker.errors import DockerException, ImageNotFound
from .config import ConfigManager
@@ -85,6 +88,87 @@ class ContainerManager:
# This ensures we don't mount the /cubbi-config volume for project-less sessions
return None
def _generate_container_config(
self,
image_name: str,
project_url: Optional[str] = None,
uid: Optional[int] = None,
gid: Optional[int] = None,
model: Optional[str] = None,
ssh: bool = False,
run_command: Optional[str] = None,
no_shell: bool = False,
mcp_list: Optional[List[str]] = None,
persistent_links: Optional[List[Dict[str, str]]] = None,
) -> Path:
"""Generate container configuration YAML file"""
providers = {}
for name, provider in self.user_config_manager.list_providers().items():
api_key = provider.get("api_key", "")
if api_key.startswith("${") and api_key.endswith("}"):
env_var = api_key[2:-1]
api_key = os.environ.get(env_var, "")
provider_config = {
"type": provider.get("type"),
"api_key": api_key,
}
if provider.get("base_url"):
provider_config["base_url"] = provider.get("base_url")
providers[name] = provider_config
mcps = []
if mcp_list:
for mcp_name in mcp_list:
mcp_config = self.mcp_manager.get_mcp(mcp_name)
if mcp_config:
mcps.append(mcp_config)
config = {
"version": "1.0",
"user": {"uid": uid or 1000, "gid": gid or 1000},
"providers": providers,
"mcps": mcps,
"project": {
"config_dir": "/cubbi-config",
"image_config_dir": f"/cubbi-config/{image_name}",
},
"ssh": {"enabled": ssh},
}
if project_url:
config["project"]["url"] = project_url
if persistent_links:
config["persistent_links"] = persistent_links
if model:
config["defaults"] = {"model": model}
if run_command:
config["run_command"] = run_command
config["no_shell"] = no_shell
config_file = Path(tempfile.mkdtemp()) / "config.yaml"
with open(config_file, "w") as f:
yaml.dump(config, f)
# Set restrictive permissions (0o600 = read/write for owner only)
config_file.chmod(0o600)
# Set ownership to cubbi user if uid/gid are provided
if uid is not None and gid is not None:
try:
os.chown(config_file, uid, gid)
except (OSError, PermissionError):
# If we can't chown (e.g., running as non-root), just log and continue
logger.warning(f"Could not set ownership of config file to {uid}:{gid}")
return config_file
def list_sessions(self) -> List[Session]:
"""List all active Cubbi sessions"""
sessions = []
@@ -161,7 +245,6 @@ class ContainerManager:
uid: Optional[int] = None,
gid: Optional[int] = None,
model: Optional[str] = None,
provider: Optional[str] = None,
ssh: bool = False,
domains: Optional[List[str]] = None,
) -> Optional[Session]:
@@ -181,8 +264,8 @@ class ContainerManager:
mcp: Optional list of MCP server names to attach to the session
uid: Optional user ID for the container process
gid: Optional group ID for the container process
model: Optional model to use
provider: Optional provider to use
model: Optional model specification in 'provider/model' format (e.g., 'anthropic/claude-3-5-sonnet')
Legacy separate model and provider parameters are also supported for backward compatibility
ssh: Whether to start the SSH server in the container (default: False)
domains: Optional list of domains to restrict network access to (uses network-filter)
"""
@@ -213,32 +296,22 @@ class ContainerManager:
# Ensure network exists
self._ensure_network()
# Prepare environment variables
# Minimal environment variables
env_vars = environment or {}
env_vars["CUBBI_CONFIG_FILE"] = "/cubbi/config.yaml"
# Add CUBBI_USER_ID and CUBBI_GROUP_ID for entrypoint script
env_vars["CUBBI_USER_ID"] = str(uid) if uid is not None else "1000"
env_vars["CUBBI_GROUP_ID"] = str(gid) if gid is not None else "1000"
# Set SSH environment variable
env_vars["CUBBI_SSH_ENABLED"] = "true" if ssh else "false"
# Pass some environment from host environment to container for local development
keys = [
"OPENAI_API_KEY",
"OPENAI_URL",
"ANTHROPIC_API_KEY",
"ANTHROPIC_AUTH_TOKEN",
"ANTHROPIC_CUSTOM_HEADERS",
"OPENROUTER_API_KEY",
"GOOGLE_API_KEY",
"LANGFUSE_INIT_PROJECT_PUBLIC_KEY",
"LANGFUSE_INIT_PROJECT_SECRET_KEY",
"LANGFUSE_URL",
]
for key in keys:
if key in os.environ and key not in env_vars:
env_vars[key] = os.environ[key]
# Forward specified environment variables from the host to the container
if (
hasattr(image, "environments_to_forward")
and image.environments_to_forward
):
for env_name in image.environments_to_forward:
env_value = os.environ.get(env_name)
if env_value is not None:
env_vars[env_name] = env_value
print(
f"Forwarding environment variable {env_name} to container"
)
# Pull image if needed
try:
@@ -294,6 +367,7 @@ class ContainerManager:
print(f"Mounting volume: {host_path} -> {container_path}")
# Set up persistent project configuration if project_name is provided
persistent_links = []
project_config_path = self._get_project_config_path(project, project_name)
if project_config_path:
print(f"Using project configuration directory: {project_config_path}")
@@ -304,13 +378,8 @@ class ContainerManager:
"mode": "rw",
}
# Add environment variables for config path
env_vars["CUBBI_CONFIG_DIR"] = "/cubbi-config"
env_vars["CUBBI_IMAGE_CONFIG_DIR"] = f"/cubbi-config/{image_name}"
# Create image-specific config directories and set up direct volume mounts
# Create image-specific config directories and collect persistent links
if image.persistent_configs:
persistent_links_data = [] # To store "source:target" pairs for symlinks
print("Setting up persistent configuration directories:")
for config in image.persistent_configs:
# Get target directory path on host
@@ -327,24 +396,19 @@ class ContainerManager:
# For files, make sure parent directory exists
elif config.type == "file":
target_dir.parent.mkdir(parents=True, exist_ok=True)
# File will be created by the container if needed
# Store the source and target paths for the init script
# Note: config.target is the path *within* /cubbi-config
persistent_links_data.append(f"{config.source}:{config.target}")
# Store persistent link data for config file
persistent_links.append(
{
"source": config.source,
"target": config.target,
"type": config.type,
}
)
print(
f" - Prepared host path {target_dir} for symlink target {config.target}"
)
# Set up persistent links
if persistent_links_data:
env_vars["CUBBI_PERSISTENT_LINKS"] = ";".join(
persistent_links_data
)
print(
f"Setting CUBBI_PERSISTENT_LINKS={env_vars['CUBBI_PERSISTENT_LINKS']}"
)
else:
print(
"No project_name provided - skipping configuration directory setup."
@@ -394,43 +458,6 @@ class ContainerManager:
# Get MCP status to extract endpoint information
mcp_status = self.mcp_manager.get_mcp_status(mcp_name)
# Add MCP environment variables with index
idx = len(mcp_names) - 1 # 0-based index for the current MCP
if mcp_config.get("type") == "remote":
# For remote MCP, set the URL and headers
env_vars[f"MCP_{idx}_URL"] = mcp_config.get("url")
if mcp_config.get("headers"):
# Serialize headers as JSON
import json
env_vars[f"MCP_{idx}_HEADERS"] = json.dumps(
mcp_config.get("headers")
)
else:
# For Docker/proxy MCP, set the connection details
# Use both the container name and the short name for internal Docker DNS resolution
container_name = self.mcp_manager.get_mcp_container_name(
mcp_name
)
# Use the short name (mcp_name) as the primary hostname
env_vars[f"MCP_{idx}_HOST"] = mcp_name
# Default port is 8080 unless specified in status
port = next(
iter(mcp_status.get("ports", {}).values()), 8080
)
env_vars[f"MCP_{idx}_PORT"] = str(port)
# Use the short name in the URL to take advantage of the network alias
env_vars[f"MCP_{idx}_URL"] = f"http://{mcp_name}:{port}/sse"
# For backward compatibility, also set the full container name URL
env_vars[f"MCP_{idx}_CONTAINER_URL"] = (
f"http://{container_name}:{port}/sse"
)
# Set type-specific information
env_vars[f"MCP_{idx}_TYPE"] = mcp_config.get("type")
env_vars[f"MCP_{idx}_NAME"] = mcp_name
except Exception as e:
print(f"Warning: Failed to start MCP server '{mcp_name}': {e}")
# Get the container name before trying to remove it from the list
@@ -445,30 +472,8 @@ class ContainerManager:
pass
elif mcp_config.get("type") == "remote":
# For remote MCP, just set environment variables
idx = len(mcp_names) - 1 # 0-based index for the current MCP
env_vars[f"MCP_{idx}_URL"] = mcp_config.get("url")
if mcp_config.get("headers"):
# Serialize headers as JSON
import json
env_vars[f"MCP_{idx}_HEADERS"] = json.dumps(
mcp_config.get("headers")
)
# Set type-specific information
env_vars[f"MCP_{idx}_TYPE"] = mcp_config.get("mcp_type", "sse")
env_vars[f"MCP_{idx}_NAME"] = mcp_name
# Set environment variables for MCP count if we have any
if mcp_names:
env_vars["MCP_COUNT"] = str(len(mcp_names))
env_vars["MCP_ENABLED"] = "true"
# Serialize all MCP names as JSON
import json
env_vars["MCP_NAMES"] = json.dumps(mcp_names)
# Remote MCP - nothing to do here, config will handle it
pass
# Add user-specified networks
# Default Cubbi network
@@ -499,39 +504,18 @@ class ContainerManager:
target_shell = "/bin/bash"
if run_command:
# Set environment variable for cubbi-init.sh to pick up
env_vars["CUBBI_RUN_COMMAND"] = run_command
# If no_shell is true, set CUBBI_NO_SHELL environment variable
if no_shell:
env_vars["CUBBI_NO_SHELL"] = "true"
logger.info(
"Setting CUBBI_NO_SHELL=true, container will exit after run command"
)
# Set the container's command to be the final shell (or exit if no_shell is true)
container_command = [target_shell]
logger.info(
f"Setting CUBBI_RUN_COMMAND and targeting shell {target_shell}"
)
logger.info(f"Using run command with shell {target_shell}")
if no_shell:
logger.info("Container will exit after run command")
else:
# Use default behavior (often defined by image's ENTRYPOINT/CMD)
# Set the container's command to be the final shell if none specified by Dockerfile CMD
# Note: Dockerfile CMD is ["tail", "-f", "/dev/null"], so this might need adjustment
# if we want interactive shell by default without --run. Let's default to bash for now.
container_command = [target_shell]
logger.info(
"Using default container entrypoint/command for interactive shell."
)
# Set default model/provider from user config if not explicitly provided
env_vars["CUBBI_MODEL"] = model or self.user_config_manager.get(
"defaults.model", ""
)
env_vars["CUBBI_PROVIDER"] = provider or self.user_config_manager.get(
"defaults.provider", ""
)
# Handle network-filter if domains are specified
network_filter_container = None
network_mode = None
@@ -615,6 +599,29 @@ class ContainerManager:
"[yellow]Warning: MCP servers may not be accessible when using domain restrictions.[/yellow]"
)
# Generate configuration file
project_url = project if is_git_repo else None
config_file_path = self._generate_container_config(
image_name=image_name,
project_url=project_url,
uid=uid,
gid=gid,
model=model,
ssh=ssh,
run_command=run_command,
no_shell=no_shell,
mcp_list=mcp_names,
persistent_links=persistent_links
if "persistent_links" in locals()
else None,
)
# Mount config file
session_volumes[str(config_file_path)] = {
"bind": "/cubbi/config.yaml",
"mode": "ro",
}
# Create container
container_params = {
"image": image.image,

128
cubbi/images/aider/aider_plugin.py
View File

@@ -1,32 +1,22 @@
#!/usr/bin/env python3
"""
Aider Plugin for Cubbi
Handles authentication setup and configuration for Aider AI pair programming
"""
import os
import stat
from pathlib import Path
from typing import Any, Dict
from typing import Dict
from cubbi_init import ToolPlugin
from cubbi_init import ToolPlugin, cubbi_config
class AiderPlugin(ToolPlugin):
"""Plugin for setting up Aider authentication and configuration"""
@property
def tool_name(self) -> str:
return "aider"
def _get_user_ids(self) -> tuple[int, int]:
"""Get the cubbi user and group IDs from environment"""
user_id = int(os.environ.get("CUBBI_USER_ID", "1000"))
group_id = int(os.environ.get("CUBBI_GROUP_ID", "1000"))
return user_id, group_id
return cubbi_config.user.uid, cubbi_config.user.gid
def _set_ownership(self, path: Path) -> None:
"""Set ownership of a path to the cubbi user"""
user_id, group_id = self._get_user_ids()
try:
os.chown(path, user_id, group_id)
@@ -34,15 +24,12 @@ class AiderPlugin(ToolPlugin):
self.status.log(f"Failed to set ownership for {path}: {e}", "WARNING")
def _get_aider_config_dir(self) -> Path:
"""Get the Aider configuration directory"""
return Path("/home/cubbi/.aider")
def _get_aider_cache_dir(self) -> Path:
"""Get the Aider cache directory"""
return Path("/home/cubbi/.cache/aider")
def _ensure_aider_dirs(self) -> tuple[Path, Path]:
"""Ensure Aider directories exist with correct ownership"""
config_dir = self._get_aider_config_dir()
cache_dir = self._get_aider_cache_dir()
@@ -59,7 +46,6 @@ class AiderPlugin(ToolPlugin):
return config_dir, cache_dir
def initialize(self) -> bool:
"""Initialize Aider configuration"""
self.status.log("Setting up Aider configuration...")
# Ensure Aider directories exist
@@ -89,31 +75,82 @@ class AiderPlugin(ToolPlugin):
return True
def _create_environment_config(self) -> Dict[str, str]:
"""Create environment variable configuration for Aider"""
env_vars = {}
# Map environment variables to Aider configuration
api_key_mappings = {
"OPENAI_API_KEY": "OPENAI_API_KEY",
"ANTHROPIC_API_KEY": "ANTHROPIC_API_KEY",
"DEEPSEEK_API_KEY": "DEEPSEEK_API_KEY",
"GEMINI_API_KEY": "GEMINI_API_KEY",
"OPENROUTER_API_KEY": "OPENROUTER_API_KEY",
}
# Configure Aider with the default model from cubbi config
provider_config = cubbi_config.get_provider_for_default_model()
if provider_config and cubbi_config.defaults.model:
_, model_name = cubbi_config.defaults.model.split("/", 1)
# Check for OpenAI API base URL
openai_url = os.environ.get("OPENAI_URL")
if openai_url:
env_vars["OPENAI_API_BASE"] = openai_url
self.status.log(f"Set OpenAI API base URL to {openai_url}")
# Set the model for Aider
env_vars["AIDER_MODEL"] = model_name
self.status.log(f"Set Aider model to {model_name}")
# Check for standard API keys
for env_var, aider_var in api_key_mappings.items():
value = os.environ.get(env_var)
if value:
env_vars[aider_var] = value
provider = env_var.replace("_API_KEY", "").lower()
self.status.log(f"Added {provider} API key")
# Set provider-specific API key and configuration
if provider_config.type == "anthropic":
env_vars["AIDER_ANTHROPIC_API_KEY"] = provider_config.api_key
self.status.log("Configured Anthropic API key for Aider")
elif provider_config.type == "openai":
env_vars["AIDER_OPENAI_API_KEY"] = provider_config.api_key
if provider_config.base_url:
env_vars["AIDER_OPENAI_API_BASE"] = provider_config.base_url
self.status.log(
f"Set Aider OpenAI API base to {provider_config.base_url}"
)
self.status.log("Configured OpenAI API key for Aider")
# Note: Aider uses different environment variable names for some providers
# We map cubbi provider types to Aider's expected variable names
elif provider_config.type == "google":
# Aider may expect GEMINI_API_KEY for Google models
env_vars["GEMINI_API_KEY"] = provider_config.api_key
self.status.log("Configured Google/Gemini API key for Aider")
elif provider_config.type == "openrouter":
env_vars["OPENROUTER_API_KEY"] = provider_config.api_key
self.status.log("Configured OpenRouter API key for Aider")
else:
self.status.log(
f"Provider type '{provider_config.type}' not directly supported by Aider plugin",
"WARNING",
)
else:
self.status.log(
"No default model or provider configured - checking legacy environment variables",
"WARNING",
)
# Fallback to legacy environment variable checking for backward compatibility
api_key_mappings = {
"OPENAI_API_KEY": "AIDER_OPENAI_API_KEY",
"ANTHROPIC_API_KEY": "AIDER_ANTHROPIC_API_KEY",
"DEEPSEEK_API_KEY": "DEEPSEEK_API_KEY",
"GEMINI_API_KEY": "GEMINI_API_KEY",
"OPENROUTER_API_KEY": "OPENROUTER_API_KEY",
}
for env_var, aider_var in api_key_mappings.items():
value = os.environ.get(env_var)
if value:
env_vars[aider_var] = value
provider = env_var.replace("_API_KEY", "").lower()
self.status.log(f"Added {provider} API key from environment")
# Check for OpenAI API base URL from legacy environment
openai_url = os.environ.get("OPENAI_URL")
if openai_url:
env_vars["AIDER_OPENAI_API_BASE"] = openai_url
self.status.log(
f"Set OpenAI API base URL to {openai_url} from environment"
)
# Legacy model configuration
model = os.environ.get("AIDER_MODEL")
if model:
env_vars["AIDER_MODEL"] = model
self.status.log(f"Set model to {model} from environment")
# Handle additional API keys from AIDER_API_KEYS
additional_keys = os.environ.get("AIDER_API_KEYS")
@@ -129,12 +166,6 @@ class AiderPlugin(ToolPlugin):
except Exception as e:
self.status.log(f"Failed to parse AIDER_API_KEYS: {e}", "WARNING")
# Add model configuration
model = os.environ.get("AIDER_MODEL")
if model:
env_vars["AIDER_MODEL"] = model
self.status.log(f"Set default model to {model}")
# Add git configuration
auto_commits = os.environ.get("AIDER_AUTO_COMMITS", "true")
if auto_commits.lower() in ["true", "false"]:
@@ -155,7 +186,6 @@ class AiderPlugin(ToolPlugin):
return env_vars
def _write_env_file(self, env_file: Path, env_vars: Dict[str, str]) -> bool:
"""Write environment variables to .env file"""
try:
content = "\n".join(f"{key}={value}" for key, value in env_vars.items())
@@ -174,19 +204,17 @@ class AiderPlugin(ToolPlugin):
return False
def setup_tool_configuration(self) -> bool:
"""Set up Aider configuration - called by base class"""
# Additional tool configuration can be added here if needed
return True
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate Aider with available MCP servers if applicable"""
if mcp_config["count"] == 0:
def integrate_mcp_servers(self) -> bool:
if not cubbi_config.mcps:
self.status.log("No MCP servers to integrate")
return True
# Aider doesn't have native MCP support like Claude Code,
# but we could potentially add custom integrations here
self.status.log(
f"Found {mcp_config['count']} MCP server(s) - no direct integration available"
f"Found {len(cubbi_config.mcps)} MCP server(s) - no direct integration available for Aider"
)
return True

93
cubbi/images/aider/cubbi_image.yaml
View File

@@ -3,75 +3,40 @@ description: Aider AI pair programming environment
version: 1.0.0
maintainer: team@monadical.com
image: monadical/cubbi-aider:latest
init:
pre_command: /cubbi-init.sh
command: /entrypoint.sh
environment:
# OpenAI Configuration
- name: OPENAI_API_KEY
description: OpenAI API key for GPT models
required: false
sensitive: true
# Anthropic Configuration
- name: ANTHROPIC_API_KEY
description: Anthropic API key for Claude models
required: false
sensitive: true
# DeepSeek Configuration
- name: DEEPSEEK_API_KEY
description: DeepSeek API key for DeepSeek models
required: false
sensitive: true
# Gemini Configuration
- name: GEMINI_API_KEY
description: Google Gemini API key
required: false
sensitive: true
# OpenRouter Configuration
- name: OPENROUTER_API_KEY
description: OpenRouter API key for various models
required: false
sensitive: true
# Generic provider API keys
- name: AIDER_API_KEYS
description: Additional API keys in format "provider1=key1,provider2=key2"
required: false
sensitive: true
persistent_configs: []
environments_to_forward:
# API Keys
- OPENAI_API_KEY
- ANTHROPIC_API_KEY
- ANTHROPIC_AUTH_TOKEN
- ANTHROPIC_CUSTOM_HEADERS
- DEEPSEEK_API_KEY
- GEMINI_API_KEY
- OPENROUTER_API_KEY
- AIDER_API_KEYS
# Model Configuration
- name: AIDER_MODEL
description: Default model to use (e.g., sonnet, o3-mini, deepseek)
required: false
- AIDER_MODEL
- CUBBI_MODEL
- CUBBI_PROVIDER
# Git Configuration
- name: AIDER_AUTO_COMMITS
description: Enable automatic commits (true/false)
required: false
default: "true"
- name: AIDER_DARK_MODE
description: Enable dark mode (true/false)
required: false
default: "false"
- AIDER_AUTO_COMMITS
- AIDER_DARK_MODE
- GIT_AUTHOR_NAME
- GIT_AUTHOR_EMAIL
- GIT_COMMITTER_NAME
- GIT_COMMITTER_EMAIL
# Proxy Configuration
- name: HTTP_PROXY
description: HTTP proxy server URL
required: false
- HTTP_PROXY
- HTTPS_PROXY
- NO_PROXY
- name: HTTPS_PROXY
description: HTTPS proxy server URL
required: false
# OpenAI Configuration
- OPENAI_URL
- OPENAI_API_BASE
- AIDER_OPENAI_API_BASE
volumes:
- mountPath: /app
description: Application directory
persistent_configs: []
# Timezone (useful for logs and timestamps)
- TZ

73
cubbi/images/claudecode/claudecode_plugin.py
View File

@@ -1,49 +1,23 @@
#!/usr/bin/env python3
"""
Claude Code Plugin for Cubbi
Handles authentication setup and configuration for Claude Code
"""
import json
import os
import stat
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Dict, Optional
from cubbi_init import ToolPlugin
# API key mappings from environment variables to Claude Code configuration
API_KEY_MAPPINGS = {
"ANTHROPIC_API_KEY": "api_key",
"ANTHROPIC_AUTH_TOKEN": "auth_token",
"ANTHROPIC_CUSTOM_HEADERS": "custom_headers",
}
# Enterprise integration environment variables
ENTERPRISE_MAPPINGS = {
"CLAUDE_CODE_USE_BEDROCK": "use_bedrock",
"CLAUDE_CODE_USE_VERTEX": "use_vertex",
"HTTP_PROXY": "http_proxy",
"HTTPS_PROXY": "https_proxy",
"DISABLE_TELEMETRY": "disable_telemetry",
}
from cubbi_init import ToolPlugin, cubbi_config
class ClaudeCodePlugin(ToolPlugin):
"""Plugin for setting up Claude Code authentication and configuration"""
@property
def tool_name(self) -> str:
return "claudecode"
def _get_user_ids(self) -> tuple[int, int]:
"""Get the cubbi user and group IDs from environment"""
user_id = int(os.environ.get("CUBBI_USER_ID", "1000"))
group_id = int(os.environ.get("CUBBI_GROUP_ID", "1000"))
return user_id, group_id
return cubbi_config.user.uid, cubbi_config.user.gid
def _set_ownership(self, path: Path) -> None:
"""Set ownership of a path to the cubbi user"""
user_id, group_id = self._get_user_ids()
try:
os.chown(path, user_id, group_id)
@@ -51,11 +25,9 @@ class ClaudeCodePlugin(ToolPlugin):
self.status.log(f"Failed to set ownership for {path}: {e}", "WARNING")
def _get_claude_dir(self) -> Path:
"""Get the Claude Code configuration directory"""
return Path("/home/cubbi/.claude")
def _ensure_claude_dir(self) -> Path:
"""Ensure Claude directory exists with correct ownership"""
claude_dir = self._get_claude_dir()
try:
@@ -69,7 +41,6 @@ class ClaudeCodePlugin(ToolPlugin):
return claude_dir
def initialize(self) -> bool:
"""Initialize Claude Code configuration"""
self.status.log("Setting up Claude Code authentication...")
# Ensure Claude directory exists
@@ -97,23 +68,30 @@ class ClaudeCodePlugin(ToolPlugin):
return True
def _create_settings(self) -> Optional[Dict]:
"""Create Claude Code settings configuration"""
settings = {}
# Core authentication
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
return None
# Get Anthropic provider configuration from cubbi_config
anthropic_provider = None
for provider_name, provider_config in cubbi_config.providers.items():
if provider_config.type == "anthropic":
anthropic_provider = provider_config
break
# Basic authentication setup
settings["apiKey"] = api_key
if not anthropic_provider or not anthropic_provider.api_key:
# Fallback to environment variable for backward compatibility
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
return None
settings["apiKey"] = api_key
else:
settings["apiKey"] = anthropic_provider.api_key
# Custom authorization token (optional)
# Custom authorization token (optional) - still from environment
auth_token = os.environ.get("ANTHROPIC_AUTH_TOKEN")
if auth_token:
settings["authToken"] = auth_token
# Custom headers (optional)
# Custom headers (optional) - still from environment
custom_headers = os.environ.get("ANTHROPIC_CUSTOM_HEADERS")
if custom_headers:
try:
@@ -124,14 +102,14 @@ class ClaudeCodePlugin(ToolPlugin):
"⚠️ Invalid ANTHROPIC_CUSTOM_HEADERS format, skipping", "WARNING"
)
# Enterprise integration settings
# Enterprise integration settings - still from environment
if os.environ.get("CLAUDE_CODE_USE_BEDROCK") == "true":
settings["provider"] = "bedrock"
if os.environ.get("CLAUDE_CODE_USE_VERTEX") == "true":
settings["provider"] = "vertex"
# Network proxy settings
# Network proxy settings - still from environment
http_proxy = os.environ.get("HTTP_PROXY")
https_proxy = os.environ.get("HTTPS_PROXY")
if http_proxy or https_proxy:
@@ -141,7 +119,7 @@ class ClaudeCodePlugin(ToolPlugin):
if https_proxy:
settings["proxy"]["https"] = https_proxy
# Telemetry settings
# Telemetry settings - still from environment
if os.environ.get("DISABLE_TELEMETRY") == "true":
settings["telemetry"] = {"enabled": False}
@@ -160,7 +138,6 @@ class ClaudeCodePlugin(ToolPlugin):
return settings
def _write_settings(self, settings_file: Path, settings: Dict) -> bool:
"""Write settings to Claude Code configuration file"""
try:
# Write settings with secure permissions
with open(settings_file, "w") as f:
@@ -177,13 +154,11 @@ class ClaudeCodePlugin(ToolPlugin):
return False
def setup_tool_configuration(self) -> bool:
"""Set up Claude Code configuration - called by base class"""
# Additional tool configuration can be added here if needed
return True
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate Claude Code with available MCP servers"""
if mcp_config["count"] == 0:
def integrate_mcp_servers(self) -> bool:
if not cubbi_config.mcps:
self.status.log("No MCP servers to integrate")
return True

60
cubbi/images/claudecode/cubbi_image.yaml
View File

@@ -3,55 +3,13 @@ description: Claude Code AI environment
version: 1.0.0
maintainer: team@monadical.com
image: monadical/cubbi-claudecode:latest
init:
pre_command: /cubbi-init.sh
command: /entrypoint.sh
environment:
# Core Anthropic Authentication
- name: ANTHROPIC_API_KEY
description: Anthropic API key for Claude
required: true
sensitive: true
# Optional Enterprise Integration
- name: ANTHROPIC_AUTH_TOKEN
description: Custom authorization token for Claude
required: false
sensitive: true
- name: ANTHROPIC_CUSTOM_HEADERS
description: Additional HTTP headers for Claude API requests
required: false
sensitive: true
# Enterprise Deployment Options
- name: CLAUDE_CODE_USE_BEDROCK
description: Use Amazon Bedrock instead of direct API
required: false
- name: CLAUDE_CODE_USE_VERTEX
description: Use Google Vertex AI instead of direct API
required: false
# Network Configuration
- name: HTTP_PROXY
description: HTTP proxy server URL
required: false
- name: HTTPS_PROXY
description: HTTPS proxy server URL
required: false
# Optional Telemetry Control
- name: DISABLE_TELEMETRY
description: Disable Claude Code telemetry
required: false
default: "false"
volumes:
- mountPath: /app
description: Application directory
persistent_configs: []
environments_to_forward:
- ANTHROPIC_API_KEY
- ANTHROPIC_AUTH_TOKEN
- ANTHROPIC_CUSTOM_HEADERS
- CLAUDE_CODE_USE_BEDROCK
- CLAUDE_CODE_USE_VERTEX
- HTTP_PROXY
- HTTPS_PROXY
- DISABLE_TELEMETRY

266
cubbi/images/crush/crush_plugin.py
View File

@@ -1,31 +1,22 @@
#!/usr/bin/env python3
"""
Crush-specific plugin for Cubbi initialization
"""
import json
import os
from pathlib import Path
from typing import Any, Dict
from cubbi_init import ToolPlugin
from cubbi_init import ToolPlugin, cubbi_config
class CrushPlugin(ToolPlugin):
"""Plugin for Crush AI coding assistant initialization"""
@property
def tool_name(self) -> str:
return "crush"
def _get_user_ids(self) -> tuple[int, int]:
"""Get the cubbi user and group IDs from environment"""
user_id = int(os.environ.get("CUBBI_USER_ID", "1000"))
group_id = int(os.environ.get("CUBBI_GROUP_ID", "1000"))
return user_id, group_id
return cubbi_config.user.uid, cubbi_config.user.gid
def _set_ownership(self, path: Path) -> None:
"""Set ownership of a path to the cubbi user"""
user_id, group_id = self._get_user_ids()
try:
os.chown(path, user_id, group_id)
@@ -33,11 +24,107 @@ class CrushPlugin(ToolPlugin):
self.status.log(f"Failed to set ownership for {path}: {e}", "WARNING")
def _get_user_config_path(self) -> Path:
"""Get the correct config path for the cubbi user"""
return Path("/home/cubbi/.config/crush")
def _map_provider_to_crush_format(
self, provider_name: str, provider_config
) -> Dict[str, Any] | None:
"""Map cubbi provider configuration to crush provider format"""
if provider_config.type == "anthropic":
return {
"name": "Anthropic",
"type": "anthropic",
"api_key": provider_config.api_key,
"base_url": provider_config.base_url or "https://api.anthropic.com/v1",
"models": [
{
"id": "claude-3-5-sonnet-20241022",
"name": "Claude 3.5 Sonnet",
"context_window": 200000,
"default_max_tokens": 4096,
},
{
"id": "claude-3-5-haiku-20241022",
"name": "Claude 3.5 Haiku",
"context_window": 200000,
"default_max_tokens": 4096,
},
],
}
elif provider_config.type == "openai":
base_url = provider_config.base_url or "https://api.openai.com/v1"
return {
"name": "OpenAI"
if base_url.startswith("https://api.openai.com")
else f"OpenAI ({base_url})",
"type": "openai",
"api_key": provider_config.api_key,
"base_url": base_url,
"models": [
{
"id": "gpt-4o",
"name": "GPT-4o",
"context_window": 128000,
"default_max_tokens": 4096,
},
{
"id": "gpt-4o-mini",
"name": "GPT-4o Mini",
"context_window": 128000,
"default_max_tokens": 16384,
},
],
}
elif provider_config.type == "google":
return {
"name": "Google",
"type": "openai", # Google Gemini uses OpenAI-compatible API
"api_key": provider_config.api_key,
"base_url": "https://generativelanguage.googleapis.com/v1beta/openai/",
"models": [
{
"id": "gemini-1.5-pro",
"name": "Gemini 1.5 Pro",
"context_window": 2000000,
"default_max_tokens": 8192,
},
{
"id": "gemini-1.5-flash",
"name": "Gemini 1.5 Flash",
"context_window": 1000000,
"default_max_tokens": 8192,
},
],
}
elif provider_config.type == "openrouter":
return {
"name": "OpenRouter",
"type": "openai",
"api_key": provider_config.api_key,
"base_url": "https://openrouter.ai/api/v1",
"models": [
{
"id": "anthropic/claude-3.5-sonnet",
"name": "Claude 3.5 Sonnet (via OpenRouter)",
"context_window": 200000,
"default_max_tokens": 4096,
},
{
"id": "openai/gpt-4o",
"name": "GPT-4o (via OpenRouter)",
"context_window": 128000,
"default_max_tokens": 4096,
},
],
}
return None
def _ensure_user_config_dir(self) -> Path:
"""Ensure config directory exists with correct ownership"""
config_dir = self._get_user_config_path()
# Create the full directory path
@@ -63,12 +150,10 @@ class CrushPlugin(ToolPlugin):
return config_dir
def initialize(self) -> bool:
"""Initialize Crush configuration"""
self._ensure_user_config_dir()
return self.setup_tool_configuration()
def setup_tool_configuration(self) -> bool:
"""Set up Crush configuration file"""
# Ensure directory exists before writing
config_dir = self._ensure_user_config_dir()
if not config_dir.exists():
@@ -78,23 +163,77 @@ class CrushPlugin(ToolPlugin):
)
return False
config_file = config_dir / "config.json"
config_file = config_dir / "crush.json"
# Load or initialize configuration
if config_file.exists():
try:
with config_file.open("r") as f:
config_data = json.load(f)
except (json.JSONDecodeError, OSError) as e:
self.status.log(f"Failed to load existing config: {e}", "WARNING")
config_data = {}
else:
config_data = {}
# Initialize Crush configuration with schema
config_data = {"$schema": "https://charm.land/crush.json", "providers": {}}
# Set default model and provider if specified
# cubbi_model = os.environ.get("CUBBI_MODEL")
# cubbi_provider = os.environ.get("CUBBI_PROVIDER")
# XXX i didn't understood yet the configuration file, tbd later.
# Get all configured providers using the new provider system
self.status.log(
f"Found {len(cubbi_config.providers)} configured providers for Crush"
)
for provider_name, provider_config in cubbi_config.providers.items():
crush_provider = self._map_provider_to_crush_format(
provider_name, provider_config
)
if crush_provider:
config_data["providers"][provider_name] = crush_provider
self.status.log(
f"Added {provider_name} provider to Crush configuration"
)
# Fallback to legacy environment variables if no providers found
if not config_data["providers"]:
self.status.log(
"No providers found via new system, falling back to legacy detection"
)
# Check for legacy environment variables
legacy_providers = {
"anthropic": "ANTHROPIC_API_KEY",
"openai": "OPENAI_API_KEY",
"google": "GOOGLE_API_KEY",
"openrouter": "OPENROUTER_API_KEY",
}
for provider_name, env_var in legacy_providers.items():
api_key = os.environ.get(env_var)
if api_key:
# Create a simple object for legacy compatibility
class LegacyProvider:
def __init__(self, provider_type, api_key, base_url=None):
self.type = provider_type
self.api_key = api_key
self.base_url = base_url
if provider_name == "openai":
openai_url = os.environ.get("OPENAI_URL")
legacy_provider = LegacyProvider("openai", api_key, openai_url)
else:
legacy_provider = LegacyProvider(provider_name, api_key)
crush_provider = self._map_provider_to_crush_format(
provider_name, legacy_provider
)
if crush_provider:
config_data["providers"][provider_name] = crush_provider
self.status.log(
f"Added {provider_name} provider from legacy environment (legacy)"
)
# Set default model from cubbi configuration
if cubbi_config.defaults.model:
# Crush expects provider/model format for default model selection
config_data["default_model"] = cubbi_config.defaults.model
self.status.log(f"Set default model to {config_data['default_model']}")
# Only write config if we have providers configured
if not config_data["providers"]:
self.status.log(
"No providers configured, skipping Crush configuration file creation"
)
return True
try:
with config_file.open("w") as f:
@@ -103,15 +242,16 @@ class CrushPlugin(ToolPlugin):
# Set ownership of the config file to cubbi user
self._set_ownership(config_file)
self.status.log(f"Updated Crush configuration at {config_file}")
self.status.log(
f"Created Crush configuration at {config_file} with {len(config_data['providers'])} providers"
)
return True
except Exception as e:
self.status.log(f"Failed to write Crush configuration: {e}", "ERROR")
return False
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate Crush with available MCP servers"""
if mcp_config["count"] == 0:
def integrate_mcp_servers(self) -> bool:
if not cubbi_config.mcps:
self.status.log("No MCP servers to integrate")
return True
@@ -124,7 +264,7 @@ class CrushPlugin(ToolPlugin):
)
return False
config_file = config_dir / "config.json"
config_file = config_dir / "crush.json"
if config_file.exists():
try:
@@ -132,37 +272,34 @@ class CrushPlugin(ToolPlugin):
config_data = json.load(f)
except (json.JSONDecodeError, OSError) as e:
self.status.log(f"Failed to load existing config: {e}", "WARNING")
config_data = {}
config_data = {
"$schema": "https://charm.land/crush.json",
"providers": {},
}
else:
config_data = {}
config_data = {"$schema": "https://charm.land/crush.json", "providers": {}}
if "mcp_servers" not in config_data:
config_data["mcp_servers"] = {}
# Crush uses "mcps" field for MCP server configuration
if "mcps" not in config_data:
config_data["mcps"] = {}
for server in mcp_config["servers"]:
server_name = server["name"]
server_host = server["host"]
server_url = server["url"]
if server_name and server_host:
mcp_url = f"http://{server_host}:8080/sse"
self.status.log(f"Adding MCP server: {server_name} - {mcp_url}")
config_data["mcp_servers"][server_name] = {
"uri": mcp_url,
"type": server.get("type", "sse"),
"enabled": True,
}
elif server_name and server_url:
self.status.log(
f"Adding remote MCP server: {server_name} - {server_url}"
)
config_data["mcp_servers"][server_name] = {
"uri": server_url,
"type": server.get("type", "sse"),
"enabled": True,
}
for mcp in cubbi_config.mcps:
if mcp.type == "remote":
if mcp.name and mcp.url:
self.status.log(f"Adding remote MCP server: {mcp.name} - {mcp.url}")
config_data["mcps"][mcp.name] = {
"transport": {"type": "sse", "url": mcp.url},
"enabled": True,
}
elif mcp.type in ["docker", "proxy"]:
if mcp.name and mcp.host:
mcp_port = mcp.port or 8080
mcp_url = f"http://{mcp.host}:{mcp_port}/sse"
self.status.log(f"Adding MCP server: {mcp.name} - {mcp_url}")
config_data["mcps"][mcp.name] = {
"transport": {"type": "sse", "url": mcp_url},
"enabled": True,
}
try:
with config_file.open("w") as f:
@@ -171,6 +308,9 @@ class CrushPlugin(ToolPlugin):
# Set ownership of the config file to cubbi user
self._set_ownership(config_file)
self.status.log(
f"Integrated {len(cubbi_config.mcps)} MCP servers into Crush configuration"
)
return True
except Exception as e:
self.status.log(f"Failed to integrate MCP servers: {e}", "ERROR")

47
cubbi/images/crush/cubbi_image.yaml
View File

@@ -3,41 +3,14 @@ description: Crush AI coding assistant environment
version: 1.0.0
maintainer: team@monadical.com
image: monadical/cubbi-crush:latest
init:
pre_command: /cubbi-init.sh
command: /entrypoint.sh
environment:
- name: OPENAI_API_KEY
description: OpenAI API key for crush
required: false
sensitive: true
- name: ANTHROPIC_API_KEY
description: Anthropic API key for crush
required: false
sensitive: true
- name: GROQ_API_KEY
description: Groq API key for crush
required: false
sensitive: true
- name: OPENAI_URL
description: Custom OpenAI-compatible API URL
required: false
- name: CUBBI_MODEL
description: AI model to use with crush
required: false
- name: CUBBI_PROVIDER
description: AI provider to use with crush
required: false
volumes:
- mountPath: /app
description: Application directory
persistent_configs: []
environments_to_forward:
# API Keys
- OPENAI_API_KEY
- ANTHROPIC_API_KEY
- ANTHROPIC_AUTH_TOKEN
- ANTHROPIC_CUSTOM_HEADERS
- DEEPSEEK_API_KEY
- GEMINI_API_KEY
- OPENROUTER_API_KEY
- AIDER_API_KEYS

306
cubbi/images/cubbi_init.py
View File

@@ -1,6 +1,6 @@
#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["ruamel.yaml"]
# dependencies = ["ruamel.yaml", "pydantic"]
# ///
"""
Standalone Cubbi initialization script
@@ -19,12 +19,91 @@ import sys
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
from ruamel.yaml import YAML
# Status Management
class UserConfig(BaseModel):
uid: int = 1000
gid: int = 1000
class ProjectConfig(BaseModel):
url: Optional[str] = None
config_dir: Optional[str] = None
image_config_dir: Optional[str] = None
class PersistentLink(BaseModel):
source: str
target: str
type: str
class ProviderConfig(BaseModel):
type: str
api_key: str
base_url: Optional[str] = None
class MCPConfig(BaseModel):
name: str
type: str
host: Optional[str] = None
port: Optional[int] = None
url: Optional[str] = None
headers: Optional[Dict[str, str]] = None
class DefaultsConfig(BaseModel):
model: Optional[str] = None
class SSHConfig(BaseModel):
enabled: bool = False
class CubbiConfig(BaseModel):
"""Central configuration for container using Pydantic BaseModel"""
version: str = "1.0"
user: UserConfig = UserConfig()
providers: Dict[str, ProviderConfig] = {}
mcps: List[MCPConfig] = []
project: ProjectConfig = ProjectConfig()
persistent_links: List[PersistentLink] = []
defaults: DefaultsConfig = DefaultsConfig()
ssh: SSHConfig = SSHConfig()
run_command: Optional[str] = None
no_shell: bool = False
def get_provider_for_default_model(self) -> Optional[ProviderConfig]:
"""Get the provider config for the default model"""
if not self.defaults.model or "/" not in self.defaults.model:
return None
provider_name = self.defaults.model.split("/")[0]
return self.providers.get(provider_name)
def load_cubbi_config() -> CubbiConfig:
"""Load configuration from file or return default"""
config_path = Path("/cubbi/config.yaml")
if not config_path.exists():
return CubbiConfig()
yaml = YAML(typ="safe")
with open(config_path, "r") as f:
config_data = yaml.load(f) or {}
return CubbiConfig(**config_data)
cubbi_config = load_cubbi_config()
class StatusManager:
"""Manages initialization status and logging"""
@@ -36,12 +115,10 @@ class StatusManager:
self._setup_logging()
def _setup_logging(self) -> None:
"""Set up logging to both stdout and log file"""
self.log_file.touch(exist_ok=True)
self.set_status(False)
def log(self, message: str, level: str = "INFO") -> None:
"""Log a message with timestamp"""
print(message)
sys.stdout.flush()
@@ -64,11 +141,8 @@ class StatusManager:
self.set_status(True)
# Configuration Management
@dataclass
class PersistentConfig:
"""Persistent configuration mapping"""
source: str
target: str
type: str = "directory"
@@ -77,14 +151,13 @@ class PersistentConfig:
@dataclass
class ImageConfig:
"""Cubbi image configuration"""
name: str
description: str
version: str
maintainer: str
image: str
persistent_configs: List[PersistentConfig] = field(default_factory=list)
environments_to_forward: List[str] = field(default_factory=list)
class ConfigParser:
@@ -95,7 +168,6 @@ class ConfigParser:
self.environment: Dict[str, str] = dict(os.environ)
def load_image_config(self) -> ImageConfig:
"""Load and parse the cubbi_image.yaml configuration"""
if not self.config_file.exists():
raise FileNotFoundError(f"Configuration file not found: {self.config_file}")
@@ -103,7 +175,6 @@ class ConfigParser:
with open(self.config_file, "r") as f:
config_data = yaml.load(f)
# Parse persistent configurations
persistent_configs = []
for pc_data in config_data.get("persistent_configs", []):
persistent_configs.append(PersistentConfig(**pc_data))
@@ -115,39 +186,10 @@ class ConfigParser:
maintainer=config_data["maintainer"],
image=config_data["image"],
persistent_configs=persistent_configs,
environments_to_forward=config_data.get("environments_to_forward", []),
)
def get_cubbi_config(self) -> Dict[str, Any]:
"""Get standard Cubbi configuration from environment"""
return {
"user_id": int(self.environment.get("CUBBI_USER_ID", "1000")),
"group_id": int(self.environment.get("CUBBI_GROUP_ID", "1000")),
"run_command": self.environment.get("CUBBI_RUN_COMMAND"),
"no_shell": self.environment.get("CUBBI_NO_SHELL", "false").lower()
== "true",
"config_dir": self.environment.get("CUBBI_CONFIG_DIR", "/cubbi-config"),
"persistent_links": self.environment.get("CUBBI_PERSISTENT_LINKS", ""),
}
def get_mcp_config(self) -> Dict[str, Any]:
"""Get MCP server configuration from environment"""
mcp_count = int(self.environment.get("MCP_COUNT", "0"))
mcp_servers = []
for idx in range(mcp_count):
server = {
"name": self.environment.get(f"MCP_{idx}_NAME"),
"type": self.environment.get(f"MCP_{idx}_TYPE"),
"host": self.environment.get(f"MCP_{idx}_HOST"),
"url": self.environment.get(f"MCP_{idx}_URL"),
}
if server["name"]: # Only add if name is present
mcp_servers.append(server)
return {"count": mcp_count, "servers": mcp_servers}
# Core Management Classes
class UserManager:
"""Manages user and group creation/modification in containers"""
@@ -156,7 +198,6 @@ class UserManager:
self.username = "cubbi"
def _run_command(self, cmd: list[str]) -> bool:
"""Run a system command and log the result"""
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
if result.stdout:
@@ -168,7 +209,6 @@ class UserManager:
return False
def setup_user_and_group(self, user_id: int, group_id: int) -> bool:
"""Set up user and group with specified IDs"""
self.status.log(
f"Setting up user '{self.username}' with UID: {user_id}, GID: {group_id}"
)
@@ -244,7 +284,6 @@ class DirectoryManager:
def create_directory(
self, path: str, user_id: int, group_id: int, mode: int = 0o755
) -> bool:
"""Create a directory with proper ownership and permissions"""
dir_path = Path(path)
try:
@@ -260,7 +299,6 @@ class DirectoryManager:
return False
def setup_standard_directories(self, user_id: int, group_id: int) -> bool:
"""Set up standard Cubbi directories"""
directories = [
("/app", 0o755),
("/cubbi-config", 0o755),
@@ -317,7 +355,6 @@ class DirectoryManager:
return success
def _chown_recursive(self, path: Path, user_id: int, group_id: int) -> None:
"""Recursively change ownership of a directory"""
try:
os.chown(path, user_id, group_id)
for item in path.iterdir():
@@ -340,7 +377,6 @@ class ConfigManager:
def create_symlink(
self, source_path: str, target_path: str, user_id: int, group_id: int
) -> bool:
"""Create a symlink with proper ownership"""
try:
source = Path(source_path)
@@ -367,7 +403,6 @@ class ConfigManager:
def _ensure_target_directory(
self, target_path: str, user_id: int, group_id: int
) -> bool:
"""Ensure the target directory exists with proper ownership"""
try:
target_dir = Path(target_path)
if not target_dir.exists():
@@ -387,7 +422,6 @@ class ConfigManager:
def setup_persistent_configs(
self, persistent_configs: List[PersistentConfig], user_id: int, group_id: int
) -> bool:
"""Set up persistent configuration symlinks from image config"""
if not persistent_configs:
self.status.log("No persistent configurations defined in image config")
return True
@@ -404,6 +438,15 @@ class ConfigManager:
return success
def setup_persistent_link(
self, source: str, target: str, link_type: str, user_id: int, group_id: int
) -> bool:
"""Setup a single persistent link"""
if not self._ensure_target_directory(target, user_id, group_id):
return False
return self.create_symlink(source, target, user_id, group_id)
class CommandManager:
"""Manages command execution and user switching"""
@@ -413,7 +456,6 @@ class CommandManager:
self.username = "cubbi"
def run_as_user(self, command: List[str], user: str = None) -> int:
"""Run a command as the specified user using gosu"""
if user is None:
user = self.username
@@ -428,7 +470,6 @@ class CommandManager:
return 1
def run_user_command(self, command: str) -> int:
"""Run user-specified command as cubbi user"""
if not command:
return 0
@@ -436,7 +477,6 @@ class CommandManager:
return self.run_as_user(["sh", "-c", command])
def exec_as_user(self, args: List[str]) -> None:
"""Execute the final command as cubbi user (replaces current process)"""
if not args:
args = ["tail", "-f", "/dev/null"]
@@ -451,7 +491,6 @@ class CommandManager:
sys.exit(1)
# Tool Plugin System
class ToolPlugin(ABC):
"""Base class for tool-specific initialization plugins"""
@@ -462,20 +501,95 @@ class ToolPlugin(ABC):
@property
@abstractmethod
def tool_name(self) -> str:
"""Return the name of the tool this plugin supports"""
pass
@abstractmethod
def initialize(self) -> bool:
"""Main tool initialization logic"""
pass
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate with available MCP servers"""
return True
def get_resolved_model(self) -> Dict[str, Any] | None:
model_spec = os.environ.get("CUBBI_MODEL_SPEC", "")
if not model_spec:
return None
# Parse provider/model format
if "/" in model_spec:
provider_name, model_name = model_spec.split("/", 1)
else:
# Legacy format - treat as provider name
provider_name = model_spec
model_name = ""
# Get provider type from CUBBI_PROVIDER env var
provider_type = os.environ.get("CUBBI_PROVIDER", provider_name)
# Get base URL if available (for OpenAI-compatible providers)
base_url = None
if provider_type == "openai":
base_url = os.environ.get("OPENAI_URL")
return {
"provider_name": provider_name,
"provider_type": provider_type,
"model_name": model_name,
"base_url": base_url,
"model_spec": model_spec,
}
def get_provider_config(self, provider_name: str) -> Dict[str, str]:
provider_config = {}
# Map provider names to their environment variables
if provider_name == "anthropic" or provider_name.startswith("anthropic"):
api_key = os.environ.get("ANTHROPIC_API_KEY")
if api_key:
provider_config["ANTHROPIC_API_KEY"] = api_key
elif provider_name == "openai" or provider_name.startswith("openai"):
api_key = os.environ.get("OPENAI_API_KEY")
base_url = os.environ.get("OPENAI_URL")
if api_key:
provider_config["OPENAI_API_KEY"] = api_key
if base_url:
provider_config["OPENAI_URL"] = base_url
elif provider_name == "google" or provider_name.startswith("google"):
api_key = os.environ.get("GOOGLE_API_KEY")
if api_key:
provider_config["GOOGLE_API_KEY"] = api_key
elif provider_name == "openrouter" or provider_name.startswith("openrouter"):
api_key = os.environ.get("OPENROUTER_API_KEY")
if api_key:
provider_config["OPENROUTER_API_KEY"] = api_key
return provider_config
def get_all_providers_config(self) -> Dict[str, Dict[str, str]]:
all_providers = {}
# Check for each standard provider
standard_providers = ["anthropic", "openai", "google", "openrouter"]
for provider_name in standard_providers:
provider_config = self.get_provider_config(provider_name)
if provider_config: # Only include providers with API keys
all_providers[provider_name] = provider_config
# Also check for custom OpenAI-compatible providers
# These would have been set up with custom names but use OpenAI env vars
openai_config = self.get_provider_config("openai")
if openai_config and "OPENAI_URL" in openai_config:
# This might be a custom provider - we could check for custom naming
# but for now, we'll just include it as openai
pass
return all_providers
# Main Initializer
class CubbiInitializer:
"""Main Cubbi initialization orchestrator"""
@@ -494,21 +608,17 @@ class CubbiInitializer:
# Load configuration
image_config = self.config_parser.load_image_config()
cubbi_config = self.config_parser.get_cubbi_config()
mcp_config = self.config_parser.get_mcp_config()
self.status.log(f"Initializing {image_config.name} v{image_config.version}")
# Core initialization
success = self._run_core_initialization(image_config, cubbi_config)
success = self._run_core_initialization(image_config)
if not success:
self.status.log("Core initialization failed", "ERROR")
sys.exit(1)
# Tool-specific initialization
success = self._run_tool_initialization(
image_config, cubbi_config, mcp_config
)
success = self._run_tool_initialization(image_config)
if not success:
self.status.log("Tool initialization failed", "ERROR")
sys.exit(1)
@@ -517,16 +627,15 @@ class CubbiInitializer:
self.status.complete_initialization()
# Handle commands
self._handle_command_execution(cubbi_config, final_args)
self._handle_command_execution(final_args)
except Exception as e:
self.status.log(f"Initialization failed with error: {e}", "ERROR")
sys.exit(1)
def _run_core_initialization(self, image_config, cubbi_config) -> bool:
"""Run core Cubbi initialization steps"""
user_id = cubbi_config["user_id"]
group_id = cubbi_config["group_id"]
def _run_core_initialization(self, image_config) -> bool:
user_id = cubbi_config.user.uid
group_id = cubbi_config.user.gid
if not self.user_manager.setup_user_and_group(user_id, group_id):
return False
@@ -534,25 +643,29 @@ class CubbiInitializer:
if not self.directory_manager.setup_standard_directories(user_id, group_id):
return False
config_path = Path(cubbi_config["config_dir"])
if not config_path.exists():
self.status.log(f"Creating config directory: {cubbi_config['config_dir']}")
try:
config_path.mkdir(parents=True, exist_ok=True)
os.chown(cubbi_config["config_dir"], user_id, group_id)
except Exception as e:
self.status.log(f"Failed to create config directory: {e}", "ERROR")
return False
if cubbi_config.project.config_dir:
config_path = Path(cubbi_config.project.config_dir)
if not config_path.exists():
self.status.log(
f"Creating config directory: {cubbi_config.project.config_dir}"
)
try:
config_path.mkdir(parents=True, exist_ok=True)
os.chown(cubbi_config.project.config_dir, user_id, group_id)
except Exception as e:
self.status.log(f"Failed to create config directory: {e}", "ERROR")
return False
if not self.config_manager.setup_persistent_configs(
image_config.persistent_configs, user_id, group_id
):
return False
# Setup persistent configs
for link in cubbi_config.persistent_links:
if not self.config_manager.setup_persistent_link(
link.source, link.target, link.type, user_id, group_id
):
return False
return True
def _run_tool_initialization(self, image_config, cubbi_config, mcp_config) -> bool:
"""Run tool-specific initialization"""
def _run_tool_initialization(self, image_config) -> bool:
# Look for a tool-specific plugin file in the same directory
plugin_name = image_config.name.lower().replace("-", "_")
plugin_file = Path(__file__).parent / f"{plugin_name}_plugin.py"
@@ -591,14 +704,7 @@ class CubbiInitializer:
return False
# Instantiate and run the plugin
plugin = plugin_class(
self.status,
{
"image_config": image_config,
"cubbi_config": cubbi_config,
"mcp_config": mcp_config,
},
)
plugin = plugin_class(self.status, {"image_config": image_config})
self.status.log(f"Running {plugin.tool_name}-specific initialization")
@@ -606,7 +712,7 @@ class CubbiInitializer:
self.status.log(f"{plugin.tool_name} initialization failed", "ERROR")
return False
if not plugin.integrate_mcp_servers(mcp_config):
if not plugin.integrate_mcp_servers():
self.status.log(f"{plugin.tool_name} MCP integration failed", "ERROR")
return False
@@ -618,22 +724,19 @@ class CubbiInitializer:
)
return False
def _handle_command_execution(self, cubbi_config, final_args):
"""Handle command execution"""
def _handle_command_execution(self, final_args):
exit_code = 0
if cubbi_config["run_command"]:
if cubbi_config.run_command:
self.status.log("--- Executing initial command ---")
exit_code = self.command_manager.run_user_command(
cubbi_config["run_command"]
)
exit_code = self.command_manager.run_user_command(cubbi_config.run_command)
self.status.log(
f"--- Initial command finished (exit code: {exit_code}) ---"
)
if cubbi_config["no_shell"]:
if cubbi_config.no_shell:
self.status.log(
"--- CUBBI_NO_SHELL=true, exiting container without starting shell ---"
"--- no_shell=true, exiting container without starting shell ---"
)
sys.exit(exit_code)
@@ -641,7 +744,6 @@ class CubbiInitializer:
def main() -> int:
"""Main CLI entry point"""
import argparse
parser = argparse.ArgumentParser(

35
cubbi/images/goose/cubbi_image.yaml
View File

@@ -3,29 +3,14 @@ description: Goose AI environment
version: 1.0.0
maintainer: team@monadical.com
image: monadical/cubbi-goose:latest
init:
pre_command: /cubbi-init.sh
command: /entrypoint.sh
environment:
- name: LANGFUSE_INIT_PROJECT_PUBLIC_KEY
description: Langfuse public key
required: false
sensitive: true
- name: LANGFUSE_INIT_PROJECT_SECRET_KEY
description: Langfuse secret key
required: false
sensitive: true
- name: LANGFUSE_URL
description: Langfuse API URL
required: false
default: https://cloud.langfuse.com
volumes:
- mountPath: /app
description: Application directory
persistent_configs: []
environments_to_forward:
# API Keys
- OPENAI_API_KEY
- ANTHROPIC_API_KEY
- ANTHROPIC_AUTH_TOKEN
- ANTHROPIC_CUSTOM_HEADERS
- DEEPSEEK_API_KEY
- GEMINI_API_KEY
- OPENROUTER_API_KEY
- AIDER_API_KEYS

191
cubbi/images/goose/goose_plugin.py
View File

@@ -1,31 +1,21 @@
#!/usr/bin/env python3
"""
Goose-specific plugin for Cubbi initialization
"""
import os
from pathlib import Path
from typing import Any, Dict
from cubbi_init import ToolPlugin
from cubbi_init import ToolPlugin, cubbi_config
from ruamel.yaml import YAML
class GoosePlugin(ToolPlugin):
"""Plugin for Goose AI tool initialization"""
@property
def tool_name(self) -> str:
return "goose"
def _get_user_ids(self) -> tuple[int, int]:
"""Get the cubbi user and group IDs from environment"""
user_id = int(os.environ.get("CUBBI_USER_ID", "1000"))
group_id = int(os.environ.get("CUBBI_GROUP_ID", "1000"))
return user_id, group_id
return cubbi_config.user.uid, cubbi_config.user.gid
def _set_ownership(self, path: Path) -> None:
"""Set ownership of a path to the cubbi user"""
user_id, group_id = self._get_user_ids()
try:
os.chown(path, user_id, group_id)
@@ -33,11 +23,9 @@ class GoosePlugin(ToolPlugin):
self.status.log(f"Failed to set ownership for {path}: {e}", "WARNING")
def _get_user_config_path(self) -> Path:
"""Get the correct config path for the cubbi user"""
return Path("/home/cubbi/.config/goose")
def _ensure_user_config_dir(self) -> Path:
"""Ensure config directory exists with correct ownership"""
config_dir = self._get_user_config_path()
# Create the full directory path
@@ -62,13 +50,64 @@ class GoosePlugin(ToolPlugin):
return config_dir
def _write_env_vars_to_profile(self, env_vars: dict) -> None:
"""Write environment variables to shell profile for interactive sessions"""
try:
# Write to cubbi user's bash profile
profile_path = Path("/home/cubbi/.bashrc")
# Create cubbi env section marker
env_section_start = "# CUBBI GOOSE ENVIRONMENT VARIABLES"
env_section_end = "# END CUBBI GOOSE ENVIRONMENT VARIABLES"
# Read existing profile or create empty
if profile_path.exists():
with open(profile_path, "r") as f:
lines = f.readlines()
else:
lines = []
# Remove existing cubbi env section
new_lines = []
skip_section = False
for line in lines:
if env_section_start in line:
skip_section = True
elif env_section_end in line:
skip_section = False
continue
elif not skip_section:
new_lines.append(line)
# Add new env vars section
if env_vars:
new_lines.append(f"\n{env_section_start}\n")
for key, value in env_vars.items():
new_lines.append(f'export {key}="{value}"\n')
new_lines.append(f"{env_section_end}\n")
# Write updated profile
profile_path.parent.mkdir(parents=True, exist_ok=True)
with open(profile_path, "w") as f:
f.writelines(new_lines)
# Set ownership
self._set_ownership(profile_path)
self.status.log(
f"Updated shell profile with {len(env_vars)} environment variables"
)
except Exception as e:
self.status.log(
f"Failed to write environment variables to profile: {e}", "ERROR"
)
def initialize(self) -> bool:
"""Initialize Goose configuration"""
self._ensure_user_config_dir()
return self.setup_tool_configuration()
def setup_tool_configuration(self) -> bool:
"""Set up Goose configuration file"""
# Ensure directory exists before writing
config_dir = self._ensure_user_config_dir()
if not config_dir.exists():
@@ -99,24 +138,52 @@ class GoosePlugin(ToolPlugin):
"type": "builtin",
}
# Update with environment variables
goose_model = os.environ.get("CUBBI_MODEL")
goose_provider = os.environ.get("CUBBI_PROVIDER")
# Configure Goose with the default model
provider_config = cubbi_config.get_provider_for_default_model()
if provider_config and cubbi_config.defaults.model:
_, model_name = cubbi_config.defaults.model.split("/", 1)
if goose_model:
config_data["GOOSE_MODEL"] = goose_model
self.status.log(f"Set GOOSE_MODEL to {goose_model}")
# Set Goose model and provider
config_data["GOOSE_MODEL"] = model_name
config_data["GOOSE_PROVIDER"] = provider_config.type
if goose_provider:
config_data["GOOSE_PROVIDER"] = goose_provider
self.status.log(f"Set GOOSE_PROVIDER to {goose_provider}")
# Set ONLY the specific API key for the selected provider
# Set both in current process AND in shell environment file
env_vars_to_set = {}
# If provider is OpenAI and OPENAI_URL is set, configure OPENAI_HOST
if goose_provider.lower() == "openai":
openai_url = os.environ.get("OPENAI_URL")
if openai_url:
config_data["OPENAI_HOST"] = openai_url
self.status.log(f"Set OPENAI_HOST to {openai_url}")
if provider_config.type == "anthropic" and provider_config.api_key:
env_vars_to_set["ANTHROPIC_API_KEY"] = provider_config.api_key
self.status.log("Set Anthropic API key for goose")
elif provider_config.type == "openai" and provider_config.api_key:
# For OpenAI-compatible providers (including litellm), goose expects OPENAI_API_KEY
env_vars_to_set["OPENAI_API_KEY"] = provider_config.api_key
self.status.log("Set OpenAI API key for goose")
# Set base URL for OpenAI-compatible providers in both env and config
if provider_config.base_url:
env_vars_to_set["OPENAI_BASE_URL"] = provider_config.base_url
config_data["OPENAI_HOST"] = provider_config.base_url
self.status.log(
f"Set OPENAI_BASE_URL and OPENAI_HOST to {provider_config.base_url}"
)
elif provider_config.type == "google" and provider_config.api_key:
env_vars_to_set["GOOGLE_API_KEY"] = provider_config.api_key
self.status.log("Set Google API key for goose")
elif provider_config.type == "openrouter" and provider_config.api_key:
env_vars_to_set["OPENROUTER_API_KEY"] = provider_config.api_key
self.status.log("Set OpenRouter API key for goose")
# Set environment variables for current process (for --run commands)
for key, value in env_vars_to_set.items():
os.environ[key] = value
# Write environment variables to shell profile for interactive sessions
self._write_env_vars_to_profile(env_vars_to_set)
self.status.log(
f"Configured Goose: model={model_name}, provider={provider_config.type}"
)
else:
self.status.log("No default model or provider configured", "WARNING")
try:
with config_file.open("w") as f:
@@ -131,9 +198,8 @@ class GoosePlugin(ToolPlugin):
self.status.log(f"Failed to write Goose configuration: {e}", "ERROR")
return False
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate Goose with available MCP servers"""
if mcp_config["count"] == 0:
def integrate_mcp_servers(self) -> bool:
if not cubbi_config.mcps:
self.status.log("No MCP servers to integrate")
return True
@@ -158,36 +224,33 @@ class GoosePlugin(ToolPlugin):
if "extensions" not in config_data:
config_data["extensions"] = {}
for server in mcp_config["servers"]:
server_name = server["name"]
server_host = server["host"]
server_url = server["url"]
if server_name and server_host:
mcp_url = f"http://{server_host}:8080/sse"
self.status.log(f"Adding MCP extension: {server_name} - {mcp_url}")
config_data["extensions"][server_name] = {
"enabled": True,
"name": server_name,
"timeout": 60,
"type": server.get("type", "sse"),
"uri": mcp_url,
"envs": {},
}
elif server_name and server_url:
self.status.log(
f"Adding remote MCP extension: {server_name} - {server_url}"
)
config_data["extensions"][server_name] = {
"enabled": True,
"name": server_name,
"timeout": 60,
"type": server.get("type", "sse"),
"uri": server_url,
"envs": {},
}
for mcp in cubbi_config.mcps:
if mcp.type == "remote":
if mcp.name and mcp.url:
self.status.log(
f"Adding remote MCP extension: {mcp.name} - {mcp.url}"
)
config_data["extensions"][mcp.name] = {
"enabled": True,
"name": mcp.name,
"timeout": 60,
"type": "sse",
"uri": mcp.url,
"envs": {},
}
elif mcp.type in ["docker", "proxy"]:
if mcp.name and mcp.host:
mcp_port = mcp.port or 8080
mcp_url = f"http://{mcp.host}:{mcp_port}/sse"
self.status.log(f"Adding MCP extension: {mcp.name} - {mcp_url}")
config_data["extensions"][mcp.name] = {
"enabled": True,
"name": mcp.name,
"timeout": 60,
"type": "sse",
"uri": mcp_url,
"envs": {},
}
try:
with config_file.open("w") as f:

20
cubbi/images/opencode/cubbi_image.yaml
View File

@@ -3,14 +3,14 @@ description: Opencode AI environment
version: 1.0.0
maintainer: team@monadical.com
image: monadical/cubbi-opencode:latest
init:
pre_command: /cubbi-init.sh
command: /entrypoint.sh
environment: []
volumes:
- mountPath: /app
description: Application directory
persistent_configs: []
environments_to_forward:
# API Keys
- OPENAI_API_KEY
- ANTHROPIC_API_KEY
- ANTHROPIC_AUTH_TOKEN
- ANTHROPIC_CUSTOM_HEADERS
- DEEPSEEK_API_KEY
- GEMINI_API_KEY
- OPENROUTER_API_KEY
- AIDER_API_KEYS

244
cubbi/images/opencode/opencode_plugin.py
View File

@@ -1,39 +1,24 @@
#!/usr/bin/env python3
"""
Opencode-specific plugin for Cubbi initialization
"""
import json
import os
from pathlib import Path
from typing import Any, Dict
from cubbi_init import ToolPlugin
from cubbi_init import ToolPlugin, cubbi_config
# Map of environment variables to provider names in auth.json
API_KEY_MAPPINGS = {
"ANTHROPIC_API_KEY": "anthropic",
"GOOGLE_API_KEY": "google",
"OPENAI_API_KEY": "openai",
"OPENROUTER_API_KEY": "openrouter",
}
# Standard providers that OpenCode supports natively
STANDARD_PROVIDERS = ["anthropic", "openai", "google", "openrouter"]
class OpencodePlugin(ToolPlugin):
"""Plugin for Opencode AI tool initialization"""
@property
def tool_name(self) -> str:
return "opencode"
def _get_user_ids(self) -> tuple[int, int]:
"""Get the cubbi user and group IDs from environment"""
user_id = int(os.environ.get("CUBBI_USER_ID", "1000"))
group_id = int(os.environ.get("CUBBI_GROUP_ID", "1000"))
return user_id, group_id
return cubbi_config.user.uid, cubbi_config.user.gid
def _set_ownership(self, path: Path) -> None:
"""Set ownership of a path to the cubbi user"""
user_id, group_id = self._get_user_ids()
try:
os.chown(path, user_id, group_id)
@@ -41,15 +26,12 @@ class OpencodePlugin(ToolPlugin):
self.status.log(f"Failed to set ownership for {path}: {e}", "WARNING")
def _get_user_config_path(self) -> Path:
"""Get the correct config path for the cubbi user"""
return Path("/home/cubbi/.config/opencode")
def _get_user_data_path(self) -> Path:
"""Get the correct data path for the cubbi user"""
return Path("/home/cubbi/.local/share/opencode")
def _ensure_user_config_dir(self) -> Path:
"""Ensure config directory exists with correct ownership"""
config_dir = self._get_user_config_path()
# Create the full directory path
@@ -75,7 +57,6 @@ class OpencodePlugin(ToolPlugin):
return config_dir
def _ensure_user_data_dir(self) -> Path:
"""Ensure data directory exists with correct ownership"""
data_dir = self._get_user_data_path()
# Create the full directory path
@@ -98,72 +79,15 @@ class OpencodePlugin(ToolPlugin):
return data_dir
def _create_auth_file(self) -> bool:
"""Create auth.json file with configured API keys"""
# Ensure data directory exists
data_dir = self._ensure_user_data_dir()
if not data_dir.exists():
self.status.log(
f"Data directory {data_dir} does not exist and could not be created",
"ERROR",
)
return False
auth_file = data_dir / "auth.json"
auth_data = {}
# Check each API key and add to auth data if present
for env_var, provider in API_KEY_MAPPINGS.items():
api_key = os.environ.get(env_var)
if api_key:
auth_data[provider] = {"type": "api", "key": api_key}
# Add custom endpoint URL for OpenAI if available
if provider == "openai":
openai_url = os.environ.get("OPENAI_URL")
if openai_url:
auth_data[provider]["baseURL"] = openai_url
self.status.log(
f"Added OpenAI custom endpoint URL: {openai_url}"
)
self.status.log(f"Added {provider} API key to auth configuration")
# Only write file if we have at least one API key
if not auth_data:
self.status.log("No API keys found, skipping auth.json creation")
return True
try:
with auth_file.open("w") as f:
json.dump(auth_data, f, indent=2)
# Set ownership of the auth file to cubbi user
self._set_ownership(auth_file)
# Set secure permissions (readable only by owner)
auth_file.chmod(0o600)
self.status.log(f"Created OpenCode auth configuration at {auth_file}")
return True
except Exception as e:
self.status.log(f"Failed to create auth configuration: {e}", "ERROR")
return False
def initialize(self) -> bool:
"""Initialize Opencode configuration"""
self._ensure_user_config_dir()
# Create auth.json file with API keys
auth_success = self._create_auth_file()
# Set up tool configuration
# Set up tool configuration with new provider format
config_success = self.setup_tool_configuration()
return auth_success and config_success
return config_success
def setup_tool_configuration(self) -> bool:
"""Set up Opencode configuration file"""
# Ensure directory exists before writing
config_dir = self._ensure_user_config_dir()
if not config_dir.exists():
@@ -175,23 +99,101 @@ class OpencodePlugin(ToolPlugin):
config_file = config_dir / "config.json"
# Load or initialize configuration
if config_file.exists():
with config_file.open("r") as f:
config_data = json.load(f) or {}
else:
config_data = {}
# Initialize configuration with schema
config_data = {"$schema": "https://opencode.ai/config.json"}
# Set default theme to system
config_data.setdefault("theme", "system")
config_data["theme"] = "system"
# Update with environment variables
opencode_model = os.environ.get("CUBBI_MODEL")
opencode_provider = os.environ.get("CUBBI_PROVIDER")
# Add providers configuration
config_data["provider"] = {}
if opencode_model and opencode_provider:
config_data["model"] = f"{opencode_provider}/{opencode_model}"
self.status.log(f"Set model to {config_data['model']}")
# Configure all available providers
for provider_name, provider_config in cubbi_config.providers.items():
# Check if this is a custom provider (has baseURL)
if provider_config.base_url:
# Custom provider - include baseURL and name
provider_entry = {
"options": {
"apiKey": provider_config.api_key,
"baseURL": provider_config.base_url,
},
"models": {},
}
# Add npm package and name for custom providers
if provider_config.type in STANDARD_PROVIDERS:
# Standard provider with custom URL - determine npm package
if provider_config.type == "anthropic":
provider_entry["npm"] = "@ai-sdk/anthropic"
provider_entry["name"] = f"Anthropic ({provider_name})"
elif provider_config.type == "openai":
provider_entry["npm"] = "@ai-sdk/openai-compatible"
provider_entry["name"] = f"OpenAI Compatible ({provider_name})"
elif provider_config.type == "google":
provider_entry["npm"] = "@ai-sdk/google"
provider_entry["name"] = f"Google ({provider_name})"
elif provider_config.type == "openrouter":
provider_entry["npm"] = "@ai-sdk/openai-compatible"
provider_entry["name"] = f"OpenRouter ({provider_name})"
else:
# Non-standard provider with custom URL
provider_entry["npm"] = "@ai-sdk/openai-compatible"
provider_entry["name"] = provider_name.title()
config_data["provider"][provider_name] = provider_entry
self.status.log(
f"Added {provider_name} custom provider to OpenCode configuration"
)
else:
# Standard provider without custom URL - minimal config
if provider_config.type in STANDARD_PROVIDERS:
config_data["provider"][provider_name] = {
"options": {"apiKey": provider_config.api_key},
"models": {},
}
self.status.log(
f"Added {provider_name} standard provider to OpenCode configuration"
)
# Set default model and add it only to the default provider
if cubbi_config.defaults.model:
config_data["model"] = cubbi_config.defaults.model
self.status.log(f"Set default model to {config_data['model']}")
# Add the specific model only to the provider that matches the default model
provider_name, model_name = cubbi_config.defaults.model.split("/", 1)
if provider_name in config_data["provider"]:
config_data["provider"][provider_name]["models"] = {
model_name: {"name": model_name}
}
self.status.log(
f"Added default model {model_name} to {provider_name} provider"
)
else:
# Fallback to legacy environment variables
opencode_model = os.environ.get("CUBBI_MODEL")
opencode_provider = os.environ.get("CUBBI_PROVIDER")
if opencode_model and opencode_provider:
config_data["model"] = f"{opencode_provider}/{opencode_model}"
self.status.log(f"Set model to {config_data['model']} (legacy)")
# Add the legacy model to the provider if it exists
if opencode_provider in config_data["provider"]:
config_data["provider"][opencode_provider]["models"] = {
opencode_model: {"name": opencode_model}
}
# Only write config if we have providers configured
if not config_data["provider"]:
self.status.log(
"No providers configured, using minimal OpenCode configuration"
)
config_data = {
"$schema": "https://opencode.ai/config.json",
"theme": "system",
}
try:
with config_file.open("w") as f:
@@ -200,15 +202,16 @@ class OpencodePlugin(ToolPlugin):
# Set ownership of the config file to cubbi user
self._set_ownership(config_file)
self.status.log(f"Updated Opencode configuration at {config_file}")
self.status.log(
f"Updated OpenCode configuration at {config_file} with {len(config_data.get('provider', {}))} providers"
)
return True
except Exception as e:
self.status.log(f"Failed to write Opencode configuration: {e}", "ERROR")
self.status.log(f"Failed to write OpenCode configuration: {e}", "ERROR")
return False
def integrate_mcp_servers(self, mcp_config: Dict[str, Any]) -> bool:
"""Integrate Opencode with available MCP servers"""
if mcp_config["count"] == 0:
def integrate_mcp_servers(self) -> bool:
if not cubbi_config.mcps:
self.status.log("No MCP servers to integrate")
return True
@@ -232,28 +235,25 @@ class OpencodePlugin(ToolPlugin):
if "mcp" not in config_data:
config_data["mcp"] = {}
for server in mcp_config["servers"]:
server_name = server["name"]
server_host = server.get("host")
server_url = server.get("url")
if server_name and server_host:
mcp_url = f"http://{server_host}:8080/sse"
self.status.log(f"Adding MCP extension: {server_name} - {mcp_url}")
config_data["mcp"][server_name] = {
"type": "remote",
"url": mcp_url,
}
elif server_name and server_url:
self.status.log(
f"Adding remote MCP extension: {server_name} - {server_url}"
)
config_data["mcp"][server_name] = {
"type": "remote",
"url": server_url,
}
for mcp in cubbi_config.mcps:
if mcp.type == "remote":
if mcp.name and mcp.url:
self.status.log(
f"Adding remote MCP extension: {mcp.name} - {mcp.url}"
)
config_data["mcp"][mcp.name] = {
"type": "remote",
"url": mcp.url,
}
elif mcp.type in ["docker", "proxy"]:
if mcp.name and mcp.host:
mcp_port = mcp.port or 8080
mcp_url = f"http://{mcp.host}:{mcp_port}/sse"
self.status.log(f"Adding MCP extension: {mcp.name} - {mcp_url}")
config_data["mcp"][mcp.name] = {
"type": "remote",
"url": mcp_url,
}
try:
with config_file.open("w") as f:

13
cubbi/models.py
View File

@@ -33,26 +33,15 @@ class PersistentConfig(BaseModel):
description: str = ""
class VolumeMount(BaseModel):
mountPath: str
description: str = ""
class ImageInit(BaseModel):
pre_command: Optional[str] = None
command: str
class Image(BaseModel):
name: str
description: str
version: str
maintainer: str
image: str
init: Optional[ImageInit] = None
environment: List[ImageEnvironmentVariable] = []
volumes: List[VolumeMount] = []
persistent_configs: List[PersistentConfig] = []
environments_to_forward: List[str] = []
class RemoteMCP(BaseModel):

368
cubbi/user_config.py
View File

@@ -8,8 +8,28 @@ from typing import Any, Dict, List, Optional, Tuple
import yaml
# Define the environment variable mappings
ENV_MAPPINGS = {
# Define the environment variable mappings for auto-discovery
STANDARD_PROVIDERS = {
"anthropic": {
"type": "anthropic",
"env_key": "ANTHROPIC_API_KEY",
},
"openai": {
"type": "openai",
"env_key": "OPENAI_API_KEY",
},
"google": {
"type": "google",
"env_key": "GOOGLE_API_KEY",
},
"openrouter": {
"type": "openrouter",
"env_key": "OPENROUTER_API_KEY",
},
}
# Legacy environment variable mappings (kept for backward compatibility)
LEGACY_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",
@@ -44,6 +64,10 @@ class UserConfigManager:
self.config_path.parent.mkdir(parents=True, exist_ok=True)
# Create default config
default_config = self._get_default_config()
# Auto-discover and add providers from environment for new configs
self._auto_discover_providers(default_config)
# Save to file
with open(self.config_path, "w") as f:
yaml.safe_dump(default_config, f)
@@ -85,7 +109,12 @@ class UserConfigManager:
config = {}
# Merge with defaults for any missing fields
return self._merge_with_defaults(config)
config = self._merge_with_defaults(config)
# Auto-discover and add providers from environment
self._auto_discover_providers(config)
return config
def _get_default_config(self) -> Dict[str, Any]:
"""Get the default configuration."""
@@ -98,15 +127,11 @@ class UserConfigManager:
"volumes": [], # Default volumes to mount, format: "source:dest"
"ports": [], # Default ports to forward, format: list of integers
"mcps": [], # Default MCP servers to connect to
"model": "claude-3-5-sonnet-latest", # Default LLM model to use
"provider": "anthropic", # Default LLM provider to use
"model": "anthropic/claude-3-5-sonnet-latest", # Default LLM model (provider/model format)
},
"providers": {}, # LLM providers configuration
"services": {
"langfuse": {},
"openai": {},
"anthropic": {},
"openrouter": {},
"google": {},
"langfuse": {}, # Keep langfuse in services as it's not an LLM provider
},
"docker": {
"network": "cubbi-network",
@@ -148,7 +173,7 @@ class UserConfigManager:
and not key_path.startswith("services.")
and not any(
key_path.startswith(section + ".")
for section in ["defaults", "docker", "remote", "ui"]
for section in ["defaults", "docker", "remote", "ui", "providers"]
)
):
service, setting = key_path.split(".", 1)
@@ -177,7 +202,7 @@ class UserConfigManager:
and not key_path.startswith("services.")
and not any(
key_path.startswith(section + ".")
for section in ["defaults", "docker", "remote", "ui"]
for section in ["defaults", "docker", "remote", "ui", "providers"]
)
):
service, setting = key_path.split(".", 1)
@@ -247,13 +272,22 @@ class UserConfigManager:
def get_environment_variables(self) -> Dict[str, str]:
"""Get environment variables from the configuration.
NOTE: API keys are now handled by cubbi_init plugins, not passed from host.
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():
# Process the legacy service configurations and map to environment variables
# BUT EXCLUDE API KEYS - they're now handled by cubbi_init
for config_path, env_var in LEGACY_ENV_MAPPINGS.items():
# Skip API key environment variables - let cubbi_init handle them
if any(
key_word in env_var.upper() for key_word in ["API_KEY", "SECRET_KEY"]
):
continue
value = self.get(config_path)
if value:
# Handle environment variable references
@@ -267,6 +301,68 @@ class UserConfigManager:
env_vars[env_var] = str(value)
# NOTE: Provider API keys are no longer passed as environment variables
# They are now handled by cubbi_init plugins based on selected model
# This prevents unused API keys from being exposed in containers
return env_vars
def get_provider_environment_variables(self, provider_name: str) -> Dict[str, str]:
"""Get environment variables for a specific provider.
Args:
provider_name: Name of the provider to get environment variables for
Returns:
Dictionary of environment variables for the provider
"""
env_vars = {}
provider_config = self.get_provider(provider_name)
if not provider_config:
return env_vars
provider_type = provider_config.get("type", provider_name)
api_key = provider_config.get("api_key", "")
base_url = provider_config.get("base_url")
# Resolve environment variable references
if api_key.startswith("${") and api_key.endswith("}"):
env_var_name = api_key[2:-1]
resolved_api_key = os.environ.get(env_var_name, "")
else:
resolved_api_key = api_key
if not resolved_api_key:
return env_vars
# Add environment variables based on provider type
if provider_type == "anthropic":
env_vars["ANTHROPIC_API_KEY"] = resolved_api_key
elif provider_type == "openai":
env_vars["OPENAI_API_KEY"] = resolved_api_key
if base_url:
env_vars["OPENAI_URL"] = base_url
elif provider_type == "google":
env_vars["GOOGLE_API_KEY"] = resolved_api_key
elif provider_type == "openrouter":
env_vars["OPENROUTER_API_KEY"] = resolved_api_key
return env_vars
def get_all_providers_environment_variables(self) -> Dict[str, str]:
"""Get environment variables for all configured providers.
Returns:
Dictionary of all provider environment variables
"""
env_vars = {}
providers = self.get("providers", {})
for provider_name in providers.keys():
provider_env = self.get_provider_environment_variables(provider_name)
env_vars.update(provider_env)
return env_vars
def list_config(self) -> List[Tuple[str, Any]]:
@@ -295,3 +391,247 @@ class UserConfigManager:
_flatten_dict(self.config)
return sorted(result)
def _auto_discover_providers(self, config: Dict[str, Any]) -> None:
"""Auto-discover providers from environment variables."""
if "providers" not in config:
config["providers"] = {}
for provider_name, provider_info in STANDARD_PROVIDERS.items():
# Skip if provider already configured
if provider_name in config["providers"]:
continue
# Check if environment variable exists
api_key = os.environ.get(provider_info["env_key"])
if api_key:
config["providers"][provider_name] = {
"type": provider_info["type"],
"api_key": f"${{{provider_info['env_key']}}}", # Reference to env var
}
def get_provider(self, provider_name: str) -> Optional[Dict[str, Any]]:
"""Get a provider configuration by name."""
return self.get(f"providers.{provider_name}")
def list_providers(self) -> Dict[str, Dict[str, Any]]:
"""Get all configured providers."""
return self.get("providers", {})
def add_provider(
self,
name: str,
provider_type: str,
api_key: str,
base_url: Optional[str] = None,
env_key: Optional[str] = None,
) -> None:
"""Add a new provider configuration.
Args:
name: Provider name/identifier
provider_type: Type of provider (anthropic, openai, etc.)
api_key: API key value or environment variable reference
base_url: Custom base URL for API calls (optional)
env_key: If provided, use env reference instead of direct api_key
"""
provider_config = {
"type": provider_type,
"api_key": f"${{{env_key}}}" if env_key else api_key,
}
if base_url:
provider_config["base_url"] = base_url
self.set(f"providers.{name}", provider_config)
def remove_provider(self, name: str) -> bool:
"""Remove a provider configuration.
Returns:
True if provider was removed, False if it didn't exist
"""
providers = self.get("providers", {})
if name in providers:
del providers[name]
self.set("providers", providers)
return True
return False
def resolve_model(self, model_spec: str) -> Optional[Dict[str, Any]]:
"""Resolve a model specification (provider/model) to provider config.
Args:
model_spec: Model specification in format "provider/model"
Returns:
Dictionary with resolved provider config and model name
"""
if "/" not in model_spec:
# Legacy format - try to use as provider name with empty model
provider_name = model_spec
model_name = ""
else:
provider_name, model_name = model_spec.split("/", 1)
provider_config = self.get_provider(provider_name)
if not provider_config:
return None
# Resolve environment variable references in API key
api_key = provider_config.get("api_key", "")
if api_key.startswith("${") and api_key.endswith("}"):
env_var_name = api_key[2:-1]
resolved_api_key = os.environ.get(env_var_name, "")
else:
resolved_api_key = api_key
return {
"provider_name": provider_name,
"provider_type": provider_config.get("type", provider_name),
"model_name": model_name,
"api_key": resolved_api_key,
"base_url": provider_config.get("base_url"),
}
# Resource management methods
def list_mcps(self) -> List[str]:
"""Get all configured default MCP servers."""
return self.get("defaults.mcps", [])
def add_mcp(self, name: str) -> None:
"""Add a new default MCP server."""
mcps = self.list_mcps()
if name not in mcps:
mcps.append(name)
self.set("defaults.mcps", mcps)
def remove_mcp(self, name: str) -> bool:
"""Remove a default MCP server.
Returns:
True if MCP was removed, False if it didn't exist
"""
mcps = self.list_mcps()
if name in mcps:
mcps.remove(name)
self.set("defaults.mcps", mcps)
return True
return False
def list_mcp_configurations(self) -> List[Dict[str, Any]]:
"""Get all configured MCP server configurations."""
return self.get("mcps", [])
def get_mcp_configuration(self, name: str) -> Optional[Dict[str, Any]]:
"""Get an MCP configuration by name."""
mcps = self.list_mcp_configurations()
for mcp in mcps:
if mcp.get("name") == name:
return mcp
return None
def add_mcp_configuration(self, mcp_config: Dict[str, Any]) -> None:
"""Add a new MCP server configuration."""
mcps = self.list_mcp_configurations()
# Remove existing MCP with the same name if it exists
mcps = [mcp for mcp in mcps if mcp.get("name") != mcp_config.get("name")]
# Add the new MCP
mcps.append(mcp_config)
# Save the configuration
self.set("mcps", mcps)
def remove_mcp_configuration(self, name: str) -> bool:
"""Remove an MCP server configuration.
Returns:
True if MCP was removed, False if it didn't exist
"""
mcps = self.list_mcp_configurations()
original_length = len(mcps)
# Filter out the MCP with the specified name
mcps = [mcp for mcp in mcps if mcp.get("name") != name]
if len(mcps) < original_length:
self.set("mcps", mcps)
# Also remove from defaults if it's there
self.remove_mcp(name)
return True
return False
def list_networks(self) -> List[str]:
"""Get all configured default networks."""
return self.get("defaults.networks", [])
def add_network(self, name: str) -> None:
"""Add a new default network."""
networks = self.list_networks()
if name not in networks:
networks.append(name)
self.set("defaults.networks", networks)
def remove_network(self, name: str) -> bool:
"""Remove a default network.
Returns:
True if network was removed, False if it didn't exist
"""
networks = self.list_networks()
if name in networks:
networks.remove(name)
self.set("defaults.networks", networks)
return True
return False
def list_volumes(self) -> List[str]:
"""Get all configured default volumes."""
return self.get("defaults.volumes", [])
def add_volume(self, volume: str) -> None:
"""Add a new default volume mapping."""
volumes = self.list_volumes()
if volume not in volumes:
volumes.append(volume)
self.set("defaults.volumes", volumes)
def remove_volume(self, volume: str) -> bool:
"""Remove a default volume mapping.
Returns:
True if volume was removed, False if it didn't exist
"""
volumes = self.list_volumes()
if volume in volumes:
volumes.remove(volume)
self.set("defaults.volumes", volumes)
return True
return False
def list_ports(self) -> List[int]:
"""Get all configured default ports."""
return self.get("defaults.ports", [])
def add_port(self, port: int) -> None:
"""Add a new default port."""
ports = self.list_ports()
if port not in ports:
ports.append(port)
self.set("defaults.ports", ports)
def remove_port(self, port: int) -> bool:
"""Remove a default port.
Returns:
True if port was removed, False if it didn't exist
"""
ports = self.list_ports()
if port in ports:
ports.remove(port)
self.set("defaults.ports", ports)
return True
return False

8
pyproject.toml
View File

@@ -14,6 +14,7 @@ dependencies = [
"pyyaml>=6.0.1",
"rich>=13.6.0",
"pydantic>=2.5.0",
"questionary>=2.0.0",
]
classifiers = [
"Development Status :: 3 - Alpha",
@@ -45,6 +46,13 @@ cubbix = "cubbi.cli:session_create_entry_point"
line-length = 88
target-version = "py312"
[tool.pytest.ini_options]
# Exclude integration tests by default
addopts = "-v --tb=short -m 'not integration'"
markers = [
"integration: marks tests as integration tests (deselected by default)",
]
[tool.mypy]
python_version = "3.12"
warn_return_any = true

208
test.sh Executable file
View File

@@ -0,0 +1,208 @@
#!/bin/bash
# Comprehensive test script for all cubbi images with different model combinations
# Tests single prompt/response functionality for each tool
set -e
# Configuration
TIMEOUT="180s"
TEST_PROMPT="What is 2+2?"
LOG_FILE="test_results.log"
TEMP_DIR="/tmp/cubbi_test_$$"
# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# Test matrix
declare -a IMAGES=("goose" "aider" "claudecode" "opencode" "crush")
declare -a MODELS=(
"anthropic/claude-sonnet-4-20250514"
"openai/gpt-4o"
"openrouter/openai/gpt-4o"
"litellm/gpt-oss:120b"
)
# Command templates for each tool (based on research)
declare -A COMMANDS=(
["goose"]="goose run -t '$TEST_PROMPT' --no-session --quiet"
["aider"]="aider --message '$TEST_PROMPT' --yes-always --no-fancy-input --no-check-update --no-auto-commits"
["claudecode"]="claude -p '$TEST_PROMPT'"
["opencode"]="opencode run -m %MODEL% '$TEST_PROMPT'"
["crush"]="crush run '$TEST_PROMPT'"
)
# Initialize results tracking
declare -A RESULTS
TOTAL_TESTS=0
PASSED_TESTS=0
FAILED_TESTS=0
# Setup
echo -e "${BLUE}=== Cubbi Plugin Configuration Test Suite ===${NC}"
echo "Starting comprehensive test at $(date)"
echo "Test prompt: '$TEST_PROMPT'"
echo "Timeout: $TIMEOUT"
echo ""
mkdir -p "$TEMP_DIR"
> "$LOG_FILE"
# Function to log with timestamp
log() {
echo "$(date '+%Y-%m-%d %H:%M:%S') $1" >> "$LOG_FILE"
}
# Function to run a single test
run_test() {
local image="$1"
local model="$2"
local command="$3"
# Replace %MODEL% placeholder in command
command="${command//%MODEL%/$model}"
local test_name="${image}_${model//\//_}"
local log_file="${TEMP_DIR}/${test_name}.log"
echo -ne "Testing ${BLUE}$image${NC} with ${YELLOW}$model${NC}... "
log "Starting test: $test_name"
log "Command: $command"
# Run the test with timeout
local start_time=$(date +%s)
if timeout "$TIMEOUT" uv run -m cubbi.cli session create \
-i "$image" \
-m "$model" \
--no-connect \
--no-shell \
--run "$command" > "$log_file" 2>&1; then
local end_time=$(date +%s)
local duration=$((end_time - start_time))
# Check if we got a meaningful response
if grep -q "Initial command finished (exit code: 0)" "$log_file" &&
grep -q "Command execution complete" "$log_file"; then
echo -e "${GREEN}PASS${NC} (${duration}s)"
RESULTS["$test_name"]="PASS"
((PASSED_TESTS++))
log "Test passed in ${duration}s"
else
echo -e "${RED}FAIL${NC} (no valid output)"
RESULTS["$test_name"]="FAIL_NO_OUTPUT"
((FAILED_TESTS++))
log "Test failed - no valid output"
fi
else
local end_time=$(date +%s)
local duration=$((end_time - start_time))
echo -e "${RED}FAIL${NC} (timeout/error after ${duration}s)"
RESULTS["$test_name"]="FAIL_TIMEOUT"
((FAILED_TESTS++))
log "Test failed - timeout or error after ${duration}s"
fi
((TOTAL_TESTS++))
# Save detailed log
log "=== Test output for $test_name ==="
cat "$log_file" >> "$LOG_FILE"
log "=== End test output ==="
log ""
}
# Function to print test matrix header
print_matrix_header() {
echo ""
echo -e "${BLUE}=== Test Results Matrix ===${NC}"
printf "%-15s" "Image/Model"
for model in "${MODELS[@]}"; do
# Shorten model name for display
short_model=$(echo "$model" | sed 's/.*\///')
printf "%-20s" "$short_model"
done
echo ""
printf "%-15s" "==============="
for model in "${MODELS[@]}"; do
printf "%-20s" "===================="
done
echo ""
}
# Function to print test matrix row
print_matrix_row() {
local image="$1"
printf "%-15s" "$image"
for model in "${MODELS[@]}"; do
local test_name="${image}_${model//\//_}"
local result="${RESULTS[$test_name]}"
case "$result" in
"PASS")
printf "%-20s" "$(echo -e "${GREEN}PASS${NC}")"
;;
"FAIL_NO_OUTPUT")
printf "%-20s" "$(echo -e "${RED}FAIL (no output)${NC}")"
;;
"FAIL_TIMEOUT")
printf "%-20s" "$(echo -e "${RED}FAIL (timeout)${NC}")"
;;
*)
printf "%-20s" "$(echo -e "${YELLOW}UNKNOWN${NC}")"
;;
esac
done
echo ""
}
# Main test execution
echo -e "${YELLOW}Running ${#IMAGES[@]} images × ${#MODELS[@]} models = $((${#IMAGES[@]} * ${#MODELS[@]})) total tests${NC}"
echo ""
for image in "${IMAGES[@]}"; do
echo -e "${BLUE}--- Testing $image ---${NC}"
for model in "${MODELS[@]}"; do
command="${COMMANDS[$image]}"
run_test "$image" "$model" "$command"
done
echo ""
done
# Print results summary
print_matrix_header
for image in "${IMAGES[@]}"; do
print_matrix_row "$image"
done
echo ""
echo -e "${BLUE}=== Final Summary ===${NC}"
echo "Total tests: $TOTAL_TESTS"
echo -e "Passed: ${GREEN}$PASSED_TESTS${NC}"
echo -e "Failed: ${RED}$FAILED_TESTS${NC}"
if [ $FAILED_TESTS -eq 0 ]; then
echo -e "${GREEN}All tests passed! 🎉${NC}"
exit_code=0
else
echo -e "${RED}$FAILED_TESTS tests failed${NC}"
exit_code=1
fi
echo ""
echo "Detailed logs saved to: $LOG_FILE"
echo "Test completed at $(date)"
# Cleanup
rm -rf "$TEMP_DIR"
log "Test suite completed. Total: $TOTAL_TESTS, Passed: $PASSED_TESTS, Failed: $FAILED_TESTS"
exit $exit_code

83
tests/README_integration.md Normal file
View File

@@ -0,0 +1,83 @@
# Integration Tests
This directory contains integration tests for cubbi images with different model combinations.
## Test Matrix
The integration tests cover:
- **5 Images**: goose, aider, claudecode, opencode, crush
- **4 Models**: anthropic/claude-sonnet-4-20250514, openai/gpt-4o, openrouter/openai/gpt-4o, litellm/gpt-oss:120b
- **Total**: 20 image/model combinations + additional tests
## Running Tests
### Default (Skip Integration)
```bash
# Regular tests only (integration tests excluded by default)
uv run -m pytest
# Specific test file (excluding integration)
uv run -m pytest tests/test_cli.py
```
### Integration Tests Only
```bash
# Run all integration tests (20 combinations + helpers)
uv run -m pytest -m integration
# Run specific image with all models
uv run -m pytest -m integration -k "goose"
# Run specific model with all images
uv run -m pytest -m integration -k "anthropic"
# Run single combination
uv run -m pytest -m integration -k "goose and anthropic"
# Verbose output with timing
uv run -m pytest -m integration -v -s
```
### Combined Tests
```bash
# Run both regular and integration tests
uv run -m pytest -m "not slow" # or remove the default marker exclusion
```
## Test Structure
### `test_image_model_combination`
- Parametrized test with all image/model combinations
- Tests single prompt/response functionality
- Uses appropriate command syntax for each tool
- Verifies successful completion and basic output
### `test_image_help_command`
- Tests help command for each image
- Ensures basic functionality works
### `test_all_images_available`
- Verifies all required images are built and available
## Command Templates
Each image uses its specific command syntax:
- **goose**: `goose run -t 'prompt' --no-session --quiet`
- **aider**: `aider --message 'prompt' --yes-always --no-fancy-input --no-check-update --no-auto-commits`
- **claudecode**: `claude -p 'prompt'`
- **opencode**: `opencode run -m MODEL 'prompt'`
- **crush**: `crush run 'prompt'`
## Expected Results
All tests should pass when:
1. Images are built (`uv run -m cubbi.cli image build [IMAGE]`)
2. API keys are configured (`uv run -m cubbi.cli configure`)
3. Models are accessible and working
## Debugging Failed Tests
If tests fail, check:
1. Image availability: `uv run -m cubbi.cli image list`
2. Configuration: `uv run -m cubbi.cli config list`
3. Manual test: `uv run -m cubbi.cli session create -i IMAGE -m MODEL --run "COMMAND"`

135
tests/test_integration.py Normal file
View File

@@ -0,0 +1,135 @@
"""Integration tests for cubbi images with different model combinations."""
import subprocess
import pytest
from typing import Dict
IMAGES = ["goose", "aider", "opencode"] # fixme: crush
MODELS = [
"anthropic/claude-sonnet-4-20250514",
"openai/gpt-4o",
"openrouter/openai/gpt-4o",
"litellm/gpt-oss:120b",
]
# Command templates for each tool (based on research)
COMMANDS: Dict[str, str] = {
"goose": "goose run -t '{prompt}' --no-session --quiet",
"aider": "aider --message '{prompt}' --yes-always --no-fancy-input --no-check-update --no-auto-commits",
"opencode": "opencode run '{prompt}'",
"crush": "crush run '{prompt}'",
}
def run_cubbi_command(
image: str, model: str, command: str, timeout: int = 20
) -> subprocess.CompletedProcess:
"""Run a cubbi command with specified image, model, and command."""
full_command = [
"uv",
"run",
"-m",
"cubbi.cli",
"session",
"create",
"-i",
image,
"-m",
model,
"--no-connect",
"--no-shell",
"--run",
command,
]
return subprocess.run(
full_command,
capture_output=True,
text=True,
timeout=timeout,
cwd="/home/tito/code/monadical/cubbi",
)
def is_successful_response(result: subprocess.CompletedProcess) -> bool:
"""Check if the cubbi command completed successfully."""
# Check for successful completion markers
return (
result.returncode == 0
and "Initial command finished (exit code: 0)" in result.stdout
and "Command execution complete" in result.stdout
)
@pytest.mark.integration
@pytest.mark.parametrize("image", IMAGES)
@pytest.mark.parametrize("model", MODELS)
def test_image_model_combination(image: str, model: str):
"""Test each image with each model using appropriate command syntax."""
prompt = "What is 2+2?"
# Get the command template for this image
command_template = COMMANDS[image]
# For opencode, we need to substitute the model in the command
if image == "opencode":
command = command_template.format(prompt=prompt, model=model)
else:
command = command_template.format(prompt=prompt)
# Run the test with timeout handling
try:
result = run_cubbi_command(image, model, command)
except subprocess.TimeoutExpired:
pytest.fail(f"Test timed out after 20s for {image} with {model}")
# Check if the command was successful
assert is_successful_response(result), (
f"Failed to run {image} with {model}. "
f"Return code: {result.returncode}\n"
f"Stdout: {result.stdout}\n"
f"Stderr: {result.stderr}"
)
@pytest.mark.integration
def test_all_images_available():
"""Test that all required images are available for testing."""
# Run image list command
result = subprocess.run(
["uv", "run", "-m", "cubbi.cli", "image", "list"],
capture_output=True,
text=True,
timeout=30,
cwd="/home/tito/code/monadical/cubbi",
)
assert result.returncode == 0, f"Failed to list images: {result.stderr}"
for image in IMAGES:
assert image in result.stdout, f"Image {image} not found in available images"
@pytest.mark.integration
def test_claudecode():
"""Test Claude Code without model preselection since it only supports Anthropic."""
command = "claude -p hello"
try:
result = run_cubbi_command("claudecode", MODELS[0], command, timeout=20)
except subprocess.TimeoutExpired:
pytest.fail("Claude Code help command timed out after 20s")
assert is_successful_response(result), (
f"Failed to run Claude Code help command. "
f"Return code: {result.returncode}\n"
f"Stdout: {result.stdout}\n"
f"Stderr: {result.stderr}"
)
if __name__ == "__main__":
# Allow running the test file directly for development
pytest.main([__file__, "-v", "-m", "integration"])

14
tests/test_integration_docker.py
View File

@@ -24,20 +24,6 @@ def execute_command_in_container(container_id, command):
def wait_for_container_init(container_id, timeout=5.0, poll_interval=0.1):
"""
Wait for a Cubbi container to complete initialization by polling /cubbi/init.status.
Args:
container_id: Docker container ID
timeout: Maximum time to wait in seconds (default: 5.0)
poll_interval: Time between polls in seconds (default: 0.1)
Returns:
bool: True if initialization completed, False if timed out
Raises:
subprocess.CalledProcessError: If docker exec command fails
"""
start_time = time.time()
while time.time() - start_time < timeout:

37
uv.lock generated
View File

@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 2
requires-python = ">=3.12"
[[package]]
@@ -84,6 +84,7 @@ dependencies = [
{ name = "docker" },
{ name = "pydantic" },
{ name = "pyyaml" },
{ name = "questionary" },
{ name = "rich" },
{ name = "typer" },
]
@@ -107,6 +108,7 @@ requires-dist = [
{ name = "pydantic", specifier = ">=2.5.0" },
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=7.4.0" },
{ name = "pyyaml", specifier = ">=6.0.1" },
{ name = "questionary", specifier = ">=2.0.0" },
{ name = "rich", specifier = ">=13.6.0" },
{ name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.9" },
{ name = "typer", specifier = ">=0.9.0" },
@@ -221,6 +223,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/88/5f/e351af9a41f866ac3f1fac4ca0613908d9a41741cfcf2228f4ad853b697d/pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669", size = 20556, upload-time = "2024-04-20T21:34:40.434Z" },
]
[[package]]
name = "prompt-toolkit"
version = "3.0.51"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "wcwidth" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bb/6e/9d084c929dfe9e3bfe0c6a47e31f78a25c54627d64a66e884a8bf5474f1c/prompt_toolkit-3.0.51.tar.gz", hash = "sha256:931a162e3b27fc90c86f1b48bb1fb2c528c2761475e57c9c06de13311c7b54ed", size = 428940, upload-time = "2025-04-15T09:18:47.731Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ce/4f/5249960887b1fbe561d9ff265496d170b55a735b76724f10ef19f9e40716/prompt_toolkit-3.0.51-py3-none-any.whl", hash = "sha256:52742911fde84e2d423e2f9a4cf1de7d7ac4e51958f648d9540e0fb8db077b07", size = 387810, upload-time = "2025-04-15T09:18:44.753Z" },
]
[[package]]
name = "pydantic"
version = "2.10.6"
@@ -337,6 +351,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/fa/de/02b54f42487e3d3c6efb3f89428677074ca7bf43aae402517bc7cca949f3/PyYAML-6.0.2-cp313-cp313-win_amd64.whl", hash = "sha256:8388ee1976c416731879ac16da0aff3f63b286ffdd57cdeb95f3f2e085687563", size = 156446, upload-time = "2024-08-06T20:33:04.33Z" },
]
[[package]]
name = "questionary"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "prompt-toolkit" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a8/b8/d16eb579277f3de9e56e5ad25280fab52fc5774117fb70362e8c2e016559/questionary-2.1.0.tar.gz", hash = "sha256:6302cdd645b19667d8f6e6634774e9538bfcd1aad9be287e743d96cacaf95587", size = 26775, upload-time = "2024-12-29T11:49:17.802Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ad/3f/11dd4cd4f39e05128bfd20138faea57bec56f9ffba6185d276e3107ba5b2/questionary-2.1.0-py3-none-any.whl", hash = "sha256:44174d237b68bc828e4878c763a9ad6790ee61990e0ae72927694ead57bab8ec", size = 36747, upload-time = "2024-12-29T11:49:16.734Z" },
]
[[package]]
name = "requests"
version = "2.32.3"
@@ -431,3 +457,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/aa/63/e53da845320b757bf
wheels = [
{ url = "https://files.pythonhosted.org/packages/c8/19/4ec628951a74043532ca2cf5d97b7b14863931476d117c471e8e2b1eb39f/urllib3-2.3.0-py3-none-any.whl", hash = "sha256:1cee9ad369867bfdbbb48b7dd50374c0967a0bb7710050facf0dd6911440e3df", size = 128369, upload-time = "2024-12-22T07:47:28.074Z" },
]
[[package]]
name = "wcwidth"
version = "0.2.13"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/6c/63/53559446a878410fc5a5974feb13d31d78d752eb18aeba59c7fef1af7598/wcwidth-0.2.13.tar.gz", hash = "sha256:72ea0c06399eb286d978fdedb6923a9eb47e1c486ce63e9b4e64fc18303972b5", size = 101301, upload-time = "2024-01-06T02:10:57.829Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fd/84/fd2ba7aafacbad3c4201d395674fc6348826569da3c0937e75505ead3528/wcwidth-0.2.13-py2.py3-none-any.whl", hash = "sha256:3da69048e4540d84af32131829ff948f1e022c1c6bdb8d6102117aac784f6859", size = 34166, upload-time = "2024-01-06T02:10:55.763Z" },
]