"""Application-backed MCP liveness parsing and user-facing status descriptions.""" from __future__ import annotations import logging from dataclasses import dataclass, replace from typing import Any, Literal from hermes_platform import declaration from hermes_platform.host import facts from hermes_platform.resolver.app import AppDef, AppResolver from hermes_platform.resolver.availability import Availability, availability from hermes_platform.resolver.base import Effort from hermes_platform.resolver.core import CheckState logger = logging.getLogger(__name__) LivenessKind = Literal["static", "server_json", "interactive_session"] LivenessState = Literal[ "app_not_running", "hermes_not_connected", "endpoint_unavailable", "no_interactive_session", "version_too_old", "missing_app", ] Retry = Literal["after_user_action", "never_here"] @dataclass(frozen=True) class Liveness: kind: LivenessKind path: str = "" url_field: str = "http" token_field: str = "token" pid_field: str = "pid" def app_definition(self, definition: AppDef) -> AppDef: if self.kind != "server_json": return definition return replace( definition, liveness_kind="server_json", liveness_path=self.path, liveness_pid_key=self.pid_field, liveness_url_key=self.url_field, liveness_token_key=self.token_field, ) @dataclass(frozen=True) class Status: state: LivenessState availability: Availability liveness: Liveness user_action: str retry: Retry def parse_liveness(raw: Any) -> Liveness: """Parse one portable-plugin liveness declaration.""" if not isinstance(raw, dict): raise ValueError("liveness must be an object") kind = raw.get("kind") if kind == "static": if set(raw) != {"kind"}: raise ValueError("static liveness only accepts 'kind'") return Liveness("static") if kind == "interactive_session": if set(raw) != {"kind"}: raise ValueError("interactive_session liveness only accepts 'kind'") return Liveness("interactive_session") if kind != "server_json": raise ValueError(f"unknown liveness kind: {kind!r}") if set(raw) - {"kind", "path", "fields"}: raise ValueError("server_json liveness has unknown fields") path = raw.get("path") fields = {"url": "http", "token": "token", "pid": "pid", **(raw.get("fields") or {})} if not isinstance(path, str) or not path.strip(): raise ValueError("server_json liveness requires a non-empty path") if set(fields) != {"url", "token", "pid"}: raise ValueError("server_json liveness fields may only override url, token, and pid") if any(not isinstance(value, str) or not value.strip() for value in fields.values()): raise ValueError("server_json liveness field names must be non-empty strings") return Liveness( "server_json", path=path.strip(), url_field=fields["url"].strip(), token_field=fields["token"].strip(), pid_field=fields["pid"].strip(), ) def liveness_for(server_name: str) -> Liveness: """Return a server's registered liveness declaration, defaulting to static.""" try: from hermes_cli.agent_plugins import liveness_for as registered_liveness except ImportError: return Liveness("static") raw = registered_liveness(server_name) if raw is None: return Liveness("static") try: return parse_liveness(raw) except ValueError as exc: logger.warning("MCP server '%s' has an invalid liveness declaration (%s); treating it as static", server_name, exc) return Liveness("static") def _action(state: LivenessState, app_name: str) -> tuple[str, Retry]: actions: dict[LivenessState, tuple[str, Retry]] = { "app_not_running": (f"Start {app_name}, then try again.", "after_user_action"), "hermes_not_connected": (f"Reconnect {app_name} in Hermes, then try again.", "after_user_action"), "endpoint_unavailable": (f"Open {app_name} and enable its local connection, then try again.", "after_user_action"), "no_interactive_session": (f"Open an interactive desktop session and start {app_name}, then try again.", "never_here"), "version_too_old": (f"Update {app_name}, then try again.", "after_user_action"), "missing_app": (f"Install {app_name}, then try again.", "after_user_action"), } return actions[state] def status(server_name: str) -> Status | None: """Return the current unavailable state for a registered declaration.""" decl = declaration.lookup(server_name) if decl is None: return None available = availability(decl) live = liveness_for(server_name) if available.state in {"missing_app", "unsupported_os"}: state: LivenessState = "missing_app" elif available.state == "version_too_old": state = "version_too_old" elif live.kind == "interactive_session" and not facts.interactive_session(): state = "no_interactive_session" elif live.kind == "server_json": definition = decl.app_for(facts.os_family()) if definition is None: state = "missing_app" else: probe = AppResolver(live.app_definition(definition)).probe( AppResolver(live.app_definition(definition)).locate(), effort=Effort.LOCAL ) if probe.running.value is not True: state = "app_not_running" elif probe.endpoint.state is not CheckState.PRESENT: state = "endpoint_unavailable" else: # The app runs and its endpoint answers: the only thing missing is Hermes' own # MCP connection to it. Telling the user to "start" an app that IS running is # the wrong instruction (#119975). state = "hermes_not_connected" else: # static / unknown liveness kinds cannot observe the app, so they cannot conclude it # is not running; the honest answer is that Hermes is not connected (#119975). state = "hermes_not_connected" action, retry = _action(state, decl.name) return Status(state, available, live, action, retry) def describe(decl: declaration.Declaration, available: Availability, liveness_state: LivenessState, display_name: str | None = None) -> str: """Compose one unavailable-state sentence with one user action. ``display_name`` overrides the declaration's slug when the caller knows the plugin's catalog/manifest title — the Plugins tab does, and a raw slug reads like an error code.""" app_name = display_name or decl.name action, _retry = _action(liveness_state, app_name) if liveness_state == "missing_app": reason = f"{app_name} is not installed." elif liveness_state == "version_too_old": found = f" version {available.version}" if available.version else "" minimum = f"; version {available.min_version} or newer is required" if available.min_version else "" reason = f"{app_name}{found} is too old{minimum}." elif liveness_state == "no_interactive_session": reason = f"{app_name} needs an interactive desktop session." elif liveness_state == "endpoint_unavailable": reason = f"{app_name}'s local endpoint is unavailable." elif liveness_state == "hermes_not_connected": reason = f"{app_name}'s MCP connection is missing." else: reason = f"{app_name} is not running." return f"{reason} {action}" def unavailable_details(server_name: str) -> tuple[declaration.Declaration, Status, str] | None: """Return declaration, structured state, and its composed sentence.""" decl = declaration.lookup(server_name) current = status(server_name) if decl is None or current is None: return None return decl, current, describe(decl, current.availability, current.state)