96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
# web/mcp_registrations.py
|
|
"""Read MCP server registrations from host config files mounted in /host."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
HOST_DIR = Path("/host")
|
|
|
|
|
|
def _redact_env_keys(env: dict[str, Any] | None) -> list[str]:
|
|
if not env or not isinstance(env, dict):
|
|
return []
|
|
return sorted(env.keys())
|
|
|
|
|
|
def _read_claude_registrations() -> list[dict[str, Any]]:
|
|
"""Read MCP registrations from /host/.claude.json."""
|
|
path = HOST_DIR / ".claude.json"
|
|
if not path.exists():
|
|
return []
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
logger.warning("claude_json_parse_failed: %s", e)
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
# Top-level mcpServers
|
|
for name, cfg in (data.get("mcpServers") or {}).items():
|
|
out.append(_normalize("Claude Code (global)", name, cfg))
|
|
# Per-project mcpServers
|
|
for project_path, project_cfg in (data.get("projects") or {}).items():
|
|
for name, cfg in (project_cfg.get("mcpServers") or {}).items():
|
|
out.append(_normalize(f"Claude Code ({project_path})", name, cfg))
|
|
return out
|
|
|
|
|
|
def _read_paperclip_registrations() -> list[dict[str, Any]]:
|
|
"""Read MCP registrations from /host/.paperclip/instances/*/mcp.json."""
|
|
base = HOST_DIR / ".paperclip" / "instances"
|
|
if not base.exists():
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for instance_dir in sorted(base.iterdir()):
|
|
if not instance_dir.is_dir():
|
|
continue
|
|
mcp_json = instance_dir / "mcp.json"
|
|
if not mcp_json.exists():
|
|
continue
|
|
try:
|
|
data = json.loads(mcp_json.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
logger.warning(
|
|
"paperclip_mcp_json_parse_failed: %s %s",
|
|
instance_dir.name, e,
|
|
)
|
|
continue
|
|
for name, cfg in (data.get("mcpServers") or data or {}).items():
|
|
if not isinstance(cfg, dict):
|
|
continue
|
|
out.append(_normalize(f"Paperclip ({instance_dir.name})", name, cfg))
|
|
return out
|
|
|
|
|
|
def _normalize(client: str, server_name: str, cfg: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"client": client,
|
|
"server_name": server_name,
|
|
"command": cfg.get("command", ""),
|
|
"args": cfg.get("args") or [],
|
|
"cwd": cfg.get("cwd") or cfg.get("workingDirectory") or "",
|
|
"env_keys": _redact_env_keys(cfg.get("env")),
|
|
"transport": cfg.get("transport") or "stdio",
|
|
}
|
|
|
|
|
|
def list_registrations() -> dict[str, Any]:
|
|
"""Return all MCP registrations + status."""
|
|
if not HOST_DIR.exists():
|
|
return {
|
|
"registrations": [],
|
|
"error": "host_path_unavailable",
|
|
"message": "תיקיית /host לא mounted. ראה runbook להגדרת volumes ב-Coolify.",
|
|
}
|
|
return {
|
|
"registrations": (
|
|
_read_claude_registrations() + _read_paperclip_registrations()
|
|
),
|
|
"error": None,
|
|
}
|