admin
ab1cfc88da
Hermes Agent plugin that drives WordPress sites running the WordVibe plugin over the site MCP server (POST /wp-json/wp-ide/v1/mcp, Bearer key from Settings -> MCP). 8 lean tools (sites, site_info, tools, call, read_file, write_file, edit_file, search) + /wordvibe slash command + bundled skill. The site exposes ~100 tools; the rest are reachable via wordvibe_call to keep the per-turn schema small. Verified against a mock of the plugin JSON-RPC server: 15/15 scenarios (initialize, tools/list, tools/call, blocked tools, unknown tool, 403 when MCP off, connection refused, plain-permalink ?rest_route= fallback).
215 lines
7.3 KiB
Python
215 lines
7.3 KiB
Python
"""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)
|