"""Kubernetes Discovery Source. Discovers MCP mcp_servers from Kubernetes pods/services using annotations. Supports both in-cluster and out-of-cluster configuration. Annotation Prefix: mcp-hangar.io/* Example Pod Annotations: mcp-hangar.io/enabled: "true" mcp-hangar.io/name: "my-mcp_server" mcp-hangar.io/mode: "http" mcp-hangar.io/port: "8080" mcp-hangar.io/group: "data-team" mcp-hangar.io/health-path: "/health" mcp-hangar.io/path: "/mcp" """ from dataclasses import dataclass from mcp_hangar.domain.discovery.discovered_mcp_server import DiscoveredMcpServer from mcp_hangar.domain.discovery.discovery_source import DiscoveryMode, DiscoverySource, SourcePolicyViolation from ...logging_config import get_logger logger = get_logger(__name__) # Optional Kubernetes dependency try: from kubernetes import client, config from kubernetes.client.rest import ApiException KUBERNETES_AVAILABLE = True except ImportError: KUBERNETES_AVAILABLE = False ApiException = Exception # Fallback for type hints client = None config = None logger.debug("kubernetes package not installed, KubernetesDiscoverySource unavailable") @dataclass(frozen=True) class NamespacePolicy: """Which namespaces this source may take servers from. A value object rather than four attributes on the source, for one practical reason: the source itself cannot be constructed without the `kubernetes` package, which is optional and absent from CI. Testing these rules through the source would mean skipping them there -- and "green because it never ran" is the worst possible answer for a security rule that just moved house. Denied wins over allowed, as it did when the core owned these checks. """ allowed: frozenset[str] = frozenset() denied: frozenset[str] = frozenset({"kube-system", "default"}) def violation(self, namespace: str) -> SourcePolicyViolation | None: """The reason to refuse this namespace, or None to accept it.""" if namespace in self.denied: return SourcePolicyViolation( reason=f"Namespace '{namespace}' is in denied list", details={"namespace": namespace, "denied_namespaces": sorted(self.denied)}, ) if self.allowed and namespace not in self.allowed: return SourcePolicyViolation( reason=f"Namespace '{namespace}' is not in allowed list", details={"namespace": namespace, "allowed_namespaces": sorted(self.allowed)}, ) return None class KubernetesDiscoverySource(DiscoverySource): """Discover MCP mcp_servers from Kubernetes pods/services. Uses pod annotations to discover and configure MCP mcp_servers. Supports namespace filtering and label selectors. Attributes: ANNOTATION_PREFIX: Prefix for all MCP annotations """ ANNOTATION_PREFIX = "mcp-hangar.io/" def __init__( self, mode: DiscoveryMode = DiscoveryMode.AUTHORITATIVE, namespaces: list[str] | None = None, label_selector: str | None = None, in_cluster: bool = True, kubeconfig_path: str | None = None, default_ttl: int = 90, allowed_namespaces: set[str] | None = None, denied_namespaces: set[str] | None = None, ): """Initialize Kubernetes discovery source. Args: mode: Discovery mode (default: authoritative for K8s) namespaces: List of namespaces to watch (None = all) label_selector: Kubernetes label selector in_cluster: Whether running inside cluster kubeconfig_path: Path to kubeconfig (for out-of-cluster) default_ttl: Default TTL for discovered mcp_servers """ super().__init__(mode) if not KUBERNETES_AVAILABLE: raise ImportError( "kubernetes package is required for KubernetesDiscoverySource. Install with: pip install kubernetes" ) self.namespaces = namespaces or [] self.label_selector = label_selector self.in_cluster = in_cluster self.kubeconfig_path = kubeconfig_path self.default_ttl = default_ttl # This source's own policy, in this source's own vocabulary. It used to # live in the core's SecurityConfig and run behind an # `if source_type == "kubernetes"` -- so a security component knew the # names of sources, and a new source either escaped these checks # silently or made its author edit security code. self._namespace_policy = NamespacePolicy( allowed=frozenset(allowed_namespaces or ()), denied=frozenset(denied_namespaces) if denied_namespaces is not None else NamespacePolicy.denied, ) self._v1: client.CoreV1Api | None = None self._initialized = False def _ensure_initialized(self) -> None: """Ensure Kubernetes client is initialized.""" if self._initialized: return try: if self.in_cluster: config.load_incluster_config() else: config.load_kube_config(config_file=self.kubeconfig_path) self._v1 = client.CoreV1Api() self._initialized = True logger.info( f"Kubernetes discovery initialized " f"(in_cluster={self.in_cluster}, namespaces={self.namespaces or 'all'})" ) except Exception as e: # noqa: BLE001 -- infra-boundary: k8s init failure is non-fatal logger.error(f"Failed to initialize Kubernetes client: {e}") raise @property def source_type(self) -> str: return "kubernetes" async def discover(self) -> list[DiscoveredMcpServer]: """Discover mcp_servers from pod annotations. Returns: List of discovered mcp_servers """ self._ensure_initialized() mcp_servers = [] namespaces = self.namespaces or await self._get_all_namespaces() for namespace in namespaces: try: assert self._v1 is not None pods = self._v1.list_namespaced_pod(namespace=namespace, label_selector=self.label_selector) for pod in pods.items: mcp_server = self._parse_pod(pod, namespace) if mcp_server: mcp_servers.append(mcp_server) await self.on_mcp_server_discovered(mcp_server) except ApiException as e: logger.warning(f"Failed to list pods in {namespace}: {e.reason}") except Exception as e: # noqa: BLE001 -- infra-boundary: skip namespace on discovery error logger.error(f"Error discovering in namespace {namespace}: {e}") logger.debug(f"Kubernetes discovery found {len(mcp_servers)} mcp_servers") return mcp_servers def _parse_pod(self, pod, namespace: str) -> DiscoveredMcpServer | None: """Parse pod annotations into DiscoveredMcpServer. Args: pod: Kubernetes pod object namespace: Pod namespace Returns: DiscoveredMcpServer or None if not MCP-enabled """ annotations = pod.metadata.annotations or {} # Check if MCP discovery is enabled enabled = annotations.get(f"{self.ANNOTATION_PREFIX}enabled", "false") if enabled.lower() != "true": return None # Extract mcp_server config name = annotations.get(f"{self.ANNOTATION_PREFIX}name", pod.metadata.name) mode = annotations.get(f"{self.ANNOTATION_PREFIX}mode", "http") port = annotations.get(f"{self.ANNOTATION_PREFIX}port", "8080") group = annotations.get(f"{self.ANNOTATION_PREFIX}group") health_path = annotations.get(f"{self.ANNOTATION_PREFIX}health-path", "/health") # The MCP endpoint's path (#1208). Without this the built endpoint was # always `http://:` with nothing after it -- a 404 against # every server mounted the way the SDK's own reference server and # `mcp-proxy` both default to. Defaults to that same convention. path = annotations.get(f"{self.ANNOTATION_PREFIX}path", "/mcp") ttl = int(annotations.get(f"{self.ANNOTATION_PREFIX}ttl", str(self.default_ttl))) # Get pod IP pod_ip = pod.status.pod_ip if pod.status else None if not pod_ip: logger.debug(f"Pod {pod.metadata.name} has no IP, skipping") return None # Check pod phase phase = pod.status.phase if pod.status else "Unknown" if phase != "Running": logger.debug(f"Pod {pod.metadata.name} not running (phase={phase}), skipping") return None # Build connection info connection_info = { "host": pod_ip, "port": int(port), "path": path, "health_path": health_path, } # Handle subprocess mode if mode == "subprocess" or mode == "stdio": command = annotations.get(f"{self.ANNOTATION_PREFIX}command") if command: connection_info["command"] = command.split() metadata = { "namespace": namespace, "pod_name": pod.metadata.name, "pod_uid": pod.metadata.uid, "group": group, "labels": pod.metadata.labels or {}, "annotations": {k: v for k, v in annotations.items() if k.startswith(self.ANNOTATION_PREFIX)}, "node_name": pod.spec.node_name if pod.spec else None, "phase": phase, # What the API server says this pod's address is. Registration # refuses an endpoint that resolves anywhere else, so an annotation # cannot point Hangar at a neighbouring pod or an internal service # and ride discovery's trust to it (#771). "runtime_addresses": [pod_ip], } return DiscoveredMcpServer.create( name=name, source_type=self.source_type, mode=mode, connection_info=connection_info, metadata=metadata, ttl_seconds=ttl, ) async def health_check(self) -> bool: """Check Kubernetes API availability. Returns: True if API is accessible """ try: self._ensure_initialized() assert self._v1 is not None self._v1.get_api_resources() return True except Exception as e: # noqa: BLE001 -- infra-boundary: health check returns unhealthy on error logger.warning(f"Kubernetes health check failed: {e}") return False async def _get_all_namespaces(self) -> list[str]: """Get all namespace names. Returns: List of namespace names """ try: assert self._v1 is not None namespaces = self._v1.list_namespace() return [ns.metadata.name for ns in namespaces.items] except ApiException as e: logger.error(f"Failed to list namespaces: {e.reason}") return [] async def start(self) -> None: """Start the Kubernetes discovery source.""" self._ensure_initialized() logger.info("Kubernetes discovery source started") async def stop(self) -> None: """Stop the Kubernetes discovery source.""" self._initialized = False self._v1 = None logger.info("Kubernetes discovery source stopped") def policy_violation(self, mcp_server: DiscoveredMcpServer) -> SourcePolicyViolation | None: """Refuse a server from a namespace this source may not take from.""" return self._namespace_policy.violation(mcp_server.metadata.get("namespace", ""))