unsloth/unsloth_cli/commands/connect.py

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0

"""`unsloth connect` — launch a coding agent against a running Studio server."""

import json
import os
import re
import shlex
import shutil
import signal
import subprocess
import urllib.error
import urllib.request
from pathlib import Path
from typing import NoReturn, Optional

import typer

from unsloth_cli._inference import (
    _USER_AGENT,
    _studio_token,
    ensure_studio_backend_path,
    find_studio_server,
)

connect_app = typer.Typer(
    help = "Connect a coding agent to a running Studio server.",
    no_args_is_help = True,
    context_settings = {"help_option_names": ["-h", "--help"]},
)

_CODEX_PROFILE = "unsloth_api"
_CODEX_ENV_KEY = "UNSLOTH_STUDIO_AUTH_TOKEN"
_HERMES_ENV_KEY = "UNSLOTH_API_KEY"
_HERMES_PROVIDER = "unsloth"
_PROVIDER_HEADER = f"[model_providers.{_CODEX_PROFILE}]"
_PASSTHROUGH = {"allow_extra_args": True, "ignore_unknown_options": True}
_CLAUDE_ENV_UNSET = ("ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN")

# Shared by every agent command; only the config/env/command differ.
_MODEL_OPTION = typer.Option(
    None, "--model", "-m", help = "Model for the agent; defaults to the one loaded in Studio."
)
_KEY_OPTION = typer.Option(
    None,
    "--api-key",
    envvar = "UNSLOTH_API_KEY",
    help = "Studio API key; minted automatically when omitted. Keys are remembered, so passing one once is enough.",
)
_LAUNCH_OPTION = typer.Option(
    True,
    "--launch/--no-launch",
    help = "--no-launch prints the env and command instead (remote shells, WSL).",
)


def _fail(message: str) -> NoReturn:
    typer.echo(message, err = True)
    raise typer.Exit(code = 1)


def _http_error_detail(exc: urllib.error.HTTPError) -> str:
    try:
        body = json.loads(exc.read().decode())
        return body.get("detail") or body["error"]["message"]
    except Exception:
        return str(exc)


def _http_json(
    method: str,
    url: str,
    token: str,
    payload = None,
    timeout = 30,
    error = None,
):
    """On HTTPError: raise if `error` is None, else fail with `error` plus the server's detail."""
    request = urllib.request.Request(
        url,
        data = None if payload is None else json.dumps(payload).encode(),
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "User-Agent": _USER_AGENT,
        },
        method = method,
    )
    try:
        with urllib.request.urlopen(request, timeout = timeout) as response:
            return json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as exc:
        if error is None:
            raise
        _fail(f"{error}: {_http_error_detail(exc)}")
    except (urllib.error.URLError, TimeoutError) as exc:
        if error is None:
            raise
        _fail(f"{error}: {getattr(exc, 'reason', None) or exc}")


def _require_studio() -> str:
    base = find_studio_server()
    if base is None:
        expected = os.environ.get("UNSLOTH_STUDIO_URL", "http://127.0.0.1:8888")
        _fail(
            f"No running Studio server found at {expected}. Start one with "
            "`unsloth studio`, or point UNSLOTH_STUDIO_URL at a remote server."
        )
    return base


def _key_cache_path() -> Path:
    ensure_studio_backend_path()
    from utils.paths import auth_root
    return auth_root() / "agent_api_key.json"


def _cached_keys(cache: Path) -> list:
    try:
        data = json.loads(cache.read_text())
    except Exception:
        return []
    if not isinstance(data, dict):
        return []
    keys = [k for k in data.get("keys", []) if isinstance(k, str)]
    legacy = data.get("key")  # pre-multi-key cache format
    if isinstance(legacy, str) and legacy not in keys:
        keys.append(legacy)
    return keys


def _write_private_json(path: Path, data: dict) -> None:
    # O_CREAT with 0o600 so a file holding an API key is never world-readable,
    # even briefly (existing files keep whatever perms the user set).
    path.parent.mkdir(parents = True, exist_ok = True, mode = 0o700)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as handle:
        handle.write(json.dumps(data, indent = 2) + "\n")


def _read_json_object(path: Path) -> Optional[dict]:
    # {} when missing, None when it can't be parsed as an object (so the caller
    # leaves a user-managed file untouched rather than clobbering it).
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding = "utf-8"))
    except (ValueError, OSError):
        return None
    return data if isinstance(data, dict) else None


def _subdict(parent: dict, key: str) -> dict:
    child = parent.get(key)
    if not isinstance(child, dict):
        child = parent[key] = {}
    return child


def _remember_key(cache: Path, key: str) -> None:
    existing = _cached_keys(cache)
    keys = ([key] + [k for k in existing if k != key])[:8]
    if keys == existing:
        return
    try:
        _write_private_json(cache, {"keys": keys})
    except OSError:
        pass  # worst case the next launch mints another key


def _agent_api_key(base: str, explicit: Optional[str]) -> str:
    cache = _key_cache_path()
    if explicit:
        _remember_key(cache, explicit)
        return explicit

    # Keys are per-server, so when switching between Studios (local one day,
    # an SSH-tunnelled remote the next) the right key is whichever validates.
    for key in _cached_keys(cache):
        try:
            _http_json("GET", f"{base}/v1/models", key)
        except Exception:
            continue
        _remember_key(cache, key)
        return key

    token = _studio_token()
    auth_help = (
        "Couldn't authenticate with the Studio server automatically (it may be "
        "remote, or running as a different OS user). Create an API key in "
        "Studio → Settings → API and pass it once with --api-key; it is "
        "remembered for next time."
    )
    if token is None:
        _fail(auth_help)
    try:
        key = _http_json(
            "POST",
            f"{base}/api/auth/api-keys",
            token,
            {"name": "Coding agents (unsloth connect)"},
        )["key"]
    except urllib.error.HTTPError as exc:
        # A self-issued token only validates against a local, same-OS-user
        # server; a remote Studio signs with a different secret and rejects it
        # (401/403). Point at --api-key instead of the raw "expired token".
        if exc.code in (401, 403):
            _fail(auth_help)
        _fail(f"Couldn't create an API key: {_http_error_detail(exc)}")
    except (urllib.error.URLError, TimeoutError) as exc:
        _fail(f"Couldn't create an API key: {getattr(exc, 'reason', None) or exc}")
    _remember_key(cache, key)
    return key


def _loaded_models(base: str, key: str) -> list:
    return _http_json("GET", f"{base}/v1/models", key, error = "Couldn't list models").get("data", [])


def _resolve_model(base: str, key: str, requested: Optional[str]) -> dict:
    models = _loaded_models(base, key)
    match = next((m for m in models if m["id"] == requested), None)
    if requested and match is None:
        typer.echo(f"Loading {requested} on the Studio server (this can take a while)…")
        loaded = _http_json(
            "POST",
            f"{base}/api/inference/load",
            key,
            {"model_path": requested},
            timeout = 3600,
            error = "Model load failed",
        )
        # Studio registers the model under a canonical id (resolved identifier,
        # casing) that /v1/models echoes but which may differ from the path we
        # passed; match on the id the load reports so we don't silently fall
        # through to models[0] and connect to a different loaded model.
        wanted = {requested}
        if isinstance(loaded, dict):
            wanted |= {loaded.get("model"), loaded.get("display_name")} - {None}
        models = _loaded_models(base, key)
        match = next((m for m in models if m["id"] in wanted), None)
    if match is not None:
        return match
    if requested:
        # We asked Studio to load it and it didn't surface in /v1/models; don't
        # silently hand back an unrelated loaded model.
        _fail(
            f"Studio didn't report '{requested}' as loaded. Double-check the model "
            "id, or load it from the model dropdown in the UI."
        )
    if not models:
        _fail(
            "No model is loaded in Studio. Load one from the model dropdown in "
            "the UI, or pass --model <hf-id-or-path> to load it from here."
        )
    return models[0]


def _require_gguf_for_codex(base: str, key: str, model_id: str) -> None:
    # Codex always streams, and Studio only streams /v1/responses from llama-server.
    try:
        status = _http_json("GET", f"{base}/api/inference/status", key)
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            return  # older server without the endpoint; don't block the launch
        raise
    if status.get("is_gguf"):
        return
    hint = model_id if "gguf" in model_id.lower() else f"{model_id}-GGUF"
    _fail(
        f"Codex needs a GGUF model served by llama-server, but {model_id} is on "
        f"the transformers backend. Try: unsloth connect codex --model {hint}"
    )


def claude_settings_path() -> Path:
    return Path.home() / ".claude" / "settings.json"


def ensure_claude_attribution_header() -> None:
    # The header invalidates the llama.cpp KV cache (~90% slower) and Claude
    # Code only honors this setting from settings.json, not the env var.
    path = claude_settings_path()
    settings = {}
    if path.exists():
        try:
            settings = json.loads(path.read_text(encoding = "utf-8"))
        except (ValueError, OSError):
            settings = None
        if not isinstance(settings, dict):
            typer.echo(
                f"Warning: couldn't parse {path} — set CLAUDE_CODE_ATTRIBUTION_HEADER "
                'to "0" in its "env" section yourself, or local inference will be much slower.',
                err = True,
            )
            return
    env = settings.get("env")
    if not isinstance(env, dict):
        env = settings["env"] = {}
    if str(env.get("CLAUDE_CODE_ATTRIBUTION_HEADER")) == "0":
        return
    env["CLAUDE_CODE_ATTRIBUTION_HEADER"] = "0"
    try:
        path.parent.mkdir(parents = True, exist_ok = True)
        path.write_text(json.dumps(settings, indent = 2) + "\n", encoding = "utf-8")
    except OSError:
        typer.echo(
            f"Warning: couldn't write {path} — set CLAUDE_CODE_ATTRIBUTION_HEADER "
            'to "0" in its "env" section yourself, or local inference will be much slower.',
            err = True,
        )
        return
    typer.echo(f"Disabled Claude Code's attribution header in {path} (it breaks KV-cache reuse).")


_DYNAMIC_SECTIONS_FLAG = "--exclude-dynamic-system-prompt-sections"


def _claude_cache_flags() -> list:
    # The flag moves per-machine context (cwd, env info, git status) out of
    # the system prompt, where it changes every session and defeats llama.cpp
    # prefix caching. As of 2.1.175 it only takes effect in print mode (`-p`
    # passed through ctx.args); interactive sessions accept and ignore it.
    # Claude Code < 2.1.98 aborts on the unknown flag, so check the version
    # first; no local binary means a --no-launch printout for another machine.
    executable = shutil.which("claude")
    if executable is None:
        return [_DYNAMIC_SECTIONS_FLAG]
    try:
        result = subprocess.run(
            [executable, "--version"], capture_output = True, text = True, timeout = 10
        )
        version = tuple(int(part) for part in result.stdout.split()[0].split("."))
    except Exception:
        return []
    return [_DYNAMIC_SECTIONS_FLAG] if version >= (2, 1, 98) else []


def codex_home() -> Path:
    return Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex")


def _merge_codex_config(existing: str, base: str) -> str:
    chunks = re.split(r"(?m)^(?=\[)", existing)  # preamble, then one chunk per table
    if not re.search(r"(?m)^\s*oss_provider\s*=", chunks[0]):
        if chunks[0] and not chunks[0].endswith("\n"):
            chunks[0] += "\n"
        chunks[0] += f'oss_provider = "{_CODEX_PROFILE}"\n'
    # Drop the provider table and any stale [model_providers.unsloth_api.*] subtables.
    stale = (_PROVIDER_HEADER, _PROVIDER_HEADER[:-1] + ".")
    text = "".join(c for c in chunks if not c.startswith(stale))
    if not text.endswith("\n"):
        text += "\n"
    if not text.endswith("\n\n"):
        text += "\n"
    return text + (
        f"{_PROVIDER_HEADER}\n"
        'name = "Unsloth Studio"\n'
        f"base_url = {json.dumps(base + '/v1')}\n"
        f'env_key = "{_CODEX_ENV_KEY}"\n'
        'wire_api = "responses"\n'
        "requires_openai_auth = false\n"
    )


def write_codex_config(base: str, model: dict) -> None:
    home = codex_home()
    home.mkdir(parents = True, exist_ok = True)

    config = home / "config.toml"
    existing = config.read_text(encoding = "utf-8") if config.exists() else ""
    merged = _merge_codex_config(existing, base)
    if merged != existing:
        config.write_text(merged, encoding = "utf-8")
        typer.echo(f"Updated {config}")

    # oss_provider here too: codex --oss picks the provider from it, and the
    # profile layer must beat a user-set value (e.g. "ollama") in config.toml.
    profile_text = (
        f'oss_provider = "{_CODEX_PROFILE}"\n'
        f'model_provider = "{_CODEX_PROFILE}"\n'
        f"model = {json.dumps(model['id'])}\n"
    )
    window = model.get("context_length") or model.get("max_context_length")
    if window:
        profile_text += f"model_context_window = {int(window)}\n"
    profile = home / f"{_CODEX_PROFILE}.config.toml"
    if not profile.exists() or profile.read_text(encoding = "utf-8") != profile_text:
        profile.write_text(profile_text, encoding = "utf-8")
        typer.echo(f"Updated {profile}")


def _wsl_windows_executable(command: list) -> Optional[str]:
    if os.name == "nt" or not os.environ.get("WSL_DISTRO_NAME"):
        return None
    executable = shutil.which(command[0])
    if executable and executable.startswith("/mnt/"):
        return executable
    return None


def _merge_wslenv(current: str, names: tuple) -> str:
    entries = [entry for entry in current.split(":") if entry]
    existing = {entry.split("/", 1)[0] for entry in entries}
    entries.extend(name for name in names if name not in existing)
    return ":".join(entries)


def _print_env(
    env: dict,
    command: list,
    unset_env: tuple = (),
    wsl_env_bridge: tuple = (),
) -> None:
    if os.name == "nt":
        for name in unset_env:
            typer.echo(f"Remove-Item Env:{name} -ErrorAction SilentlyContinue")
        for name, value in env.items():
            # PowerShell: ` is the escape char, and $ triggers expansion inside "".
            escaped = value.replace("`", "``").replace('"', '`"').replace("$", "`$")
            typer.echo(f'$env:{name} = "{escaped}"')
        typer.echo(subprocess.list2cmdline(command))
        return
    for name in unset_env:
        typer.echo(f"export {name}=" if wsl_env_bridge else f"unset {name}")
    for name, value in env.items():
        typer.echo(f"export {name}={shlex.quote(value)}")
    if wsl_env_bridge:
        typer.echo(
            f"export WSLENV={shlex.quote(_merge_wslenv(os.environ.get('WSLENV', ''), wsl_env_bridge))}"
        )
    typer.echo(shlex.join(command))


def _launch(
    command: list,
    env: dict,
    install_hint: str,
    unset_env: tuple = (),
) -> NoReturn:
    executable = shutil.which(command[0])
    if executable is None:
        _fail(f"`{command[0]}` not found on PATH. Install it with: {install_hint}")
    wsl_env_bridge = (
        tuple(dict.fromkeys((*env.keys(), *unset_env))) if _wsl_windows_executable(command) else ()
    )
    child_env = dict(os.environ)
    if wsl_env_bridge:
        child_env["WSLENV"] = _merge_wslenv(child_env.get("WSLENV", ""), wsl_env_bridge)
        for name in unset_env:
            child_env[name] = ""
    else:
        for name in unset_env:
            child_env.pop(name, None)
    child_env.update(env)
    # Ctrl+C cancels a turn inside the agent; don't let it kill this wrapper.
    previous = signal.signal(signal.SIGINT, signal.SIG_IGN)
    try:
        code = subprocess.run([executable, *command[1:]], env = child_env).returncode
    finally:
        signal.signal(signal.SIGINT, previous)
    # Negative returncode means killed by signal N; shells expect 128+N.
    raise typer.Exit(code = code if code >= 0 else 128 - code)


def _connect(api_key: Optional[str], model: Optional[str]) -> tuple:
    base = _require_studio()
    key = _agent_api_key(base, api_key)
    return base, key, _resolve_model(base, key, model)


def _run(
    base: str,
    entry: dict,
    env: dict,
    command: list,
    *,
    launch: bool,
    install_hint: str,
    unset_env: tuple = (),
) -> None:
    typer.echo(f"Studio {base} · model {entry['id']}")
    wsl_env_bridge = (
        tuple(dict.fromkeys((*env.keys(), *unset_env))) if _wsl_windows_executable(command) else ()
    )
    if not launch:
        _print_env(env, command, unset_env = unset_env, wsl_env_bridge = wsl_env_bridge)
        return
    _launch(command, env, install_hint = install_hint, unset_env = unset_env)


def openclaw_config_path() -> Path:
    return Path.home() / ".openclaw" / "openclaw.json"


def write_openclaw_config(base: str, key: str, model: dict) -> None:
    path = openclaw_config_path()
    config = _read_json_object(path)
    if config is None:
        typer.echo(
            f"Warning: couldn't parse {path} — add an 'unsloth' provider there "
            "yourself, or move the file aside and re-run.",
            err = True,
        )
        return
    before = json.dumps(config, sort_keys = True)
    # Studio is a generic OpenAI-compatible /v1 endpoint (the vLLM/LM Studio path).
    provider_model = {"id": model["id"], "name": model["id"]}
    window = model.get("context_length") or model.get("max_context_length")
    if window:
        provider_model["contextWindow"] = int(window)
    models = _subdict(config, "models")
    models.setdefault("mode", "merge")
    _subdict(models, "providers")["unsloth"] = {
        "baseUrl": f"{base}/v1",
        "apiKey": key,
        "api": "openai-completions",
        "models": [provider_model],
    }
    # Pin a default model, else OpenClaw drops into its setup agent ("no models available").
    defaults = _subdict(_subdict(config, "agents"), "defaults")
    _subdict(defaults, "model")["primary"] = f"unsloth/{model['id']}"
    # Unauthenticated loopback gateway: without auth.mode=none the client won't open
    # the websocket. The daemon must still be started separately (`openclaw gateway`).
    gateway = _subdict(config, "gateway")
    gateway.setdefault("mode", "local")
    _subdict(gateway, "auth").setdefault("mode", "none")
    if json.dumps(config, sort_keys = True) != before:
        _write_private_json(path, config)
        typer.echo(f"Updated {path}")


def opencode_config_path() -> Path:
    config_home = os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config"
    return Path(config_home) / "opencode" / "opencode.json"


def write_opencode_config(base: str, key: str, model: dict) -> None:
    path = opencode_config_path()
    config = _read_json_object(path)
    if config is None:
        typer.echo(
            f"Warning: couldn't parse {path} — add an 'unsloth' provider there "
            "yourself, or move the file aside and re-run.",
            err = True,
        )
        return
    before = json.dumps(config, sort_keys = True)
    config.setdefault("$schema", "https://opencode.ai/config.json")
    _subdict(config, "provider")["unsloth"] = {
        "npm": "@ai-sdk/openai-compatible",
        "name": "Unsloth Studio",
        "options": {"baseURL": f"{base}/v1", "apiKey": key},
        "models": {model["id"]: {"name": model["id"]}},
    }
    # OpenCode selects a model by "<providerID>/<modelID>".
    config["model"] = f"unsloth/{model['id']}"
    if json.dumps(config, sort_keys = True) != before:
        _write_private_json(path, config)
        typer.echo(f"Updated {path}")


def hermes_config_path() -> Path:
    return Path.home() / ".hermes" / "config.yaml"


def write_hermes_config(base: str, model: dict) -> None:
    import yaml

    path = hermes_config_path()
    config: dict = {}
    if path.exists():
        try:
            loaded = yaml.safe_load(path.read_text(encoding = "utf-8"))
        except (yaml.YAMLError, OSError):
            typer.echo(
                f"Warning: couldn't parse {path} — configure the custom endpoint "
                "there yourself, or move the file aside and re-run.",
                err = True,
            )
            return
        if isinstance(loaded, dict):
            config = loaded
        elif loaded is not None:
            # Non-empty, non-mapping YAML is a user-managed file; leave it.
            typer.echo(
                f"Warning: couldn't parse {path} — configure the custom endpoint "
                "there yourself, or move the file aside and re-run.",
                err = True,
            )
            return
    # Hermes only reads the key for a *named* custom provider (a bare
    # `provider: custom` ignores it), so register it under providers.*.
    _subdict(config, "model").update(
        provider = f"custom:{_HERMES_PROVIDER}",
        default = model["id"],
        api_mode = "openai",
    )
    _subdict(config, "providers")[_HERMES_PROVIDER] = {
        "base_url": f"{base}/v1",
        "api_mode": "openai",
        "key_env": _HERMES_ENV_KEY,
    }
    text = yaml.safe_dump(config, sort_keys = False)
    if not path.exists() or path.read_text(encoding = "utf-8") != text:
        path.parent.mkdir(parents = True, exist_ok = True)
        path.write_text(text, encoding = "utf-8")
        typer.echo(f"Updated {path}")


@connect_app.command("claude", context_settings = _PASSTHROUGH)
def claude(
    ctx: typer.Context,
    model: Optional[str] = _MODEL_OPTION,
    api_key: Optional[str] = _KEY_OPTION,
    launch: bool = _LAUNCH_OPTION,
):
    """Point Claude Code at the running Studio server and start it."""
    base, key, entry = _connect(api_key, model)
    model_id = entry["id"]
    ensure_claude_attribution_header()

    env = {
        "ANTHROPIC_BASE_URL": base,
        "ANTHROPIC_AUTH_TOKEN": key,
        "ANTHROPIC_MODEL": model_id,
        # Update checks, beta features, and other background requests either
        # stall against a local server or evict the conversation from
        # llama-server's KV-cache slots, so turn off everything nonessential.
        "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
        "CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1",
    }
    command = ["claude", "--model", model_id, *_claude_cache_flags(), *ctx.args]
    install_hint = (
        "irm https://claude.ai/install.ps1 | iex"
        if os.name == "nt"
        else "curl -fsSL https://claude.ai/install.sh | bash"
    )
    _run(
        base,
        entry,
        env,
        command,
        launch = launch,
        install_hint = install_hint,
        unset_env = _CLAUDE_ENV_UNSET,
    )


@connect_app.command("codex", context_settings = _PASSTHROUGH)
def codex(
    ctx: typer.Context,
    model: Optional[str] = _MODEL_OPTION,
    api_key: Optional[str] = _KEY_OPTION,
    launch: bool = _LAUNCH_OPTION,
):
    """Point OpenAI Codex at the running Studio server and start it."""
    base, key, entry = _connect(api_key, model)
    _require_gguf_for_codex(base, key, entry["id"])
    write_codex_config(base, entry)

    env = {_CODEX_ENV_KEY: key}
    command = ["codex", "--oss", "--profile", _CODEX_PROFILE, *ctx.args]
    _run(base, entry, env, command, launch = launch, install_hint = "npm install -g @openai/codex")


@connect_app.command("openclaw", context_settings = _PASSTHROUGH)
def openclaw(
    ctx: typer.Context,
    model: Optional[str] = _MODEL_OPTION,
    api_key: Optional[str] = _KEY_OPTION,
    launch: bool = _LAUNCH_OPTION,
):
    """Point OpenClaw at the running Studio server and start it."""
    base, key, entry = _connect(api_key, model)
    write_openclaw_config(base, key, entry)  # key lives in the config, not the env

    command = ["openclaw", *ctx.args]
    install_hint = (
        "iwr -useb https://openclaw.ai/install.ps1 | iex"
        if os.name == "nt"
        else "curl -fsSL https://openclaw.ai/install.sh | bash"
    )
    _run(base, entry, {}, command, launch = launch, install_hint = install_hint)


@connect_app.command("opencode", context_settings = _PASSTHROUGH)
def opencode(
    ctx: typer.Context,
    model: Optional[str] = _MODEL_OPTION,
    api_key: Optional[str] = _KEY_OPTION,
    launch: bool = _LAUNCH_OPTION,
):
    """Point OpenCode at the running Studio server and start it."""
    base, key, entry = _connect(api_key, model)
    write_opencode_config(base, key, entry)  # key lives in the config, not the env

    command = ["opencode", *ctx.args]
    _run(base, entry, {}, command, launch = launch, install_hint = "npm install -g opencode-ai")


@connect_app.command("hermes", context_settings = _PASSTHROUGH)
def hermes(
    ctx: typer.Context,
    model: Optional[str] = _MODEL_OPTION,
    api_key: Optional[str] = _KEY_OPTION,
    launch: bool = _LAUNCH_OPTION,
):
    """Point Hermes (Nous Research) at the running Studio server and start it."""
    base, key, entry = _connect(api_key, model)
    write_hermes_config(base, entry)

    env = {_HERMES_ENV_KEY: key}
    command = ["hermes", *ctx.args]
    install_hint = (
        "curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent"
        "/main/scripts/install.sh | bash"
    )
    _run(base, entry, env, command, launch = launch, install_hint = install_hint)