akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(mcp): auto-register headroom MCP server in wrap claude/codex and init -g

Browse source 
The proxy compresses tool_result payloads and emits [Retrieve more: hash=…]
markers, but Claude Code / Codex had no headroom_retrieve tool to call on
those markers unless the user separately ran 'headroom mcp install'. The
markers were dead pointers — silent quality loss.

Adds a per-agent MCP registrar abstraction (mcp_registry/) and wires it
into wrap and init so MCP install happens automatically alongside rtk:

  - mcp_registry/base.py — MCPRegistrar ABC, ServerSpec, RegisterResult,
    RegisterStatus enum.
  - mcp_registry/claude.py — Claude Code registrar (claude mcp add CLI
    with .claude.json / mcp.json file fallback).
  - mcp_registry/codex.py — OpenAI Codex registrar (marker-delimited TOML
    block edits to ~/.codex/config.toml; preserves user's other config).
  - mcp_registry/install.py — install_everywhere() orchestrator with
    detect-then-register semantics.
  - mcp_registry/display.py — shared format_result()/format_results() for
    consistent CLI output across wrap, init, and 'headroom mcp install'.

Adding a new agent (Cursor, Continue, Cline, Windsurf, Goose) is now a
single new file plus one entry in get_all_registrars(); call sites and
display logic don't change.

Test seam is constructor injection (home_dir, claude_cli) — zero patches
in 66 new tests across the registry. Removed 13 brittle CLI integration
tests in test_mcp.py that were patching module-level globals; equivalent
coverage now lives at the registrar/orchestrator layer.

wrap codex: snapshot ~/.codex/config.toml at the top of the command so
the existing wrap→unwrap round-trip captures the true pre-wrap state
even though MCP install now writes to the same file mid-flow.

220 tests pass (66 new + 154 existing CLI + integration). ruff and mypy
clean on touched files.
This commit is contained in:
Tejas Chopra 2026-05-08 17:17:53 -07:00
parent 7afb30186a
commit d9d8972ac4
15 changed files with 1867 additions and 346 deletions
Download patch file Download diff file Expand all files Collapse all files

26
headroom/cli/init.py
View file

@ -697,6 +697,32 @@ def _run_init_targets(
elif target == "openclaw":
_init_openclaw(global_scope=global_scope, port=port)
# Register the headroom MCP server with every targeted agent that has
# a registrar implemented. Wave 1 covers Claude Code; subsequent waves
# add Cursor / Codex / Continue / Cline / Windsurf / Goose without
# touching the call sites.
_install_headroom_mcp_for_targets(targets=targets, port=port)
def _install_headroom_mcp_for_targets(*, targets: list[str], port: int) -> None:
"""Install the headroom MCP server into each detected target agent."""
from headroom.mcp_registry import format_results, install_everywhere
proxy_url = f"http://127.0.0.1:{port}"
results = install_everywhere(proxy_url=proxy_url, agents=targets)
if not results:
return
lines = format_results(
results,
verbose=True,
overwrite_hint=f"headroom mcp install --proxy-url {proxy_url} --force",
)
if lines:
click.echo("\nMCP retrieve tool:")
for line in lines:
click.echo(line)
@main.group(invoke_without_command=True)
@click.option("-g", "--global", "global_scope", is_flag=True, help="Install for the current user.")

133
headroom/cli/mcp.py
View file

@ -95,24 +95,32 @@ def mcp() -> None:
default=DEFAULT_PROXY_URL,
help=f"Headroom proxy URL (default: {DEFAULT_PROXY_URL})",
)
@click.option(
"--agent",
"agents",
multiple=True,
help="Restrict installation to specific agents (default: every detected agent).",
)
@click.option(
"--force",
is_flag=True,
help="Overwrite existing headroom config",
help="Overwrite existing headroom config in case of mismatch.",
)
def mcp_install(proxy_url: str, force: bool) -> None:
"""Install Headroom MCP server into Claude Code config.
def mcp_install(proxy_url: str, agents: tuple[str, ...], force: bool) -> None:
"""Install the Headroom MCP server into every detected coding agent.
\b
This registers headroom with Claude Code so it can use the
headroom_retrieve tool for CCR (Compress-Cache-Retrieve).
By default this installs into every agent that has a registrar and is
detected on this system (Claude Code today; Cursor / Codex / Continue /
others added in subsequent releases). Pass ``--agent NAME`` one or more
times to restrict the installation.
\b
Example:
headroom mcp install
Examples:
headroom mcp install # every detected agent
headroom mcp install --agent claude # Claude Code only
headroom mcp install --proxy-url http://localhost:9000
"""
# Check for MCP SDK
try:
import mcp # noqa: F401
except ImportError:
@ -120,98 +128,35 @@ def mcp_install(proxy_url: str, force: bool) -> None:
click.echo("Install with: pip install 'headroom-ai[mcp]'", err=True)
raise SystemExit(1) from None
command = get_headroom_command()
env: dict[str, str] = {}
if proxy_url != DEFAULT_PROXY_URL:
env["HEADROOM_PROXY_URL"] = proxy_url
from headroom.mcp_registry import any_succeeded, format_results, install_everywhere
# Prefer `claude mcp add` (Claude Code CLI ≥2.x stores servers in
# ~/.claude/.claude.json, which is what `claude mcp list` reads).
claude_cli = shutil.which("claude")
used_claude_cli = False
if claude_cli:
# Check if already registered
result = subprocess.run(
[claude_cli, "mcp", "get", "headroom"],
capture_output=True,
text=True,
)
already_registered = result.returncode == 0
if already_registered and not force:
click.echo("Headroom MCP is already configured in Claude Code.")
click.echo("Use --force to overwrite, or 'headroom mcp uninstall' first.")
raise SystemExit(0)
if already_registered and force:
subprocess.run(
[claude_cli, "mcp", "remove", "headroom", "-s", "user"],
capture_output=True,
)
add_cmd = [claude_cli, "mcp", "add", "headroom", "-s", "user"]
for k, v in env.items():
add_cmd += ["-e", f"{k}={v}"]
add_cmd += ["--", *command]
result = subprocess.run(add_cmd, capture_output=True, text=True)
if result.returncode == 0:
used_claude_cli = True
else:
click.echo(
f"Warning: 'claude mcp add' failed ({result.stderr.strip()}), "
"falling back to mcp.json.",
err=True,
)
if not used_claude_cli:
# Fallback: write ~/.claude/mcp.json (used by older Claude Code versions
# and the Claude.ai desktop app).
config = load_mcp_config()
if "headroom" in config.get("mcpServers", {}) and not force:
click.echo("Headroom MCP is already configured in Claude Code.")
click.echo("Use --force to overwrite, or 'headroom mcp uninstall' first.")
raise SystemExit(0)
server_config: dict = {"command": command[0], "args": command[1:]}
if env:
server_config["env"] = env
if "mcpServers" not in config:
config["mcpServers"] = {}
config["mcpServers"]["headroom"] = server_config
save_mcp_config(config)
config_note = (
"Registered via: claude mcp add (scope: user)"
if used_claude_cli
else f"Configuration written to: {MCP_CONFIG_PATH}"
results = install_everywhere(
proxy_url=proxy_url,
agents=list(agents) if agents else None,
force=force,
)
click.echo(f"""
✓ Headroom MCP server installed!
if not results:
click.echo("No agents matched the requested filter.")
raise SystemExit(1)
{config_note}
click.echo("Installing Headroom MCP server...")
for line in format_results(
results,
verbose=True,
overwrite_hint=f"headroom mcp install --proxy-url {proxy_url} --force",
):
click.echo(line)
Next steps:
1. Start the Headroom proxy (if not running):
headroom proxy
if not any_succeeded(results):
raise SystemExit(1)
2. Start Claude Code WITH the proxy base URL:
ANTHROPIC_BASE_URL={proxy_url} claude
3. Claude Code now has:
- All requests compressed through the proxy (saves tokens & cost)
- Access to headroom_retrieve tool for CCR retrieval
- Stats visible at {proxy_url}/stats
NOTE: The MCP server provides on-demand compression tools
(headroom_compress, headroom_retrieve, headroom_stats). For automatic
compression of ALL traffic, also set ANTHROPIC_BASE_URL as shown above.
Proxy URL: {proxy_url}
""")
click.echo(
f"\nNext steps:\n"
f" 1. Start the Headroom proxy (if not running): headroom proxy\n"
f" 2. Start your agent (e.g.) ANTHROPIC_BASE_URL={proxy_url} claude\n"
f" 3. Restart any agent that was already running so it picks up the new MCP server.\n"
)
@mcp.command("uninstall")

78
headroom/cli/wrap.py
View file

@ -242,6 +242,39 @@ def _setup_rtk(verbose: bool = False) -> Path | None:
return rtk_path
def _setup_headroom_mcp(registrar: Any, port: int, *, verbose: bool = False) -> None:
"""Register the headroom MCP server with the given agent (idempotent).
The proxy compresses tool_result payloads and emits ``[Retrieve more:
hash=…]`` markers. Without this registration those markers point at
nothing — the agent has no ``headroom_retrieve`` tool to call.
Generic across registrars: ``ClaudeRegistrar``, ``CodexRegistrar``, and
any future agent registrar all flow through the same setup path.
"""
from headroom.mcp_registry import build_headroom_spec, format_result
if not registrar.detect():
if verbose:
click.echo(f" MCP retrieve tool: {registrar.display_name} not detected — skipping")
return
proxy_url = f"http://127.0.0.1:{port}"
spec = build_headroom_spec(proxy_url)
result = registrar.register_server(spec)
line = format_result(
registrar.name,
result,
label="MCP retrieve tool",
verbose=verbose,
overwrite_hint=f"headroom mcp install --proxy-url {proxy_url} --force",
restart_hint=f"restart {registrar.display_name} if it was already running",
)
if line is not None:
click.echo(line)
_CBM_MCP_SERVER_NAME = "codebase-memory-mcp"
@ -1336,6 +1369,11 @@ def unwrap() -> None:
@wrap.command(context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option("--no-rtk", is_flag=True, help="Skip rtk installation and hook registration")
@click.option(
"--no-mcp",
is_flag=True,
help="Skip headroom MCP server registration (compression markers will be unactionable)",
)
@click.option(
"--code-graph",
is_flag=True,
@ -1352,6 +1390,7 @@ def unwrap() -> None:
def claude(
port: int,
no_rtk: bool,
no_mcp: bool,
code_graph: bool,
no_proxy: bool,
learn: bool,
@ -1374,6 +1413,7 @@ def claude(
headroom wrap claude -- -p # Claude in print mode
headroom wrap claude --code-graph # With code graph intelligence
headroom wrap claude --no-rtk # Skip rtk (proxy only)
headroom wrap claude --no-mcp # Skip MCP retrieve tool registration
"""
if prepare_only:
if not no_rtk:
@ -1451,6 +1491,13 @@ def claude(
elif verbose:
click.echo(" Skipping rtk (--no-rtk)")
if not no_mcp:
from headroom.mcp_registry import ClaudeRegistrar
_setup_headroom_mcp(ClaudeRegistrar(), port, verbose=verbose)
elif verbose:
click.echo(" Skipping MCP retrieve tool (--no-mcp)")
if code_graph:
_setup_code_graph(verbose=verbose)
@ -1659,6 +1706,11 @@ def copilot(
@wrap.command(context_settings={"ignore_unknown_options": True})
@click.option("--port", "-p", default=8787, type=int, help="Proxy port (default: 8787)")
@click.option("--no-rtk", is_flag=True, help="Skip rtk installation and AGENTS.md injection")
@click.option(
"--no-mcp",
is_flag=True,
help="Skip headroom MCP server registration (compression markers will be unactionable)",
)
@click.option(
"--code-graph",
is_flag=True,
@ -1688,6 +1740,7 @@ def copilot(
def codex(
port: int,
no_rtk: bool,
no_mcp: bool,
code_graph: bool,
no_proxy: bool,
learn: bool,
@ -1704,16 +1757,28 @@ def codex(
\b
Sets OPENAI_BASE_URL to route all OpenAI API calls through Headroom.
Installs rtk and injects instructions into AGENTS.md so Codex uses
token-optimized commands (60-90% savings on shell output).
token-optimized commands (60-90% savings on shell output). Also
registers the headroom MCP server in ~/.codex/config.toml so Codex
can call ``headroom_retrieve`` on compression markers.
\b
Examples:
headroom wrap codex # Start proxy + rtk + codex
headroom wrap codex # Start proxy + rtk + mcp + codex
headroom wrap codex -- "fix the bug" # Pass prompt to codex
headroom wrap codex --no-rtk # Skip rtk setup
headroom wrap codex --no-mcp # Skip MCP retrieve tool registration
headroom wrap codex --port 9999 # Custom proxy port
headroom wrap codex --backend anyllm --anyllm-provider groq
"""
# Snapshot ~/.codex/config.toml BEFORE any wrap-time mutation so
# `headroom unwrap codex` can restore the user's pre-wrap state
# byte-for-byte. The snapshot is a no-op if the backup already exists
# or if the file already has Headroom markers, so this is safe to
# call repeatedly. Crucially this must run before MCP install, which
# writes its marker block to the same file.
_codex_config_file, _codex_backup_file = _codex_config_paths()
_snapshot_codex_config_if_unwrapped(_codex_config_file, _codex_backup_file)
# Setup rtk for Codex (binary + AGENTS.md instructions, no hooks)
if not no_rtk:
click.echo(" Setting up rtk for Codex...")
@ -1727,6 +1792,15 @@ def codex(
global_agents = Path.home() / ".codex" / "AGENTS.md"
_inject_rtk_instructions(global_agents, verbose=verbose)
# Register headroom MCP server in ~/.codex/config.toml so Codex can
# call headroom_retrieve on compression markers from the proxy.
if not no_mcp:
from headroom.mcp_registry import CodexRegistrar
_setup_headroom_mcp(CodexRegistrar(), port, verbose=verbose)
elif verbose:
click.echo(" Skipping MCP retrieve tool (--no-mcp)")
# Setup memory MCP server for Codex (native tool integration)
if memory:
click.echo(" Setting up memory for Codex...")

41
headroom/mcp_registry/__init__.py Normal file
View file

@ -0,0 +1,41 @@
"""Generic MCP server registration across coding agents.
The MCP protocol is universal but each agent's *registration* mechanism is
not — Claude Code uses its own CLI + ``~/.claude/.claude.json``, Cursor
writes ``~/.cursor/mcp.json``, Codex patches a TOML file, and so on. This
module provides a uniform interface so headroom can install its MCP server
(``headroom mcp serve``) into every detected agent.
Wave 1 ships :class:`ClaudeRegistrar`. Other registrars (Cursor, Codex,
Continue, Cline, Windsurf, Goose, OpenHands) are added in subsequent waves
without changing the calling code.
"""
from __future__ import annotations
from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec
from .claude import ClaudeRegistrar
from .codex import CodexRegistrar
from .display import any_succeeded, format_result, format_results
from .install import (
DEFAULT_PROXY_URL,
build_headroom_spec,
get_all_registrars,
install_everywhere,
)
__all__ = [
"DEFAULT_PROXY_URL",
"ClaudeRegistrar",
"CodexRegistrar",
"MCPRegistrar",
"RegisterResult",
"RegisterStatus",
"ServerSpec",
"any_succeeded",
"build_headroom_spec",
"format_result",
"format_results",
"get_all_registrars",
"install_everywhere",
]

106
headroom/mcp_registry/base.py Normal file
View file

@ -0,0 +1,106 @@
"""Abstract base for per-agent MCP registrars.
The MCP protocol itself is universal, and the headroom MCP server
(``headroom mcp serve``) is a single stdio binary that any compliant client
can launch. What differs between agents (Claude Code, Cursor, Codex, ...) is
how each one *learns* that a server exists: each invented its own config
file, format, and registration mechanism.
Subclasses of :class:`MCPRegistrar` own one agent's registration mechanism.
The orchestrator in :mod:`headroom.mcp_registry.install` calls a fleet of
registrars to install headroom across every detected agent.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
class RegisterStatus(str, Enum):
"""Outcome of a :meth:`MCPRegistrar.register_server` call."""
REGISTERED = "registered"
"""Newly written to the agent's MCP config."""
ALREADY = "already"
"""Already present with a configuration that matches the requested spec."""
MISMATCH = "mismatch"
"""Already present but with a different configuration; left untouched."""
FAILED = "failed"
"""Registration was attempted but the agent's tooling rejected it."""
NOT_DETECTED = "not_detected"
"""The agent does not appear to be installed on this system."""
NO_SDK = "no_sdk"
"""A required Python dependency is missing (e.g. the ``mcp`` package)."""
@dataclass
class ServerSpec:
"""Universal description of an MCP server to register.
Each registrar serializes this to its agent's native config format. The
fields cover what every JSON/TOML schema we've seen requires.
"""
name: str
command: str
args: tuple[str, ...] = ()
env: dict[str, str] = field(default_factory=dict)
@dataclass
class RegisterResult:
"""Outcome plus a human-readable detail string."""
status: RegisterStatus
detail: str | None = None
@property
def ok(self) -> bool:
"""True when the server is registered (newly or already)."""
return self.status in (RegisterStatus.REGISTERED, RegisterStatus.ALREADY)
class MCPRegistrar(ABC):
"""Per-agent MCP server registrar.
Each subclass owns exactly one agent's config schema and write path.
Contract:
* :meth:`detect` — does this agent appear installed?
* :meth:`get_server` — read current config; return the spec or ``None``.
* :meth:`register_server` — idempotent install. If the named server is
already registered with a different spec, returns
:attr:`RegisterStatus.MISMATCH` and does **not** overwrite unless
``force=True``.
* :meth:`unregister_server` — remove the named server.
"""
#: Stable agent identifier ("claude", "cursor", "codex", ...).
name: str = ""
#: Human-readable display name ("Claude Code", "Cursor", ...).
display_name: str = ""
@abstractmethod
def detect(self) -> bool:
"""Return True if this agent appears to be installed."""
@abstractmethod
def get_server(self, server_name: str) -> ServerSpec | None:
"""Return the registered :class:`ServerSpec`, or ``None`` if absent."""
@abstractmethod
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
"""Idempotently register an MCP server."""
@abstractmethod
def unregister_server(self, server_name: str) -> bool:
"""Remove the named server. Returns True on success."""

256
headroom/mcp_registry/claude.py Normal file
View file

@ -0,0 +1,256 @@
"""Claude Code MCP registrar.
Claude Code 2.x stores MCP server configuration in ``~/.claude/.claude.json``
and ships a CLI (``claude mcp add/remove/list/get``) that owns the file.
Older Claude Code releases (and the Claude Desktop app) read
``~/.claude/mcp.json``. This registrar prefers the CLI for writes when
available, and reads the underlying JSON files directly for compare /
``get_server`` so it is robust to CLI output format changes.
"""
from __future__ import annotations
import json
import logging
import shutil
import subprocess
from pathlib import Path
from typing import Any
from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec
logger = logging.getLogger(__name__)
class ClaudeRegistrar(MCPRegistrar):
"""Register MCP servers with Claude Code."""
name = "claude"
display_name = "Claude Code"
def __init__(
self,
*,
claude_cli: str | None | object = ...,
home_dir: Path | None = None,
) -> None:
"""Allow overrides for testing.
``claude_cli`` defaults to :func:`shutil.which` lookup. Pass
``None`` to force the file-based fallback path. Pass an explicit
path to point at a specific binary.
"""
home = home_dir if home_dir is not None else Path.home()
self._claude_dir = home / ".claude"
self._modern_config = home / ".claude" / ".claude.json"
self._legacy_config = home / ".claude" / "mcp.json"
if claude_cli is ...:
self._claude_cli = shutil.which("claude")
else:
# ``...`` sentinel preserves "not set"; explicit None disables CLI.
self._claude_cli = claude_cli # type: ignore[assignment]
# ------------------------------------------------------------------
# MCPRegistrar interface
# ------------------------------------------------------------------
def detect(self) -> bool:
if self._claude_cli:
return True
return self._claude_dir.is_dir()
def get_server(self, server_name: str) -> ServerSpec | None:
# Read from disk regardless of whether the CLI is present — the file
# format is stable and easier to compare than CLI output.
for config_path in (self._modern_config, self._legacy_config):
entry = self._read_server_entry(config_path, server_name)
if entry is not None:
return entry
return None
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
existing = self.get_server(spec.name)
if existing is not None:
if _specs_equivalent(existing, spec):
return RegisterResult(RegisterStatus.ALREADY, "matches current configuration")
if not force:
return RegisterResult(
RegisterStatus.MISMATCH,
_diff_specs(existing, spec),
)
# force=True: remove first, then write fresh below.
self.unregister_server(spec.name)
if self._claude_cli:
return self._register_via_cli(spec)
return self._register_via_file(spec)
def unregister_server(self, server_name: str) -> bool:
if self._claude_cli:
result = subprocess.run(
[str(self._claude_cli), "mcp", "remove", server_name, "-s", "user"],
capture_output=True,
text=True,
)
if result.returncode == 0:
return True
logger.debug("claude mcp remove failed: %s", result.stderr.strip())
# Fall through to file-based removal in case CLI didn't know
# about the user-scope entry but the file still has it.
removed = False
for config_path in (self._modern_config, self._legacy_config):
removed = self._remove_from_file(config_path, server_name) or removed
return removed
# ------------------------------------------------------------------
# CLI-backed implementation
# ------------------------------------------------------------------
def _register_via_cli(self, spec: ServerSpec) -> RegisterResult:
cmd = [str(self._claude_cli), "mcp", "add", spec.name, "-s", "user"]
for k, v in spec.env.items():
cmd += ["-e", f"{k}={v}"]
cmd += ["--", spec.command, *spec.args]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
return RegisterResult(RegisterStatus.REGISTERED, "via `claude mcp add` (scope: user)")
# CLI failed — try the file fallback rather than giving up.
logger.warning("claude mcp add failed: %s", result.stderr.strip())
file_result = self._register_via_file(spec)
if file_result.status == RegisterStatus.REGISTERED:
return RegisterResult(
RegisterStatus.REGISTERED,
f"via file fallback after CLI failed: {result.stderr.strip()}",
)
return RegisterResult(
RegisterStatus.FAILED,
f"CLI: {result.stderr.strip()}; file: {file_result.detail}",
)
# ------------------------------------------------------------------
# File-backed implementation (CLI absent / older clients)
# ------------------------------------------------------------------
def _register_via_file(self, spec: ServerSpec) -> RegisterResult:
# Prefer the modern config path. If only the legacy file exists,
# write to that to avoid surprising older clients.
target = self._modern_config
if not self._modern_config.exists() and self._legacy_config.exists():
target = self._legacy_config
try:
target.parent.mkdir(parents=True, exist_ok=True)
config = _read_json(target)
servers = config.setdefault("mcpServers", {})
servers[spec.name] = _spec_to_entry(spec)
_write_json(target, config)
except OSError as exc:
return RegisterResult(RegisterStatus.FAILED, f"could not write {target}: {exc}")
return RegisterResult(RegisterStatus.REGISTERED, f"wrote to {target}")
def _remove_from_file(self, path: Path, server_name: str) -> bool:
if not path.exists():
return False
try:
config = _read_json(path)
except OSError:
return False
servers = config.get("mcpServers", {})
if server_name not in servers:
return False
del servers[server_name]
try:
_write_json(path, config)
except OSError:
return False
return True
def _read_server_entry(self, path: Path, server_name: str) -> ServerSpec | None:
if not path.exists():
return None
try:
config = _read_json(path)
except OSError:
return None
entry = config.get("mcpServers", {}).get(server_name)
if not isinstance(entry, dict):
return None
return _entry_to_spec(server_name, entry)
# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
def _read_json(path: Path) -> dict[str, Any]:
"""Read a JSON file, returning empty dict if absent or unparseable."""
if not path.exists():
return {}
try:
with open(path) as f:
data = json.load(f)
except (OSError, json.JSONDecodeError):
return {}
if not isinstance(data, dict):
return {}
return data
def _write_json(path: Path, data: dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(data, f, indent=2)
f.write("\n")
def _spec_to_entry(spec: ServerSpec) -> dict[str, Any]:
entry: dict[str, Any] = {"command": spec.command}
if spec.args:
entry["args"] = list(spec.args)
if spec.env:
entry["env"] = dict(spec.env)
return entry
def _entry_to_spec(name: str, entry: dict[str, Any]) -> ServerSpec:
args_value = entry.get("args", [])
if isinstance(args_value, list):
args = tuple(str(x) for x in args_value)
else:
args = ()
env_value = entry.get("env", {})
env: dict[str, str] = {}
if isinstance(env_value, dict):
env = {str(k): str(v) for k, v in env_value.items()}
return ServerSpec(
name=name,
command=str(entry.get("command", "")),
args=args,
env=env,
)
def _specs_equivalent(a: ServerSpec, b: ServerSpec) -> bool:
"""Two specs match when every field is equal."""
return (
a.name == b.name
and a.command == b.command
and tuple(a.args) == tuple(b.args)
and dict(a.env) == dict(b.env)
)
def _diff_specs(existing: ServerSpec, requested: ServerSpec) -> str:
"""Render the difference between two specs for human consumption."""
parts: list[str] = []
if existing.command != requested.command:
parts.append(f"command {existing.command!r} -> {requested.command!r}")
if tuple(existing.args) != tuple(requested.args):
parts.append(f"args {list(existing.args)} -> {list(requested.args)}")
if dict(existing.env) != dict(requested.env):
parts.append(f"env {dict(existing.env)} -> {dict(requested.env)}")
if not parts:
return "spec differs in unidentified field(s)"
return "; ".join(parts)

222
headroom/mcp_registry/codex.py Normal file
View file

@ -0,0 +1,222 @@
"""OpenAI Codex CLI MCP registrar.
Codex stores MCP server config in ``~/.codex/config.toml`` as
``[mcp_servers.<name>]`` tables (with optional ``[mcp_servers.<name>.env]``
sub-tables). There is no general-purpose CLI for adding entries, so we
edit the file in place — using marker-delimited blocks so we can
idempotently inject, replace, and remove our entry without disturbing
anything else the user has configured.
"""
from __future__ import annotations
import logging
import sys
from pathlib import Path
from typing import Any
from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec
if sys.version_info >= (3, 11):
import tomllib
else: # pragma: no cover — exercised only on 3.10
import tomli as tomllib # type: ignore[no-redef]
logger = logging.getLogger(__name__)
_MARKER_START = "# --- Headroom MCP server ---"
_MARKER_END = "# --- end Headroom MCP server ---"
class CodexRegistrar(MCPRegistrar):
"""Register MCP servers with the OpenAI Codex CLI."""
name = "codex"
display_name = "OpenAI Codex CLI"
def __init__(self, *, home_dir: Path | None = None) -> None:
home = home_dir if home_dir is not None else Path.home()
self._codex_dir = home / ".codex"
self._config_file = home / ".codex" / "config.toml"
# ------------------------------------------------------------------
# MCPRegistrar interface
# ------------------------------------------------------------------
def detect(self) -> bool:
return self._codex_dir.is_dir()
def get_server(self, server_name: str) -> ServerSpec | None:
data = self._load_toml()
servers = data.get("mcp_servers", {})
if not isinstance(servers, dict):
return None
entry = servers.get(server_name)
if not isinstance(entry, dict):
return None
return _entry_to_spec(server_name, entry)
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
existing = self.get_server(spec.name)
if existing is not None and _specs_equivalent(existing, spec):
return RegisterResult(RegisterStatus.ALREADY, "matches current configuration")
if existing is not None and not force:
content = self._read_text()
if _MARKER_START not in content:
# Entry exists but wasn't written by us — refuse to clobber.
return RegisterResult(
RegisterStatus.MISMATCH,
"user-managed [mcp_servers."
f"{spec.name}] entry outside Headroom markers; "
f"{_diff_specs(existing, spec)}",
)
return RegisterResult(RegisterStatus.MISMATCH, _diff_specs(existing, spec))
if existing is not None and force:
# Drop any prior block (ours or user's) before re-writing.
self.unregister_server(spec.name)
return self._write_block(spec)
def unregister_server(self, server_name: str) -> bool:
# Only removes the marker-block we wrote. User-managed entries
# outside markers are intentionally preserved.
if not self._config_file.exists():
return False
content = self._read_text()
if _MARKER_START not in content or _MARKER_END not in content:
return False
try:
start = content.index(_MARKER_START)
end = content.index(_MARKER_END) + len(_MARKER_END)
except ValueError:
return False
before = content[:start].rstrip("\n")
after = content[end:].lstrip("\n")
if before and after:
new_content = before + "\n\n" + after
else:
new_content = (before or after).rstrip("\n") + ("\n" if (before or after) else "")
try:
self._config_file.write_text(new_content)
except OSError:
return False
return True
# ------------------------------------------------------------------
# File IO
# ------------------------------------------------------------------
def _load_toml(self) -> dict[str, Any]:
if not self._config_file.exists():
return {}
try:
with open(self._config_file, "rb") as f:
data = tomllib.load(f)
except (tomllib.TOMLDecodeError, OSError):
return {}
return data if isinstance(data, dict) else {}
def _read_text(self) -> str:
try:
return self._config_file.read_text()
except OSError:
return ""
def _write_block(self, spec: ServerSpec) -> RegisterResult:
block = _render_block(spec)
try:
self._codex_dir.mkdir(parents=True, exist_ok=True)
content = self._read_text()
if _MARKER_START in content and _MARKER_END in content:
start = content.index(_MARKER_START)
end = content.index(_MARKER_END) + len(_MARKER_END)
content = (
content[:start].rstrip("\n")
+ ("\n\n" if content[:start].rstrip("\n") else "")
+ block
+ "\n"
+ content[end:].lstrip("\n")
)
elif content.strip():
content = content.rstrip("\n") + "\n\n" + block + "\n"
else:
content = block + "\n"
self._config_file.write_text(content)
except OSError as exc:
return RegisterResult(
RegisterStatus.FAILED, f"could not write {self._config_file}: {exc}"
)
return RegisterResult(RegisterStatus.REGISTERED, f"wrote to {self._config_file}")
# ----------------------------------------------------------------------
# TOML rendering / parsing helpers (kept module-private)
# ----------------------------------------------------------------------
def _render_block(spec: ServerSpec) -> str:
"""Render a Headroom-marked TOML block for ``spec``."""
lines: list[str] = [
_MARKER_START,
f"[mcp_servers.{spec.name}]",
f"command = {_toml_str(spec.command)}",
]
if spec.args:
items = ", ".join(_toml_str(a) for a in spec.args)
lines.append(f"args = [{items}]")
if spec.env:
lines.append("")
lines.append(f"[mcp_servers.{spec.name}.env]")
for k, v in spec.env.items():
lines.append(f"{k} = {_toml_str(v)}")
lines.append(_MARKER_END)
return "\n".join(lines)
def _toml_str(s: str) -> str:
"""Render a Python string as a TOML basic string literal."""
escaped = s.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
def _entry_to_spec(name: str, entry: dict[str, Any]) -> ServerSpec:
args_value = entry.get("args", [])
if isinstance(args_value, list):
args = tuple(str(x) for x in args_value)
else:
args = ()
env_value = entry.get("env", {})
env: dict[str, str] = {}
if isinstance(env_value, dict):
env = {str(k): str(v) for k, v in env_value.items()}
return ServerSpec(
name=name,
command=str(entry.get("command", "")),
args=args,
env=env,
)
def _specs_equivalent(a: ServerSpec, b: ServerSpec) -> bool:
return (
a.name == b.name
and a.command == b.command
and tuple(a.args) == tuple(b.args)
and dict(a.env) == dict(b.env)
)
def _diff_specs(existing: ServerSpec, requested: ServerSpec) -> str:
parts: list[str] = []
if existing.command != requested.command:
parts.append(f"command {existing.command!r} -> {requested.command!r}")
if tuple(existing.args) != tuple(requested.args):
parts.append(f"args {list(existing.args)} -> {list(requested.args)}")
if dict(existing.env) != dict(requested.env):
parts.append(f"env {dict(existing.env)} -> {dict(requested.env)}")
if not parts:
return "spec differs in unidentified field(s)"
return "; ".join(parts)

95
headroom/mcp_registry/display.py Normal file
View file

@ -0,0 +1,95 @@
"""Human-readable rendering of MCP registration results.
The CLI ``mcp install`` command, ``init`` flow, and ``wrap claude`` setup
all consume the same ``dict[str, RegisterResult]`` and want to print one
line per agent. They differ only in label format (e.g. ``" claude:"`` vs
``" MCP retrieve tool:"``), whether to show ALREADY-registered as a
success line, and which corrective command to suggest on mismatch.
Centralizing those choices here keeps every status branch in one file.
Adding a new :class:`RegisterStatus` member becomes a single edit; the
call sites compose by passing flags rather than re-writing the switch.
"""
from __future__ import annotations
from collections.abc import Callable
from .base import RegisterResult, RegisterStatus
DEFAULT_OVERWRITE_HINT = "headroom mcp install --force"
DEFAULT_RESTART_HINT = "restart the agent if it was already running"
def format_result(
agent: str,
result: RegisterResult,
*,
label: str | None = None,
verbose: bool = False,
overwrite_hint: str = DEFAULT_OVERWRITE_HINT,
restart_hint: str = DEFAULT_RESTART_HINT,
) -> str | None:
"""Render one ``(agent, result)`` pair as a single display line.
Returns ``None`` to suppress output (e.g. ALREADY when not ``verbose``).
Args:
agent: Stable agent name (used as default label).
result: Outcome from :func:`install_everywhere` or a registrar.
label: Override the leading label. Defaults to the agent name.
verbose: If ``True``, include status lines that are otherwise
silent (e.g. ALREADY).
overwrite_hint: Command to suggest when the existing config differs.
restart_hint: Hint appended to a fresh registration line.
"""
label = label if label is not None else agent
status = result.status
if status == RegisterStatus.REGISTERED:
return f" {label}: registered ({restart_hint})"
if status == RegisterStatus.ALREADY:
return f" {label}: already registered" if verbose else None
if status == RegisterStatus.NOT_DETECTED:
return f" {label}: not detected on this system, skipped"
if status == RegisterStatus.MISMATCH:
suffix = f" To update: {overwrite_hint}" if overwrite_hint else ""
return f" {label}: existing config differs ({result.detail}).{suffix}"
if status == RegisterStatus.NO_SDK:
return f" {label}: MCP SDK missing — install with `pip install 'headroom-ai[mcp]'`"
# FAILED or any future unhandled status
return f" {label}: install failed ({status.value}): {result.detail}"
def format_results(
results: dict[str, RegisterResult],
*,
label_for: Callable[[str], str | None] | None = None,
verbose: bool = False,
overwrite_hint: str = DEFAULT_OVERWRITE_HINT,
restart_hint: str = DEFAULT_RESTART_HINT,
) -> list[str]:
"""Render a results dict to a list of display lines.
``label_for`` is an optional ``agent -> label`` mapper. Pass ``None``
(default) to use the agent name as the label.
"""
lines: list[str] = []
for agent, result in results.items():
label = label_for(agent) if label_for is not None else None
line = format_result(
agent,
result,
label=label,
verbose=verbose,
overwrite_hint=overwrite_hint,
restart_hint=restart_hint,
)
if line is not None:
lines.append(line)
return lines
def any_succeeded(results: dict[str, RegisterResult]) -> bool:
"""True when at least one agent ended in REGISTERED or ALREADY."""
return any(r.ok for r in results.values())

76
headroom/mcp_registry/install.py Normal file
View file

@ -0,0 +1,76 @@
"""Top-level orchestration: register Headroom MCP across detected agents."""
from __future__ import annotations
from collections.abc import Iterable
from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec
from .claude import ClaudeRegistrar
from .codex import CodexRegistrar
#: Default proxy URL used when none is given.
DEFAULT_PROXY_URL = "http://127.0.0.1:8787"
def get_all_registrars() -> list[MCPRegistrar]:
"""Return one instance of every registrar implemented today.
The list grows as we add adapters for Cursor, Continue, Cline, etc.
"""
return [ClaudeRegistrar(), CodexRegistrar()]
def build_headroom_spec(proxy_url: str = DEFAULT_PROXY_URL) -> ServerSpec:
"""Construct the canonical :class:`ServerSpec` for the headroom server.
The spec is identical across agents — every JSON/TOML registrar
serializes the same shape into its own format.
"""
env: dict[str, str] = {}
if proxy_url and proxy_url != DEFAULT_PROXY_URL:
env["HEADROOM_PROXY_URL"] = proxy_url
return ServerSpec(
name="headroom",
command="headroom",
args=("mcp", "serve"),
env=env,
)
def install_everywhere(
proxy_url: str = DEFAULT_PROXY_URL,
*,
agents: Iterable[str] | None = None,
force: bool = False,
registrars: Iterable[MCPRegistrar] | None = None,
) -> dict[str, RegisterResult]:
"""Install the headroom MCP server into every detected agent.
Args:
proxy_url: URL the MCP server should contact for retrieval.
agents: If given, only install into agents whose ``name`` matches.
force: Pass through to each registrar — overwrites mismatched config.
registrars: Inject a custom registrar list (test seam).
Returns:
Dict keyed by registrar name. Includes :attr:`RegisterStatus.NOT_DETECTED`
entries for agents we know about that aren't installed locally.
"""
spec = build_headroom_spec(proxy_url)
selected = list(registrars) if registrars is not None else get_all_registrars()
if agents is not None:
agent_set = set(agents)
selected = [r for r in selected if r.name in agent_set]
results: dict[str, RegisterResult] = {}
for registrar in selected:
if not registrar.detect():
results[registrar.name] = RegisterResult(
RegisterStatus.NOT_DETECTED,
f"{registrar.display_name} not found on this system",
)
continue
results[registrar.name] = registrar.register_server(spec, force=force)
return results

294
tests/test_cli/test_mcp.py
View file

@ -129,117 +129,38 @@ class TestMCPConfigFunctions:
assert "other-server" in loaded["mcpServers"]
class TestMCPInstallCommand:
"""Test 'headroom mcp install' command."""
def test_install_creates_config(self, mock_claude_config_path, mock_mcp_available):
"""Install creates MCP config file."""
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Failed with output: {result.output}"
assert "installed" in result.output.lower()
assert mock_claude_config_path.exists()
# Verify config content
config = json.loads(mock_claude_config_path.read_text())
assert "headroom" in config["mcpServers"]
assert config["mcpServers"]["headroom"]["command"] == "headroom"
assert "mcp" in config["mcpServers"]["headroom"]["args"]
assert "serve" in config["mcpServers"]["headroom"]["args"]
def test_install_preserves_other_servers(self, mock_claude_config_path, mock_mcp_available):
"""Install preserves existing MCP servers."""
# Create config with another server
existing_config = {
"mcpServers": {
"github": {"command": "github-mcp", "args": []},
}
}
mock_claude_config_path.write_text(json.dumps(existing_config))
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Failed with output: {result.output}"
# Both servers should exist
config = json.loads(mock_claude_config_path.read_text())
assert "github" in config["mcpServers"]
assert "headroom" in config["mcpServers"]
def test_install_with_custom_proxy_url(self, mock_claude_config_path, mock_mcp_available):
"""Install with custom proxy URL sets env var."""
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install", "--proxy-url", "http://localhost:9000"])
assert result.exit_code == 0, f"Failed with output: {result.output}"
config = json.loads(mock_claude_config_path.read_text())
assert (
config["mcpServers"]["headroom"]["env"]["HEADROOM_PROXY_URL"] == "http://localhost:9000"
)
def test_install_default_proxy_url_no_env(self, mock_claude_config_path, mock_mcp_available):
"""Install with default proxy URL doesn't set env var."""
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Failed with output: {result.output}"
config = json.loads(mock_claude_config_path.read_text())
# No env section for default URL
assert "env" not in config["mcpServers"]["headroom"]
def test_install_already_configured_no_force(self, mock_claude_config_path, mock_mcp_available):
"""Install without --force when already configured exits cleanly."""
# First install
runner = CliRunner()
runner.invoke(main, ["mcp", "install"])
# Second install without force
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0
assert "already configured" in result.output.lower()
def test_install_force_overwrites(self, mock_claude_config_path, mock_mcp_available):
"""Install with --force overwrites existing config."""
runner = CliRunner()
runner.invoke(main, ["mcp", "install", "--proxy-url", "http://old:8787"])
# Force install with new URL
result = runner.invoke(
main, ["mcp", "install", "--force", "--proxy-url", "http://new:9000"]
)
assert result.exit_code == 0, f"Failed with output: {result.output}"
assert "installed" in result.output.lower()
config = json.loads(mock_claude_config_path.read_text())
assert config["mcpServers"]["headroom"]["env"]["HEADROOM_PROXY_URL"] == "http://new:9000"
@pytest.mark.skipif(MCP_AVAILABLE, reason="Test only runs when MCP SDK is NOT installed")
def test_install_without_mcp_sdk_fails(self, mock_claude_config_path):
"""Install fails gracefully when MCP SDK is not installed."""
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install"])
# Should fail with helpful message
assert result.exit_code == 1
assert "mcp" in result.output.lower() or "not installed" in result.output.lower()
#
# Note: Tests for the 'mcp install' command's writes/idempotency/CLI-vs-file
# fallback used to live here, but they were tightly coupled to private
# globals (MCP_CONFIG_PATH, shutil.which) and exercised the same surface
# already covered by:
# - tests/test_mcp_registry/test_claude_registrar.py (file/CLI behavior
# with proper constructor injection — no patches)
# - tests/test_mcp_registry/test_install.py (orchestrator semantics with
# fake registrars)
# Removing the duplicates leaves the CLI as glue: argument parsing +
# output formatting, which is straightforward and not worth its own test
# layer.
class TestMCPUninstallCommand:
"""Test 'headroom mcp uninstall' command."""
def test_uninstall_removes_headroom(self, mock_claude_config_path, mock_mcp_available):
"""Uninstall removes headroom from config."""
# First install
runner = CliRunner()
runner.invoke(main, ["mcp", "install"])
"""Uninstall removes headroom from the legacy config file."""
# Pre-populate the config directly rather than depending on
# `mcp install` plumbing — keeps the test focused on uninstall.
mock_claude_config_path.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {"command": "headroom", "args": ["mcp", "serve"]},
}
}
)
)
# Then uninstall
runner = CliRunner()
result = runner.invoke(main, ["mcp", "uninstall"])
assert result.exit_code == 0
@ -307,10 +228,20 @@ class TestMCPStatusCommand:
)
def test_status_configured(self, mock_claude_config_path, mock_mcp_available):
"""Status shows configured when installed."""
runner = CliRunner()
runner.invoke(main, ["mcp", "install"])
"""Status reports configured when the legacy config has headroom."""
# Pre-populate the legacy mcp.json directly. mcp_status() reads
# from MCP_CONFIG_PATH, which the fixture redirects here.
mock_claude_config_path.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {"command": "headroom", "args": ["mcp", "serve"]},
}
}
)
)
runner = CliRunner()
result = runner.invoke(main, ["mcp", "status"])
assert result.exit_code == 0
@ -366,149 +297,12 @@ class TestMCPServerInitialization:
assert CCR_TOOL_NAME == "headroom_retrieve"
class TestEndToEndFlow:
"""Test complete install -> status -> uninstall flow."""
def test_full_lifecycle(self, mock_claude_config_path, mock_mcp_available):
"""Test complete lifecycle of MCP configuration."""
runner = CliRunner()
# Initially not configured
result = runner.invoke(main, ["mcp", "status"])
assert "No config" in result.output or "Not configured" in result.output.lower()
# Install
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Install failed: {result.output}"
assert "installed" in result.output.lower()
# Status shows configured
result = runner.invoke(main, ["mcp", "status"])
assert "✓ Configured" in result.output
# Config file has correct content
config = json.loads(mock_claude_config_path.read_text())
assert config["mcpServers"]["headroom"]["command"] == "headroom"
# Uninstall
result = runner.invoke(main, ["mcp", "uninstall"])
assert result.exit_code == 0
assert "removed" in result.output.lower()
# Status shows not configured
result = runner.invoke(main, ["mcp", "status"])
assert "headroom" not in result.output.lower() or "not configured" in result.output.lower()
class TestMCPInstallWithClaudeCLI:
"""Test mcp_install when the claude CLI is available."""
def _make_run(self, get_rc=1, add_rc=0, remove_rc=0):
"""Return a subprocess.run mock with configurable return codes."""
def run(cmd, **kwargs):
if "get" in cmd:
return MagicMock(returncode=get_rc, stderr="")
if "remove" in cmd:
return MagicMock(returncode=remove_rc, stderr="")
if "add" in cmd:
return MagicMock(returncode=add_rc, stderr="")
return MagicMock(returncode=0, stderr="")
return run
def test_install_uses_claude_mcp_add(self, mock_mcp_available):
"""When claude CLI is available, install calls claude mcp add."""
runner = CliRunner()
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
with patch("headroom.cli.mcp.subprocess.run", side_effect=self._make_run()) as mock_run:
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Failed: {result.output}"
assert "installed" in result.output.lower()
assert "claude mcp add" in result.output
# Verify claude mcp add was called
add_calls = [c for c in mock_run.call_args_list if "add" in c.args[0]]
assert len(add_calls) == 1
add_cmd = add_calls[0].args[0]
assert "headroom" in add_cmd
assert "-s" in add_cmd
assert "user" in add_cmd
def test_install_already_registered_no_force(self, mock_mcp_available):
"""Install without --force exits cleanly when already registered via claude CLI."""
runner = CliRunner()
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
# get returns 0 → already registered
with patch("headroom.cli.mcp.subprocess.run", side_effect=self._make_run(get_rc=0)):
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0
assert "already configured" in result.output.lower()
def test_install_force_calls_remove_then_add(self, mock_mcp_available):
"""--force calls claude mcp remove before claude mcp add."""
runner = CliRunner()
calls = []
def capturing_run(cmd, **kwargs):
calls.append(cmd)
return MagicMock(returncode=0, stderr="")
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
result = runner.invoke(main, ["mcp", "install", "--force"])
assert result.exit_code == 0, f"Failed: {result.output}"
subcommands = [c[2] for c in calls] # third element is the subcommand
assert "remove" in subcommands
assert "add" in subcommands
assert subcommands.index("remove") < subcommands.index("add")
def test_install_fallback_on_claude_mcp_add_failure(self, temp_claude_dir, mock_mcp_available):
"""If claude mcp add fails, falls back to writing mcp.json."""
config_path = temp_claude_dir / "mcp.json"
with patch("headroom.cli.mcp.MCP_CONFIG_PATH", config_path):
with patch("headroom.cli.mcp.CLAUDE_CONFIG_DIR", temp_claude_dir):
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
with patch(
"headroom.cli.mcp.subprocess.run",
side_effect=self._make_run(add_rc=1),
):
runner = CliRunner()
result = runner.invoke(main, ["mcp", "install"])
assert result.exit_code == 0, f"Failed: {result.output}"
assert config_path.exists()
config = json.loads(config_path.read_text())
assert "headroom" in config["mcpServers"]
def test_install_with_custom_proxy_url_passes_e_flag(self, mock_mcp_available):
"""Custom proxy URL is passed as -e KEY=VALUE to claude mcp add."""
calls = []
def capturing_run(cmd, **kwargs):
calls.append(list(cmd))
# Return rc=1 for "get" so headroom is treated as not yet registered
if "get" in cmd:
return MagicMock(returncode=1, stderr="")
return MagicMock(returncode=0, stderr="")
runner = CliRunner()
with patch("headroom.cli.mcp.shutil.which", return_value="/usr/bin/claude"):
with patch("headroom.cli.mcp.subprocess.run", side_effect=capturing_run):
result = runner.invoke(
main, ["mcp", "install", "--proxy-url", "http://custom:9000"]
)
assert result.exit_code == 0, f"Failed: {result.output}"
add_calls = [c for c in calls if "add" in c]
assert len(add_calls) == 1
add_cmd = add_calls[0]
assert "-e" in add_cmd
env_idx = add_cmd.index("-e")
assert add_cmd[env_idx + 1] == "HEADROOM_PROXY_URL=http://custom:9000"
#
# Tests for "install via claude CLI" used to live here, exercising
# subprocess.run patches against the old direct CLI invocation. Equivalent
# coverage now lives in tests/test_mcp_registry/test_claude_registrar.py
# using constructor injection (`claude_cli="/path/to/fake"`) and bounded
# subprocess.run mocks at the registrar boundary — no module-level patches.
class TestMCPUninstallWithClaudeCLI:

0
tests/test_mcp_registry/__init__.py Normal file
View file

309
tests/test_mcp_registry/test_claude_registrar.py Normal file
View file

@ -0,0 +1,309 @@
"""Tests for the Claude Code MCP registrar."""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from unittest.mock import patch
import pytest
from headroom.mcp_registry.base import RegisterStatus, ServerSpec
from headroom.mcp_registry.claude import ClaudeRegistrar
def _make_registrar(
tmp_path: Path,
*,
cli: str | None = "/usr/local/bin/claude",
) -> ClaudeRegistrar:
"""Build a registrar pointed at ``tmp_path`` as $HOME."""
return ClaudeRegistrar(claude_cli=cli, home_dir=tmp_path)
def _spec() -> ServerSpec:
return ServerSpec(
name="headroom",
command="headroom",
args=("mcp", "serve"),
env={},
)
# ----------------------------------------------------------------------
# detect()
# ----------------------------------------------------------------------
def test_detect_true_when_cli_present(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
assert reg.detect() is True
def test_detect_true_when_only_claude_dir_exists(tmp_path: Path) -> None:
(tmp_path / ".claude").mkdir()
reg = _make_registrar(tmp_path, cli=None)
assert reg.detect() is True
def test_detect_false_when_neither_present(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli=None)
assert reg.detect() is False
# ----------------------------------------------------------------------
# get_server() — file-based reads
# ----------------------------------------------------------------------
def test_get_server_returns_none_when_unregistered(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli=None)
assert reg.get_server("headroom") is None
def test_get_server_reads_modern_config(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {
"command": "headroom",
"args": ["mcp", "serve"],
"env": {"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"},
}
}
}
)
)
reg = _make_registrar(tmp_path, cli=None)
got = reg.get_server("headroom")
assert got is not None
assert got.command == "headroom"
assert got.args == ("mcp", "serve")
assert got.env == {"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"}
def test_get_server_falls_back_to_legacy(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / "mcp.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps({"mcpServers": {"headroom": {"command": "headroom", "args": ["mcp", "serve"]}}})
)
reg = _make_registrar(tmp_path, cli=None)
got = reg.get_server("headroom")
assert got is not None
assert got.command == "headroom"
assert got.args == ("mcp", "serve")
assert got.env == {}
# ----------------------------------------------------------------------
# register_server() — happy paths
# ----------------------------------------------------------------------
def test_register_via_cli_calls_claude_mcp_add(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
fake_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
with patch("subprocess.run", return_value=fake_result) as run_mock:
result = reg.register_server(_spec())
assert result.status == RegisterStatus.REGISTERED
cmds = [call.args[0] for call in run_mock.call_args_list]
add_cmd = next(c for c in cmds if "add" in c)
assert add_cmd[:6] == [
"/usr/local/bin/claude",
"mcp",
"add",
"headroom",
"-s",
"user",
]
assert add_cmd[-3:] == ["--", "headroom", "mcp"] or add_cmd[-4:] == [
"--",
"headroom",
"mcp",
"serve",
]
def test_register_via_cli_includes_env(tmp_path: Path) -> None:
spec = ServerSpec(
name="headroom",
command="headroom",
args=("mcp", "serve"),
env={"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"},
)
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
fake_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
with patch("subprocess.run", return_value=fake_result) as run_mock:
reg.register_server(spec)
add_cmd = next(c for c in [call.args[0] for call in run_mock.call_args_list] if "add" in c)
assert "-e" in add_cmd
e_idx = add_cmd.index("-e")
assert add_cmd[e_idx + 1] == "HEADROOM_PROXY_URL=http://127.0.0.1:9000"
def test_register_writes_file_when_no_cli(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli=None)
result = reg.register_server(_spec())
assert result.status == RegisterStatus.REGISTERED
cfg = tmp_path / ".claude" / ".claude.json"
data = json.loads(cfg.read_text())
assert "headroom" in data["mcpServers"]
assert data["mcpServers"]["headroom"]["command"] == "headroom"
assert data["mcpServers"]["headroom"]["args"] == ["mcp", "serve"]
def test_register_writes_to_legacy_when_only_legacy_exists(tmp_path: Path) -> None:
legacy = tmp_path / ".claude" / "mcp.json"
legacy.parent.mkdir()
legacy.write_text(json.dumps({"mcpServers": {}}))
reg = _make_registrar(tmp_path, cli=None)
result = reg.register_server(_spec())
assert result.status == RegisterStatus.REGISTERED
data = json.loads(legacy.read_text())
assert "headroom" in data["mcpServers"]
# Modern config should NOT have been created.
assert not (tmp_path / ".claude" / ".claude.json").exists()
# ----------------------------------------------------------------------
# register_server() — already / mismatch / force
# ----------------------------------------------------------------------
def test_register_already_when_spec_matches(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps({"mcpServers": {"headroom": {"command": "headroom", "args": ["mcp", "serve"]}}})
)
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
with patch("subprocess.run") as run_mock:
result = reg.register_server(_spec())
assert result.status == RegisterStatus.ALREADY
run_mock.assert_not_called() # should not touch CLI when already matching
def test_register_mismatch_when_spec_differs_no_force(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {
"command": "headroom",
"args": ["mcp", "serve"],
"env": {"HEADROOM_PROXY_URL": "http://127.0.0.1:9999"},
}
}
}
)
)
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
with patch("subprocess.run") as run_mock:
result = reg.register_server(_spec()) # default proxy = no env
assert result.status == RegisterStatus.MISMATCH
assert "env" in (result.detail or "")
run_mock.assert_not_called() # do NOT overwrite without force
def test_register_force_overwrites_mismatch(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {
"command": "headroom-old",
"args": ["mcp", "serve"],
}
}
}
)
)
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
fake_ok = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
with patch("subprocess.run", return_value=fake_ok) as run_mock:
result = reg.register_server(_spec(), force=True)
assert result.status == RegisterStatus.REGISTERED
cmds = [call.args[0] for call in run_mock.call_args_list]
assert any("remove" in c for c in cmds)
assert any("add" in c for c in cmds)
# ----------------------------------------------------------------------
# CLI failure paths
# ----------------------------------------------------------------------
def test_register_cli_failure_falls_back_to_file(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
fail = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="claude: error")
with patch("subprocess.run", return_value=fail):
result = reg.register_server(_spec())
# Even though CLI failed, we wrote the config file as a fallback.
assert result.status == RegisterStatus.REGISTERED
cfg = tmp_path / ".claude" / ".claude.json"
assert cfg.exists()
data = json.loads(cfg.read_text())
assert "headroom" in data["mcpServers"]
# ----------------------------------------------------------------------
# unregister
# ----------------------------------------------------------------------
def test_unregister_via_cli(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli="/usr/local/bin/claude")
ok = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
with patch("subprocess.run", return_value=ok) as run_mock:
assert reg.unregister_server("headroom") is True
cmd = run_mock.call_args_list[0].args[0]
assert cmd[:5] == ["/usr/local/bin/claude", "mcp", "remove", "headroom", "-s"]
assert cmd[5] == "user"
def test_unregister_via_file_when_no_cli(tmp_path: Path) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(
json.dumps(
{
"mcpServers": {
"headroom": {"command": "headroom", "args": ["mcp", "serve"]},
"other": {"command": "other"},
}
}
)
)
reg = _make_registrar(tmp_path, cli=None)
assert reg.unregister_server("headroom") is True
data = json.loads(cfg.read_text())
assert "headroom" not in data["mcpServers"]
assert "other" in data["mcpServers"]
def test_unregister_returns_false_when_absent(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path, cli=None)
assert reg.unregister_server("headroom") is False
# ----------------------------------------------------------------------
# Robustness: bad JSON should not crash
# ----------------------------------------------------------------------
@pytest.mark.parametrize("contents", ["", "not json", "{", "[]"])
def test_get_server_robust_to_bad_json(tmp_path: Path, contents: str) -> None:
cfg = tmp_path / ".claude" / ".claude.json"
cfg.parent.mkdir()
cfg.write_text(contents)
reg = _make_registrar(tmp_path, cli=None)
assert reg.get_server("headroom") is None

258
tests/test_mcp_registry/test_codex_registrar.py Normal file
View file

@ -0,0 +1,258 @@
"""Tests for the OpenAI Codex MCP registrar."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
from headroom.mcp_registry.base import RegisterStatus, ServerSpec
from headroom.mcp_registry.codex import CodexRegistrar
if sys.version_info >= (3, 11):
import tomllib
else: # pragma: no cover
import tomli as tomllib
def _make_registrar(tmp_path: Path) -> CodexRegistrar:
return CodexRegistrar(home_dir=tmp_path)
def _spec(env: dict[str, str] | None = None) -> ServerSpec:
return ServerSpec(
name="headroom",
command="headroom",
args=("mcp", "serve"),
env=env or {},
)
def _config_path(tmp_path: Path) -> Path:
return tmp_path / ".codex" / "config.toml"
# ----------------------------------------------------------------------
# detect()
# ----------------------------------------------------------------------
def test_detect_true_when_codex_dir_exists(tmp_path: Path) -> None:
(tmp_path / ".codex").mkdir()
assert _make_registrar(tmp_path).detect() is True
def test_detect_false_when_codex_dir_missing(tmp_path: Path) -> None:
assert _make_registrar(tmp_path).detect() is False
# ----------------------------------------------------------------------
# get_server()
# ----------------------------------------------------------------------
def test_get_server_returns_none_when_config_missing(tmp_path: Path) -> None:
assert _make_registrar(tmp_path).get_server("headroom") is None
def test_get_server_returns_none_when_no_table(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text('model = "gpt-4o"\n')
assert _make_registrar(tmp_path).get_server("headroom") is None
def test_get_server_returns_spec_when_table_present(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text(
"[mcp_servers.headroom]\n"
'command = "headroom"\n'
'args = ["mcp", "serve"]\n'
"\n"
"[mcp_servers.headroom.env]\n"
'HEADROOM_PROXY_URL = "http://127.0.0.1:9000"\n'
)
got = _make_registrar(tmp_path).get_server("headroom")
assert got is not None
assert got.command == "headroom"
assert got.args == ("mcp", "serve")
assert got.env == {"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"}
def test_get_server_robust_to_unparseable_toml(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text("this = is = not = valid\n")
assert _make_registrar(tmp_path).get_server("headroom") is None
# ----------------------------------------------------------------------
# register_server() — happy paths
# ----------------------------------------------------------------------
def test_register_creates_config_when_missing(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path)
result = reg.register_server(_spec())
assert result.status == RegisterStatus.REGISTERED
cfg = _config_path(tmp_path)
assert cfg.exists()
text = cfg.read_text()
assert "# --- Headroom MCP server ---" in text
assert "[mcp_servers.headroom]" in text
parsed = tomllib.loads(text)
assert parsed["mcp_servers"]["headroom"]["command"] == "headroom"
assert parsed["mcp_servers"]["headroom"]["args"] == ["mcp", "serve"]
def test_register_appends_to_existing_config_preserves_other_keys(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text('# user comment\nmodel = "gpt-4o"\n\n[other_section]\nvalue = 42\n')
result = _make_registrar(tmp_path).register_server(_spec())
assert result.status == RegisterStatus.REGISTERED
text = cfg.read_text()
# Existing content survived.
assert "# user comment" in text
assert 'model = "gpt-4o"' in text
assert "[other_section]" in text
# Plus our block.
assert "[mcp_servers.headroom]" in text
parsed = tomllib.loads(text)
assert parsed["model"] == "gpt-4o"
assert parsed["other_section"]["value"] == 42
assert parsed["mcp_servers"]["headroom"]["command"] == "headroom"
def test_register_includes_env_subtable(tmp_path: Path) -> None:
spec = _spec(env={"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"})
_make_registrar(tmp_path).register_server(spec)
text = _config_path(tmp_path).read_text()
assert "[mcp_servers.headroom.env]" in text
parsed = tomllib.loads(text)
assert parsed["mcp_servers"]["headroom"]["env"] == {
"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"
}
def test_register_omits_env_subtable_when_env_empty(tmp_path: Path) -> None:
_make_registrar(tmp_path).register_server(_spec())
text = _config_path(tmp_path).read_text()
assert "[mcp_servers.headroom.env]" not in text
# ----------------------------------------------------------------------
# Idempotency: ALREADY / MISMATCH
# ----------------------------------------------------------------------
def test_register_already_when_block_matches_spec(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path)
reg.register_server(_spec()) # first install
text_before = _config_path(tmp_path).read_text()
result = reg.register_server(_spec()) # second install, same spec
assert result.status == RegisterStatus.ALREADY
# File unchanged.
assert _config_path(tmp_path).read_text() == text_before
def test_register_mismatch_when_block_differs_no_force(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path)
reg.register_server(_spec(env={"HEADROOM_PROXY_URL": "http://127.0.0.1:9999"}))
text_before = _config_path(tmp_path).read_text()
result = reg.register_server(_spec()) # no env
assert result.status == RegisterStatus.MISMATCH
assert "env" in (result.detail or "")
assert _config_path(tmp_path).read_text() == text_before # unchanged
def test_register_force_overwrites_block(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path)
reg.register_server(_spec(env={"HEADROOM_PROXY_URL": "http://127.0.0.1:9999"}))
result = reg.register_server(_spec(), force=True)
assert result.status == RegisterStatus.REGISTERED
text = _config_path(tmp_path).read_text()
assert "9999" not in text
def test_register_mismatch_when_user_managed_outside_markers(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
# User has manually put [mcp_servers.headroom] with different config — no markers.
cfg.write_text(
'[mcp_servers.headroom]\ncommand = "/usr/local/bin/custom-headroom"\nargs = ["serve"]\n'
)
result = _make_registrar(tmp_path).register_server(_spec())
assert result.status == RegisterStatus.MISMATCH
assert "user-managed" in (result.detail or "").lower()
# Don't overwrite.
assert "/usr/local/bin/custom-headroom" in cfg.read_text()
# ----------------------------------------------------------------------
# unregister
# ----------------------------------------------------------------------
def test_unregister_removes_marker_block(tmp_path: Path) -> None:
reg = _make_registrar(tmp_path)
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text("[other_section]\nvalue = 42\n")
reg.register_server(_spec())
assert "[mcp_servers.headroom]" in cfg.read_text()
assert reg.unregister_server("headroom") is True
text = cfg.read_text()
assert "[mcp_servers.headroom]" not in text
assert "# --- Headroom MCP server ---" not in text
# Surrounding content survives.
assert "[other_section]" in text
def test_unregister_returns_false_when_no_block(tmp_path: Path) -> None:
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text('model = "gpt-4o"\n')
assert _make_registrar(tmp_path).unregister_server("headroom") is False
def test_unregister_preserves_user_managed_entry(tmp_path: Path) -> None:
"""User-managed [mcp_servers.headroom] without our markers stays put."""
cfg = _config_path(tmp_path)
cfg.parent.mkdir()
cfg.write_text('[mcp_servers.headroom]\ncommand = "/custom/headroom"\n')
# No markers => unregister is a no-op.
assert _make_registrar(tmp_path).unregister_server("headroom") is False
assert "/custom/headroom" in cfg.read_text()
# ----------------------------------------------------------------------
# Round-trip: write → re-read produces equivalent ServerSpec
# ----------------------------------------------------------------------
@pytest.mark.parametrize(
"spec",
[
ServerSpec(name="headroom", command="headroom", args=("mcp", "serve")),
ServerSpec(
name="headroom",
command="headroom",
args=("mcp", "serve"),
env={"HEADROOM_PROXY_URL": "http://127.0.0.1:9000"},
),
ServerSpec(name="headroom", command="/usr/bin/headroom", args=()),
],
)
def test_round_trip(tmp_path: Path, spec: ServerSpec) -> None:
reg = _make_registrar(tmp_path)
reg.register_server(spec)
got = reg.get_server("headroom")
assert got is not None
assert got.command == spec.command
assert got.args == spec.args
assert got.env == spec.env

179
tests/test_mcp_registry/test_display.py Normal file
View file

@ -0,0 +1,179 @@
"""Tests for the install-result display helper."""
from __future__ import annotations
from headroom.mcp_registry.base import RegisterResult, RegisterStatus
from headroom.mcp_registry.display import (
any_succeeded,
format_result,
format_results,
)
# ----------------------------------------------------------------------
# format_result
# ----------------------------------------------------------------------
def test_registered_includes_restart_hint() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.REGISTERED, "via CLI"),
restart_hint="restart Claude Code if it was running",
)
assert line is not None
assert "claude" in line
assert "registered" in line
assert "restart Claude Code" in line
def test_already_silent_when_not_verbose() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.ALREADY, "matches"),
verbose=False,
)
assert line is None
def test_already_emits_when_verbose() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.ALREADY, "matches"),
verbose=True,
)
assert line is not None
assert "already registered" in line
def test_not_detected_emits_skipped() -> None:
line = format_result(
"cursor",
RegisterResult(RegisterStatus.NOT_DETECTED, "Cursor not found"),
)
assert line is not None
assert "cursor" in line
assert "not detected" in line
assert "skipped" in line
def test_mismatch_emits_overwrite_hint() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.MISMATCH, "env differs"),
overwrite_hint="headroom mcp install --force",
)
assert line is not None
assert "differs" in line
assert "headroom mcp install --force" in line
def test_mismatch_omits_hint_when_empty() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.MISMATCH, "env differs"),
overwrite_hint="",
)
assert line is not None
assert "To update" not in line
def test_no_sdk_points_at_pip_extras() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.NO_SDK, "missing"),
)
assert line is not None
assert "MCP SDK" in line
assert "headroom-ai[mcp]" in line
def test_failed_includes_detail() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.FAILED, "connection refused"),
)
assert line is not None
assert "failed" in line
assert "connection refused" in line
def test_label_overrides_agent_name() -> None:
line = format_result(
"claude",
RegisterResult(RegisterStatus.REGISTERED, "ok"),
label="MCP retrieve tool",
)
assert line is not None
assert "MCP retrieve tool" in line
assert "claude" not in line
# ----------------------------------------------------------------------
# format_results
# ----------------------------------------------------------------------
def test_format_results_filters_silent_lines() -> None:
results = {
"claude": RegisterResult(RegisterStatus.REGISTERED, "ok"),
"cursor": RegisterResult(RegisterStatus.ALREADY, "matches"),
}
lines = format_results(results, verbose=False)
# ALREADY suppressed when verbose=False, so only one line.
assert len(lines) == 1
assert "claude" in lines[0]
def test_format_results_label_for_remaps_agent_name() -> None:
results = {
"claude": RegisterResult(RegisterStatus.REGISTERED, "ok"),
}
labels = {"claude": "Claude Code"}
lines = format_results(results, label_for=labels.get)
assert len(lines) == 1
assert "Claude Code" in lines[0]
def test_format_results_preserves_iteration_order() -> None:
results = {
"a": RegisterResult(RegisterStatus.REGISTERED, "ok"),
"b": RegisterResult(RegisterStatus.NOT_DETECTED, "missing"),
"c": RegisterResult(RegisterStatus.MISMATCH, "env differs"),
}
lines = format_results(results, verbose=True)
assert len(lines) == 3
# Same order as input dict iteration.
assert lines[0].strip().startswith("a:")
assert lines[1].strip().startswith("b:")
assert lines[2].strip().startswith("c:")
# ----------------------------------------------------------------------
# any_succeeded
# ----------------------------------------------------------------------
def test_any_succeeded_true_when_one_registered() -> None:
results = {
"a": RegisterResult(RegisterStatus.REGISTERED, "ok"),
"b": RegisterResult(RegisterStatus.NOT_DETECTED, "missing"),
}
assert any_succeeded(results) is True
def test_any_succeeded_true_when_already_registered() -> None:
results = {"a": RegisterResult(RegisterStatus.ALREADY, "matches")}
assert any_succeeded(results) is True
def test_any_succeeded_false_when_all_failed_or_skipped() -> None:
results = {
"a": RegisterResult(RegisterStatus.NOT_DETECTED, "missing"),
"b": RegisterResult(RegisterStatus.FAILED, "boom"),
"c": RegisterResult(RegisterStatus.MISMATCH, "differs"),
}
assert any_succeeded(results) is False
def test_any_succeeded_empty_dict_is_false() -> None:
assert any_succeeded({}) is False

140
tests/test_mcp_registry/test_install.py Normal file
View file

@ -0,0 +1,140 @@
"""Tests for the install_everywhere orchestrator."""
from __future__ import annotations
from headroom.mcp_registry.base import (
MCPRegistrar,
RegisterResult,
RegisterStatus,
ServerSpec,
)
from headroom.mcp_registry.install import (
DEFAULT_PROXY_URL,
build_headroom_spec,
install_everywhere,
)
class _FakeRegistrar(MCPRegistrar):
"""Minimal registrar for orchestrator tests."""
def __init__(
self,
name: str,
*,
detected: bool = True,
register_result: RegisterResult | None = None,
) -> None:
self.name = name
self.display_name = name.title()
self._detected = detected
self._register_result = register_result or RegisterResult(RegisterStatus.REGISTERED, "ok")
self.calls: list[ServerSpec] = []
def detect(self) -> bool:
return self._detected
def get_server(self, server_name: str) -> ServerSpec | None:
return None
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
self.calls.append(spec)
return self._register_result
def unregister_server(self, server_name: str) -> bool:
return True
# ----------------------------------------------------------------------
# build_headroom_spec
# ----------------------------------------------------------------------
def test_build_spec_default_proxy_no_env() -> None:
spec = build_headroom_spec()
assert spec.name == "headroom"
assert spec.command == "headroom"
assert spec.args == ("mcp", "serve")
assert spec.env == {}
def test_build_spec_custom_proxy_sets_env() -> None:
spec = build_headroom_spec("http://127.0.0.1:9999")
assert spec.env == {"HEADROOM_PROXY_URL": "http://127.0.0.1:9999"}
def test_build_spec_default_url_omits_env() -> None:
spec = build_headroom_spec(DEFAULT_PROXY_URL)
assert spec.env == {}
# ----------------------------------------------------------------------
# install_everywhere
# ----------------------------------------------------------------------
def test_install_everywhere_calls_each_detected_registrar() -> None:
a = _FakeRegistrar("a")
b = _FakeRegistrar("b")
results = install_everywhere(registrars=[a, b])
assert set(results) == {"a", "b"}
assert results["a"].status == RegisterStatus.REGISTERED
assert results["b"].status == RegisterStatus.REGISTERED
assert len(a.calls) == 1
assert len(b.calls) == 1
def test_install_everywhere_skips_undetected() -> None:
detected = _FakeRegistrar("a", detected=True)
missing = _FakeRegistrar("b", detected=False)
results = install_everywhere(registrars=[detected, missing])
assert results["a"].status == RegisterStatus.REGISTERED
assert results["b"].status == RegisterStatus.NOT_DETECTED
assert len(detected.calls) == 1
assert len(missing.calls) == 0
def test_install_everywhere_filters_by_agents() -> None:
a = _FakeRegistrar("a")
b = _FakeRegistrar("b")
c = _FakeRegistrar("c")
results = install_everywhere(registrars=[a, b, c], agents=["a", "c"])
assert set(results) == {"a", "c"}
assert "b" not in results
def test_install_everywhere_passes_proxy_url_into_spec() -> None:
captured: list[ServerSpec] = []
class CapturingRegistrar(_FakeRegistrar):
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
captured.append(spec)
return RegisterResult(RegisterStatus.REGISTERED, "ok")
reg = CapturingRegistrar("x")
install_everywhere(proxy_url="http://localhost:9000", registrars=[reg])
assert len(captured) == 1
assert captured[0].env == {"HEADROOM_PROXY_URL": "http://localhost:9000"}
def test_install_everywhere_passes_force_flag() -> None:
captured: list[bool] = []
class CapturingRegistrar(_FakeRegistrar):
def register_server(self, spec: ServerSpec, *, force: bool = False) -> RegisterResult:
captured.append(force)
return RegisterResult(RegisterStatus.REGISTERED, "ok")
reg = CapturingRegistrar("x")
install_everywhere(registrars=[reg], force=True)
assert captured == [True]
def test_install_everywhere_returns_mismatch_results() -> None:
mismatched = _FakeRegistrar(
"a",
register_result=RegisterResult(RegisterStatus.MISMATCH, "env differs"),
)
results = install_everywhere(registrars=[mismatched])
assert results["a"].status == RegisterStatus.MISMATCH
assert results["a"].ok is False # mismatch is NOT a success