#!/usr/bin/env python3 from __future__ import annotations import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os import pathlib import platform import subprocess import sys import threading import time import urllib.error import urllib.request import webbrowser PORTAL_URL = "https://apiclo.com" CONFIG_PATH = pathlib.Path.home() / ".apiclaude" / "desktop-bridge.json" VERSION = "0.3.0" CLI_ROUTER_DEFAULT_PORT = 18441 CLI_ROUTER_HOST = "127.0.0.1" CODEX_CONNECTOR_PROVIDER_ID = "apiclaude" CODEX_CONNECTOR_START = "# >>> ApiClo Codex Connector" CODEX_CONNECTOR_END = "# <<< ApiClo Codex Connector" CODEX_MODEL_CONTEXT_WINDOW = 200000 CLI_ROUTER_ALLOWED_POST_PATHS = { "/v1/chat/completions": "/v1/chat/completions", "/chat/completions": "/v1/chat/completions", "/v1/messages": "/v1/messages", "/messages": "/v1/messages", "/v1/responses": "/v1/responses", "/responses": "/v1/responses", } CLI_CLIENTS = [ { "key": "codex", "name": "Codex", "download_url": "https://developers.openai.com/codex/quickstart", "install": 'powershell -ExecutionPolicy ByPass -c "irm https://chatgpt.com/codex/install.ps1 | iex"', "config_target": "ApiClo local profile", "setup_note": "Prepares an ApiClo profile only; Codex settings are never changed", }, { "key": "opencode", "name": "OpenCode", "download_url": "https://opencode.ai/docs/", "install": "npm install -g opencode-ai@latest", "config_target": "OpenCode OpenAI-compatible provider", "setup_note": "Prepares an ApiClo local profile; existing OpenCode config is not changed", }, { "key": "cline", "name": "Cline", "download_url": "https://docs.cline.bot/usage/cli-overview", "install": "npm install -g cline", "config_target": "Cline OpenAI-compatible provider", "setup_note": "Prepares an ApiClo local profile; existing Cline settings are not changed", }, { "key": "claude-code", "name": "Claude Code", "download_url": "https://code.claude.com/docs/en/quickstart", "install": "irm https://claude.ai/install.ps1 | iex", "config_target": "Claude Code environment", "setup_note": "Prepares a safe ApiClo profile; existing settings are not changed", }, { "key": "gemini-cli", "name": "Gemini CLI", "download_url": "https://geminicli.com/docs/get-started/installation/", "install": "npm install -g @google/gemini-cli", "config_target": "Gemini custom provider settings", "setup_note": "Prepares a safe ApiClo provider profile", }, { "key": "hermes", "name": "Hermes", "download_url": "https://hermes-agent.nousresearch.com/docs/getting-started/installation", "install": "iex (irm https://hermes-agent.nousresearch.com/install.ps1)", "config_target": "Hermes custom provider config", "setup_note": "Prepares a safe ApiClo provider profile", }, ] TERMINAL_ACTIONS = {"terminal", "terminal.run", "run_command"} ACTION_PERMISSION_BY_TYPE = { "browser": "browser", "open_url": "browser", "browser.open_url": "browser", "folder_write": "folder_write", "write_file": "folder_write", "folder.write_file": "folder_write", "terminal": "terminal", "terminal.run": "terminal", "run_command": "terminal", } TEXT_EXTENSIONS = { ".txt", ".md", ".py", ".js", ".ts", ".tsx", ".jsx", ".json", ".yaml", ".yml", ".html", ".css", ".csv", ".log", } def env_bool(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def env_int(name: str, default: int, minimum: int, maximum: int) -> int: try: parsed = int(os.getenv(name, str(default))) except (TypeError, ValueError): parsed = default return max(minimum, min(maximum, parsed)) def post_json(path: str, payload: dict, token: str | None = None) -> dict: data = json.dumps(payload).encode("utf-8") request = urllib.request.Request( PORTAL_URL.rstrip("/") + path, data=data, headers={ "Content-Type": "application/json", "User-Agent": "ApiClaudeDesktopBridge/%s (%s)" % (VERSION, platform.system()), }, method="POST", ) if token: request.add_header("Authorization", "Bearer " + token) with urllib.request.urlopen(request, timeout=30) as response: body = response.read().decode("utf-8", "replace") return json.loads(body or "{}") def get_json(path: str, token: str | None = None) -> dict: request = urllib.request.Request( PORTAL_URL.rstrip("/") + path, headers={"User-Agent": "ApiClaudeDesktopBridge/%s (%s)" % (VERSION, platform.system())}, method="GET", ) if token: request.add_header("Authorization", "Bearer " + token) with urllib.request.urlopen(request, timeout=30) as response: body = response.read().decode("utf-8", "replace") return json.loads(body or "{}") def load_config() -> dict: if not CONFIG_PATH.exists(): return {} try: return json.loads(CONFIG_PATH.read_text(encoding="utf-8")) except Exception: return {} def save_config(config: dict) -> None: CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) CONFIG_PATH.write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8") try: os.chmod(CONFIG_PATH, 0o600) except Exception: pass def masked_key(value: str | None) -> str: key = str(value or "").strip() if len(key) <= 10: return "***" if key else "" return key[:6] + "..." + key[-4:] def app_resource_path(name: str) -> pathlib.Path | None: candidates = [] frozen_root = getattr(sys, "_MEIPASS", "") if frozen_root: candidates.append(pathlib.Path(frozen_root) / name) source_root = pathlib.Path(__file__).resolve().parent candidates.extend( [ source_root / name, source_root / "static" / name, source_root / "portal" / "app" / "static" / name, source_root.parent / "portal" / "app" / "static" / name, ] ) for candidate in candidates: if candidate.is_file(): return candidate return None def extract_model_ids(response: dict) -> list[str]: data = response.get("data") if isinstance(response, dict) else None if not isinstance(data, list): return [] models = [] for item in data: if isinstance(item, dict): model_id = str(item.get("id") or "").strip() else: model_id = str(item or "").strip() if model_id: models.append(model_id) return sorted(dict.fromkeys(models)) def cli_router_config(config: dict) -> dict: value = config.get("cli_router") return dict(value) if isinstance(value, dict) else {} def save_cli_router_config(config: dict, api_key: str, remember_key: bool, port: int, model: str = "") -> dict: router = cli_router_config(config) router["port"] = max(1024, min(65535, int(port or CLI_ROUTER_DEFAULT_PORT))) router["model"] = str(model or "").strip() router["remember_key"] = bool(remember_key) if remember_key: router["api_key"] = str(api_key or "").strip() else: router.pop("api_key", None) config["cli_router"] = router save_config(config) return router def terminal_enabled() -> bool: return env_bool("APICLAUDE_BRIDGE_TERMINAL", False) def configured_folder(config: dict, prompt: bool = False) -> pathlib.Path | None: raw = ( os.getenv("APICLAUDE_BRIDGE_FOLDER", "") or os.getenv("APICLAUDE_BRIDGE_WRITE_ROOT", "") or str(config.get("allowed_folder") or "") ).strip() if not raw and prompt and sys.stdin.isatty(): raw = input("Allowed folder for this bridge session (blank to skip): ").strip() if not raw: return None root = pathlib.Path(raw).expanduser().resolve() if not root.is_dir(): print("Allowed folder does not exist: %s" % root, file=sys.stderr) return None if config.get("allowed_folder") != str(root): config["allowed_folder"] = str(root) save_config(config) return root def action_permission_for_type(action_type: str | None) -> str: return ACTION_PERMISSION_BY_TYPE.get(str(action_type or "").strip().lower(), "") def action_granted(manifest: dict, action: dict) -> bool: permission_type = str(action.get("permission_type") or action_permission_for_type(action.get("action_type"))).strip().lower() if not permission_type: return False return any(item.get("type") == permission_type for item in manifest.get("action_permissions", [])) def machine_payload(actions_enabled: bool = False) -> dict: action_handlers = ["browser.open_url", "folder.write_file"] if terminal_enabled(): action_handlers.append("terminal.run") return { "device_name": os.getenv("APICLAUDE_BRIDGE_DEVICE_NAME") or platform.node() or "My computer", "platform": platform.system() or sys.platform, "version": VERSION, "capabilities": { "safe_mode": True, "mode": "lite", "actions_enabled": actions_enabled, "local_confirmation_required": True, "python": platform.python_version(), "data_collectors": ["selected_files", "selected_folder", "clipboard_snapshot", "browser_context"], "action_handlers": action_handlers, "terminal_enabled": terminal_enabled(), }, } def register(actions_enabled: bool = False, pairing_code: str | None = None, exit_on_error: bool = True) -> dict: code = pairing_code or os.getenv("APICLAUDE_BRIDGE_PAIRING_CODE") if not code and sys.stdin.isatty(): code = input("ApiClo pairing code: ").strip() code = str(code or "").strip() if not code: message = "Pairing code is required." if exit_on_error: print(message, file=sys.stderr) sys.exit(2) raise RuntimeError(message) payload = machine_payload(actions_enabled) payload["pairing_code"] = code response = post_json("/api/agent-bridge/register", payload) if not response.get("ok") or not response.get("device_token"): message = json.dumps(response, ensure_ascii=False, indent=2) if exit_on_error: print(message, file=sys.stderr) sys.exit(1) raise RuntimeError(message) existing = load_config() config = { "portal_url": PORTAL_URL, "device_id": response["device"]["id"], "device_token": response["device_token"], "heartbeat_url": response.get("heartbeat_url") or "/api/agent-bridge/heartbeat", "manifest_url": response.get("manifest_url") or "/api/agent-bridge/manifest", "data_url": response.get("data_url") or "/api/agent-bridge/data-batches", "actions_poll_url": response.get("actions_poll_url") or "/api/agent-bridge/actions/poll", } if existing.get("allowed_folder"): config["allowed_folder"] = existing["allowed_folder"] if isinstance(existing.get("cli_router"), dict): config["cli_router"] = existing["cli_router"] save_config(config) print("Desktop Bridge paired: %s" % response["device"].get("name", "computer")) return config def heartbeat(config: dict, actions_enabled: bool = False, exit_on_error: bool = True) -> dict: payload = machine_payload(actions_enabled) payload["device_id"] = config.get("device_id") response = post_json(config.get("heartbeat_url") or "/api/agent-bridge/heartbeat", payload, config.get("device_token")) if not response.get("ok"): message = json.dumps(response, ensure_ascii=False, indent=2) if exit_on_error: print(message, file=sys.stderr) sys.exit(1) raise RuntimeError(message) permissions = ", ".join(item.get("type", "") for item in response.get("permissions", [])) or "none" print("Bridge heartbeat ok. Permissions: %s" % permissions) return response def has_grant(manifest: dict, permission_type: str) -> bool: return any(item.get("type") == permission_type for item in manifest.get("data_grants", [])) def read_text_file(path: pathlib.Path, max_chars: int) -> str: if not path.is_file(): return "" try: if path.stat().st_size > max_chars * 8: return path.read_text(encoding="utf-8", errors="replace")[:max_chars] return path.read_text(encoding="utf-8", errors="replace")[:max_chars] except Exception: return "" def collect_selected_files(manifest: dict) -> list[dict]: if not has_grant(manifest, "selected_files"): return [] raw = os.getenv("APICLAUDE_BRIDGE_FILES", "") if not raw and sys.stdin.isatty(): raw = input("Files to send, separated by %r (blank to skip): " % os.pathsep).strip() if not raw: return [] max_chars = env_int("APICLAUDE_BRIDGE_ITEM_CHARS", 12000, 1000, 80000) items = [] for value in raw.split(os.pathsep): path = pathlib.Path(value).expanduser() content = read_text_file(path, max_chars) if content: items.append( { "permission_type": "selected_files", "source_type": "selected_files", "source_label": str(path), "content": content, "metadata": {"sha256": hashlib.sha256(content.encode("utf-8")).hexdigest()}, } ) return items def collect_selected_folder(config: dict, manifest: dict) -> list[dict]: if not has_grant(manifest, "selected_folder"): return [] root = configured_folder(config, prompt=True) if root is None: return [] max_files = env_int("APICLAUDE_BRIDGE_FOLDER_MAX_FILES", 8, 1, 50) max_chars = env_int("APICLAUDE_BRIDGE_ITEM_CHARS", 12000, 1000, 80000) items = [] for path in sorted(root.rglob("*")): if len(items) >= max_files: break if not path.is_file() or path.suffix.lower() not in TEXT_EXTENSIONS: continue content = read_text_file(path, max_chars) if content: items.append( { "permission_type": "selected_folder", "source_type": "selected_folder", "source_label": str(path), "content": content, "metadata": {"folder": str(root), "relative_path": str(path.relative_to(root))}, } ) return items def clipboard_text() -> str: raw = os.getenv("APICLAUDE_BRIDGE_CLIPBOARD_TEXT", "") if raw: return raw if not env_bool("APICLAUDE_BRIDGE_CLIPBOARD", False): return "" try: import tkinter # type: ignore root = tkinter.Tk() root.withdraw() value = root.clipboard_get() root.destroy() return str(value or "") except Exception: return "" def collect_clipboard(manifest: dict) -> list[dict]: if not has_grant(manifest, "clipboard_snapshot"): return [] text = clipboard_text().strip() if not text: return [] return [ { "permission_type": "clipboard_snapshot", "source_type": "clipboard_snapshot", "source_label": "clipboard", "content": text[: env_int("APICLAUDE_BRIDGE_ITEM_CHARS", 12000, 1000, 80000)], "metadata": {}, } ] def collect_browser_context(manifest: dict) -> list[dict]: if not has_grant(manifest, "browser_context"): return [] text = os.getenv("APICLAUDE_BRIDGE_BROWSER_CONTEXT", "") label = os.getenv("APICLAUDE_BRIDGE_BROWSER_LABEL", "browser context") context_file = os.getenv("APICLAUDE_BRIDGE_BROWSER_CONTEXT_FILE", "") if context_file: path = pathlib.Path(context_file).expanduser() text = read_text_file(path, env_int("APICLAUDE_BRIDGE_ITEM_CHARS", 12000, 1000, 80000)) label = str(path) if not text.strip(): return [] return [ { "permission_type": "browser_context", "source_type": "browser_context", "source_label": label, "content": text[: env_int("APICLAUDE_BRIDGE_ITEM_CHARS", 12000, 1000, 80000)], "metadata": {}, } ] def upload_data(config: dict, manifest: dict) -> dict: items = [] items.extend(collect_selected_files(manifest)) items.extend(collect_selected_folder(config, manifest)) items.extend(collect_clipboard(manifest)) items.extend(collect_browser_context(manifest)) if not items: print("No local data selected for upload.") return {"ok": True, "accepted": [], "rejected": [], "skipped": True} payload = {"client_batch_id": str(int(time.time())), "items": items} response = post_json(config.get("data_url") or "/api/agent-bridge/data-batches", payload, config.get("device_token")) print("Bridge data upload: accepted=%s rejected=%s" % (len(response.get("accepted", [])), len(response.get("rejected", [])))) if not response.get("ok"): print(json.dumps(response, ensure_ascii=False, indent=2), file=sys.stderr) return response def confirm_action(action: dict) -> bool: policy = str(action.get("approval_policy") or "ask") action_type = str(action.get("action_type") or "") payload = action.get("payload") if isinstance(action.get("payload"), dict) else {} if policy == "allow_session" and action_type not in {"terminal", "terminal.run", "run_command"}: return True if env_bool("APICLAUDE_BRIDGE_AUTO_CONFIRM", False) and policy != "confirm_sensitive": return True print("\nApiClo asks to run local action:") print(json.dumps({"action_type": action_type, "payload": payload}, ensure_ascii=False, indent=2)) answer = input("Approve this action? Type yes: ").strip().lower() return answer == "yes" def safe_write_path(payload: dict, config: dict) -> pathlib.Path: root = configured_folder(config, prompt=True) if root is None: raise RuntimeError("A local allowed folder is required for folder writes") relative = str(payload.get("relative_path") or payload.get("path") or "").strip() if not relative: raise RuntimeError("write path is required") if pathlib.PurePosixPath(relative).is_absolute() or pathlib.PureWindowsPath(relative).is_absolute(): raise RuntimeError("write path must be relative to the allowed folder") target = (root / relative).resolve() if root != target and root not in target.parents: raise RuntimeError("write path escapes allowed folder") return target def execute_action(action: dict, manifest: dict, config: dict) -> dict: if not action_granted(manifest, action): return {"ok": False, "status": "cancelled", "error": "action is not granted for this bridge session", "result": {}} if not confirm_action(action): return {"ok": False, "status": "cancelled", "error": "rejected locally", "result": {}} action_type = str(action.get("action_type") or "").strip().lower() payload = action.get("payload") if isinstance(action.get("payload"), dict) else {} if action_type in {"open_url", "browser.open_url", "browser"}: url = str(payload.get("url") or "").strip() if not url.startswith(("http://", "https://")): raise RuntimeError("Only http/https URLs can be opened") opened = webbrowser.open(url) return {"ok": bool(opened), "status": "succeeded" if opened else "failed", "result": {"url": url, "opened": opened}} if action_type in {"write_file", "folder.write_file", "folder_write"}: target = safe_write_path(payload, config) content = str(payload.get("content") or "") max_bytes = env_int("APICLAUDE_BRIDGE_WRITE_MAX_BYTES", 250000, 1, 1000000) if len(content.encode("utf-8")) > max_bytes: raise RuntimeError("write content is larger than APICLAUDE_BRIDGE_WRITE_MAX_BYTES") target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return {"ok": True, "status": "succeeded", "result": {"path": str(target), "chars": len(content)}} if action_type in TERMINAL_ACTIONS: if not terminal_enabled(): raise RuntimeError("terminal actions are disabled; set APICLAUDE_BRIDGE_TERMINAL=1 to enable them locally") command = str(payload.get("command") or "").strip() if not command: raise RuntimeError("command is required") if len(command) > env_int("APICLAUDE_BRIDGE_COMMAND_MAX_CHARS", 1000, 80, 4000): raise RuntimeError("command is too long") cwd = configured_folder(config, prompt=True) if cwd is None: raise RuntimeError("A local allowed folder is required for terminal actions") timeout = env_int("APICLAUDE_BRIDGE_COMMAND_TIMEOUT", 30, 1, 300) completed = subprocess.run( command, cwd=str(cwd), shell=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout, ) output = (completed.stdout or "")[-8000:] return { "ok": completed.returncode == 0, "status": "succeeded" if completed.returncode == 0 else "failed", "result": {"returncode": completed.returncode, "output": output}, } raise RuntimeError("Unsupported action type: %s" % action_type) def poll_actions(config: dict, manifest: dict) -> None: response = post_json(config.get("actions_poll_url") or "/api/agent-bridge/actions/poll", {}, config.get("device_token")) manifest = response.get("manifest") if isinstance(response.get("manifest"), dict) else manifest actions = response.get("actions") if isinstance(response.get("actions"), list) else [] if not actions: print("No pending bridge actions.") return for action in actions: try: result = execute_action(action, manifest, config) except Exception as exc: result = {"ok": False, "status": "failed", "error": str(exc), "result": {}} post_json("/api/agent-bridge/actions/%s/result" % action.get("id"), result, config.get("device_token")) print("Action %s -> %s" % (action.get("id"), result.get("status"))) def should_sync(args: set[str], manifest: dict) -> bool: if "--sync" in args or env_bool("APICLAUDE_BRIDGE_SYNC", False): return True if "--no-sync" in args or not manifest.get("data_access_enabled"): return False if sys.stdin.isatty(): answer = input("Send selected local data to the agent now? [y/N]: ").strip().lower() return answer in {"y", "yes"} return False def normalize_router_path(path: str) -> str: clean = str(path or "").split("?", 1)[0].strip() or "/" if clean != "/" and clean.endswith("/"): clean = clean[:-1] return clean class ApiClaudeCliRouterHandler(BaseHTTPRequestHandler): server_version = "ApiClaudeCliRouter/%s" % VERSION def log_message(self, _format: str, *_args) -> None: return def _send_json(self, status: int, payload: dict) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "authorization,content-type") self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS") self.end_headers() self.wfile.write(body) def do_OPTIONS(self) -> None: self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Headers", "authorization,content-type") self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS") self.end_headers() def do_GET(self) -> None: path = normalize_router_path(self.path) if path in {"/", "/health", "/v1/health"}: self._send_json(200, {"ok": True, "service": "apiclaude-cli-router", "version": VERSION}) return if path in {"/models", "/v1/models"}: self._proxy_to_portal("GET", "/v1/models", b"") return self._send_json(404, {"error": {"message": "Unsupported local router path", "path": path}}) def do_POST(self) -> None: path = normalize_router_path(self.path) upstream_path = CLI_ROUTER_ALLOWED_POST_PATHS.get(path) if not upstream_path: self._send_json(404, {"error": {"message": "Unsupported local router path", "path": path}}) return content_length = int(self.headers.get("Content-Length") or "0") if content_length > 25 * 1024 * 1024: self._send_json(413, {"error": {"message": "Request body is too large"}}) return self._proxy_to_portal("POST", upstream_path, self.rfile.read(content_length) if content_length else b"{}") def _local_api_key(self) -> str: key = str(getattr(self.server, "api_key", "") or "").strip() if key: return key header = str(self.headers.get("Authorization") or "").strip() if header.lower().startswith("bearer "): return header[7:].strip() return "" def _proxy_to_portal(self, method: str, upstream_path: str, body: bytes) -> None: api_key = self._local_api_key() if not api_key: self._send_json(401, {"error": {"message": "ApiClo API key is required"}}) return headers = { "Authorization": "Bearer " + api_key, "User-Agent": "ApiClaudeCliRouter/%s (%s)" % (VERSION, platform.system()), } content_type = self.headers.get("Content-Type") if content_type: headers["Content-Type"] = content_type request = urllib.request.Request( PORTAL_URL.rstrip("/") + upstream_path, data=body if method != "GET" else None, headers=headers, method=method, ) try: with urllib.request.urlopen(request, timeout=300) as response: self.send_response(getattr(response, "status", 200)) self.send_header("Access-Control-Allow-Origin", "*") content_type = response.headers.get("Content-Type") if content_type: self.send_header("Content-Type", content_type) self.end_headers() while True: chunk = response.read(8192) if not chunk: break self.wfile.write(chunk) self.wfile.flush() except urllib.error.HTTPError as exc: detail = exc.read() self.send_response(exc.code) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Type", exc.headers.get("Content-Type") or "application/json; charset=utf-8") self.end_headers() self.wfile.write(detail or json.dumps({"error": {"message": str(exc)}}).encode("utf-8")) except Exception as exc: self._send_json(502, {"error": {"message": "Portal request failed", "detail": str(exc)[:240]}}) class ApiClaudeCliRouterServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True def __init__(self, address: tuple[str, int], api_key: str): super().__init__(address, ApiClaudeCliRouterHandler) self.api_key = api_key def start_cli_router(api_key: str, port: int) -> ApiClaudeCliRouterServer: key = str(api_key or "").strip() if not key: raise RuntimeError("ApiClo API key is required") server = ApiClaudeCliRouterServer((CLI_ROUTER_HOST, int(port or CLI_ROUTER_DEFAULT_PORT)), key) threading.Thread(target=server.serve_forever, name="apiclaude-cli-router", daemon=True).start() return server def cli_client_by_key(key: str) -> dict: needle = str(key or "").strip() for item in CLI_CLIENTS: if item["key"] == needle: return item raise KeyError("Unknown CLI client: %s" % needle) def cli_client_install_text(key: str) -> str: client = cli_client_by_key(key) install = client["install"] return "%s install\n\nPaste this into your command app:\n%s\n\nOfficial guide:\n%s" % ( client["name"], install, client["download_url"], ) def cli_client_config_text(key: str, openai_base_url: str, model: str) -> str: client = cli_client_by_key(key) profile = local_provider_profile(openai_base_url, model) profile["client"] = key return "%s ApiClo local profile\nExisting tool configs are not changed by Desktop Bridge.\n\n%s\n" % ( client["name"], json.dumps(profile, ensure_ascii=False, indent=2), ) def local_provider_profile(openai_base_url: str, model: str) -> dict: openai_url = str(openai_base_url or "").rstrip("/") or "http://127.0.0.1:%s/v1" % CLI_ROUTER_DEFAULT_PORT anthropic_url = openai_url[:-3] if openai_url.endswith("/v1") else openai_url return { "name": "ApiClo Local", "openai_base_url": openai_url, "anthropic_base_url": anthropic_url, "api_key": "local-router", "model": str(model or "").strip() or "claude-sonnet-4-6", "router_host": CLI_ROUTER_HOST, "router_port": CLI_ROUTER_DEFAULT_PORT, } def write_text_with_backup(path: pathlib.Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): backup = path.with_suffix(path.suffix + ".apiclaude-backup-%s" % time.strftime("%Y%m%d-%H%M%S")) backup.write_text(path.read_text(encoding="utf-8", errors="replace"), encoding="utf-8") path.write_text(content, encoding="utf-8") def codex_config_path() -> pathlib.Path: return pathlib.Path.home() / ".codex" / "config.toml" def codex_backup_dir() -> pathlib.Path: return pathlib.Path.home() / ".apiclaude" / "codex-backups" def codex_model_catalog_path() -> pathlib.Path: return pathlib.Path.home() / ".apiclaude" / "codex-models.json" def toml_string(value: str) -> str: return '"' + str(value).replace("\\", "\\\\").replace('"', '\\"') + '"' def codex_model_catalog(model: str) -> dict: selected = str(model or "").strip() or "claude-sonnet-4-6" return { "models": [ { "slug": selected, "display_name": selected, "description": "ApiClo local route through Desktop Bridge.", "default_reasoning_level": "medium", "supported_reasoning_levels": [ {"effort": "low", "description": "Fast responses with lighter reasoning"}, {"effort": "medium", "description": "Balances speed and reasoning depth"}, {"effort": "high", "description": "Greater reasoning depth for complex tasks"}, {"effort": "xhigh", "description": "Extra reasoning depth for large tasks"}, ], "shell_type": "shell_command", "visibility": "list", "supported_in_api": True, "priority": 50, "base_instructions": "You are Codex, a coding agent. Follow the user's instructions and the repository guidance.", "model_messages": { "instructions_template": "You are Codex, a coding agent. Follow the user's instructions and the repository guidance.\n\n{{ personality }}", "instructions_variables": { "personality_default": "", "personality_friendly": "", "personality_pragmatic": "", }, }, "context_window": CODEX_MODEL_CONTEXT_WINDOW, "max_context_window": CODEX_MODEL_CONTEXT_WINDOW, "effective_context_window_percent": 95, "supports_reasoning_summaries": True, "default_reasoning_summary": "none", "support_verbosity": True, "default_verbosity": "low", "truncation_policy": {"mode": "tokens", "limit": 10000}, "apply_patch_tool_type": "freeform", "supports_parallel_tool_calls": True, "supports_image_detail_original": True, "input_modalities": ["text"], "experimental_supported_tools": [], "supports_search_tool": False, "use_responses_lite": False, } ] } def write_codex_model_catalog(model: str, path: pathlib.Path | None = None) -> pathlib.Path: target = path or codex_model_catalog_path() target.parent.mkdir(parents=True, exist_ok=True) target.write_text(json.dumps(codex_model_catalog(model), ensure_ascii=False, indent=2), encoding="utf-8") return target def codex_connector_block(openai_base_url: str, model: str) -> str: profile = local_provider_profile(openai_base_url, model) catalog = codex_model_catalog_path() return """%s model = %s model_provider = %s model_catalog_json = %s [model_providers.%s] name = %s base_url = %s wire_api = "responses" %s""" % ( CODEX_CONNECTOR_START, toml_string(profile["model"]), toml_string(CODEX_CONNECTOR_PROVIDER_ID), toml_string(str(catalog)), CODEX_CONNECTOR_PROVIDER_ID, toml_string("ApiClo"), toml_string(profile["openai_base_url"]), CODEX_CONNECTOR_END, ) def strip_codex_connector_config(text: str) -> str: lines = str(text or "").splitlines() kept = [] in_managed = False in_table = False skip_provider_table = False provider_table = "[model_providers.%s]" % CODEX_CONNECTOR_PROVIDER_ID for line in lines: stripped = line.strip() if stripped == CODEX_CONNECTOR_START: in_managed = True continue if in_managed: if stripped == CODEX_CONNECTOR_END: in_managed = False continue if stripped.startswith("[") and stripped.endswith("]"): skip_provider_table = stripped == provider_table in_table = True if skip_provider_table: continue elif skip_provider_table: continue if not in_table and ( stripped.startswith("model = ") or stripped.startswith("model_provider = ") or stripped.startswith("model_catalog_json = ") ): continue kept.append(line) return "\n".join(kept).strip() def codex_connector_config(existing: str, openai_base_url: str, model: str) -> str: block = codex_connector_block(openai_base_url, model).strip() clean = strip_codex_connector_config(existing) if not clean: return block + "\n" return block + "\n\n" + clean + "\n" def create_codex_config_backup(path: pathlib.Path | None = None) -> pathlib.Path: target = path or codex_config_path() backup_root = codex_backup_dir() backup_root.mkdir(parents=True, exist_ok=True) stamp = "%s-%03d" % (time.strftime("%Y%m%d-%H%M%S"), int((time.time() % 1) * 1000)) if target.exists(): backup = backup_root / ("config.toml.apiclaude-backup-%s.toml" % stamp) backup.write_text(target.read_text(encoding="utf-8", errors="replace"), encoding="utf-8") else: backup = backup_root / ("config.toml.apiclaude-backup-%s.missing" % stamp) backup.write_text("", encoding="utf-8") return backup def latest_codex_config_backup() -> pathlib.Path | None: backup_root = codex_backup_dir() if not backup_root.is_dir(): return None backups = list(backup_root.glob("config.toml.apiclaude-backup-*")) if not backups: return None return sorted(backups, key=lambda item: item.stat().st_mtime)[-1] def apply_codex_connector(openai_base_url: str, model: str) -> dict: path = codex_config_path() existing = path.read_text(encoding="utf-8", errors="replace") if path.exists() else "" backup = create_codex_config_backup(path) catalog = write_codex_model_catalog(local_provider_profile(openai_base_url, model)["model"]) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(codex_connector_config(existing, openai_base_url, model), encoding="utf-8") return { "ok": True, "config_path": str(path), "backup_path": str(backup), "catalog_path": str(catalog), "model": local_provider_profile(openai_base_url, model)["model"], } def preview_codex_connector(openai_base_url: str, model: str) -> str: path = codex_config_path() return "Codex config preview\nTarget: %s\nBackup folder: %s\n\n%s\n" % ( path, codex_backup_dir(), codex_connector_block(openai_base_url, model), ) def restore_latest_codex_connector_backup() -> dict: backup = latest_codex_config_backup() if backup is None: raise RuntimeError("No ApiClo Codex backup found.") path = codex_config_path() if backup.suffix == ".missing": if path.exists(): path.unlink() else: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(backup.read_text(encoding="utf-8", errors="replace"), encoding="utf-8") return {"ok": True, "config_path": str(path), "backup_path": str(backup), "restored_missing": backup.suffix == ".missing"} def apply_codex_profile(openai_base_url: str, model: str) -> str: return apply_generic_cli_profile("codex", openai_base_url, model) def apply_claude_code_profile(openai_base_url: str, model: str) -> str: return apply_generic_cli_profile("claude-code", openai_base_url, model) def apply_generic_cli_profile(key: str, openai_base_url: str, model: str) -> str: client = cli_client_by_key(key) profile = local_provider_profile(openai_base_url, model) profile["client"] = key profile_path = pathlib.Path.home() / ".apiclaude" / "profiles" / ("%s.json" % key) write_text_with_backup(profile_path, json.dumps(profile, ensure_ascii=False, indent=2)) return "%s profile prepared. Existing tool config was not changed." % client["name"] def apply_cli_client_profile(key: str, openai_base_url: str, model: str) -> str: if key == "codex": return apply_codex_profile(openai_base_url, model) if key == "claude-code": return apply_claude_code_profile(openai_base_url, model) return apply_generic_cli_profile(key, openai_base_url, model) def app_mode_requested() -> bool: args = set(sys.argv[1:]) return "--app" in args or "--gui" in args or pathlib.Path(sys.argv[0]).suffix.lower() == ".pyw" def default_app_folder(config: dict) -> str: raw = os.getenv("APICLAUDE_BRIDGE_FOLDER", "") or str(config.get("allowed_folder") or "") if raw: return str(pathlib.Path(raw).expanduser()) documents = pathlib.Path.home() / "Documents" return str(documents if documents.is_dir() else pathlib.Path.home()) def autostart_command() -> str: if getattr(sys, "frozen", False): return '"%s" --app' % sys.executable script = pathlib.Path(sys.argv[0]).resolve() return '"%s" "%s" --app' % (sys.executable, script) def windows_autostart_enabled() -> bool: if platform.system().lower() != "windows": return False try: import winreg # type: ignore with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run") as key: winreg.QueryValueEx(key, "ApiClaudeDesktopBridge") return True except Exception: return False def set_windows_autostart(enabled: bool) -> None: if platform.system().lower() != "windows": raise RuntimeError("Autostart is only supported on Windows.") import winreg # type: ignore with winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run", 0, winreg.KEY_SET_VALUE) as key: if enabled: winreg.SetValueEx(key, "ApiClaudeDesktopBridge", 0, winreg.REG_SZ, autostart_command()) else: try: winreg.DeleteValue(key, "ApiClaudeDesktopBridge") except FileNotFoundError: pass def start_tray_icon(root, show_callback, sync_callback, quit_callback): try: import pystray # type: ignore from PIL import Image, ImageDraw # type: ignore except Exception: return None image = Image.new("RGB", (64, 64), "#111111") draw = ImageDraw.Draw(image) draw.rounded_rectangle((8, 8, 56, 56), radius=12, fill="#d1765c") draw.text((24, 20), "A", fill="white") def run_on_tk(callback): root.after(0, callback) icon = pystray.Icon( "ApiClaudeDesktopBridge", image, "ApiClo Desktop Bridge", menu=pystray.Menu( pystray.MenuItem("Show", lambda _icon, _item: run_on_tk(show_callback)), pystray.MenuItem("Sync now", lambda _icon, _item: run_on_tk(sync_callback)), pystray.MenuItem("Quit", lambda _icon, _item: run_on_tk(quit_callback)), ), ) threading.Thread(target=icon.run, name="apiclaude-tray", daemon=True).start() return icon def run_app() -> None: try: import tkinter as tk # type: ignore from tkinter import filedialog, messagebox, ttk # type: ignore except Exception as exc: print("Desktop app UI is unavailable: %s" % exc, file=sys.stderr) main() return config = load_config() router_settings = cli_router_config(config) root = tk.Tk() root.title("ApiClo Desktop") root.geometry("980x1000") root.minsize(860, 780) BG = "#f6f1e8" CARD = "#fffdf8" CARD_SOFT = "#fff6ef" FIELD = "#fbfaf6" TEXT = "#181512" MUTED = "#746b62" BRAND = "#cc785c" BRAND_DARK = "#a9583e" INK = "#111111" BORDER = "#e6dbcf" root.configure(bg=BG) root.option_add("*Font", "{Segoe UI} 10") try: style = ttk.Style(root) style.theme_use("clam") style.configure("ApiClo.TCombobox", fieldbackground=FIELD, background=FIELD, foreground=TEXT, bordercolor=BORDER) except Exception: pass status_var = tk.StringVar(value="Ready") folder_var = tk.StringVar(value=default_app_folder(config)) pairing_var = tk.StringVar(value=os.getenv("APICLAUDE_BRIDGE_PAIRING_CODE", "")) autostart_var = tk.BooleanVar(value=windows_autostart_enabled()) sync_var = tk.BooleanVar(value=True) interval_var = tk.IntVar(value=env_int("APICLAUDE_BRIDGE_HEARTBEAT_SECONDS", 30, 15, 3600)) running_var = tk.BooleanVar(value=False) router_key_var = tk.StringVar(value=os.getenv("APICLAUDE_API_KEY", "") or str(router_settings.get("api_key") or "")) router_remember_var = tk.BooleanVar(value=bool(router_settings.get("remember_key"))) try: router_port = int(router_settings.get("port") or CLI_ROUTER_DEFAULT_PORT) except (TypeError, ValueError): router_port = CLI_ROUTER_DEFAULT_PORT router_port_var = tk.IntVar(value=router_port) router_model_var = tk.StringVar(value=str(router_settings.get("model") or "")) router_status_var = tk.StringVar(value="Local route: stopped") stop_event = threading.Event() cycle_lock = threading.Lock() worker_ref = {"thread": None} router_ref = {"server": None} tray_ref = {"icon": None} def ui(callback, *args): try: root.after(0, lambda: callback(*args)) except Exception: pass def append_log(message: str) -> None: timestamp = time.strftime("%H:%M:%S") log_box.configure(state="normal") log_box.insert("end", "[%s] %s\n" % (timestamp, message)) log_box.see("end") log_box.configure(state="disabled") def set_status(message: str) -> None: status_var.set(message) append_log(message) def persist_folder() -> None: raw = folder_var.get().strip() if not raw: return folder = pathlib.Path(raw).expanduser().resolve() if not folder.is_dir(): raise RuntimeError("Selected folder does not exist: %s" % folder) config["allowed_folder"] = str(folder) save_config(config) def choose_folder() -> None: selected = filedialog.askdirectory(initialdir=folder_var.get() or str(pathlib.Path.home())) if selected: folder_var.set(selected) try: persist_folder() set_status("Allowed folder saved.") except Exception as exc: messagebox.showerror("ApiClo Desktop Bridge", str(exc)) def bridge_cycle() -> None: nonlocal config with cycle_lock: persist_folder() if not config.get("device_token"): config = register(False, pairing_var.get(), exit_on_error=False) ui(set_status, "Paired with ApiClo.") manifest = heartbeat(config, False, exit_on_error=False) permissions = ", ".join(item.get("type", "") for item in manifest.get("data_grants", [])) or "none" ui(set_status, "Connected. Data permissions: %s" % permissions) if sync_var.get(): response = upload_data(config, manifest) accepted = len(response.get("accepted", [])) rejected = len(response.get("rejected", [])) if response.get("skipped"): ui(set_status, "No local data selected for sync.") else: ui(set_status, "Synced local context: accepted=%s rejected=%s" % (accepted, rejected)) def worker_loop() -> None: while not stop_event.is_set(): try: bridge_cycle() except Exception as exc: ui(set_status, "Bridge error: %s" % str(exc)[:240]) stop_event.wait(max(15, int(interval_var.get() or 30))) ui(set_status, "Stopped.") ui(running_var.set, False) def start_bridge() -> None: if running_var.get(): return if not load_config().get("device_token") and not pairing_var.get().strip() and not config.get("device_token"): messagebox.showwarning("ApiClo Desktop Bridge", "Paste the pairing code from ApiClo first.") return stop_event.clear() running_var.set(True) set_status("Starting...") thread = threading.Thread(target=worker_loop, name="apiclaude-bridge-loop", daemon=True) worker_ref["thread"] = thread thread.start() def stop_bridge() -> None: stop_event.set() running_var.set(False) def sync_now() -> None: set_status("Sync requested.") threading.Thread(target=lambda: bridge_cycle_safe(), name="apiclaude-bridge-sync-now", daemon=True).start() def bridge_cycle_safe() -> None: try: bridge_cycle() except Exception as exc: ui(set_status, "Bridge error: %s" % str(exc)[:240]) def toggle_autostart() -> None: try: set_windows_autostart(bool(autostart_var.get())) set_status("Autostart %s." % ("enabled" if autostart_var.get() else "disabled")) except Exception as exc: autostart_var.set(windows_autostart_enabled()) messagebox.showerror("ApiClo Desktop Bridge", str(exc)) def local_router_url() -> str: return "http://%s:%s/v1" % (CLI_ROUTER_HOST, int(router_port_var.get() or CLI_ROUTER_DEFAULT_PORT)) def persist_router_settings() -> None: nonlocal config model = router_model_var.get().strip() save_cli_router_config( config, router_key_var.get(), bool(router_remember_var.get()), int(router_port_var.get() or CLI_ROUTER_DEFAULT_PORT), model, ) def set_router_status(message: str) -> None: router_status_var.set(message) append_log(message) def fetch_models_worker(key: str) -> None: try: if not key: raise RuntimeError("Paste ApiClo API key first.") ui(set_router_status, "Local route: loading models...") response = get_json("/v1/models", key) models = extract_model_ids(response) if not models: raise RuntimeError("No models returned by /v1/models.") ui(apply_models, models, key) except Exception as exc: ui(set_router_status, "Local route error: %s" % str(exc)[:240]) def apply_models(models: list[str], key: str) -> None: if not router_model_var.get().strip(): router_model_var.set(models[0]) model_combo.configure(values=models) persist_router_settings() set_router_status("Local route: %s models loaded for %s" % (len(models), masked_key(key))) def fetch_models() -> None: key = router_key_var.get().strip() if not key: messagebox.showwarning("ApiClo Desktop Bridge", "Paste ApiClo API key first.") return threading.Thread(target=lambda: fetch_models_worker(key), name="apiclaude-cli-router-models", daemon=True).start() def start_router_ui() -> None: if router_ref.get("server"): set_router_status("Local route: already running at %s" % local_router_url()) return key = router_key_var.get().strip() if not key: messagebox.showwarning("ApiClo Desktop Bridge", "Paste ApiClo API key first.") return try: persist_router_settings() server = start_cli_router(key, int(router_port_var.get() or CLI_ROUTER_DEFAULT_PORT)) router_ref["server"] = server set_router_status("Local route: running at %s" % local_router_url()) except Exception as exc: set_router_status("Local route error: %s" % str(exc)[:240]) def ensure_router_running_for_setup() -> bool: if router_ref.get("server"): return True key = router_key_var.get().strip() if not key: messagebox.showwarning("ApiClo Desktop", "Paste your ApiClo key first.") return False try: persist_router_settings() server = start_cli_router(key, int(router_port_var.get() or CLI_ROUTER_DEFAULT_PORT)) router_ref["server"] = server set_router_status("Local router is running at %s" % local_router_url()) return True except Exception as exc: set_router_status("Local router error: %s" % str(exc)[:240]) return False def apply_client_setup(key: str) -> None: if not ensure_router_running_for_setup(): return client = cli_client_by_key(key) try: message = apply_cli_client_profile(key, local_router_url(), router_model_var.get()) set_router_status("%s: %s" % (client["name"], message)) except Exception as exc: set_router_status("%s setup error: %s" % (client["name"], str(exc)[:220])) def stop_router_ui() -> None: server = router_ref.get("server") if not server: set_router_status("Local route: stopped") return router_ref["server"] = None try: server.shutdown() server.server_close() except Exception: pass set_router_status("Local route: stopped") def copy_router_url() -> None: root.clipboard_clear() root.clipboard_append(local_router_url()) set_router_status("Local route URL copied: %s" % local_router_url()) def copy_text_to_clipboard(label: str, text: str) -> None: root.clipboard_clear() root.clipboard_append(text) set_router_status("%s copied." % label) def open_client_download(key: str) -> None: client = cli_client_by_key(key) webbrowser.open(client["download_url"]) set_router_status("Opened %s download guide." % client["name"]) def copy_client_install(key: str) -> None: client = cli_client_by_key(key) root.clipboard_clear() root.clipboard_append(cli_client_install_text(key)) set_router_status("%s manual install text copied." % client["name"]) def copy_client_config(key: str) -> None: client = cli_client_by_key(key) root.clipboard_clear() root.clipboard_append(cli_client_config_text(key, local_router_url(), router_model_var.get())) set_router_status("%s manual setup text copied." % client["name"]) def preview_codex_connector_ui() -> None: root.clipboard_clear() root.clipboard_append(preview_codex_connector(local_router_url(), router_model_var.get())) set_router_status("Codex connector preview copied.") def connect_codex_ui() -> None: if not ensure_router_running_for_setup(): return model = router_model_var.get().strip() or "claude-sonnet-4-6" path = codex_config_path() if not messagebox.askyesno( "Connect Codex", "Create a backup and update Codex config for ApiClo?\n\nTarget:\n%s\n\nModel: %s" % (path, model), ): set_router_status("Codex connector cancelled.") return try: result = apply_codex_connector(local_router_url(), model) set_router_status("Codex connected. Backup: %s" % result["backup_path"]) except Exception as exc: set_router_status("Codex connector error: %s" % str(exc)[:220]) def rollback_codex_ui() -> None: backup = latest_codex_config_backup() if backup is None: set_router_status("No ApiClo Codex backup found.") return if not messagebox.askyesno( "Rollback Codex", "Restore latest ApiClo Codex backup?\n\n%s" % backup, ): set_router_status("Codex rollback cancelled.") return try: result = restore_latest_codex_connector_backup() if result.get("restored_missing"): set_router_status("Codex config removed; restored previous missing state.") else: set_router_status("Codex config restored from backup.") except Exception as exc: set_router_status("Codex rollback error: %s" % str(exc)[:220]) def show_window() -> None: root.deiconify() root.lift() def quit_app() -> None: stop_bridge() stop_router_ui() icon = tray_ref.get("icon") if icon: try: icon.stop() except Exception: pass root.destroy() def on_close() -> None: if tray_ref.get("icon"): root.withdraw() else: root.iconify() root.protocol("WM_DELETE_WINDOW", on_close) def make_label(parent, text="", textvariable=None, size=10, weight="normal", fg=TEXT, bg=None, wraplength=None): return tk.Label( parent, text=text, textvariable=textvariable, bg=bg or parent.cget("bg"), fg=fg, font=("Segoe UI", size, weight), anchor="w", justify="left", wraplength=wraplength, ) def make_entry(parent, variable, show=None, width=None): entry = tk.Entry( parent, textvariable=variable, show=show, width=width or 20, bg=FIELD, fg=TEXT, insertbackground=TEXT, relief="flat", bd=0, highlightthickness=1, highlightbackground=BORDER, highlightcolor=BRAND, ) bind_entry_clipboard(entry) return entry def is_control_pressed(event) -> bool: return bool(int(getattr(event, "state", 0) or 0) & 0x4) def selected_entry_text(entry) -> str: try: return str(entry.selection_get() or "") except tk.TclError: return "" def paste_into_entry(entry, label="Text", trim=False): try: text = str(root.clipboard_get() or "") except tk.TclError: status_var.set("Clipboard is empty.") return "break" if trim: text = text.strip() if not text: status_var.set("Clipboard is empty.") return "break" try: if entry.selection_present(): entry.delete("sel.first", "sel.last") except tk.TclError: pass entry.insert("insert", text) status_var.set("%s pasted." % label) return "break" def copy_from_entry(entry): text = selected_entry_text(entry) if not text: return "break" root.clipboard_clear() root.clipboard_append(text) return "break" def cut_from_entry(entry): text = selected_entry_text(entry) if not text: return "break" root.clipboard_clear() root.clipboard_append(text) try: entry.delete("sel.first", "sel.last") except tk.TclError: pass return "break" def select_all_entry(entry): entry.select_range(0, "end") entry.icursor("end") return "break" def handle_entry_shortcut(event): if not is_control_pressed(event): return None keycode = int(getattr(event, "keycode", 0) or 0) keysym = str(getattr(event, "keysym", "") or "").lower() widget = event.widget if keycode == 86 or keysym in {"v", "cyrillic_em"}: return paste_into_entry(widget) if keycode == 67 or keysym in {"c", "cyrillic_es"}: return copy_from_entry(widget) if keycode == 88 or keysym in {"x", "cyrillic_che"}: return cut_from_entry(widget) if keycode == 65 or keysym in {"a", "cyrillic_ef"}: return select_all_entry(widget) return None def bind_entry_clipboard(entry): entry.bind("", handle_entry_shortcut, add="+") menu = tk.Menu(entry, tearoff=0) menu.add_command(label="Paste", command=lambda: paste_into_entry(entry)) menu.add_command(label="Copy", command=lambda: copy_from_entry(entry)) menu.add_command(label="Cut", command=lambda: cut_from_entry(entry)) menu.add_separator() menu.add_command(label="Select all", command=lambda: select_all_entry(entry)) def show_menu(event): entry.focus_set() menu.tk_popup(event.x_root, event.y_root) return "break" entry.bind("", show_menu, add="+") def make_button(parent, text, command, variant="soft", width=None): palette = { "primary": (INK, "#ffffff", "#2a2622"), "brand": (BRAND, "#ffffff", BRAND_DARK), "soft": (CARD_SOFT, BRAND_DARK, "#ffe8dc"), "plain": (CARD, TEXT, "#f0e7dc"), } bg, fg, active = palette.get(variant, palette["soft"]) return tk.Button( parent, text=text, command=command, width=width, bg=bg, fg=fg, activebackground=active, activeforeground=fg, relief="flat", bd=0, padx=12, pady=7, cursor="hand2", font=("Segoe UI", 10, "bold"), ) def make_card(parent): frame = tk.Frame(parent, bg=CARD, padx=16, pady=14, highlightthickness=1, highlightbackground=BORDER) return frame def section_title(parent, title, subtitle=""): make_label(parent, title, size=13, weight="bold").pack(anchor="w") if subtitle: make_label(parent, subtitle, size=9, fg=MUTED, wraplength=360).pack(anchor="w", pady=(2, 10)) def load_logo_image(size=44): logo_path = app_resource_path("apiclo-social-avatar.png") if not logo_path: return None try: image = tk.PhotoImage(file=str(logo_path)) factor = max(1, int(max(image.width(), image.height()) / size)) if factor > 1: image = image.subsample(factor, factor) return image except Exception: return None def draw_mark(parent): canvas = tk.Canvas(parent, width=46, height=46, bg=BG, highlightthickness=0) canvas.create_oval(4, 4, 42, 42, fill=INK, outline="") canvas.create_arc(12, 12, 34, 34, start=205, extent=292, style="arc", outline=BRAND, width=4) canvas.create_oval(29, 10, 37, 18, fill=BRAND, outline="") canvas.create_oval(13, 29, 20, 36, fill="#30c3d6", outline="") return canvas outer = tk.Frame(root, bg=BG, padx=20, pady=16) outer.pack(fill="both", expand=True) header = tk.Frame(outer, bg=BG) header.pack(fill="x", pady=(0, 16)) logo_image = load_logo_image() if logo_image: logo_label = tk.Label(header, image=logo_image, bg=BG) logo_label.image = logo_image logo_label.pack(side="left", padx=(0, 12)) else: draw_mark(header).pack(side="left", padx=(0, 12)) brand = tk.Frame(header, bg=BG) brand.pack(side="left", fill="x", expand=True) wordmark = tk.Frame(brand, bg=BG) wordmark.pack(anchor="w") make_label(wordmark, "Api", size=20, weight="bold", bg=BG).pack(side="left") make_label(wordmark, "Claude", size=20, weight="bold", fg=BRAND, bg=BG).pack(side="left") make_label(brand, "Desktop bridge and local model route", size=10, fg=MUTED, bg=BG).pack(anchor="w", pady=(1, 0)) status_pill = tk.Label( header, textvariable=status_var, bg=CARD_SOFT, fg=BRAND_DARK, font=("Segoe UI", 9, "bold"), padx=12, pady=8, wraplength=260, justify="left", ) status_pill.pack(side="right", anchor="n") grid = tk.Frame(outer, bg=BG) grid.pack(fill="x", pady=(0, 12)) grid.columnconfigure(0, weight=1, uniform="main") grid.columnconfigure(1, weight=1, uniform="main") connect_card = make_card(grid) connect_card.grid(row=0, column=0, sticky="nsew", padx=(0, 10)) section_title(connect_card, "Computer access", "Pair this device and choose what folder can be used.") make_label(connect_card, "Pairing code", size=9, weight="bold", fg=MUTED).pack(anchor="w") pairing_row = tk.Frame(connect_card, bg=CARD) pairing_row.pack(fill="x", pady=(4, 10)) pairing_entry = make_entry(pairing_row, pairing_var) pairing_entry.pack(side="left", fill="x", expand=True, ipady=6) make_button(pairing_row, "Paste", lambda: paste_into_entry(pairing_entry, "Pairing code", True), "soft").pack(side="left", padx=(8, 0)) make_label(connect_card, "Allowed folder", size=9, weight="bold", fg=MUTED).pack(anchor="w") folder_row = tk.Frame(connect_card, bg=CARD) folder_row.pack(fill="x", pady=(4, 10)) folder_entry = make_entry(folder_row, folder_var) folder_entry.pack(side="left", fill="x", expand=True, ipady=6) make_button(folder_row, "Browse", choose_folder, "soft").pack(side="left", padx=(8, 0)) checks = tk.Frame(connect_card, bg=CARD) checks.pack(fill="x", pady=(0, 10)) tk.Checkbutton( checks, text="Sync local context automatically", variable=sync_var, bg=CARD, activebackground=CARD, selectcolor=CARD, fg=TEXT, ).pack(anchor="w") autostart_state = "normal" if platform.system().lower() == "windows" else "disabled" tk.Checkbutton( checks, text="Start with Windows", variable=autostart_var, command=toggle_autostart, state=autostart_state, bg=CARD, activebackground=CARD, selectcolor=CARD, fg=TEXT, ).pack(anchor="w") bridge_buttons = tk.Frame(connect_card, bg=CARD) bridge_buttons.pack(fill="x") make_button(bridge_buttons, "Start", start_bridge, "brand").pack(side="left", fill="x", expand=True) make_button(bridge_buttons, "Stop", stop_bridge, "soft").pack(side="left", fill="x", expand=True, padx=(8, 0)) make_button(bridge_buttons, "Sync now", sync_now, "soft").pack(side="left", fill="x", expand=True, padx=(8, 0)) router_frame = make_card(grid) router_frame.grid(row=0, column=1, sticky="nsew", padx=(10, 0)) section_title(router_frame, "Local router", "Paste your ApiClo key once. The app will keep the local route ready for tools.") router_status = tk.Label( router_frame, textvariable=router_status_var, bg=CARD_SOFT, fg=BRAND_DARK, font=("Segoe UI", 9, "bold"), padx=10, pady=8, anchor="w", justify="left", wraplength=330, ) router_status.pack(fill="x", pady=(0, 10)) make_label(router_frame, "1. ApiClo key", size=9, weight="bold", fg=MUTED).pack(anchor="w") router_key_row = tk.Frame(router_frame, bg=CARD) router_key_row.pack(fill="x", pady=(4, 10)) router_key_entry = make_entry(router_key_row, router_key_var, show="*") router_key_entry.pack(side="left", fill="x", expand=True, ipady=6) make_button(router_key_row, "Paste", lambda: paste_into_entry(router_key_entry, "ApiClo key", True), "soft").pack(side="left", padx=(8, 0)) model_row = tk.Frame(router_frame, bg=CARD) model_row.pack(fill="x") model_col = tk.Frame(model_row, bg=CARD) model_col.pack(side="left", fill="x", expand=True) make_label(model_col, "2. Model", size=9, weight="bold", fg=MUTED).pack(anchor="w") model_combo = ttk.Combobox(model_col, textvariable=router_model_var, state="normal", style="ApiClo.TCombobox") model_combo.pack(fill="x", ipady=4, pady=(4, 10)) port_col = tk.Frame(model_row, bg=CARD) port_col.pack(side="left", padx=(10, 0)) make_label(port_col, "Port", size=9, weight="bold", fg=MUTED).pack(anchor="w") make_entry(port_col, router_port_var, width=7).pack(ipady=6, pady=(4, 10)) tk.Checkbutton( router_frame, text="Remember key on this computer", variable=router_remember_var, command=persist_router_settings, bg=CARD, activebackground=CARD, selectcolor=CARD, fg=TEXT, ).pack(anchor="w", pady=(0, 10)) router_buttons = tk.Frame(router_frame, bg=CARD) router_buttons.pack(fill="x") make_button(router_buttons, "Load models", fetch_models, "soft").pack(side="left", fill="x", expand=True) make_button(router_buttons, "Start local route", start_router_ui, "primary").pack(side="left", fill="x", expand=True, padx=(8, 0)) make_button(router_buttons, "Stop", stop_router_ui, "soft").pack(side="left", fill="x", expand=True, padx=(8, 0)) router_buttons_2 = tk.Frame(router_frame, bg=CARD) router_buttons_2.pack(fill="x", pady=(8, 0)) make_button(router_buttons_2, "Copy router URL", copy_router_url, "plain").pack(side="left", fill="x", expand=True) make_button(router_buttons_2, "Open guide", lambda: webbrowser.open(PORTAL_URL.rstrip("/") + "/knowledge"), "plain").pack(side="left", fill="x", expand=True, padx=(8, 0)) codex_card = make_card(outer) codex_card.pack(fill="x", pady=(0, 12)) section_title( codex_card, "Codex connector", "Creates a backup, then adds ApiClo as the Codex provider for new Codex sessions.", ) codex_row = tk.Frame(codex_card, bg=CARD) codex_row.pack(fill="x") codex_text = tk.Frame(codex_row, bg=CARD) codex_text.pack(side="left", fill="x", expand=True) make_label(codex_text, "Target: Codex user config", size=10, weight="bold").pack(anchor="w") make_label( codex_text, "Use Preview first. Rollback restores the latest ApiClo backup.", size=8, fg=MUTED, wraplength=520, ).pack(anchor="w", pady=(1, 0)) make_button(codex_row, "Preview", preview_codex_connector_ui, "plain", width=10).pack(side="left", padx=(8, 0)) make_button(codex_row, "Connect Codex", connect_codex_ui, "primary", width=14).pack(side="left", padx=(8, 0)) make_button(codex_row, "Rollback", rollback_codex_ui, "soft", width=10).pack(side="left", padx=(8, 0)) clients_card = make_card(outer) clients_card.pack(fill="x", pady=(0, 12)) section_title( clients_card, "Use ApiClo in your tools", "Pick a tool and press Prepare. Existing tool configs are not changed.", ) for index, client in enumerate(CLI_CLIENTS): row = tk.Frame(clients_card, bg=CARD) row.pack(fill="x", pady=(0 if index == 0 else 8, 0)) client_text = tk.Frame(row, bg=CARD) client_text.pack(side="left", fill="x", expand=True) make_label(client_text, client["name"], size=10, weight="bold").pack(anchor="w") make_label( client_text, client.get("setup_note", "Saves an ApiClo local profile"), size=8, fg=MUTED, wraplength=500, ).pack(anchor="w", pady=(1, 0)) make_button(row, "Prepare", lambda key=client["key"]: apply_client_setup(key), "primary", width=11).pack(side="left", padx=(8, 0)) make_button(row, "Open", lambda key=client["key"]: open_client_download(key), "plain", width=9).pack(side="left", padx=(8, 0)) make_button(row, "Advanced", lambda key=client["key"]: copy_client_config(key), "soft", width=9).pack(side="left", padx=(8, 0)) log_card = make_card(outer) log_card.pack(fill="both", expand=True) log_header = tk.Frame(log_card, bg=CARD) log_header.pack(fill="x", pady=(0, 8)) make_label(log_header, "Activity", size=13, weight="bold").pack(side="left") make_button(log_header, "Open chat", lambda: webbrowser.open(PORTAL_URL.rstrip("/") + "/chat"), "plain").pack(side="right") log_frame = tk.Frame(log_card, bg=INK) log_frame.pack(fill="both", expand=True) log_scrollbar = tk.Scrollbar(log_frame) log_scrollbar.pack(side="right", fill="y") log_box = tk.Text( log_frame, height=5, state="disabled", bg=INK, fg="#f7efe6", insertbackground="#f7efe6", relief="flat", bd=0, font=("Consolas", 9), yscrollcommand=log_scrollbar.set, ) log_box.pack(side="left", fill="both", expand=True) log_scrollbar.configure(command=log_box.yview) bottom = tk.Frame(outer, bg=BG) bottom.pack(fill="x", pady=(12, 0)) make_label(bottom, "Close hides the app to tray when tray support is available.", size=9, fg=MUTED, bg=BG).pack(side="left") make_button(bottom, "Quit", quit_app, "primary", width=10).pack(side="right") tray_ref["icon"] = start_tray_icon(root, show_window, sync_now, quit_app) append_log("App ready. Close hides the window when tray support is available.") if os.getenv("APICLAUDE_BRIDGE_PAIRING_CODE") or config.get("device_token"): root.after(500, start_bridge) root.mainloop() def main() -> None: args = set(sys.argv[1:]) actions_enabled = "--actions" in args or env_bool("APICLAUDE_BRIDGE_ACTIONS", False) loop_enabled = "--loop" in args or env_bool("APICLAUDE_BRIDGE_LOOP", False) config = load_config() if not config.get("device_token"): config = register(actions_enabled) manifest = heartbeat(config, actions_enabled) if should_sync(args, manifest): upload_data(config, manifest) if actions_enabled: poll_actions(config, manifest) while loop_enabled: time.sleep(max(15, env_int("APICLAUDE_BRIDGE_HEARTBEAT_SECONDS", 30, 15, 3600))) manifest = heartbeat(config, actions_enabled) if actions_enabled: poll_actions(config, manifest) if __name__ == "__main__": try: if app_mode_requested(): run_app() else: main() except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace") print("ApiClo Bridge HTTP %s: %s" % (exc.code, detail), file=sys.stderr) sys.exit(1)