"""Tool handlers — the code that runs when the LLM calls a wordvibe_* tool. Rules (per the Hermes plugin contract): 1. signature: def handler(args: dict, **kwargs) -> str 2. always return a JSON string — success and errors alike 3. never raise """ import json from .mcp_client import WordVibeMCP, load_sites, resolve_site, result_text # Keep tool results context-friendly. MAX_CHARS = 40000 def _ok(payload: dict) -> str: return json.dumps(payload, default=str) def _err(message: str, **extra) -> str: payload = {"error": message} payload.update(extra) return json.dumps(payload, default=str) def _clip(text: str, limit: int = MAX_CHARS) -> str: if len(text) <= limit: return text return text[:limit] + f"\n…[truncated at {limit} chars — narrow the query or use line ranges]" def _site_for(args: dict): """Resolve the requested site, or raise LookupError with a readable message.""" return resolve_site(args.get("site") or "") def _call_tool(site_ref: str, tool: str, arguments: dict) -> str: """Call one site tool and return clipped text (or an error string).""" try: site = resolve_site(site_ref) except LookupError as exc: return "ERROR: " + str(exc) client = WordVibeMCP(site) try: response = client.call_tool(tool, arguments or {}) except Exception as exc: # noqa: BLE001 - belt and braces; rpc() already guards return f"ERROR: {type(exc).__name__}: {exc}" return _clip(result_text(response)) # -------------------------------------------------------------------------- # Site management / discovery # -------------------------------------------------------------------------- def sites(args: dict, **kwargs) -> str: """List configured sites, optionally probing each MCP endpoint.""" try: configured = load_sites() except Exception as exc: # noqa: BLE001 return _err(f"Could not read site config: {exc}") if not configured: return _err( "No site configured. Set WORDVIBE_SITE_URL + WORDVIBE_MCP_KEY " "(or WORDVIBE_MCP_SITES for several) in the Hermes .env, then restart Hermes." ) probe = bool(args.get("probe")) out = [] for s in configured: row = {"name": s.name, "url": s.url, "endpoint": s.endpoint, "key_set": bool(s.key)} if probe: client = WordVibeMCP(s) init = client.initialize() if isinstance(init, dict) and init.get("error"): row["reachable"] = False row["error"] = (init.get("error") or {}).get("message", "unknown error") else: info = ((init or {}).get("result") or {}).get("serverInfo") or {} row["reachable"] = True row["server"] = info.get("name", "wordvibe-ide") row["plugin_version"] = info.get("version", "?") out.append(row) return _ok({"sites": out, "count": len(out)}) def site_info(args: dict, **kwargs) -> str: """MCP server identity + the site's own site_info tool.""" try: site = _site_for(args) except LookupError as exc: return _err(str(exc)) client = WordVibeMCP(site) init = client.initialize() if isinstance(init, dict) and init.get("error"): return _err( "Could not reach the MCP server on %s: %s" % (site.url, (init.get("error") or {}).get("message")), hint="Enable MCP Server Mode in WordVibe → Settings → MCP, and check WORDVIBE_MCP_KEY.", ) result = (init or {}).get("result") or {} payload = { "site": site.name, "url": site.url, "endpoint": site.endpoint, "mcp_protocol": result.get("protocolVersion"), "server": result.get("serverInfo"), } # The plugin also exposes a `site_info` tool with WordPress-level details. wp = client.call_tool("site_info", {}) wp_text = result_text(wp) if not wp_text.startswith("ERROR:"): payload["wordpress"] = _clip(wp_text, 6000) return _ok(payload) def tools(args: dict, **kwargs) -> str: """List the tools the site exposes over MCP (the plugin's agent tool registry).""" try: site = _site_for(args) except LookupError as exc: return _err(str(exc)) client = WordVibeMCP(site) response = client.list_tools() if isinstance(response, dict) and response.get("error"): return _err( "tools/list failed on %s: %s" % (site.url, (response.get("error") or {}).get("message")) ) items = ((response or {}).get("result") or {}).get("tools") or [] needle = (args.get("filter") or "").strip().lower() listed = [] for t in items: if not isinstance(t, dict): continue name = str(t.get("name") or "") desc = str(t.get("description") or "").strip() if needle and needle not in name.lower() and needle not in desc.lower(): continue listed.append({"name": name, "description": desc[:220]}) return _ok({"site": site.name, "count": len(listed), "tools": listed}) def call(args: dict, **kwargs) -> str: """Generic escape hatch: invoke any site tool by name.""" tool = str(args.get("tool") or "").strip() if not tool: return _err("Missing 'tool'. Use wordvibe_tools to list what the site exposes.") arguments = args.get("arguments") if arguments is None: arguments = {} if not isinstance(arguments, dict): return _err("'arguments' must be an object of the site tool's parameters.") return _call_tool(args.get("site") or "", tool, arguments) # -------------------------------------------------------------------------- # Convenience wrappers over the most useful site tools # -------------------------------------------------------------------------- def read_file(args: dict, **kwargs) -> str: path = str(args.get("path") or "").strip() if not path: return _err("Missing 'path'.") payload = {"path": path} lines = str(args.get("lines") or "").strip() if lines: payload["lines"] = lines return _call_tool(args.get("site") or "", "file_read", payload) def write_file(args: dict, **kwargs) -> str: path = str(args.get("path") or "").strip() content = args.get("content") if not path: return _err("Missing 'path'.") if not isinstance(content, str): return _err("Missing 'content' (must be the full file body as a string).") return _call_tool(args.get("site") or "", "file_write", {"path": path, "content": content}) def edit_file(args: dict, **kwargs) -> str: path = str(args.get("path") or "").strip() search = args.get("search") replace = args.get("replace") if not path or not isinstance(search, str) or replace is None: return _err("Need 'path', 'search' and 'replace'.") return _call_tool( args.get("site") or "", "file_edit", {"path": path, "search": search, "replace": replace} ) def search(args: dict, **kwargs) -> str: query = str(args.get("query") or "").strip() if not query: return _err("Missing 'query'.") payload = {"query": query} for key in ("path", "include"): val = str(args.get(key) or "").strip() if val: payload[key] = val if args.get("regex"): payload["regex"] = True return _call_tool(args.get("site") or "", "codebase_search", payload) def build_page(args: dict, **kwargs) -> str: """Build/update a Word Page Builder page (wraps the site's wpb_build_page tool).""" title = str(args.get("title") or "").strip() if not title: return _err("Missing 'title'.") sections = args.get("sections") if isinstance(sections, str): try: sections = json.loads(sections) except Exception as exc: # noqa: BLE001 return _err(f"'sections' must be a JSON array: {exc}") if not isinstance(sections, list) or not sections: return _err("'sections' must be a non-empty array of {kind, html}.") clean = [] for item in sections: if isinstance(item, str): clean.append({"kind": "custom", "html": item}) continue if not isinstance(item, dict) or "html" not in item: return _err("Each section needs {kind, html}.") clean.append({"kind": str(item.get("kind") or "custom"), "html": str(item.get("html") or "")}) payload = {"title": title, "sections": clean} for key in ("slug", "status"): val = str(args.get(key) or "").strip() if val: payload[key] = val if args.get("post_id"): try: payload["post_id"] = int(args["post_id"]) except Exception: # noqa: BLE001 return _err("'post_id' must be an integer.") return _call_tool(args.get("site") or "", "wpb_build_page", payload)