"""
Provider abstraction — one class per AI vendor. Each shells out to
the vendor's own coding-agent CLI so authentication piggy-backs on
the user's chat subscription (Claude.ai Pro, ChatGPT Plus/Pro/Team,
Google account) — no separate API billing.
* Anthropic Claude → the `claude` CLI ("Claude Code")
* OpenAI ChatGPT → the `codex` CLI
* Google Gemini → the `gemini` CLI
Each provider:
is_installed() — is the CLI on PATH?
is_logged_in() — best-effort check; falls back to "assume yes if
installed" (the actual auth error surfaces on
the first stream chunk).
stream_chat() — spawn the CLI subprocess, yield stdout chunks.
Conversation context is carried by concatenating the full message
history into each prompt (simplest approach that works uniformly
across all three CLIs). For subscription users token count is not a
concern.
"""
from __future__ import annotations
import os
import shutil
import subprocess
import sys
from abc import ABC, abstractmethod
from typing import Dict, Iterator, List, Optional
[docs]
class ChatProvider(ABC):
"""Abstract base for AI chat providers that shell out to a vendor CLI.
Subclasses set the ``name``/``label``/``cli_name``/``install_hint``/
``login_command`` class attributes and implement :meth:`stream_chat`.
:ivar name: short id ("claude" / "codex" / "gemini").
:ivar label: human-readable label shown in the UI.
:ivar cli_name: executable expected on ``PATH``.
:ivar install_hint: shell one-liner suggested for installation.
:ivar login_command: shell one-liner the user runs to authenticate.
"""
[docs]
name: str = "" # short id: "claude" / "codex" / "gemini"
[docs]
label: str = "" # human-readable label
[docs]
cli_name: str = "" # the executable on PATH
[docs]
install_hint: str = "" # shell one-liner to install
[docs]
login_command: str = "" # shell one-liner the user should run
def __init__(self):
# Tracks the currently-running child process so cancel_stream()
# can actually terminate it — otherwise `for line in proc.stdout`
# blocks indefinitely and the worker thread never exits.
self._current_proc: Optional[subprocess.Popen] = None
[docs]
def is_installed(self) -> bool:
"""Return True when the provider's CLI executable is on ``PATH``."""
return shutil.which(self.cli_name) is not None
[docs]
def is_logged_in(self) -> bool:
"""Best-effort — override per provider if a cheap check exists.
Default: assume yes when installed. The real auth error will
surface as a normal subprocess failure on the first send."""
return self.is_installed()
[docs]
def source_of_key(self) -> str:
"""Compat string for the old KeysDialog — now describes the
CLI's install/login state."""
if not self.is_installed():
return "CLI not installed"
return f"CLI found at {shutil.which(self.cli_name)}"
[docs]
def cancel_stream(self) -> None:
"""Kill the running subprocess (if any).
This is the ONLY reliable way to unblock a stream that's stuck
waiting on stdout — flipping a Python flag would only unblock
between chunks, which may never come."""
proc = self._current_proc
if proc is None:
return
try:
proc.terminate()
try:
proc.wait(timeout=1)
except subprocess.TimeoutExpired:
proc.kill()
try:
proc.wait(timeout=1)
except subprocess.TimeoutExpired:
pass
except Exception:
pass
@abstractmethod
[docs]
def stream_chat(self, messages: List[Dict], system: str = "",
model: Optional[str] = None) -> Iterator[str]:
"""Yield text chunks streaming from the CLI subprocess."""
# ---------------------------------------------------------------------------
# Shared subprocess helper
# ---------------------------------------------------------------------------
# Noise the vendor CLIs emit that we drop before showing to the user.
# Match on line prefix (case-sensitive).
_NOISE_LINE_PREFIXES = (
"Permission deny rule",
"Permission allow rule",
"Permission ask rule",
)
def _stream_process(argv: List[str], stdin_text: Optional[str] = None,
env_extra: Optional[Dict[str, str]] = None,
provider: Optional["ChatProvider"] = None,
) -> Iterator[str]:
"""Spawn a subprocess and yield stdout as it arrives.
Reads line-by-line so noise-filtering can drop specific warnings
(e.g. Claude Code's per-file permission-rule reminders from the
user's ~/.claude/settings.json). Merges stderr into stdout so
real errors show up inline.
If `provider` is passed we register the Popen on it so that
provider.cancel_stream() can terminate the subprocess and unblock
the caller's iteration. Without this, a stream that hangs on
a `for line in proc.stdout` read can never be cancelled and the
worker QThread will outlive its Python reference on quit — which
is exactly the crash the user reported.
"""
env = os.environ.copy()
if env_extra:
env.update(env_extra)
try:
proc = subprocess.Popen(
argv,
stdin=subprocess.PIPE if stdin_text is not None else None,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1, # line-buffered
env=env,
)
except FileNotFoundError as e:
raise RuntimeError(
f"Could not run {argv[0]!r} — is the CLI installed and on PATH?"
) from e
if provider is not None:
provider._current_proc = proc
try:
if stdin_text is not None and proc.stdin is not None:
try:
proc.stdin.write(stdin_text)
proc.stdin.close()
except BrokenPipeError:
pass
assert proc.stdout is not None
for line in proc.stdout:
if any(line.startswith(prefix) for prefix in _NOISE_LINE_PREFIXES):
continue
yield line
finally:
# Always tear the child down cleanly — cancel_stream() may have
# already terminated it; ok to call terminate again defensively.
try:
proc.stdout.close()
except Exception:
pass
try:
proc.wait(timeout=1)
except Exception:
try:
proc.terminate()
try:
proc.wait(timeout=1)
except Exception:
proc.kill()
except Exception:
pass
if provider is not None:
provider._current_proc = None
def _format_conversation(messages: List[Dict], system: str = "") -> str:
"""Flatten the {role, content} history into a single prompt.
Used by CLIs whose non-interactive mode takes one prompt string
per invocation. Prior turns get simple role prefixes so the model
knows who said what.
"""
parts: List[str] = []
if system:
parts.append(f"System:\n{system}\n")
for m in messages[:-1]:
role = m.get("role", "user")
prefix = "User" if role == "user" else "Assistant"
parts.append(f"{prefix}:\n{m.get('content','')}\n")
if messages:
last = messages[-1]
role = last.get("role", "user")
prefix = "User" if role == "user" else "Assistant"
parts.append(f"{prefix}:\n{last.get('content','')}")
return "\n".join(parts)
# ---------------------------------------------------------------------------
# Anthropic Claude — via `claude` (Claude Code)
# ---------------------------------------------------------------------------
[docs]
class ClaudeCliProvider(ChatProvider):
"""Anthropic Claude via the ``claude`` (Claude Code) CLI."""
[docs]
label = "Claude (via Claude Code)"
[docs]
install_hint = (
"curl -fsSL https://claude.ai/install.sh | bash # or "
"npm install -g @anthropic-ai/claude-code"
)
[docs]
login_command = "claude setup-token"
[docs]
def stream_chat(self, messages: List[Dict], system: str = "",
model: Optional[str] = None) -> Iterator[str]:
"""Stream a chat completion from the ``claude`` CLI.
:param messages: conversation history as ``{role, content}`` dicts.
:param system: optional system prompt appended via
``--append-system-prompt``.
:param model: optional model override passed via ``--model``.
:returns: iterator yielding stdout text chunks.
"""
prompt = _format_conversation(messages, system=system)
argv = ["claude", "-p", prompt]
if system:
argv += ["--append-system-prompt", system]
if model:
argv += ["--model", model]
yield from _stream_process(argv, provider=self)
# ---------------------------------------------------------------------------
# OpenAI ChatGPT — via `codex` (OpenAI Codex CLI)
# ---------------------------------------------------------------------------
[docs]
class CodexCliProvider(ChatProvider):
"""OpenAI ChatGPT via the ``codex`` CLI."""
[docs]
label = "ChatGPT (via Codex CLI)"
[docs]
install_hint = (
"npm install -g @openai/codex # or brew install codex"
)
[docs]
login_command = "codex login"
[docs]
def stream_chat(self, messages: List[Dict], system: str = "",
model: Optional[str] = None) -> Iterator[str]:
"""Stream a chat completion from the ``codex`` CLI.
:param messages: conversation history as ``{role, content}`` dicts.
:param system: optional system prompt folded into the prompt body.
:param model: optional model override passed via ``--model``.
:returns: iterator yielding stdout text chunks.
"""
prompt = _format_conversation(messages, system=system)
argv = ["codex", "exec", prompt]
if model:
argv += ["--model", model]
yield from _stream_process(argv, provider=self)
# ---------------------------------------------------------------------------
# Google Gemini — via `gemini` CLI
# ---------------------------------------------------------------------------
[docs]
class GeminiCliProvider(ChatProvider):
"""Google Gemini via the ``gemini`` CLI."""
[docs]
label = "Gemini (via Gemini CLI)"
[docs]
install_hint = (
"npm install -g @google/gemini-cli # or brew install gemini-cli"
)
[docs]
login_command = "gemini"
[docs]
def stream_chat(self, messages: List[Dict], system: str = "",
model: Optional[str] = None) -> Iterator[str]:
"""Stream a chat completion from the ``gemini`` CLI.
:param messages: conversation history as ``{role, content}`` dicts.
:param system: optional system prompt folded into the prompt body.
:param model: optional model override passed via ``-m``.
:returns: iterator yielding stdout text chunks.
"""
prompt = _format_conversation(messages, system=system)
argv = ["gemini", "-p", prompt]
if model:
argv += ["-m", model]
yield from _stream_process(argv, provider=self)
# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------
_PROVIDERS: List[ChatProvider] = [
ClaudeCliProvider(),
CodexCliProvider(),
GeminiCliProvider(),
]
[docs]
def list_providers() -> List[ChatProvider]:
"""Return every registered provider, regardless of install state."""
return list(_PROVIDERS)
[docs]
def get_provider(name: str) -> Optional[ChatProvider]:
"""Look up a registered provider by its short id.
:param name: provider id (``"claude"``, ``"codex"``, ``"gemini"``).
:returns: the matching provider, or ``None`` if no such id.
"""
for p in _PROVIDERS:
if p.name == name:
return p
return None