#!/usr/bin/env python3 """CodeAct executor for Hyperlight Sandbox. Discovers available host tools (mirroring Copilot CLI built-in tools), registers them in a Hyperlight micro-VM, and executes Python code that chains tools via call_tool(). Tool names match the Copilot CLI built-in tools so the agent can use familiar names inside the sandbox: Copilot CLI tool | Sandbox call_tool() name -------------------|--------------------------- view | view create | create edit | edit glob | glob grep / rg | grep bash | bash web_fetch | web_fetch sql | sql github-mcp-server | github_api Modes: --discover List available tools as a JSON manifest. --instructions Print call_tool() reference for discovered tools. --code / --code-file Execute sandboxed code with registered tools. Examples: python3 codeact.py --discover python3 codeact.py --instructions python3 codeact.py --auto --code 'print(call_tool("view", path="README.md")[:100])' python3 codeact.py --auto --workspace . --code-file analysis.py """ from __future__ import annotations import argparse import json import os import re import shutil import sqlite3 import subprocess import sys from pathlib import Path from typing import Any # --------------------------------------------------------------------------- # Tool discovery # --------------------------------------------------------------------------- def discover_tools() -> list[dict[str, Any]]: """Discover tools available on the host, named to match Copilot CLI.""" tools: list[dict[str, Any]] = [] # -- view (equivalent to Copilot CLI "view" tool) -- tools.append({ "name": "view", "cli_equivalent": "view", "description": "Read file contents or list a directory. " "Pass view_range=[start,end] to read specific lines.", "parameters": { "path": {"type": "string", "required": True, "description": "File or directory path."}, "view_range": {"type": "array", "required": False, "description": "[start_line, end_line] 1-indexed. " "Omit to read the full file."}, }, "implementation": {"type": "builtin"}, }) # -- create (equivalent to Copilot CLI "create" tool) -- tools.append({ "name": "create", "cli_equivalent": "create", "description": "Create a new file with the given content. " "Fails if the file already exists.", "parameters": { "path": {"type": "string", "required": True, "description": "Path for the new file."}, "file_text": {"type": "string", "required": True, "description": "Content to write."}, }, "implementation": {"type": "builtin"}, }) # -- edit (equivalent to Copilot CLI "edit" tool) -- tools.append({ "name": "edit", "cli_equivalent": "edit", "description": "Replace exactly one occurrence of old_str with new_str " "in a file. Use for surgical edits.", "parameters": { "path": {"type": "string", "required": True, "description": "File to edit."}, "old_str": {"type": "string", "required": True, "description": "Exact text to find (must match once)."}, "new_str": {"type": "string", "required": True, "description": "Replacement text."}, }, "implementation": {"type": "builtin"}, }) # -- glob (equivalent to Copilot CLI "glob" tool) -- tools.append({ "name": "glob", "cli_equivalent": "glob", "description": "Find files matching a glob pattern (max 200 results).", "parameters": { "pattern": {"type": "string", "required": True, "description": "Glob pattern, e.g. '**/*.py'."}, "paths": {"type": "string", "required": False, "default": ".", "description": "Base directory for the search."}, }, "implementation": {"type": "builtin"}, }) # -- bash (equivalent to Copilot CLI "bash" tool) -- tools.append({ "name": "bash", "cli_equivalent": "bash", "description": "Execute a shell command. Returns dict with stdout, " "stderr, returncode. HIGH RISK: allows arbitrary commands.", "parameters": { "command": {"type": "string", "required": True, "description": "Shell command to run."}, "timeout": {"type": "number", "required": False, "default": 30, "description": "Max seconds before kill."}, }, "implementation": {"type": "builtin"}, "risk": "high", }) # -- sql (equivalent to Copilot CLI "sql" tool) -- tools.append({ "name": "sql", "cli_equivalent": "sql", "description": "Execute a SQL query against a SQLite database. " "Returns rows as list of dicts.", "parameters": { "query": {"type": "string", "required": True, "description": "SQL query to execute."}, "db_path": {"type": "string", "required": False, "default": ":memory:", "description": "Path to SQLite database file."}, }, "implementation": {"type": "builtin"}, }) # -- grep (equivalent to Copilot CLI "grep" / "rg" tool) -- if shutil.which("rg"): tools.append({ "name": "grep", "cli_equivalent": "grep / rg", "description": "Search file contents with ripgrep. Returns matching lines.", "parameters": { "pattern": {"type": "string", "required": True, "description": "Regex search pattern."}, "paths": {"type": "string", "required": False, "default": ".", "description": "Directory to search."}, "glob": {"type": "string", "required": False, "description": "File-type glob filter, e.g. '*.py'."}, "context_lines": {"type": "number", "required": False, "default": 0, "description": "Lines of context around each match."}, }, "implementation": {"type": "builtin"}, }) # -- web_fetch (equivalent to Copilot CLI "web_fetch" tool) -- if shutil.which("curl"): tools.append({ "name": "web_fetch", "cli_equivalent": "web_fetch", "description": "Fetch a URL. HTML is auto-converted to plain text " "and capped at max_length chars.", "parameters": { "url": {"type": "string", "required": True, "description": "URL to fetch (http/https only)."}, "method": {"type": "string", "required": False, "default": "GET", "description": "HTTP method."}, "headers": {"type": "object", "required": False, "description": "Dict of HTTP headers."}, "data": {"type": "string", "required": False, "description": "Request body."}, "max_length": {"type": "number", "required": False, "default": 20000}, }, "implementation": {"type": "builtin"}, }) # -- github_api (bridges to github-mcp-server tools via gh CLI) -- if shutil.which("gh"): tools.append({ "name": "github_api", "cli_equivalent": "github-mcp-server-*", "description": "Call the GitHub REST API via the gh CLI. " "Covers repos, issues, PRs, commits, actions, search.", "parameters": { "endpoint": {"type": "string", "required": True, "description": "API path, e.g. '/repos/owner/repo/issues'."}, "method": {"type": "string", "required": False, "default": "GET", "description": "HTTP method (GET, POST, PATCH, DELETE)."}, "body": {"type": "string", "required": False, "description": "JSON request body for POST/PATCH."}, }, "implementation": {"type": "builtin"}, }) # -- mcp_call (bridge to MCP servers when .mcp.json exists) -- mcp_cfg = _load_mcp_config() if mcp_cfg.get("servers"): server_names = list(mcp_cfg["servers"].keys()) tools.append({ "name": "mcp_call", "cli_equivalent": "MCP servers", "description": ( "Call a tool on an MCP server. Available servers: " + ", ".join(server_names) + ". Use call_tool('mcp_call', server='name', tool='tool_name', key=val) " "to invoke. Returns the tool result as a string." ), "parameters": { "server": {"type": "string", "required": True, "description": f"MCP server name. One of: {', '.join(server_names)}"}, "tool": {"type": "string", "required": True, "description": "Tool name on the MCP server."}, }, "implementation": {"type": "builtin"}, }) return tools def discover_mcp_servers() -> list[dict[str, Any]]: """Scan well-known locations for MCP server configs.""" servers: list[dict[str, Any]] = [] search_paths = [ Path(".mcp.json"), Path(".vscode/mcp.json"), Path(".github/copilot/mcp.json"), Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "mcp/mcp.json", ] for p in search_paths: if not p.exists(): continue try: cfg = json.loads(p.read_text()) for name, scfg in cfg.get("servers", cfg.get("mcpServers", {})).items(): servers.append({"name": name, "source": str(p), "config": scfg}) except Exception as exc: print(f"⚠ failed to load MCP config {p}: {exc}", file=sys.stderr)# --------------------------------------------------------------------------- # User config: enable/disable + custom tool loading # --------------------------------------------------------------------------- def _user_config_dir() -> Path: """Resolve the user config directory (CODEACT_CONFIG_DIR overrides).""" override = os.environ.get("CODEACT_CONFIG_DIR") if override: return Path(override) base = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config") return Path(base) / "codeact" def _split_csv(value: str | None) -> list[str]: if not value: return [] return [s.strip() for s in value.split(",") if s.strip()] def _load_user_tool(py_file: Path) -> dict[str, Any] | None: """Load a single user tool .py file. Returns a tool def, or None on error.""" import importlib.util name = py_file.stem try: spec = importlib.util.spec_from_file_location(f"codeact_user_{name}", py_file) if spec is None or spec.loader is None: return None mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) except Exception as exc: print(f"⚠ failed to load custom tool {py_file}: {exc}", file=sys.stderr) return None meta = getattr(mod, "TOOL", {}) or {} func_name = meta.get("function", "run") func = getattr(mod, func_name, None) if not callable(func): print(f"⚠ custom tool {py_file} has no callable '{func_name}'", file=sys.stderr) return None return { "name": meta.get("name", name), "cli_equivalent": "user", "description": meta.get("description", (mod.__doc__ or "User tool").strip()), "parameters": meta.get("parameters", {}), "implementation": { "type": "user", "module_path": str(py_file), "function": func_name, }, } def apply_user_config(tools: list[dict[str, Any]]) -> list[dict[str, Any]]: """Filter built-in tools per allow/deny config and append custom tools.""" cfg_dir = _user_config_dir() cfg_file = cfg_dir / "config.json" cfg: dict[str, Any] = {} if cfg_file.is_file(): try: cfg = json.loads(cfg_file.read_text()) except Exception as exc: print(f"⚠ invalid {cfg_file}: {exc}", file=sys.stderr) enabled = set(_split_csv(os.environ.get("CODEACT_TOOLS")) or cfg.get("enabled", [])) disabled = set(_split_csv(os.environ.get("CODEACT_DISABLE")) or cfg.get("disabled", [])) filtered = [] for t in tools: n = t["name"] if disabled and n in disabled: continue if enabled and n not in enabled: continue filtered.append(t) tools_dir = cfg_dir / "tools" if tools_dir.is_dir(): for py in sorted(tools_dir.glob("*.py")): if py.name.startswith("_"): continue tdef = _load_user_tool(py) if tdef is None: continue if disabled and tdef["name"] in disabled: continue if enabled and tdef["name"] not in enabled: continue filtered.append(tdef) return filtered # --------------------------------------------------------------------------- # Built-in host-side tool handlers # --------------------------------------------------------------------------- _WORKSPACE_ROOT: Path | None = None _SQLITE_CONNECTIONS: dict[str, sqlite3.Connection] = {} def _check_workspace(p: Path) -> Path: """Resolve a path and verify it falls inside the workspace root.""" resolved = p.expanduser().resolve() if _WORKSPACE_ROOT is not None: ws = str(_WORKSPACE_ROOT) rp = str(resolved) if rp != ws and not rp.startswith(ws + os.sep): raise PermissionError( f"Path {resolved} is outside workspace {_WORKSPACE_ROOT}") return resolved _MAX_VIEW_BYTES = 50 * 1024 * 1024 # 50 MB def _view(path: str = "", view_range: list[int] | None = None) -> str: p = _check_workspace(Path(path)) if p.is_dir(): entries = sorted(p.iterdir()) return "\n".join(e.name + ("/" if e.is_dir() else "") for e in entries) if p.stat().st_size > _MAX_VIEW_BYTES: raise ValueError(f"File too large ({p.stat().st_size} bytes, max {_MAX_VIEW_BYTES})") text = p.read_text() if view_range and len(view_range) == 2: lines = text.splitlines(keepends=True) start = max(view_range[0] - 1, 0) end = view_range[1] if view_range[1] != -1 else len(lines) return "".join(lines[start:end]) return text def _create(path: str = "", file_text: str = "") -> str: p = _check_workspace(Path(path)) if p.exists(): raise FileExistsError(f"{path} already exists — use edit to modify.") p.parent.mkdir(parents=True, exist_ok=True) p.write_text(file_text) return f"Created {path} ({len(file_text)} bytes)" def _edit(path: str = "", old_str: str = "", new_str: str = "") -> str: p = _check_workspace(Path(path)) text = p.read_text() count = text.count(old_str) if count == 0: raise ValueError("old_str not found in file.") if count > 1: raise ValueError(f"old_str matches {count} times — must be unique.") p.write_text(text.replace(old_str, new_str, 1)) return f"Edited {path}" def _glob(pattern: str = "**/*", paths: str = ".") -> list[str]: base = _check_workspace(Path(paths)) # Support brace expansion: {a,b} → run multiple globs and merge if "{" in pattern and "}" in pattern: prefix = pattern[:pattern.index("{")] rest = pattern[pattern.index("{"):] brace_end = rest.index("}") + 1 alternatives = rest[1:brace_end-1].split(",") suffix = rest[brace_end:] expanded = [prefix + alt + suffix for alt in alternatives] all_matches: list[str] = [] for p in expanded: all_matches.extend(str(m) for m in base.glob(p) if m.is_file()) matches = sorted(set(all_matches))[:200] else: matches = sorted(str(p) for p in base.glob(pattern) if p.is_file())[:200] # Return workspace-relative paths if _WORKSPACE_ROOT is not None: root = str(_WORKSPACE_ROOT) + "/" matches = [m[len(root):] if m.startswith(root) else m for m in matches] return matches def _bash(command: str = "", timeout: int = 30) -> dict[str, Any]: r = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=int(timeout)) return {"stdout": r.stdout, "stderr": r.stderr, "returncode": r.returncode} def _sql(query: str = "", db_path: str = ":memory:") -> list[dict[str, Any]]: if db_path != ":memory:": _check_workspace(Path(db_path)) key = db_path if key not in _SQLITE_CONNECTIONS: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row _SQLITE_CONNECTIONS[key] = conn conn = _SQLITE_CONNECTIONS[key] cur = conn.execute(query) conn.commit() if cur.description: rows = cur.fetchmany(10000) # cap at 10K rows to prevent OOM return [dict(row) for row in rows] return [{"rows_affected": cur.rowcount}] def _grep(pattern: str = "", paths: str = ".", glob: str = "", context_lines: int = 0) -> str: cmd = ["rg", "--no-heading", "--line-number", pattern, paths] if glob: cmd += ["--glob", glob] if context_lines: cmd += ["-C", str(int(context_lines))] r = subprocess.run(cmd, capture_output=True, text=True, timeout=30) return r.stdout def _web_fetch(url: str = "", method: str = "GET", headers: dict[str, str] | None = None, data: str = "", max_length: int = 20000) -> str: if url and not (url.startswith("http://") or url.startswith("https://")): raise ValueError(f"Blocked URL scheme (only http/https allowed): {url[:50]}") cmd = ["curl", "-sS", "-L", "--max-time", "30", "--max-redirs", "5", "-X", method] for k, v in (headers or {}).items(): cmd += ["-H", f"{k}: {v}"] if data: cmd += ["-d", data] cmd.append(url) r = subprocess.run(cmd, capture_output=True, text=True, timeout=35) body = r.stdout if " int(max_length): body = body[:int(max_length)] + f"\n\n[truncated at {max_length} chars]" return body def _html_to_text(html: str) -> str: """Convert HTML to plain text using stdlib html.parser.""" from html.parser import HTMLParser class _Extractor(HTMLParser): def __init__(self): super().__init__() self._parts: list[str] = [] self._skip = False def handle_starttag(self, tag, attrs): if tag in ("script", "style"): self._skip = True def handle_endtag(self, tag): if tag in ("script", "style"): self._skip = False def handle_data(self, data): if not self._skip: self._parts.append(data) parser = _Extractor() try: parser.feed(html) except Exception: text = re.sub(r"<[^>]*>", " ", html) return re.sub(r"\s+", " ", text).strip() text = " ".join(parser._parts) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n\s*\n", "\n\n", text) return text.strip() def _github_api(endpoint: str = "", method: str = "GET", body: str = "") -> str: cmd = ["gh", "api", "-X", method, endpoint] if body and method.upper() in ("POST", "PATCH", "PUT"): cmd += ["--input", "-"] r = subprocess.run(cmd, input=body, capture_output=True, text=True, timeout=30) else: r = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if r.returncode != 0: raise RuntimeError(f"gh api failed: {r.stderr.strip()}") return r.stdout # --------------------------------------------------------------------------- # MCP bridge # --------------------------------------------------------------------------- _MCP_CONFIG: dict[str, Any] | None = None def _load_mcp_config() -> dict[str, Any]: """Lazy-load MCP config from the bridge module.""" global _MCP_CONFIG if _MCP_CONFIG is not None: return _MCP_CONFIG bridge_path = Path(__file__).resolve().parent.parent.parent.parent / "scripts" / "mcp-bridge.py" if not bridge_path.is_file(): _MCP_CONFIG = {"servers": {}} return _MCP_CONFIG import importlib.util spec = importlib.util.spec_from_file_location("mcp_bridge", bridge_path) if spec is None or spec.loader is None: _MCP_CONFIG = {"servers": {}} return _MCP_CONFIG mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) _MCP_CONFIG = mod._load_mcp_config() _MCP_CONFIG["_bridge_mod"] = mod return _MCP_CONFIG def _mcp_call(server: str = "", tool: str = "", **kwargs) -> str: """Call an MCP server tool. Available when .mcp.json defines servers.""" config = _load_mcp_config() bridge = config.get("_bridge_mod") if bridge is None: return json.dumps({"error": "MCP bridge not available"}) return bridge.call_mcp(config, server, tool, kwargs) _BUILTIN_HANDLERS: dict[str, Any] = { "view": _view, "create": _create, "edit": _edit, "glob": _glob, "bash": _bash, "sql": _sql, "grep": _grep, "web_fetch": _web_fetch, "github_api": _github_api, "mcp_call": _mcp_call, } # --------------------------------------------------------------------------- # Handler factory for custom tool definitions # --------------------------------------------------------------------------- def _make_handler(tool_def: dict[str, Any]): """Return a host-side callback for a tool definition.""" impl = tool_def.get("implementation", {}) impl_type = impl.get("type", "builtin") if impl_type == "builtin": name = tool_def["name"] if name in _BUILTIN_HANDLERS: return _BUILTIN_HANDLERS[name] raise ValueError(f"No built-in handler for '{name}'") if impl_type == "shell": tpl = impl["command_template"] def _shell(**kw: Any) -> str: cmd = tpl.format(**kw) r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=impl.get("timeout", 30)) if r.returncode != 0: raise RuntimeError(r.stderr.strip() or f"exit {r.returncode}") return r.stdout return _shell if impl_type == "python": code = impl["code"] def _py(**kw: Any) -> Any: ns: dict[str, Any] = dict(kw) exec(code, {"__builtins__": __builtins__, "Path": Path, "json": json, "re": re}, ns) return ns.get("result") return _py if impl_type == "user": import importlib.util path = Path(impl["module_path"]) func_name = impl.get("function", "run") spec = importlib.util.spec_from_file_location( f"codeact_user_{path.stem}", path) if spec is None or spec.loader is None: raise RuntimeError(f"could not load user tool {path}") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) fn = getattr(mod, func_name) def _user(**kw: Any) -> Any: return fn(**kw) return _user raise ValueError(f"Unknown implementation type: {impl_type}") # --------------------------------------------------------------------------- # Instruction builder # --------------------------------------------------------------------------- def build_instructions(tools: list[dict[str, Any]]) -> str: """Generate compact call_tool() reference for LLM prompts.""" lines = ["### Sandbox tools (use `call_tool(name, **kwargs)` — no import needed)"] lines.append("") for t in tools: sig_parts = [] for pname, pdef in t.get("parameters", {}).items(): if pdef.get("required"): sig_parts.append(f"{pname}=...") else: sig_parts.append(f"{pname}={pdef.get('default', '...')!r}") sig = ", ".join(sig_parts) desc = t.get("description", "").split(".")[0] lines.append(f"- `call_tool(\"{t['name']}\", {sig})` — {desc}") # Always document mcp_call even if no .mcp.json at install time tool_names = {t["name"] for t in tools} if "mcp_call" not in tool_names: lines.append('- `call_tool("mcp_call", server=..., tool=..., **kwargs)` — Call an MCP server tool (available when .mcp.json is configured)') lines.append("") lines.append("Return types: `glob`→**list of strings**, `view`→**string**, " "`bash`→**dict** (stdout/stderr/returncode), `mcp_call`→**string**") lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Dependency installer # --------------------------------------------------------------------------- def _try_install(package: str, import_name: str): """Prompt the user to install a missing package. Returns the module or None.""" import importlib print(f"\n⚠ {package} is not installed.", file=sys.stderr) # Try uv first if shutil.which("uv"): answer = input(f"Install {package} with uv? [Y/n] ").strip().lower() if answer in ("", "y", "yes"): r = subprocess.run( ["uv", "pip", "install", package], capture_output=True, text=True, ) if r.returncode == 0: print(f"✓ Installed {package}", file=sys.stderr) return importlib.import_module(import_name) print(f"uv install failed: {r.stderr.strip()}", file=sys.stderr) else: # Offer to install uv itself answer = input("uv is not installed. Install uv? (recommended) [Y/n] ").strip().lower() if answer in ("", "y", "yes"): r = subprocess.run( ["bash", "-c", "curl -LsSf https://astral.sh/uv/install.sh | sh"], capture_output=True, text=True, ) if r.returncode == 0: uv_path = Path.home() / ".local" / "bin" / "uv" if uv_path.exists(): print("✓ Installed uv", file=sys.stderr) r2 = subprocess.run( [str(uv_path), "pip", "install", package], capture_output=True, text=True, ) if r2.returncode == 0: print(f"✓ Installed {package}", file=sys.stderr) return importlib.import_module(import_name) print("uv install failed. Trying pip...", file=sys.stderr) # Fallback to pip answer = input(f"Install {package} with pip? [Y/n] ").strip().lower() if answer in ("", "y", "yes"): r = subprocess.run( [sys.executable, "-m", "pip", "install", package], capture_output=True, text=True, ) if r.returncode == 0: print(f"✓ Installed {package}", file=sys.stderr) return importlib.import_module(import_name) print(f"pip install failed: {r.stderr.strip()}", file=sys.stderr) print(f"\n✗ Could not install {package}. Install manually:", file=sys.stderr) print(f" uv pip install {package}", file=sys.stderr) print(f" pip install {package}", file=sys.stderr) return None # Minimum version that exports the Sandbox class and Wasm backend. _MIN_VERSION = "0.3.0" # Maximum Python version with pre-built Wasm backend wheels. _MAX_PYTHON = (3, 13) def _check_python_version() -> None: """Warn if the current Python is too new for pre-built Wasm wheels.""" if sys.version_info[:2] > _MAX_PYTHON: max_str = f"{_MAX_PYTHON[0]}.{_MAX_PYTHON[1]}" cur_str = f"{sys.version_info.major}.{sys.version_info.minor}" print( f"\n⚠ Python {cur_str} detected. The hyperlight-sandbox Wasm backend " f"only ships wheels for Python ≤{max_str}.\n" f" Re-run with: uv run --python {max_str} " f"--with 'hyperlight-sandbox[wasm,python_guest]>={_MIN_VERSION}' " f"python3 scripts/codeact.py ...\n", file=sys.stderr, ) sys.exit(1) def _import_sandbox(): """Import the Sandbox class, giving actionable errors on failure.""" # 1. Check Python version first — avoids confusing resolution errors. _check_python_version() # 2. Try the import. try: from hyperlight_sandbox import Sandbox return Sandbox except ImportError: pass # 3. Check if hyperlight_sandbox is installed but too old (stub package). try: import hyperlight_sandbox as _hl ver = getattr(_hl, "__version__", "0.0.0") if not hasattr(_hl, "Sandbox"): print( f"\n✗ hyperlight-sandbox {ver} is installed but does not " f"export Sandbox (likely a stub package).\n" f" Install version ≥{_MIN_VERSION}:\n" f" uv run --python {_MAX_PYTHON[0]}.{_MAX_PYTHON[1]} " f"--with 'hyperlight-sandbox[wasm,python_guest]>={_MIN_VERSION}' " f"python3 scripts/codeact.py ...\n", file=sys.stderr, ) sys.exit(1) except ImportError: pass # 4. Not installed at all — try interactive install. _mod = _try_install( f"hyperlight-sandbox[wasm,python_guest]>={_MIN_VERSION}", "hyperlight_sandbox", ) if _mod is None: sys.exit(1) return _mod.Sandbox # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main() -> None: ap = argparse.ArgumentParser( description="CodeAct executor — discover Copilot CLI-equivalent tools, " "register them in a Hyperlight sandbox, and run chained code.") ap.add_argument("--discover", action="store_true", help="Print a JSON manifest of available host tools.") ap.add_argument("--instructions", action="store_true", help="Print call_tool() reference for LLM prompts.") ap.add_argument("--manifest", help="Path to a JSON tool manifest (from --discover).") ap.add_argument("--code", help="Inline Python code to execute.") ap.add_argument("--code-file", help="Path to a .py file to execute.") ap.add_argument("--auto", action="store_true", help="Auto-discover tools before running code.") ap.add_argument("--stdin", action="store_true", help="Read JSON config (tools + code) from stdin.") ap.add_argument("--output", help="Write discovery output to this file.") ap.add_argument("--module-path", help="Path to the Hyperlight guest .aot / .wasm module.") ap.add_argument("--heap-size", help="Sandbox heap (e.g. '25Mi').") ap.add_argument("--stack-size", help="Sandbox stack (e.g. '35Mi').") ap.add_argument("--allowed-domains", nargs="*", default=[], help="Domains reachable via sandbox http_get/http_post.") ap.add_argument("--raw", action="store_true", help="On success, print stdout/stderr directly instead of JSON envelope.") ap.add_argument("--workspace", help="Restrict file/sql tools to this directory tree.") args = ap.parse_args() # ---- discovery / instructions ---- if args.discover or args.instructions: tools = apply_user_config(discover_tools()) mcp = discover_mcp_servers() if args.instructions: print(build_instructions(tools)) return manifest: dict[str, Any] = {"tools": tools} if mcp: manifest["mcp_servers"] = mcp blob = json.dumps(manifest, indent=2) if args.output: Path(args.output).write_text(blob) print(f"Manifest written to {args.output}", file=sys.stderr) else: print(blob) return # ---- execution ---- Sandbox = _import_sandbox() global _WORKSPACE_ROOT if args.workspace: _WORKSPACE_ROOT = Path(args.workspace).resolve() # Clear global state from any prior run for conn in _SQLITE_CONNECTIONS.values(): try: conn.close() except Exception: pass _SQLITE_CONNECTIONS.clear() config: dict[str, Any] = {} if args.stdin: config = json.load(sys.stdin) elif args.manifest: config = json.loads(Path(args.manifest).read_text()) elif args.auto: config["tools"] = apply_user_config(discover_tools()) code = args.code if args.code_file: code = Path(args.code_file).read_text() elif not code and "code" in config: code = config["code"] if not code: ap.error("No code provided. Use --code, --code-file, or 'code' in JSON.") tools = config.get("tools", []) try: sb_kw: dict[str, Any] = {"backend": "wasm"} if args.module_path: sb_kw["module_path"] = args.module_path else: sb_kw["module"] = "python_guest.path" if args.heap_size: sb_kw["heap_size"] = args.heap_size if args.stack_size: sb_kw["stack_size"] = args.stack_size sandbox = Sandbox(**sb_kw) for tdef in tools: handler = _make_handler(tdef) sandbox.register_tool(tdef["name"], handler) for domain in args.allowed_domains + config.get("allowed_domains", []): if isinstance(domain, list): sandbox.allow_domain(domain[0], methods=domain[1:]) else: sandbox.allow_domain(domain) result = sandbox.run(code) output = { "stdout": result.stdout, "stderr": result.stderr, "exit_code": result.exit_code, "success": result.success, } except Exception as exc: output = { "stdout": "", "stderr": f"Executor error: {exc}", "exit_code": 1, "success": False, } if args.raw and output["success"]: sys.stdout.write(output["stdout"]) if output["stderr"]: sys.stderr.write(output["stderr"]) sys.exit(0) elif args.raw and not output["success"]: if output["stdout"]: sys.stdout.write(output["stdout"]) sys.stderr.write(output["stderr"]) sys.exit(1) else: print(json.dumps(output, indent=2)) if __name__ == "__main__": main()