Coverage for node / src / stigmem_node / cli / mcp.py: 56%

254 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-05-25 01:49 +0000

1"""MCP editor integration CLI handlers.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import datetime as dt 

7import json 

8import os 

9import shutil 

10import subprocess # nosec B404 - fixed-command local MCP diagnostics only. 

11from dataclasses import asdict, dataclass 

12from pathlib import Path 

13from typing import Literal 

14 

15ConfigFormat = Literal["toml", "json", "jsonc"] 

16ValidationTier = Literal["validated", "caveated", "experimental"] 

17DEFAULT_API_KEY_PLACEHOLDER = "<your-api-key>" 

18 

19 

20@dataclass(frozen=True) 

21class EditorConfig: 

22 editor: str 

23 label: str 

24 validation_tier: ValidationTier 

25 config_path: str 

26 config_format: ConfigFormat 

27 docs_link: str 

28 config_template: str 

29 

30 

31EDITOR_CONFIGS: dict[str, EditorConfig] = { 

32 "codex-cli": EditorConfig( 

33 editor="codex-cli", 

34 label="Codex CLI", 

35 validation_tier="validated", 

36 config_path="~/.codex/config.toml", 

37 config_format="toml", 

38 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/codex-cli", 

39 config_template="""[mcp_servers.stigmem] 

40command = "stigmem-mcp" 

41 

42[mcp_servers.stigmem.env] 

43STIGMEM_URL = "{stigmem_url}" 

44STIGMEM_API_KEY = "{stigmem_api_key}" 

45""", 

46 ), 

47 "claude-code": EditorConfig( 

48 editor="claude-code", 

49 label="Claude Code", 

50 validation_tier="validated", 

51 config_path="~/.claude/mcp_servers.json", 

52 config_format="json", 

53 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/claude-code", 

54 config_template="""{ 

55 "mcpServers": { 

56 "stigmem": { 

57 "command": "stigmem-mcp", 

58 "env": { 

59 "STIGMEM_URL": "{stigmem_url}", 

60 "STIGMEM_API_KEY": "{stigmem_api_key}" 

61 } 

62 } 

63 } 

64} 

65""", 

66 ), 

67 "gemini-cli": EditorConfig( 

68 editor="gemini-cli", 

69 label="Gemini CLI", 

70 validation_tier="caveated", 

71 config_path="~/.gemini/settings.json", 

72 config_format="json", 

73 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/gemini-cli", 

74 config_template="""{ 

75 "mcpServers": { 

76 "stigmem": { 

77 "command": "stigmem-mcp", 

78 "env": { 

79 "STIGMEM_URL": "{stigmem_url}", 

80 "STIGMEM_API_KEY": "{stigmem_api_key}" 

81 } 

82 } 

83 } 

84} 

85""", 

86 ), 

87 "continue-dev": EditorConfig( 

88 editor="continue-dev", 

89 label="Continue.dev", 

90 validation_tier="experimental", 

91 config_path="~/.continue/config.json", 

92 config_format="json", 

93 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/continue-dev", 

94 config_template="""{ 

95 "mcpServers": [ 

96 { 

97 "name": "stigmem", 

98 "command": "stigmem-mcp", 

99 "env": { 

100 "STIGMEM_URL": "{stigmem_url}", 

101 "STIGMEM_API_KEY": "{stigmem_api_key}" 

102 } 

103 } 

104 ] 

105} 

106""", 

107 ), 

108 "cursor": EditorConfig( 

109 editor="cursor", 

110 label="Cursor", 

111 validation_tier="experimental", 

112 config_path="~/.cursor/mcp.json", 

113 config_format="json", 

114 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/cursor", 

115 config_template="""{ 

116 "mcpServers": { 

117 "stigmem": { 

118 "command": "stigmem-mcp", 

119 "env": { 

120 "STIGMEM_URL": "{stigmem_url}", 

121 "STIGMEM_API_KEY": "{stigmem_api_key}" 

122 } 

123 } 

124 } 

125} 

126""", 

127 ), 

128 "zed": EditorConfig( 

129 editor="zed", 

130 label="Zed", 

131 validation_tier="experimental", 

132 config_path="~/.config/zed/settings.json", 

133 config_format="jsonc", 

134 docs_link="https://docs.stigmem.dev/en/latest/docs/integrations/mcp/zed", 

135 config_template="""{ 

136 "mcp_servers": { 

137 "stigmem": { 

138 "command": "stigmem-mcp", 

139 "env": { 

140 "STIGMEM_URL": "{stigmem_url}", 

141 "STIGMEM_API_KEY": "{stigmem_api_key}" 

142 } 

143 } 

144 } 

145} 

146""", 

147 ), 

148} 

149 

150 

151def editor_catalog() -> list[dict[str, str]]: 

152 """Return the supported MCP editor catalog as JSON-safe dictionaries.""" 

153 return [ 

154 { 

155 "editor": config.editor, 

156 "label": config.label, 

157 "validation_tier": config.validation_tier, 

158 "docs_link": config.docs_link, 

159 "config_path": config.config_path, 

160 "config_format": config.config_format, 

161 } 

162 for config in EDITOR_CONFIGS.values() 

163 ] 

164 

165 

166def _config_for(editor: str) -> EditorConfig | None: 

167 return EDITOR_CONFIGS.get(editor) 

168 

169 

170def _render_snippet(config: EditorConfig, stigmem_url: str, stigmem_api_key: str) -> str: 

171 return config.config_template.format( 

172 stigmem_url=stigmem_url, 

173 stigmem_api_key=stigmem_api_key, 

174 ) 

175 

176 

177def _json_stigmem_entry(stigmem_url: str, stigmem_api_key: str) -> dict[str, object]: 

178 return { 

179 "command": "stigmem-mcp", 

180 "env": { 

181 "STIGMEM_URL": stigmem_url, 

182 "STIGMEM_API_KEY": stigmem_api_key, 

183 }, 

184 } 

185 

186 

187def _merge_json_config( 

188 editor: str, 

189 existing: str, 

190 stigmem_url: str, 

191 stigmem_api_key: str, 

192 fallback: str, 

193) -> str: 

194 if not existing.strip(): 

195 return fallback 

196 try: 

197 data = json.loads(existing) 

198 except json.JSONDecodeError: 

199 return f"{existing.rstrip()}\n\n{fallback}" 

200 if not isinstance(data, dict): 

201 return fallback 

202 if editor == "continue-dev": 

203 current = data.get("mcpServers") 

204 server_list = current if isinstance(current, list) else [] 

205 server_list = [server for server in server_list if not _server_named_stigmem(server)] 

206 server_list.append({"name": "stigmem", **_json_stigmem_entry(stigmem_url, stigmem_api_key)}) 

207 data["mcpServers"] = server_list 

208 elif editor == "zed": 

209 current = data.get("mcp_servers") 

210 server_map = current if isinstance(current, dict) else {} 

211 server_map["stigmem"] = _json_stigmem_entry(stigmem_url, stigmem_api_key) 

212 data["mcp_servers"] = server_map 

213 else: 

214 current = data.get("mcpServers") 

215 server_map = current if isinstance(current, dict) else {} 

216 server_map["stigmem"] = _json_stigmem_entry(stigmem_url, stigmem_api_key) 

217 data["mcpServers"] = server_map 

218 return json.dumps(data, indent=2, sort_keys=True) + "\n" 

219 

220 

221def _server_named_stigmem(server: object) -> bool: 

222 return isinstance(server, dict) and server.get("name") == "stigmem" 

223 

224 

225def _merge_config( 

226 config: EditorConfig, 

227 existing: str, 

228 stigmem_url: str, 

229 stigmem_api_key: str, 

230) -> str: 

231 snippet = _render_snippet(config, stigmem_url, stigmem_api_key) 

232 if not existing.strip(): 

233 return snippet 

234 if config.config_format in {"json", "jsonc"}: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 return _merge_json_config(config.editor, existing, stigmem_url, stigmem_api_key, snippet) 

236 return f"{existing.rstrip()}\n\n# Added by stigmem mcp install at {_iso_timestamp()}\n{snippet}" 

237 

238 

239def _api_key_from_environment(api_key: str) -> bool: 

240 env_value = os.environ.get("STIGMEM_API_KEY", "") 

241 return bool(env_value and api_key == env_value) 

242 

243 

244def _render_dry_run_preview(config: EditorConfig, stigmem_url: str) -> str: 

245 if config.config_format == "toml": 245 ↛ 253line 245 didn't jump to line 253 because the condition on line 245 was always true

246 return f"""[mcp_servers.stigmem] 

247command = "stigmem-mcp" 

248 

249[mcp_servers.stigmem.env] 

250STIGMEM_URL = "{stigmem_url}" 

251# credential field omitted from dry-run output 

252""" 

253 if config.editor == "continue-dev": 

254 return json.dumps( 

255 { 

256 "mcpServers": [ 

257 { 

258 "name": "stigmem", 

259 "command": "stigmem-mcp", 

260 "env": { 

261 "STIGMEM_URL": stigmem_url, 

262 "_credential_omitted": True, 

263 }, 

264 } 

265 ] 

266 }, 

267 indent=2, 

268 sort_keys=True, 

269 ) 

270 key = "mcp_servers" if config.editor == "zed" else "mcpServers" 

271 return json.dumps( 

272 { 

273 key: { 

274 "stigmem": { 

275 "command": "stigmem-mcp", 

276 "env": { 

277 "STIGMEM_URL": stigmem_url, 

278 "_credential_omitted": True, 

279 }, 

280 } 

281 } 

282 }, 

283 indent=2, 

284 sort_keys=True, 

285 ) 

286 

287 

288def _iso_timestamp() -> str: 

289 return dt.datetime.now(dt.UTC).strftime("%Y-%m-%dT%H-%M-%SZ") 

290 

291 

292def _config_path(config: EditorConfig) -> Path: 

293 return Path(config.config_path).expanduser() 

294 

295 

296def _read_existing(path: Path) -> str: 

297 try: 

298 return path.read_text() 

299 except FileNotFoundError: 

300 return "" 

301 except OSError as exc: 

302 raise RuntimeError(f"cannot read existing config at {path}: {exc}") from exc 

303 

304 

305def _detect_editors() -> dict[str, dict[str, object]]: 

306 detected: dict[str, dict[str, object]] = {} 

307 for editor, config in EDITOR_CONFIGS.items(): 

308 path = _config_path(config) 

309 content = "" 

310 if path.exists(): 

311 try: 

312 content = path.read_text() 

313 except OSError: 

314 content = "" 

315 detected[editor] = { 

316 "label": config.label, 

317 "validation_tier": config.validation_tier, 

318 "config_path": str(path), 

319 "editor_dir_exists": path.parent.exists(), 

320 "config_exists": path.exists(), 

321 "has_stigmem_mcp": "stigmem-mcp" in content, 

322 } 

323 return detected 

324 

325 

326def _stigmem_mcp_version() -> str | None: 

327 binary = shutil.which("stigmem-mcp") 

328 if binary is None: 

329 return None 

330 try: 

331 result = subprocess.run( # noqa: S603 

332 [binary, "--version"], 

333 check=False, 

334 capture_output=True, 

335 text=True, 

336 timeout=5, 

337 ) # nosec B603 - fixed executable resolved from PATH for local diagnostics. 

338 except (FileNotFoundError, subprocess.TimeoutExpired): 

339 return None 

340 output = (result.stdout or result.stderr).strip() 

341 return output or None 

342 

343 

344def _mcp_report() -> dict[str, object]: 

345 binary = shutil.which("stigmem-mcp") 

346 return { 

347 "stigmem_mcp_on_path": binary is not None, 

348 "stigmem_mcp_path": binary, 

349 "stigmem_mcp_version": _stigmem_mcp_version() if binary else None, 

350 "stigmem_url_set": "STIGMEM_URL" in os.environ, 

351 "stigmem_api_key_set": "STIGMEM_API_KEY" in os.environ, 

352 "detected_editors": _detect_editors(), 

353 } 

354 

355 

356def _cmd_mcp_doctor(args: argparse.Namespace) -> int: 

357 report = _mcp_report() 

358 if args.json: 

359 print(json.dumps(report, indent=2, sort_keys=True)) 

360 return 0 

361 print("Stigmem MCP doctor:") 

362 print( 

363 " stigmem-mcp on PATH: " 

364 f"{'yes' if report['stigmem_mcp_on_path'] else 'no - run: npm install -g stigmem-mcp'}" 

365 ) 

366 if report["stigmem_mcp_path"]: 

367 print(f" path: {report['stigmem_mcp_path']}") 

368 if report["stigmem_mcp_version"]: 

369 print(f" version: {report['stigmem_mcp_version']}") 

370 print(f" STIGMEM_URL: {'set' if report['stigmem_url_set'] else 'unset'}") 

371 print(f" STIGMEM_API_KEY: {'set' if report['stigmem_api_key_set'] else 'unset'}") 

372 print("Detected editors:") 

373 for editor, state in sorted(_detect_editors().items()): 

374 marker = "configured" if state["has_stigmem_mcp"] else "not configured" 

375 print(f" {editor}: {marker} ({state['config_path']})") 

376 return 0 

377 

378 

379def _cmd_mcp_detect(args: argparse.Namespace) -> int: 

380 detected = _detect_editors() 

381 if args.json: 381 ↛ 384line 381 didn't jump to line 384 because the condition on line 381 was always true

382 print(json.dumps(detected, indent=2, sort_keys=True)) 

383 return 0 

384 for editor, state in sorted(detected.items()): 

385 if state["editor_dir_exists"] or state["config_exists"]: 

386 marker = "configured" if state["has_stigmem_mcp"] else "detected" 

387 print(f"{editor}: {marker} ({state['config_path']})") 

388 return 0 

389 

390 

391def _cmd_mcp_status(args: argparse.Namespace) -> int: 

392 detected = _detect_editors() 

393 if args.json: 

394 print(json.dumps(detected, indent=2, sort_keys=True)) 

395 return 0 

396 print("MCP integration status:") 

397 for editor, state in sorted(detected.items()): 

398 if state["config_exists"]: 

399 marker = "configured" if state["has_stigmem_mcp"] else "not configured" 

400 print(f" {editor}: {marker}") 

401 return 0 

402 

403 

404def _cmd_mcp_config(args: argparse.Namespace) -> int: 

405 if args.list: 

406 for editor_config in EDITOR_CONFIGS.values(): 

407 print( 

408 f"{editor_config.editor} " 

409 f"({editor_config.validation_tier}) - {editor_config.docs_link}" 

410 ) 

411 return 0 

412 config = _config_for(args.editor or "") 

413 if config is None: 

414 print(f"error: unknown MCP editor: {args.editor}") 

415 return 1 

416 print(f"Editor: {config.editor}") 

417 print(f"Config path: {config.config_path}") 

418 print(f"Config format: {config.config_format}") 

419 print(f"Guide: {config.docs_link}") 

420 print(f"Preview the planned Stigmem MCP entry: stigmem mcp install {config.editor} --dry-run") 

421 print(f"Apply with backup: stigmem mcp install {config.editor} --write") 

422 print( 

423 "This command intentionally does not print a config snippet; see the guide " 

424 "for manual configuration." 

425 ) 

426 return 0 

427 

428 

429def _cmd_mcp_install(args: argparse.Namespace) -> int: 

430 config = _config_for(args.editor) 

431 if config is None: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true

432 print(f"error: unknown MCP editor: {args.editor}") 

433 return 1 

434 path = _config_path(config) 

435 try: 

436 existing = _read_existing(path) 

437 except RuntimeError as exc: 

438 print(f"error: {exc}") 

439 return 1 

440 if "stigmem-mcp" in existing and not args.force: 

441 print(f"error: {path} already references stigmem-mcp; pass --force to replace it") 

442 return 1 

443 merged = _merge_config(config, existing, args.stigmem_url, args.stigmem_api_key) 

444 backup_dir = Path(args.backup_dir).expanduser() if args.backup_dir else path.parent 

445 backup_path = backup_dir / f"{path.name}.stigmem-bak-{_iso_timestamp()}" 

446 action = "WRITE" if args.write else "DRY-RUN" 

447 print(f"Editor: {config.editor}") 

448 print(f"Config path: {path}") 

449 print(f"Backup path: {backup_path}") 

450 print(f"Action: {action}") 

451 print(f"Guide: {config.docs_link}") 

452 print("Planned change: add or replace the stigmem MCP server entry.") 

453 print(f"Credential value is not echoed and will be written to: {path}") 

454 if _api_key_from_environment(args.stigmem_api_key): 

455 print("Credential source: environment variable.") 

456 print("Press Ctrl-C if you did not intend to embed that value in this file.") 

457 elif args.stigmem_api_key == DEFAULT_API_KEY_PLACEHOLDER: 

458 print("Credential source: placeholder. Edit the file before use.") 

459 else: 

460 print("Credential source: command-line flag.") 

461 if not args.write: 

462 preview = _render_dry_run_preview(config, args.stigmem_url) 

463 print() 

464 print("--- planned stigmem MCP server entry (dry-run; auth key omitted) ---") 

465 print(preview, end="" if preview.endswith("\n") else "\n") 

466 print("--- end planned stigmem MCP server entry ---") 

467 print() 

468 print( 

469 "Dry-run only. Re-run with --write to merge this entry into the editor config " 

470 "with the configured credential value." 

471 ) 

472 return 0 

473 if os.isatty(0) and not args.yes: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true

474 answer = input("Apply this change? [y/N] ").strip().lower() 

475 if answer not in {"y", "yes"}: 

476 print("Aborted.") 

477 return 0 

478 path.parent.mkdir(parents=True, exist_ok=True) 

479 if existing: 

480 backup_dir.mkdir(parents=True, exist_ok=True) 

481 backup_path.write_text(existing) 

482 if os.name == "posix": 482 ↛ 484line 482 didn't jump to line 484 because the condition on line 482 was always true

483 backup_path.chmod(0o600) 

484 print(f"Backed up existing config to {backup_path} (owner-only mode)") 

485 path.write_text(merged) 

486 print(f"Wrote {path}") 

487 print(f"Verify with: stigmem mcp smoke {config.editor}") 

488 return 0 

489 

490 

491def _cmd_mcp_smoke(args: argparse.Namespace) -> int: 

492 config = _config_for(args.editor) 

493 if config is None: 493 ↛ 494line 493 didn't jump to line 494 because the condition on line 493 was never true

494 print(f"error: unknown MCP editor: {args.editor}") 

495 return 1 

496 path = _config_path(config) 

497 try: 

498 content = path.read_text() 

499 except FileNotFoundError: 

500 print(f"error: editor config not found at {path}") 

501 return 1 

502 except OSError as exc: 

503 print(f"error: cannot read editor config at {path}: {exc}") 

504 return 1 

505 if "stigmem-mcp" not in content: 505 ↛ 506line 505 didn't jump to line 506 because the condition on line 505 was never true

506 print(f"error: editor config at {path} does not reference stigmem-mcp") 

507 return 1 

508 if shutil.which("stigmem-mcp") is None: 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true

509 print("error: stigmem-mcp is not on PATH; run: npm install -g stigmem-mcp") 

510 return 1 

511 smoke_script = ( 

512 Path(__file__).resolve().parents[4] / "adapters" / "mcp" / "tests" / "smoke.sh" 

513 ) 

514 if smoke_script.exists(): 514 ↛ 523line 514 didn't jump to line 523 because the condition on line 514 was always true

515 result = subprocess.run( # noqa: S603 

516 [str(smoke_script)], 

517 check=False, 

518 ) # nosec B603 - fixed repo-local smoke script. 

519 if result.returncode != 0: 519 ↛ 520line 519 didn't jump to line 520 because the condition on line 519 was never true

520 print(f"error: MCP protocol smoke failed with exit {result.returncode}") 

521 return result.returncode 

522 else: 

523 print("repo protocol smoke script not bundled; checked config and binary only") 

524 print(f"MCP smoke passed for {config.editor}") 

525 return 0 

526 

527 

528def catalog_asdict() -> list[dict[str, object]]: 

529 return [asdict(config) for config in EDITOR_CONFIGS.values()]