Files
MNeMoNiCuZ-ComfyUI-mnemic-n…/utils/llm_cli.py
T

496 lines
20 KiB
Python

"""Local subscription CLIs as LLM endpoints: Claude Code and Codex.
Both run the CLI installed on this machine in non-interactive mode, signed in
with your own subscription, so no API key is involved:
- Claude Code: `claude -p` with stream-json input/output. All tools are
disabled (`--tools ""`), MCP servers and user/project settings (hooks)
aren't loaded and nothing is saved as a session, so it answers as a plain
model.
- Codex: `codex exec --json --ignore-user-config`. Its tool features
(shell, browser, computer use, apps, hooks…) are disabled where this Codex
version has them, and it runs in a read-only sandbox in an empty temporary
folder. It refuses to run if the shell tools can't be confirmed off.
API-key variables (ANTHROPIC_API_KEY, OPENAI_API_KEY…) are removed from the
CLI's environment so the subscription login is what gets used.
"""
import base64
import json
import re
import os
import contextlib
import queue
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
from .env_manager import redact
CLAUDE = "claude_cli"
CODEX = "codex_cli"
CLI_PROVIDERS = (CLAUDE, CODEX)
# Other subscription CLIs considered and not added here:
# - xAI/Grok: no official CLI with subscription-based auth comparable to
# Claude Code or Codex was found.
# - Google Gemini CLI (google-gemini/gemini-cli) has subscription auth, but
# its own --output-format json is reported broken/unreliable upstream:
# https://github.com/google-gemini/gemini-cli/issues/9009
# https://github.com/google-gemini/gemini-cli/issues/8022
# A local proxy such as https://github.com/router-for-me/CLIProxyAPI wraps
# Gemini CLI, Grok, Claude Code and Codex logins behind a normal
# OpenAI-compatible /v1/chat/completions server; that already works with
# this pack's generic "openai" provider (add it in UserEndpoints.json, see
# the "CLI Proxy API" example) without a bespoke subprocess adapter here.
# Only as a fallback when nothing else is given; Claude Code's own default
# system prompt is a coding agent's.
DEFAULT_SYSTEM = "You are a helpful assistant."
_STRIP_ENV = {
CLAUDE: ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"),
CODEX: ("OPENAI_API_KEY", "CODEX_API_KEY"),
}
_CODEX_TOOL_FEATURES = (
"shell_tool", "unified_exec", "browser_use", "browser_use_external", "in_app_browser",
"computer_use", "apps", "image_generation", "plugins", "remote_plugin", "tool_suggest",
"standalone_web_search", "web_search_request", "web_search_cached",
"hooks", "multi_agent", "view_image", "goals",
)
# Model names go on the command line. On Windows an npm-installed CLI is a
# .cmd script run through cmd.exe, which Python can't quote safely, so a
# model name from a (possibly shared) workflow is restricted to plain
# characters.
_SAFE_MODEL = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@\[\]-]{0,127}$")
_CODEX_SHELL_FEATURES = {"shell_tool", "unified_exec"}
_EFFORT = {"minimal": "low", "none": "low", "low": "low", "medium": "medium", "high": "high"}
_codex_features_cache = {}
class CLIError(Exception):
def __init__(self, message, status="cli error"):
super().__init__(redact(str(message)))
self.status = status
def find_command(command):
"""Full path of the CLI, or None if it isn't installed."""
if not command:
return None
if os.path.isfile(command):
return command
return shutil.which(command)
def _env(provider):
env = dict(os.environ)
for name in _STRIP_ENV[provider]:
env.pop(name, None)
return env
def _popen(args, provider, cwd):
# Its own process group: npm installs are wrappers (node, or a .cmd on
# Windows) around the real binary, and stopping only the wrapper would
# leave the model call running.
kwargs = {}
if sys.platform == "win32":
kwargs["creationflags"] = getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
else:
kwargs["start_new_session"] = True
return subprocess.Popen(
args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=cwd, env=_env(provider), **kwargs,
)
def _terminate(proc):
"""Stop the CLI and everything it started."""
if proc.poll() is not None:
return
try:
if sys.platform == "win32":
subprocess.run(["taskkill", "/T", "/F", "/PID", str(proc.pid)], capture_output=True, timeout=10,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
else:
os.killpg(proc.pid, signal.SIGTERM) # the node wrapper forwards it
try:
proc.wait(timeout=3)
return
except subprocess.TimeoutExpired:
os.killpg(proc.pid, signal.SIGKILL)
except (OSError, subprocess.SubprocessError):
proc.kill()
with contextlib.suppress(subprocess.TimeoutExpired):
proc.wait(timeout=5)
@contextlib.contextmanager
def _work_dir():
"""A temporary folder that is removed afterwards even if a stopped CLI
still had it open (Windows refuses to delete it then)."""
path = tempfile.mkdtemp(prefix="mnemic_llm_")
try:
yield path
finally:
shutil.rmtree(path, ignore_errors=True)
def _read_lines(stream, out):
for raw in iter(stream.readline, b""):
out.put(raw)
out.put(None)
def _run(args, provider, cwd, stdin_bytes, on_line, timeout, check_interrupt):
"""Run the CLI, feeding stdin and passing each stdout line to on_line.
Returns (exit_code, stderr_text). Kills the process on timeout or
interrupt.
"""
proc = _popen(args, provider, cwd)
lines = queue.Queue()
stderr_chunks = []
stdout_reader = threading.Thread(target=_read_lines, args=(proc.stdout, lines), daemon=True)
stderr_reader = threading.Thread(target=lambda: stderr_chunks.append(proc.stderr.read()), daemon=True)
stdout_reader.start()
stderr_reader.start()
try:
proc.stdin.write(stdin_bytes)
proc.stdin.close()
except OSError:
pass
deadline = time.monotonic() + timeout
try:
while True:
if check_interrupt:
check_interrupt()
if time.monotonic() > deadline:
raise CLIError(f"no reply within {timeout} s", status="timeout")
try:
raw = lines.get(timeout=0.25)
except queue.Empty:
continue
if raw is None:
break
line = raw.decode("utf-8", errors="replace").strip()
if line:
on_line(line)
proc.wait(timeout=10)
# The error text (e.g. "not logged in") is only complete once the
# reader has finished.
stdout_reader.join(timeout=5)
stderr_reader.join(timeout=5)
except BaseException:
_terminate(proc)
raise
stderr = b"".join(c for c in stderr_chunks if c).decode("utf-8", errors="replace")
return proc.returncode, stderr
def _last_line(text):
lines = [l.strip() for l in (text or "").splitlines() if l.strip()]
return lines[-1][:500] if lines else ""
# --------------------------------------------------------------------------
# Claude Code
# --------------------------------------------------------------------------
def _claude_content(req, encoded_images):
content = [{"type": "image", "source": {"type": "base64", "media_type": mime, "data": data}}
for mime, data in encoded_images]
if req.user:
content.append({"type": "text", "text": req.user})
return content
def run_claude(command, req, result, *, encoded_images, system, timeout, on_delta, check_interrupt, log):
with _work_dir() as work:
system_file = os.path.join(work, "system.txt")
with open(system_file, "w", encoding="utf-8") as f:
f.write(system or DEFAULT_SYSTEM)
args = [command, "-p", "--input-format", "stream-json", "--output-format", "stream-json",
"--verbose", "--include-partial-messages", "--tools", "", "--strict-mcp-config",
"--no-session-persistence", "--setting-sources", "", "--system-prompt-file", system_file]
if req.model:
args += ["--model", req.model]
if req.reasoning in _EFFORT:
args += ["--effort", _EFFORT[req.reasoning]]
message = {"type": "user", "message": {"role": "user", "content": _claude_content(req, encoded_images)}}
if log:
shown = " ".join(a if a else '""' for a in args[1:])
log(f"Claude Code CLI: {shown}")
text, thinking = [], []
final = {}
last_push = [0.0]
def on_line(line):
try:
event = json.loads(line)
except ValueError:
return
kind = event.get("type")
if kind == "stream_event":
inner = event.get("event") or {}
if inner.get("type") == "content_block_delta":
delta = inner.get("delta") or {}
if delta.get("type") == "text_delta":
text.append(delta.get("text", ""))
elif delta.get("type") == "thinking_delta":
thinking.append(delta.get("thinking", ""))
now = time.monotonic()
if on_delta and now - last_push[0] > 0.08:
last_push[0] = now
on_delta("".join(text), "".join(thinking))
elif inner.get("type") == "message_start":
result.model = (inner.get("message") or {}).get("model", result.model)
elif kind == "result":
final.update(event)
code, stderr = _run(args, CLAUDE, work, (json.dumps(message) + "\n").encode("utf-8"),
on_line, timeout, check_interrupt)
if final.get("is_error") or final.get("subtype", "success") != "success" or (code and not final):
detail = final.get("result") or _last_line(stderr) or f"exit code {code}"
raise CLIError(f"Claude Code CLI: {detail}. If it asks you to log in, run `claude` once in a terminal.")
result.text = final.get("result") if isinstance(final.get("result"), str) else "".join(text)
result.thinking = "".join(thinking)
result.finish_reason = final.get("stop_reason") or "end_turn"
usage = final.get("usage") or {}
result.input_tokens = usage.get("input_tokens")
result.output_tokens = usage.get("output_tokens")
return result
# --------------------------------------------------------------------------
# Codex
# --------------------------------------------------------------------------
def _codex_features(command, work):
"""Feature flags this Codex version knows (disabling an unknown one is a
hard error). Only a successful probe is cached, per binary."""
try:
key = (command, os.path.getmtime(command))
except OSError:
key = (command, None)
if key in _codex_features_cache:
return _codex_features_cache[key]
names = set()
try:
out = subprocess.run([command, "features", "list"], capture_output=True, cwd=work,
env=_env(CODEX), timeout=60,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0)
if out.returncode == 0:
for line in out.stdout.decode("utf-8", errors="replace").splitlines():
parts = line.split()
if parts and parts[0].replace("_", "").replace(".", "").isalnum():
names.add(parts[0])
except (OSError, subprocess.SubprocessError):
pass
# Fail closed: without the shell tools turned off, a prompt could make
# Codex read files on this machine (.env, stored keys) into its reply.
if not names & _CODEX_SHELL_FEATURES:
raise CLIError("could not confirm that Codex's shell tools can be turned off "
"(`codex features list` failed or changed), so it was not run. Update Codex and try again.",
status="unsafe")
_codex_features_cache[key] = names
return names
def run_codex(command, req, result, *, encoded_images, system, timeout, on_delta, check_interrupt, log):
with _work_dir() as work:
last_message = os.path.join(work, "last_message.txt")
# --ignore-user-config: MCP servers and hooks from config.toml would
# otherwise be offered as tools (sign-in still works).
args = [command, "exec", "--json", "--skip-git-repo-check", "--ephemeral", "--ignore-user-config",
"--sandbox", "read-only", "-C", work, "-o", last_message]
known = _codex_features(command, work)
for feature in _CODEX_TOOL_FEATURES:
if feature in known:
args += ["--disable", feature]
if req.model:
args += ["-m", req.model]
if req.reasoning in _EFFORT:
args += ["-c", f'model_reasoning_effort="{_EFFORT[req.reasoning]}"']
for i, (mime, data) in enumerate(encoded_images):
path = os.path.join(work, f"image_{i}.jpg")
with open(path, "wb") as f:
f.write(base64.b64decode(data))
# --image=<file> takes exactly one value; "-i a b" would also
# swallow the "-" prompt argument as an image.
args.append(f"--image={path}")
args.append("-")
# Codex has no separate system prompt in exec mode.
prompt = f"{system}\n\n---\n\n{req.user}" if system and req.user else (system or req.user)
if not prompt and encoded_images:
prompt = "Describe the image(s)." # codex exec refuses an empty prompt
if log:
log(f"Codex CLI: {' '.join(args[1:])}")
text, thinking = [], []
failure = []
last_error = []
def on_line(line):
try:
event = json.loads(line)
except ValueError:
return
kind = event.get("type")
item = event.get("item") or {}
if kind == "item.completed":
if item.get("type") == "agent_message" and item.get("text"):
text.append(item["text"])
elif item.get("type") == "reasoning" and item.get("text"):
thinking.append(item["text"])
if on_delta:
on_delta("\n\n".join(text), "\n\n".join(thinking))
elif kind == "turn.completed":
usage = event.get("usage") or {}
result.input_tokens = usage.get("input_tokens")
result.output_tokens = usage.get("output_tokens")
elif kind == "turn.failed":
failure.append((event.get("error") or {}).get("message") or "turn failed")
elif kind == "error":
# Usually a transient "Reconnecting…" notice; kept in case
# the run fails.
last_error.append(str(event.get("message", "")))
code, stderr = _run(args, CODEX, work, prompt.encode("utf-8"), on_line, timeout, check_interrupt)
final_text = ""
if os.path.exists(last_message):
with open(last_message, "r", encoding="utf-8", errors="replace") as f:
final_text = f.read().strip()
if failure or code:
detail = (failure or last_error or [_last_line(stderr) or f"exit code {code}"])[-1]
raise CLIError(f"Codex CLI: {detail[:500]}. If it asks you to log in, run `codex login` in a terminal.")
result.text = final_text or "\n\n".join(text)
result.thinking = "\n\n".join(thinking)
result.finish_reason = "stop"
result.model = req.model or "codex default"
return result
# --------------------------------------------------------------------------
# Entry points used by llm_providers
# --------------------------------------------------------------------------
def run_cli_chat(provider, command, req, result, **kwargs):
if req.model and not _SAFE_MODEL.match(req.model):
raise CLIError("invalid model name: use letters, digits and . _ - : / @ only", status="invalid model")
runner = run_claude if provider == CLAUDE else run_codex
try:
return runner(command, req, result, **kwargs)
except FileNotFoundError:
raise CLIError("the CLI was not found; install it or set its path in .env", status="not installed") from None
except OSError as e:
raise CLIError(f"could not start the CLI ({type(e).__name__})", status="cli error") from None
# Every model here is current-generation and accepts image input, so all are
# marked vision: True. static: True marks a fixed list, not fetched live, for
# the picker's "*" indicator. Only the rolling alias is listed for each model
# (not also its current pinned full name, e.g. claude-sonnet-5): both resolve
# to the same model today, and the alias is the one that keeps working as
# Anthropic ships new versions.
CLAUDE_MODELS = [
{"id": "sonnet", "detail": "", "vision": True, "static": True},
{"id": "opus", "detail": "", "vision": True, "static": True},
{"id": "haiku", "detail": "", "vision": True, "static": True},
{"id": "fable", "detail": "", "vision": True, "static": True},
]
# Codex has no equivalent of `claude` picking up new releases under a fixed
# alias. Its app-server does expose a live model/list RPC (see
# _codex_model_list_rpc below), so this is only the fallback when that can't
# be reached (CLI not installed, too old to speak the protocol, timed out…).
# Matched against codex-cli 0.157.0's own live catalog, not a guess.
CODEX_MODELS = [
{"id": "gpt-6-astra", "detail": "", "vision": True, "static": True},
{"id": "gpt-6-sol", "detail": "", "vision": True, "static": True},
{"id": "gpt-6-luna", "detail": "", "vision": True, "static": True},
{"id": "gpt-5.6-sol", "detail": "", "vision": True, "static": True},
{"id": "gpt-5.6-terra", "detail": "", "vision": True, "static": True},
{"id": "gpt-5.6-luna", "detail": "", "vision": True, "static": True},
{"id": "gpt-5.5", "detail": "", "vision": True, "static": True},
]
def _codex_model_list_rpc(command, timeout=6):
"""The live model catalog from Codex's own `app-server` JSON-RPC daemon
(method "model/list"), so the picker matches what this install actually
offers. None on any failure (not installed, too old to speak this
protocol, no reply in time…).
"""
try:
with _work_dir() as cwd:
proc = _popen([command, "app-server"], CODEX, cwd)
lines = queue.Queue()
threading.Thread(target=_read_lines, args=(proc.stdout, lines), daemon=True).start()
try:
proc.stdin.write((json.dumps({
"jsonrpc": "2.0", "id": 1, "method": "initialize",
"params": {"clientInfo": {"name": "ComfyUI-mnemic-nodes", "version": "1.0"}},
}) + "\n").encode())
proc.stdin.write((json.dumps({
"jsonrpc": "2.0", "id": 2, "method": "model/list", "params": {},
}) + "\n").encode())
proc.stdin.flush()
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
raw = lines.get(timeout=0.25)
except queue.Empty:
continue
if raw is None:
return None
try:
msg = json.loads(raw.decode("utf-8", errors="replace"))
except ValueError:
continue
if msg.get("id") == 2:
return (msg.get("result") or {}).get("data")
return None
finally:
_terminate(proc)
except OSError:
return None
def _codex_models_from_catalog(data):
models = []
for m in data or []:
if m.get("hidden"):
continue
model_id = m.get("id") or m.get("model")
if not model_id:
continue
vision = "image" in (m.get("inputModalities") or ["text", "image"])
models.append({"id": model_id, "detail": "", "vision": vision})
return models
def list_cli_models(provider, command=""):
"""Claude Code has no documented way to enumerate installed models, so
CLAUDE_MODELS is a fixed list of current aliases/names. Codex exposes a
live catalog through its own app-server; that is queried directly, and
CODEX_MODELS is only the fallback when it can't be reached.
"""
if provider == CLAUDE:
return list(CLAUDE_MODELS)
if command:
models = _codex_models_from_catalog(_codex_model_list_rpc(command))
if models:
return models
return list(CODEX_MODELS)