mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-10 04:08:28 +08:00
Compare commits
1 Commits
fix/plugin
...
feat/codex
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
965d2fec98 |
334
agent/codex_cli_client.py
Normal file
334
agent/codex_cli_client.py
Normal file
@@ -0,0 +1,334 @@
|
||||
"""OpenAI-compatible shim that forwards Hermes requests to ``codex exec --json``.
|
||||
|
||||
This adapter lets Hermes treat the OpenAI Codex CLI as a chat-style backend.
|
||||
Each request spawns ``codex exec --json --ephemeral --dangerously-bypass-approvals-and-sandbox``,
|
||||
parses the JSONL event stream, extracts the agent message text and token usage,
|
||||
and converts the result into the minimal shape Hermes expects from an OpenAI client.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CODEX_CLI_BASE_URL = "codex-cli://local"
|
||||
_DEFAULT_TIMEOUT_SECONDS = 900.0
|
||||
|
||||
|
||||
def _resolve_command() -> str:
|
||||
return (
|
||||
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
|
||||
or os.getenv("CODEX_CLI_PATH", "").strip()
|
||||
or "codex"
|
||||
)
|
||||
|
||||
|
||||
def _resolve_args() -> list[str]:
|
||||
raw = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
|
||||
if not raw:
|
||||
return [
|
||||
"exec",
|
||||
"--json",
|
||||
"--ephemeral",
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--skip-git-repo-check",
|
||||
]
|
||||
import shlex
|
||||
return shlex.split(raw)
|
||||
|
||||
|
||||
def _build_subprocess_env() -> dict[str, str]:
|
||||
env = os.environ.copy()
|
||||
# Preserve HOME so codex can find ~/.codex/auth.json
|
||||
home = os.environ.get("HOME", "")
|
||||
if not home:
|
||||
home = os.path.expanduser("~")
|
||||
if home and home != "~":
|
||||
env["HOME"] = home
|
||||
return env
|
||||
|
||||
|
||||
def _parse_turn_completed_usage(event: dict[str, Any]) -> SimpleNamespace:
|
||||
usage = event.get("usage") or {}
|
||||
input_tokens = int(usage.get("input_tokens") or 0)
|
||||
cached_tokens = int(usage.get("cached_input_tokens") or 0)
|
||||
output_tokens = int(usage.get("output_tokens") or 0)
|
||||
reasoning_tokens = int(usage.get("reasoning_output_tokens") or 0)
|
||||
return SimpleNamespace(
|
||||
prompt_tokens=input_tokens,
|
||||
completion_tokens=output_tokens + reasoning_tokens,
|
||||
total_tokens=input_tokens + output_tokens + reasoning_tokens,
|
||||
prompt_tokens_details=SimpleNamespace(cached_tokens=cached_tokens),
|
||||
)
|
||||
|
||||
|
||||
class _CodexCLIChatCompletions:
|
||||
def __init__(self, client: "CodexCLIClient"):
|
||||
self._client = client
|
||||
|
||||
def create(self, **kwargs: Any) -> Any:
|
||||
return self._client._create_chat_completion(**kwargs)
|
||||
|
||||
|
||||
class _CodexCLIChatNamespace:
|
||||
def __init__(self, client: "CodexCLIClient"):
|
||||
self.completions = _CodexCLIChatCompletions(client)
|
||||
|
||||
|
||||
class CodexCLIClient:
|
||||
"""Minimal OpenAI-client-compatible facade for Codex CLI."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
api_key: str | None = None,
|
||||
base_url: str | None = None,
|
||||
default_headers: dict[str, str] | None = None,
|
||||
command: str | None = None,
|
||||
args: list[str] | None = None,
|
||||
**_: Any,
|
||||
):
|
||||
self.api_key = api_key or "codex-cli"
|
||||
self.base_url = base_url or _CODEX_CLI_BASE_URL
|
||||
self._default_headers = dict(default_headers or {})
|
||||
self._command = command or _resolve_command()
|
||||
self._args = list(args or _resolve_args())
|
||||
self.chat = _CodexCLIChatNamespace(self)
|
||||
self.is_closed = False
|
||||
self._active_process: subprocess.Popen[str] | None = None
|
||||
self._active_process_lock = threading.Lock()
|
||||
|
||||
def close(self) -> None:
|
||||
proc: subprocess.Popen[str] | None
|
||||
with self._active_process_lock:
|
||||
proc = self._active_process
|
||||
self._active_process = None
|
||||
self.is_closed = True
|
||||
if proc is None:
|
||||
return
|
||||
try:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=2)
|
||||
except Exception:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _build_prompt(self, messages: list[dict[str, Any]], model: str | None = None) -> str:
|
||||
sections: list[str] = [
|
||||
"You are being used as the active Codex CLI agent backend for Hermes.",
|
||||
"Respond to the user's request directly. Do NOT call tools — Hermes handles tools.",
|
||||
]
|
||||
if model:
|
||||
sections.append(f"Hermes requested model hint: {model}")
|
||||
|
||||
transcript: list[str] = []
|
||||
for message in messages:
|
||||
if not isinstance(message, dict):
|
||||
continue
|
||||
role = str(message.get("role") or "unknown").strip().lower()
|
||||
content = message.get("content")
|
||||
if content is None:
|
||||
continue
|
||||
if isinstance(content, list):
|
||||
parts = []
|
||||
for item in content:
|
||||
if isinstance(item, str):
|
||||
parts.append(item)
|
||||
elif isinstance(item, dict) and "text" in item:
|
||||
parts.append(str(item["text"]))
|
||||
content = "\n".join(parts).strip()
|
||||
if not content:
|
||||
continue
|
||||
label = {
|
||||
"system": "System",
|
||||
"user": "User",
|
||||
"assistant": "Assistant",
|
||||
"tool": "Tool",
|
||||
}.get(role, role.title())
|
||||
transcript.append(f"{label}:\n{content}")
|
||||
|
||||
if transcript:
|
||||
sections.append("Conversation transcript:\n\n" + "\n\n".join(transcript))
|
||||
|
||||
sections.append("Continue the conversation from the latest user request.")
|
||||
return "\n\n".join(s.strip() for s in sections if s and s.strip())
|
||||
|
||||
def _create_chat_completion(
|
||||
self,
|
||||
*,
|
||||
model: str | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
timeout: float | None = None,
|
||||
tools: list[dict[str, Any]] | None = None,
|
||||
tool_choice: Any = None,
|
||||
**_: Any,
|
||||
) -> Any:
|
||||
prompt_text = self._build_prompt(messages or [], model=model)
|
||||
|
||||
# Normalise timeout: run_agent.py may pass an httpx.Timeout object
|
||||
if timeout is None:
|
||||
effective_timeout = _DEFAULT_TIMEOUT_SECONDS
|
||||
elif isinstance(timeout, (int, float)):
|
||||
effective_timeout = float(timeout)
|
||||
else:
|
||||
candidates = [
|
||||
getattr(timeout, attr, None)
|
||||
for attr in ("read", "write", "connect", "pool", "timeout")
|
||||
]
|
||||
numeric = [float(v) for v in candidates if isinstance(v, (int, float))]
|
||||
effective_timeout = max(numeric) if numeric else _DEFAULT_TIMEOUT_SECONDS
|
||||
|
||||
response_text, usage = self._run_prompt(prompt_text, timeout_seconds=effective_timeout)
|
||||
|
||||
assistant_message = SimpleNamespace(
|
||||
content=response_text,
|
||||
tool_calls=[],
|
||||
reasoning=None,
|
||||
reasoning_content=None,
|
||||
reasoning_details=None,
|
||||
)
|
||||
choice = SimpleNamespace(message=assistant_message, finish_reason="stop")
|
||||
return SimpleNamespace(
|
||||
choices=[choice],
|
||||
usage=usage,
|
||||
model=model or "codex-cli",
|
||||
)
|
||||
|
||||
def _run_prompt(self, prompt_text: str, *, timeout_seconds: float) -> tuple[str, SimpleNamespace]:
|
||||
cmd = [self._command] + self._args
|
||||
# The prompt is a positional arg — pass it via stdin with pipe
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
cmd,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
env=_build_subprocess_env(),
|
||||
)
|
||||
except FileNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
f"Could not start Codex CLI command '{self._command}'. "
|
||||
"Install Codex CLI (npm install -g @openai/codex) or set "
|
||||
f"HERMES_CODEX_CLI_COMMAND / CODEX_CLI_PATH."
|
||||
) from exc
|
||||
|
||||
if proc.stdin is None or proc.stdout is None:
|
||||
proc.kill()
|
||||
raise RuntimeError("Codex CLI process did not expose stdin/stdout pipes.")
|
||||
|
||||
self.is_closed = False
|
||||
with self._active_process_lock:
|
||||
self._active_process = proc
|
||||
|
||||
response_parts: list[str] = []
|
||||
usage = SimpleNamespace(
|
||||
prompt_tokens=0,
|
||||
completion_tokens=0,
|
||||
total_tokens=0,
|
||||
prompt_tokens_details=SimpleNamespace(cached_tokens=0),
|
||||
)
|
||||
stderr_lines: list[str] = []
|
||||
|
||||
try:
|
||||
# Write prompt to stdin and close it to signal end of input
|
||||
proc.stdin.write(prompt_text)
|
||||
proc.stdin.close()
|
||||
|
||||
deadline = time.monotonic() + timeout_seconds
|
||||
stdout_thread = threading.Thread(target=lambda: None, daemon=True)
|
||||
|
||||
# Collect stdout lines
|
||||
stdout_lines: list[str] = []
|
||||
|
||||
def _read_stdout():
|
||||
if proc.stdout is None:
|
||||
return
|
||||
for line in proc.stdout:
|
||||
stdout_lines.append(line.rstrip("\n"))
|
||||
|
||||
stdout_thread = threading.Thread(target=_read_stdout, daemon=True)
|
||||
stdout_thread.start()
|
||||
|
||||
# We'll also collect stderr
|
||||
stderr_output: list[str] = []
|
||||
|
||||
def _read_stderr():
|
||||
if proc.stderr is None:
|
||||
return
|
||||
for line in proc.stderr:
|
||||
stderr_output.append(line.rstrip("\n"))
|
||||
|
||||
stderr_thread = threading.Thread(target=_read_stderr, daemon=True)
|
||||
stderr_thread.start()
|
||||
|
||||
# Wait for process to complete or timeout
|
||||
remaining = deadline - time.monotonic()
|
||||
while remaining > 0:
|
||||
if proc.poll() is not None:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
remaining = deadline - time.monotonic()
|
||||
|
||||
if proc.poll() is None:
|
||||
proc.kill()
|
||||
raise TimeoutError("Timed out waiting for Codex CLI response.")
|
||||
|
||||
# Wait for threads to finish reading
|
||||
stdout_thread.join(timeout=5)
|
||||
stderr_thread.join(timeout=5)
|
||||
|
||||
# Parse JSONL output
|
||||
agent_text = ""
|
||||
for line in stdout_lines:
|
||||
try:
|
||||
event = json.loads(line)
|
||||
except Exception:
|
||||
# Non-JSON line (banner, status) — skip
|
||||
continue
|
||||
event_type = event.get("type", "")
|
||||
if event_type == "item.completed":
|
||||
item = event.get("item") or {}
|
||||
if item.get("type") == "agent_message":
|
||||
text = item.get("text") or ""
|
||||
if text:
|
||||
agent_text += text
|
||||
elif event_type == "turn.completed":
|
||||
usage = _parse_turn_completed_usage(event)
|
||||
|
||||
if agent_text:
|
||||
response_parts.append(agent_text)
|
||||
|
||||
# Stderr with useful diagnostics
|
||||
for line in stderr_output:
|
||||
if line.strip():
|
||||
stderr_lines.append(line)
|
||||
if stderr_lines and not agent_text:
|
||||
raise RuntimeError(
|
||||
"Codex CLI produced no agent message. "
|
||||
f"stderr: {'; '.join(stderr_lines[-5:])}"
|
||||
)
|
||||
|
||||
return "\n".join(response_parts).strip(), usage
|
||||
|
||||
finally:
|
||||
if proc.poll() is None:
|
||||
try:
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
with self._active_process_lock:
|
||||
if self._active_process is proc:
|
||||
self._active_process = None
|
||||
@@ -197,6 +197,13 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
|
||||
inference_base_url=DEFAULT_COPILOT_ACP_BASE_URL,
|
||||
base_url_env_var="COPILOT_ACP_BASE_URL",
|
||||
),
|
||||
"codex-cli": ProviderConfig(
|
||||
id="codex-cli",
|
||||
name="OpenAI Codex CLI",
|
||||
auth_type="external_process",
|
||||
inference_base_url="codex-cli://local",
|
||||
base_url_env_var="CODEX_CLI_BASE_URL",
|
||||
),
|
||||
"gemini": ProviderConfig(
|
||||
id="gemini",
|
||||
name="Google AI Studio",
|
||||
@@ -1377,6 +1384,7 @@ def resolve_provider(
|
||||
"github": "copilot", "github-copilot": "copilot",
|
||||
"github-models": "copilot", "github-model": "copilot",
|
||||
"github-copilot-acp": "copilot-acp", "copilot-acp-agent": "copilot-acp",
|
||||
"codexcli": "codex-cli", "openai-codex-cli": "codex-cli",
|
||||
"aigateway": "ai-gateway", "vercel": "ai-gateway", "vercel-ai-gateway": "ai-gateway",
|
||||
"opencode": "opencode-zen", "zen": "opencode-zen",
|
||||
"qwen-portal": "qwen-oauth", "qwen-cli": "qwen-oauth", "qwen-oauth": "qwen-oauth", "google-gemini-cli": "google-gemini-cli", "gemini-cli": "google-gemini-cli", "gemini-oauth": "google-gemini-cli",
|
||||
@@ -4009,28 +4017,60 @@ def get_external_process_provider_status(provider_id: str) -> Dict[str, Any]:
|
||||
if not pconfig or pconfig.auth_type != "external_process":
|
||||
return {"configured": False}
|
||||
|
||||
command = (
|
||||
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
|
||||
or os.getenv("COPILOT_CLI_PATH", "").strip()
|
||||
or "copilot"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
|
||||
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
|
||||
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
|
||||
if not base_url:
|
||||
base_url = pconfig.inference_base_url
|
||||
if provider_id == "copilot-acp":
|
||||
command = (
|
||||
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
|
||||
or os.getenv("COPILOT_CLI_PATH", "").strip()
|
||||
or "copilot"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
|
||||
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
|
||||
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
|
||||
if not base_url:
|
||||
base_url = pconfig.inference_base_url
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
return {
|
||||
"configured": bool(resolved_command or base_url.startswith("acp+tcp://")),
|
||||
"provider": provider_id,
|
||||
"name": pconfig.name,
|
||||
"command": command,
|
||||
"args": args,
|
||||
"resolved_command": resolved_command,
|
||||
"base_url": base_url,
|
||||
"logged_in": bool(resolved_command or base_url.startswith("acp+tcp://")),
|
||||
}
|
||||
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
return {
|
||||
"configured": bool(resolved_command or base_url.startswith("acp+tcp://")),
|
||||
"provider": provider_id,
|
||||
"name": pconfig.name,
|
||||
"command": command,
|
||||
"args": args,
|
||||
"resolved_command": resolved_command,
|
||||
"base_url": base_url,
|
||||
"logged_in": bool(resolved_command or base_url.startswith("acp+tcp://")),
|
||||
}
|
||||
if provider_id == "codex-cli":
|
||||
command = (
|
||||
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
|
||||
or os.getenv("CODEX_CLI_PATH", "").strip()
|
||||
or "codex"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
|
||||
default_args = [
|
||||
"exec",
|
||||
"--json",
|
||||
"--ephemeral",
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--skip-git-repo-check",
|
||||
]
|
||||
args = shlex.split(raw_args) if raw_args else default_args
|
||||
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
|
||||
if not base_url:
|
||||
base_url = pconfig.inference_base_url
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
return {
|
||||
"configured": bool(resolved_command),
|
||||
"provider": provider_id,
|
||||
"name": pconfig.name,
|
||||
"command": command,
|
||||
"args": args,
|
||||
"resolved_command": resolved_command,
|
||||
"base_url": base_url,
|
||||
"logged_in": bool(resolved_command),
|
||||
}
|
||||
|
||||
return {"configured": False}
|
||||
|
||||
|
||||
def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
|
||||
@@ -4048,6 +4088,8 @@ def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
|
||||
return get_gemini_oauth_auth_status()
|
||||
if target == "copilot-acp":
|
||||
return get_external_process_provider_status(target)
|
||||
if target == "codex-cli":
|
||||
return get_external_process_provider_status(target)
|
||||
# API-key providers
|
||||
pconfig = PROVIDER_REGISTRY.get(target)
|
||||
if pconfig and pconfig.auth_type == "api_key":
|
||||
@@ -4121,30 +4163,69 @@ def resolve_external_process_provider_credentials(provider_id: str) -> Dict[str,
|
||||
if not base_url:
|
||||
base_url = pconfig.inference_base_url
|
||||
|
||||
command = (
|
||||
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
|
||||
or os.getenv("COPILOT_CLI_PATH", "").strip()
|
||||
or "copilot"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
|
||||
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
if not resolved_command and not base_url.startswith("acp+tcp://"):
|
||||
raise AuthError(
|
||||
f"Could not find the Copilot CLI command '{command}'. "
|
||||
"Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.",
|
||||
provider=provider_id,
|
||||
code="missing_copilot_cli",
|
||||
if provider_id == "copilot-acp":
|
||||
command = (
|
||||
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
|
||||
or os.getenv("COPILOT_CLI_PATH", "").strip()
|
||||
or "copilot"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
|
||||
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
if not resolved_command and not base_url.startswith("acp+tcp://"):
|
||||
raise AuthError(
|
||||
f"Could not find the Copilot CLI command '{command}'. "
|
||||
"Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.",
|
||||
provider=provider_id,
|
||||
code="missing_copilot_cli",
|
||||
)
|
||||
return {
|
||||
"provider": provider_id,
|
||||
"api_key": "copilot-acp",
|
||||
"base_url": base_url.rstrip("/"),
|
||||
"command": resolved_command or command,
|
||||
"args": args,
|
||||
"source": "process",
|
||||
}
|
||||
|
||||
return {
|
||||
"provider": provider_id,
|
||||
"api_key": "copilot-acp",
|
||||
"base_url": base_url.rstrip("/"),
|
||||
"command": resolved_command or command,
|
||||
"args": args,
|
||||
"source": "process",
|
||||
}
|
||||
if provider_id == "codex-cli":
|
||||
command = (
|
||||
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
|
||||
or os.getenv("CODEX_CLI_PATH", "").strip()
|
||||
or "codex"
|
||||
)
|
||||
raw_args = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
|
||||
default_args = [
|
||||
"exec",
|
||||
"--json",
|
||||
"--ephemeral",
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--skip-git-repo-check",
|
||||
]
|
||||
args = shlex.split(raw_args) if raw_args else default_args
|
||||
resolved_command = shutil.which(command) if command else None
|
||||
if not resolved_command:
|
||||
raise AuthError(
|
||||
f"Could not find the Codex CLI command '{command}'. "
|
||||
"Install Codex CLI (npm install -g @openai/codex) or set "
|
||||
"HERMES_CODEX_CLI_COMMAND / CODEX_CLI_PATH.",
|
||||
provider=provider_id,
|
||||
code="missing_codex_cli",
|
||||
)
|
||||
return {
|
||||
"provider": provider_id,
|
||||
"api_key": "codex-cli",
|
||||
"base_url": base_url.rstrip("/"),
|
||||
"command": resolved_command or command,
|
||||
"args": args,
|
||||
"source": "process",
|
||||
}
|
||||
|
||||
raise AuthError(
|
||||
f"Unknown external-process provider '{provider_id}'.",
|
||||
provider=provider_id,
|
||||
code="unknown_external_process_provider",
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
||||
@@ -8858,7 +8858,7 @@ def _build_provider_choices() -> list[str]:
|
||||
except Exception:
|
||||
# Fallback: static list guarantees the CLI always works
|
||||
return [
|
||||
"auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot",
|
||||
"auto", "openrouter", "nous", "openai-codex", "copilot-acp", "codex-cli", "copilot",
|
||||
"anthropic", "gemini", "google-gemini-cli", "xai", "bedrock", "azure-foundry",
|
||||
"ollama-cloud", "huggingface", "zai", "kimi-coding", "kimi-coding-cn",
|
||||
"stepfun", "minimax", "minimax-cn", "kilocode", "xiaomi", "arcee",
|
||||
|
||||
@@ -207,6 +207,17 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
|
||||
"copilot-acp": [
|
||||
"copilot-acp",
|
||||
],
|
||||
"codex-cli": [
|
||||
"gpt-5.5",
|
||||
"gpt-5.4",
|
||||
"gpt-5.4-mini",
|
||||
"gpt-5.3-codex",
|
||||
"gpt-5.2-codex",
|
||||
"gpt-5.1-codex-max",
|
||||
"gpt-5.1-codex-mini",
|
||||
"o3",
|
||||
"o4-mini",
|
||||
],
|
||||
"copilot": [
|
||||
"gpt-5.4",
|
||||
"gpt-5.4-mini",
|
||||
@@ -799,6 +810,7 @@ CANONICAL_PROVIDERS: list[ProviderEntry] = [
|
||||
ProviderEntry("qwen-oauth", "Qwen OAuth (Portal)", "Qwen OAuth (reuses local Qwen CLI login)"),
|
||||
ProviderEntry("copilot", "GitHub Copilot", "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"),
|
||||
ProviderEntry("copilot-acp", "GitHub Copilot ACP", "GitHub Copilot ACP (spawns `copilot --acp --stdio`)"),
|
||||
ProviderEntry("codex-cli", "OpenAI Codex CLI", "OpenAI Codex CLI (spawns `codex exec --json` — text-only MVP, Hermes tools disabled)"),
|
||||
ProviderEntry("huggingface", "Hugging Face", "Hugging Face Inference Providers (20+ open models)"),
|
||||
ProviderEntry("gemini", "Google AI Studio", "Google AI Studio (Gemini models — native Gemini API)"),
|
||||
ProviderEntry("google-gemini-cli", "Google Gemini (OAuth)", "Google Gemini via OAuth + Code Assist (free tier supported; no API key needed)"),
|
||||
@@ -858,6 +870,8 @@ _PROVIDER_ALIASES = {
|
||||
"github-model": "copilot",
|
||||
"github-copilot-acp": "copilot-acp",
|
||||
"copilot-acp-agent": "copilot-acp",
|
||||
"codexcli": "codex-cli",
|
||||
"openai-codex-cli": "codex-cli",
|
||||
"google": "gemini",
|
||||
"google-gemini": "gemini",
|
||||
"google-ai-studio": "gemini",
|
||||
|
||||
@@ -1137,6 +1137,19 @@ def resolve_runtime_provider(
|
||||
"requested_provider": requested_provider,
|
||||
}
|
||||
|
||||
if provider == "codex-cli":
|
||||
creds = resolve_external_process_provider_credentials(provider)
|
||||
return {
|
||||
"provider": "codex-cli",
|
||||
"api_mode": "chat_completions",
|
||||
"base_url": creds.get("base_url", "").rstrip("/"),
|
||||
"api_key": creds.get("api_key", ""),
|
||||
"command": creds.get("command", ""),
|
||||
"args": list(creds.get("args") or []),
|
||||
"source": creds.get("source", "process"),
|
||||
"requested_provider": requested_provider,
|
||||
}
|
||||
|
||||
# Anthropic (native Messages API)
|
||||
if provider == "anthropic":
|
||||
# Allow base URL override from config.yaml model.base_url, but only
|
||||
|
||||
22
run_agent.py
22
run_agent.py
@@ -1264,6 +1264,7 @@ class AIAgent:
|
||||
api_mode is None
|
||||
and self.api_mode == "chat_completions"
|
||||
and self.provider != "copilot-acp"
|
||||
and self.provider != "codex-cli"
|
||||
and not str(self.base_url or "").lower().startswith("acp://copilot")
|
||||
and not str(self.base_url or "").lower().startswith("acp+tcp://")
|
||||
and not self._is_azure_openai_url()
|
||||
@@ -1587,6 +1588,9 @@ class AIAgent:
|
||||
if self.provider == "copilot-acp":
|
||||
client_kwargs["command"] = self.acp_command
|
||||
client_kwargs["args"] = self.acp_args
|
||||
if self.provider == "codex-cli":
|
||||
client_kwargs["command"] = self.acp_command
|
||||
client_kwargs["args"] = self.acp_args
|
||||
effective_base = base_url
|
||||
if base_url_host_matches(effective_base, "openrouter.ai"):
|
||||
from agent.auxiliary_client import build_or_headers
|
||||
@@ -1761,6 +1765,11 @@ class AIAgent:
|
||||
disabled_toolsets=disabled_toolsets,
|
||||
quiet_mode=self.quiet_mode,
|
||||
)
|
||||
|
||||
# Codex CLI provider is text-in/text-out MVP — Hermes tools are disabled
|
||||
# because Codex handles its own tool calling internally via `codex exec`.
|
||||
if self.provider == "codex-cli":
|
||||
self.tools = []
|
||||
|
||||
# Show tool configuration and store valid tool names for validation
|
||||
self.valid_tool_names = set()
|
||||
@@ -5959,6 +5968,17 @@ class AIAgent:
|
||||
self._client_log_context(),
|
||||
)
|
||||
return client
|
||||
if self.provider == "codex-cli" or str(client_kwargs.get("base_url", "")).startswith("codex-cli://"):
|
||||
from agent.codex_cli_client import CodexCLIClient
|
||||
|
||||
client = CodexCLIClient(**client_kwargs)
|
||||
logger.info(
|
||||
"Codex CLI client created (%s, shared=%s) %s",
|
||||
reason,
|
||||
shared,
|
||||
self._client_log_context(),
|
||||
)
|
||||
return client
|
||||
if self.provider == "google-gemini-cli" or str(client_kwargs.get("base_url", "")).startswith("cloudcode-pa://"):
|
||||
from agent.gemini_cloudcode_adapter import GeminiCloudCodeClient
|
||||
|
||||
@@ -11809,8 +11829,10 @@ class AIAgent:
|
||||
# API upgrade (lines ~1083-1085).
|
||||
elif (
|
||||
self.provider == "copilot-acp"
|
||||
or self.provider == "codex-cli"
|
||||
or str(self.base_url or "").lower().startswith("acp://copilot")
|
||||
or str(self.base_url or "").lower().startswith("acp+tcp://")
|
||||
or str(self.base_url or "").lower().startswith("codex-cli://")
|
||||
):
|
||||
_use_streaming = False
|
||||
elif not self._has_stream_consumers():
|
||||
|
||||
107
tests/hermes_cli/test_codex_cli_provider.py
Normal file
107
tests/hermes_cli/test_codex_cli_provider.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""Tests for the codex-cli external-process provider."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
# CRITICAL: import directly from the module to avoid module-level side effects
|
||||
from hermes_cli.auth import (
|
||||
PROVIDER_REGISTRY,
|
||||
get_external_process_provider_status,
|
||||
get_auth_status,
|
||||
resolve_external_process_provider_credentials,
|
||||
)
|
||||
|
||||
|
||||
class TestCodexCLIProviderRegistry:
|
||||
"""Test that the codex-cli provider is correctly registered."""
|
||||
|
||||
def test_provider_registered(self):
|
||||
assert "codex-cli" in PROVIDER_REGISTRY
|
||||
pconfig = PROVIDER_REGISTRY["codex-cli"]
|
||||
assert pconfig.name == "OpenAI Codex CLI"
|
||||
assert pconfig.auth_type == "external_process"
|
||||
assert pconfig.inference_base_url == "codex-cli://local"
|
||||
assert pconfig.base_url_env_var == "CODEX_CLI_BASE_URL"
|
||||
|
||||
def test_aliases_resolve(self):
|
||||
from hermes_cli.auth import resolve_provider
|
||||
|
||||
assert resolve_provider("codexcli") == "codex-cli"
|
||||
assert resolve_provider("openai-codex-cli") == "codex-cli"
|
||||
|
||||
|
||||
class TestCodexCLIStatus:
|
||||
"""Test the external-process status helper for codex-cli."""
|
||||
|
||||
def test_status_not_configured_when_codex_missing(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
status = get_external_process_provider_status("codex-cli")
|
||||
assert status["configured"] is False
|
||||
assert status["provider"] == "codex-cli"
|
||||
|
||||
def test_status_configured_when_codex_exists(self):
|
||||
with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}):
|
||||
with patch("shutil.which", return_value="/opt/homebrew/bin/codex"):
|
||||
status = get_external_process_provider_status("codex-cli")
|
||||
assert status["configured"] is True
|
||||
assert status["provider"] == "codex-cli"
|
||||
assert status["resolved_command"] == "/opt/homebrew/bin/codex"
|
||||
assert status["command"] == "codex"
|
||||
|
||||
def test_auth_status_dispatches(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
status = get_auth_status("codex-cli")
|
||||
# Should not throw, returns a dict even when not configured
|
||||
assert isinstance(status, dict)
|
||||
assert "configured" in status or "logged_in" in status
|
||||
|
||||
def test_status_with_custom_command_env(self):
|
||||
with patch.dict(os.environ, {"HERMES_CODEX_CLI_COMMAND": "/usr/local/bin/my-codex"}, clear=False):
|
||||
status = get_external_process_provider_status("codex-cli")
|
||||
assert status["command"] == "/usr/local/bin/my-codex"
|
||||
assert status["command"] == "/usr/local/bin/my-codex"
|
||||
|
||||
def test_status_with_custom_args_env(self):
|
||||
with patch.dict(os.environ, {
|
||||
"HERMES_CODEX_CLI_ARGS": "exec --json --model gpt-5.5",
|
||||
}, clear=False):
|
||||
status = get_external_process_provider_status("codex-cli")
|
||||
assert "exec" in status["args"]
|
||||
assert "--json" in status["args"]
|
||||
assert "--model" in status["args"]
|
||||
|
||||
def test_status_unknown_provider(self):
|
||||
status = get_external_process_provider_status("nonexistent")
|
||||
assert status == {"configured": False}
|
||||
|
||||
|
||||
class TestCodexCLICredentials:
|
||||
"""Test the credential resolver for codex-cli."""
|
||||
|
||||
def test_resolves_command_path_when_available(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with patch("shutil.which", return_value="/opt/homebrew/bin/codex"):
|
||||
creds = resolve_external_process_provider_credentials("codex-cli")
|
||||
assert creds["provider"] == "codex-cli"
|
||||
assert creds["command"] == "/opt/homebrew/bin/codex"
|
||||
assert creds["api_key"] == "codex-cli"
|
||||
assert creds["base_url"] == "codex-cli://local"
|
||||
assert "--json" in creds["args"]
|
||||
assert "--ephemeral" in creds["args"]
|
||||
|
||||
def test_raises_when_command_missing(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with patch("shutil.which", return_value=None):
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
resolve_external_process_provider_credentials("codex-cli")
|
||||
assert "codex-cli" in str(exc_info.value).lower() or "codex" in str(exc_info.value).lower()
|
||||
|
||||
def test_custom_command_from_env(self):
|
||||
with patch.dict(os.environ, {"HERMES_CODEX_CLI_COMMAND": "/usr/local/bin/custom-codex"}, clear=False):
|
||||
with patch("shutil.which", return_value="/usr/local/bin/custom-codex"):
|
||||
creds = resolve_external_process_provider_credentials("codex-cli")
|
||||
assert creds["command"] == "/usr/local/bin/custom-codex"
|
||||
Reference in New Issue
Block a user