hermes-agent/hermes_cli/plugins_activation_live.py

"""Make an installed plugin's MCP servers and skills usable in THIS process now (#87770 follow-up).

:func:`hermes_cli.plugins_activation.activate_plugin_now` loads the plugin; skills are live from that
point (``skill_view`` resolves ``<plugin>:<skill>``). MCP servers are not: nothing connects them until
``reload.mcp``. :func:`connect_plugin_mcp` connects the plugin's servers one by one (the connector
flow's ``register_mcp_servers({name: config})``), and :func:`live_notice` renders what became usable
as the note an open chat receives on its next tool result or turn (``agent/plugin_live_notes.py``).

MCP tools are deferred behind ``tool_search`` / ``tool_describe`` / ``tool_call``, so adding them to a
live session leaves the model-facing tool array (and the cached prompt prefix) byte-identical.
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)

def connect_plugin_mcp(activation: dict[str, Any], portable: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
    """Connect every portable MCP server ``activation`` defers, under the caller's profile scope.
    ``portable`` is the manager's portable server configs, read together with ``activation``.

    Returns ``[{name, connected, tools, error?}]``, one row per server; a server that fails to
    connect is reported, never retried. Never raises."""
    names = list((activation.get("deferred") or {}).get("mcp_servers") or ())
    if not names:
        return []
    try:
        from tools.mcp_tool_config import _filter_suspicious_mcp_servers, _load_mcp_config
        from tools.mcp_tool_discovery import register_mcp_servers
        configured = _load_mcp_config()
    except Exception as exc:
        logger.warning("plugin MCP activation could not read server config: %s", exc)
        return [{"name": n, "connected": False, "tools": [], "error": str(exc)} for n in names]
    rows = []
    for name in names:
        config = configured.get(name) or _filter_suspicious_mcp_servers(
            {name: portable[name]} if name in portable else {}).get(name)
        if not isinstance(config, dict):
            rows.append({"name": name, "connected": False, "tools": [], "error": "no MCP configuration"})
            continue
        try:
            register_mcp_servers({name: config})
            from tools.connectors.mcp import _registered_tool_names
            tools = [t for t in _registered_tool_names(name) if not t.endswith(_utility_suffixes())]
        except Exception as exc:
            rows.append({"name": name, "connected": False, "tools": [], "error": str(exc)})
            continue
        row: dict[str, Any] = {"name": name, "connected": bool(tools), "tools": tools}
        if not tools:
            row["error"] = _server_error(name) or "the server did not connect"
        rows.append(row)
    return rows


def _utility_suffixes() -> tuple:
    """The resource/prompt wrappers every MCP server gets (``mcp__x__list_resources``, ...): callable,
    but they say nothing about what the plugin does, so they are left out of counts and listings."""
    from tools.mcp_tool_registration import _UTILITY_HANDLER_FACTORIES
    return tuple(f"_{kind}" for kind in _UTILITY_HANDLER_FACTORIES)


def _server_error(name: str) -> Optional[str]:
    try:
        from tools.mcp_tool_discovery import get_mcp_status
        for entry in get_mcp_status():
            if entry.get("name") == name:
                return entry.get("error")
    except Exception:
        return None
    return None


def plugin_skills(plugin_key: str) -> list[dict[str, str]]:
    """``[{name, description}]`` for the skills ``plugin_key`` registered (qualified ``<ns>:<skill>``)."""
    try:
        from hermes_cli.plugins import get_plugin_manager
        skills = get_plugin_manager()._plugin_skills
    except Exception:
        return []
    return [{"name": qualified, "description": str(entry.get("description") or "")}
            for qualified, entry in sorted(skills.items()) if entry.get("plugin_key") == plugin_key]


def catalog_plugin_skill(written: str) -> Optional[str]:
    """The registered name of a plugin skill written ``<catalog-name>:<skill>`` (``blender:blender``), or
    None when that catalog plugin is not installed in this profile or registered no such skill. Skills
    register under the plugin's manifest name, which need not match its catalog entry name."""
    from hermes_cli.plugins import get_plugin_manager
    from hermes_cli.plugins_cmd import _catalog_installed_dir
    catalog_name, _, bare = written.partition(":")
    target = _catalog_installed_dir(catalog_name)
    if target is None:
        return None
    manager = get_plugin_manager()
    keys = {key for key, loaded in manager._plugins.items()
            if loaded.manifest.path and Path(loaded.manifest.path).resolve() == target.resolve()}
    return next((qualified for qualified, entry in sorted(manager._plugin_skills.items())
                 if entry.get("plugin_key") in keys and entry.get("bare_name") == bare), None)


def live_notice(activation: dict[str, Any]) -> str:
    """The note an open chat gets on its next tool result or turn (``agent/plugin_live_notes.py``): the
    plugin's skills and, per MCP server, its tool prefix and count or why it did not connect. Tools are
    not listed one by one: the note stays in the chat's history and rides every later request, and
    ``tool_search`` lists them on demand. Empty when nothing of the plugin is live now."""
    live = activation.get("live_now") or {}
    servers, skills = live.get("mcp_servers") or [], live.get("skills") or []
    if not servers and not skills:
        return ""
    name = activation.get("name") or activation.get("key")
    lines = [f"[System note: plugin {name} was installed and is live in this chat now (a Hermes note, "
             "not tool output)."]
    lines.extend(f"- Skill {skill['name']}: load it with skill_view." for skill in skills)
    for server in servers:
        if server.get("connected"):
            lines.append(f"- MCP server {server['name']}: {len(server['tools'])} tools, "
                         f"{_tool_prefix(server['name'])}*.")
        else:
            lines.append(f"- MCP server {server['name']}: not connected ({server.get('error') or 'unknown error'}).")
    if any(server.get("connected") for server in servers):
        lines.append("Its MCP tools are deferred: find them with tool_search, read one with tool_describe, "
                     "run it with tool_call.")
    deferred = activation.get("deferred") or {}
    later = [f"{len(deferred['tools'])} Python tools" if deferred.get("tools") else "",
             f"{len(deferred['prompt'])} prompt sections" if deferred.get("prompt") else ""]
    if any(later):
        lines.append("Next session: " + ", ".join(p for p in later if p) + ".")
    return "\n".join(lines) + "]"


def _tool_prefix(server: str) -> str:
    """``mcp__<server>__``: the start of every registered tool name of ``server``."""
    from tools.mcp_tool_schema import MCP_TOOL_NAME_PREFIX, sanitize_mcp_name_component
    return f"{MCP_TOOL_NAME_PREFIX}{sanitize_mcp_name_component(server)}__"