cli.py
1 # Copyright 2026 Alibaba Group Holding Ltd. 2 # 3 # Licensed under the Apache License, Version 2.0 (the "License"); 4 # you may not use this file except in compliance with the License. 5 # You may obtain a copy of the License at 6 # 7 # http://www.apache.org/licenses/LICENSE-2.0 8 # 9 # Unless required by applicable law or agreed to in writing, software 10 # distributed under the License is distributed on an "AS IS" BASIS, 11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 # See the License for the specific language governing permissions and 13 # limitations under the License. 14 15 from __future__ import annotations 16 17 import argparse 18 import os 19 import shutil 20 import types 21 from importlib import resources 22 from pathlib import Path 23 from typing import Any, FrozenSet, Union, get_args, get_origin 24 25 import uvicorn 26 from pydantic import BaseModel 27 28 from opensandbox_server.config import ( 29 AgentSandboxRuntimeConfig, 30 CONFIG_ENV_VAR, 31 DEFAULT_CONFIG_PATH, 32 DockerConfig, 33 EgressConfig, 34 IngressConfig, 35 KubernetesRuntimeConfig, 36 RenewIntentConfig, 37 RuntimeConfig, 38 ServerConfig, 39 StorageConfig, 40 ) 41 42 43 def _strip_optional(annotation: Any) -> Any: 44 """Unwrap Optional / Union[..., None] to the inner type.""" 45 if annotation is None: 46 return None 47 origin = get_origin(annotation) 48 args = get_args(annotation) 49 if origin is Union or origin is types.UnionType: 50 filtered = [a for a in args if a is not type(None)] 51 if len(filtered) == 1: 52 return filtered[0] 53 return annotation 54 55 56 def _is_basemodel_type(annotation: Any) -> bool: 57 inner = _strip_optional(annotation) 58 return isinstance(inner, type) and issubclass(inner, BaseModel) 59 60 EXAMPLE_FILE_MAP = { 61 "docker": "example.config.toml", 62 "docker-zh": "example.config.zh.toml", 63 "k8s": "example.config.k8s.toml", 64 "k8s-zh": "example.config.k8s.zh.toml", 65 } 66 67 68 def _build_parser() -> argparse.ArgumentParser: 69 parser = argparse.ArgumentParser( 70 description="Run the OpenSandbox server.", 71 formatter_class=argparse.RawTextHelpFormatter, 72 ) 73 parser.add_argument( 74 "--config", 75 help="Path to the server config TOML file (overrides SANDBOX_CONFIG_PATH).", 76 ) 77 parser.add_argument( 78 "--reload", 79 action="store_true", 80 help="Enable auto-reload (development only).", 81 ) 82 83 subparsers = parser.add_subparsers(dest="command") 84 85 init_parser = subparsers.add_parser( 86 "init-config", 87 help="Generate a config file from packaged examples or the schema skeleton.", 88 ) 89 init_parser.add_argument( 90 "path", 91 nargs="?", 92 default=str(DEFAULT_CONFIG_PATH), 93 help="Destination path for the config file (default: ~/.sandbox.toml).", 94 ) 95 init_parser.add_argument( 96 "--example", 97 choices=sorted(EXAMPLE_FILE_MAP), 98 help=( 99 "Packaged example to copy (docker, docker-zh, k8s, k8s-zh). " 100 "Omit to render the full skeleton with placeholders." 101 ), 102 ) 103 init_parser.add_argument( 104 "--force", 105 action="store_true", 106 help="Overwrite existing file when generating config.", 107 ) 108 109 parser.epilog = ( 110 "Subcommands:\n" 111 " init-config [path] [--example {docker,docker-zh,k8s,k8s-zh}] [--force]\n" 112 " Generate a config file. Without --example it renders the full skeleton (placeholders only).\n" 113 " --example Copy a packaged example config.\n" 114 " --force Overwrite destination if it exists.\n" 115 ) 116 return parser 117 118 119 def copy_example_config( 120 destination: str | Path | None = None, *, force: bool = False, kind: str = "default" 121 ) -> Path: 122 """Copy a packaged example config template to the target path.""" 123 if kind not in EXAMPLE_FILE_MAP: 124 supported = ", ".join(EXAMPLE_FILE_MAP) 125 raise ValueError(f"Unsupported example kind '{kind}'. Choices: {supported}") 126 127 filename = EXAMPLE_FILE_MAP[kind] 128 dest_path = Path(destination or DEFAULT_CONFIG_PATH).expanduser() 129 dest_path.parent.mkdir(parents=True, exist_ok=True) 130 if dest_path.exists() and not force: 131 raise FileExistsError(f"Config file already exists at {dest_path}. Use --force to overwrite.") 132 133 example_resource = resources.files("opensandbox_server.examples").joinpath(filename) 134 if not example_resource.is_file(): 135 raise FileNotFoundError(f"Missing packaged example config template: {filename}") 136 137 with resources.as_file(example_resource) as src_path: 138 shutil.copyfile(src_path, dest_path) 139 return dest_path 140 141 142 def render_full_config(destination: str | Path | None = None, *, force: bool = False) -> Path: 143 """ 144 Render the most complete config skeleton from config models with comments. 145 146 No defaults are prefilled; everything is emitted as placeholders so users 147 must explicitly set values. Field comments come from pydantic Field 148 descriptions to stay in sync with the schema. 149 """ 150 151 def _placeholder_for_field(field) -> str: 152 """Return a placeholder TOML value that is intentionally empty.""" 153 ann = field.annotation 154 if ann is not None: 155 origin = getattr(ann, "__origin__", None) 156 if ann is list or origin is list: 157 return "[]" 158 return '""' # string placeholder for scalars/bool/int; user must replace 159 160 def _render_section( 161 section: str, 162 model, 163 *, 164 placeholders: dict[str, str] | None = None, 165 extra_comments: list[str] | None = None, 166 dotted_nested: FrozenSet[str] | None = None, 167 ) -> str: 168 lines: list[str] = [] 169 if extra_comments: 170 lines.extend([f"# {c}" for c in extra_comments]) 171 lines.append(f"[{section}]") 172 173 placeholders = placeholders or {} 174 dotted_nested = dotted_nested or frozenset() 175 176 for field_name, field in model.model_fields.items(): 177 if _is_basemodel_type(field.annotation): 178 continue 179 key = field.alias or field_name 180 value = placeholders.get(key, _placeholder_for_field(field)) 181 if field.description: 182 lines.append(f"# {field.description}") 183 lines.append(f"{key} = {value}") 184 lines.append("") 185 186 for field_name, field in model.model_fields.items(): 187 if field_name not in dotted_nested or not _is_basemodel_type(field.annotation): 188 continue 189 inner = _strip_optional(field.annotation) 190 if not isinstance(inner, type) or not issubclass(inner, BaseModel): 191 continue 192 for sub_name, sub_field in inner.model_fields.items(): 193 sub_key = f"{field_name}.{sub_name}" 194 value = placeholders.get(sub_key, _placeholder_for_field(sub_field)) 195 if sub_field.description: 196 lines.append(f"# {sub_field.description}") 197 lines.append(f"{sub_key} = {value}") 198 lines.append("") 199 200 nested_blocks: list[str] = [] 201 for field_name, field in model.model_fields.items(): 202 if not _is_basemodel_type(field.annotation): 203 continue 204 if field_name in dotted_nested: 205 continue 206 inner = _strip_optional(field.annotation) 207 if not isinstance(inner, type) or not issubclass(inner, BaseModel): 208 continue 209 nested_path = f"{section}.{field_name}" 210 nested_blocks.append( 211 _render_section(nested_path, inner, placeholders=None, extra_comments=None) 212 ) 213 214 if nested_blocks: 215 if lines and lines[-1] == "": 216 lines.pop() 217 lines.append("") 218 lines.extend(nested_blocks) 219 220 if lines and lines[-1] == "": 221 lines.pop() 222 return "\n".join(lines) 223 224 dest_path = Path(destination or DEFAULT_CONFIG_PATH).expanduser() 225 dest_path.parent.mkdir(parents=True, exist_ok=True) 226 if dest_path.exists() and not force: 227 raise FileExistsError(f"Config file already exists at {dest_path}. Use --force to overwrite.") 228 229 sections = [ 230 "# Generated from OpenSandbox config schema. Remove sections you do not use.", 231 _render_section("server", ServerConfig), 232 _render_section( 233 "renew_intent", 234 RenewIntentConfig, 235 extra_comments=[ 236 "Renew-intent: top-level section (not under [server]). " 237 "Redis options use dotted keys in this table (redis.enabled, redis.queue_key, …)." 238 ], 239 dotted_nested=frozenset({"redis"}), 240 ), 241 _render_section("runtime", RuntimeConfig), 242 _render_section("docker", DockerConfig), 243 _render_section( 244 "egress", 245 EgressConfig, 246 extra_comments=["Used when networkPolicy is provided. Requires docker.network_mode = \"bridge\"."], 247 ), 248 _render_section( 249 "kubernetes", 250 KubernetesRuntimeConfig, 251 extra_comments=["Only used when runtime.type = \"kubernetes\""], 252 ), 253 _render_section( 254 "agent_sandbox", 255 AgentSandboxRuntimeConfig, 256 extra_comments=["Requires kubernetes.workload_provider = \"agent-sandbox\""], 257 ), 258 _render_section("ingress", IngressConfig), 259 _render_section("storage", StorageConfig), 260 ] 261 262 content = "\n\n".join(sections) + "\n" 263 dest_path.write_text(content, encoding="utf-8") 264 return dest_path 265 266 267 def main() -> None: 268 parser = _build_parser() 269 args = parser.parse_args() 270 271 if args.command == "init-config": 272 try: 273 if args.example: 274 dest = copy_example_config(args.path, force=args.force, kind=args.example) 275 print(f"Wrote example config ({args.example}) to {dest}\n") 276 else: 277 dest = render_full_config(args.path, force=args.force) 278 print(f"Wrote full config skeleton to {dest}\n") 279 except Exception as exc: # noqa: BLE001 280 print(f"Failed to write config template: {exc}\n") 281 raise SystemExit(1) 282 return 283 284 if args.config: 285 os.environ[CONFIG_ENV_VAR] = args.config 286 287 from opensandbox_server import main as server_main # local import after env is set 288 289 uvicorn.run( 290 "opensandbox_server.main:app", 291 host=server_main.app_config.server.host, 292 port=server_main.app_config.server.port, 293 reload=args.reload, 294 log_config=server_main._log_config, 295 timeout_keep_alive=server_main.app_config.server.timeout_keep_alive, 296 ) 297 298 299 if __name__ == "__main__": 300 main()