Initial asterisk-mcp: MCP server for vessel troubleshoot stack.

Part of GeneseasX golden-vessel tooling (see homelab-vault Projects/vessel-troubleshoot-stack).

src/asterisk_mcp/targets.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+from asterisk_mcp.client import AsteriskClient, ConnectionInfo
+from asterisk_mcp.config import settings
+from asterisk_mcp.guards import GuardError
+
+
+@dataclass
+class TargetConfig:
+    name: str
+    ssh_host: str
+    ssh_user: str = "root"
+    ssh_port: int = 22
+    ssh_key_path: str = ""
+    container: str = "asterisk"
+    description: str = ""
+    registered_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+
+    def to_connection(self) -> ConnectionInfo:
+        return ConnectionInfo(
+            ssh_host=self.ssh_host,
+            ssh_user=self.ssh_user,
+            ssh_port=self.ssh_port,
+            ssh_key_path=self.ssh_key_path,
+            container=self.container,
+        )
+
+    def public_view(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            **self.to_connection().public_view(),
+            "description": self.description,
+            "registered_at": self.registered_at,
+        }
+
+
+class TargetRegistry:
+    def __init__(self) -> None:
+        self._targets: dict[str, TargetConfig] = {}
+        self._active: str | None = None
+        self._loading: bool = False
+
+    @staticmethod
+    def _normalize_name(name: str) -> str:
+        name = re.sub(r"[^a-z0-9_-]+", "-", name.strip().lower())
+        name = re.sub(r"-+", "-", name).strip("-")
+        if not name:
+            raise GuardError("Target name must contain at least one alphanumeric character.")
+        return name
+
+    def register(
+        self,
+        *,
+        name: str = "",
+        ssh_host: str,
+        ssh_user: str = "root",
+        ssh_port: int = 22,
+        ssh_key_path: str = "",
+        container: str = "asterisk",
+        description: str = "",
+        registered_at: str = "",
+        set_active: bool = True,
+    ) -> TargetConfig:
+        if not ssh_host.strip():
+            raise GuardError("ssh_host is required.")
+        target_name = self._normalize_name(name) if name.strip() else self._normalize_name(ssh_host)
+        cfg_kwargs: dict[str, Any] = {
+            "name": target_name,
+            "ssh_host": ssh_host.strip(),
+            "ssh_user": ssh_user,
+            "ssh_port": ssh_port,
+            "ssh_key_path": ssh_key_path,
+            "container": container,
+            "description": description,
+        }
+        if registered_at.strip():
+            cfg_kwargs["registered_at"] = registered_at.strip()
+        cfg = TargetConfig(**cfg_kwargs)
+        self._targets[target_name] = cfg
+        if set_active:
+            self._active = target_name
+        self._persist()
+        return cfg
+
+    def unregister(self, name: str) -> None:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        del self._targets[key]
+        if self._active == key:
+            self._active = next(iter(self._targets), None)
+        self._persist()
+
+    def set_active(self, name: str) -> TargetConfig:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        self._active = key
+        self._persist()
+        return self._targets[key]
+
+    def active(self) -> TargetConfig | None:
+        if self._active and self._active in self._targets:
+            return self._targets[self._active]
+        return None
+
+    def list(self) -> list[dict[str, Any]]:
+        active = self._active
+        rows = []
+        for cfg in self._targets.values():
+            row = cfg.public_view()
+            row["active"] = cfg.name == active
+            rows.append(row)
+        return sorted(rows, key=lambda r: r["name"])
+
+    def resolve(
+        self,
+        *,
+        target: str = "",
+        ssh_host: str = "",
+        ssh_user: str = "",
+        ssh_port: int = 0,
+        ssh_key_path: str = "",
+        container: str = "",
+    ) -> TargetConfig:
+        if ssh_host.strip():
+            return TargetConfig(
+                name="__inline__",
+                ssh_host=ssh_host,
+                ssh_user=ssh_user or settings.ssh_user,
+                ssh_port=ssh_port or settings.ssh_port,
+                ssh_key_path=ssh_key_path or settings.ssh_key_path,
+                container=container or settings.container,
+            )
+        if target.strip():
+            return self.get(target)
+        active = self.active()
+        if active:
+            return active
+        if settings.ssh_host:
+            return TargetConfig(
+                name="__env__",
+                ssh_host=settings.ssh_host,
+                ssh_user=settings.ssh_user,
+                ssh_port=settings.ssh_port,
+                ssh_key_path=settings.ssh_key_path,
+                container=settings.container,
+            )
+        raise GuardError("No Asterisk target connected. Call asterisk_connect first.")
+
+    def _persist(self) -> None:
+        if self._loading:
+            return
+        from asterisk_mcp.targets_store import save_registry
+
+        save_registry(self)
+
+    def get(self, name: str) -> TargetConfig:
+        key = self._normalize_name(name)
+        if key not in self._targets:
+            raise GuardError(f"Unknown target '{name}'.")
+        return self._targets[key]
+
+
+registry = TargetRegistry()
+
+
+def _load_persisted_targets() -> None:
+    from asterisk_mcp.targets_store import load_registry
+
+    load_registry(registry)
+
+
+_load_persisted_targets()
+
+
+def resolve_connection(**kwargs: Any) -> ConnectionInfo:
+    return registry.resolve(**kwargs).to_connection()