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).
302 lines
10 KiB
Python
302 lines
10 KiB
Python
"""Minimal JSON-RPC 2.0 client for the WordVibe MCP server (stdlib only).
|
|
|
|
The WordVibe WordPress plugin ships an MCP server at:
|
|
|
|
POST {site}/wp-json/wp-ide/v1/mcp (pretty permalinks)
|
|
POST {site}/index.php?rest_route=/wp-ide/v1/mcp (plain permalinks)
|
|
Auth: Authorization: Bearer <wv_mcp_server_key>
|
|
Gate: option wv_mcp_server_enabled must be true (WordVibe → Settings → MCP)
|
|
|
|
Both URL shapes are accepted, and a 404 on the /wp-json/ form automatically
|
|
retries via ?rest_route= (the form the settings screen shows on plain-permalink
|
|
sites such as flexi.com.et).
|
|
|
|
Supported methods: initialize, initialized, tools/list, tools/call, ping.
|
|
Tools exposed = the plugin's own agent tool registry (`wv_ai_tools()`),
|
|
minus `php_eval` and `db_mutate`, which the server blocks remotely.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
MCP_PATH = "/wp-json/wp-ide/v1/mcp"
|
|
MCP_REST_ROUTE = "index.php?rest_route=/wp-ide/v1/mcp"
|
|
PROTOCOL_VERSION = "2024-11-05"
|
|
DEFAULT_TIMEOUT = 120
|
|
|
|
|
|
def _endpoint(site_url: str) -> str:
|
|
"""Normalise a site root (or a full MCP URL) into the MCP endpoint.
|
|
|
|
Accepts every form the WordVibe settings screen hands out:
|
|
https://site.com
|
|
https://site.com/wp-json/wp-ide/v1/mcp
|
|
https://site.com/index.php?rest_route=/wp-ide/v1/mcp (plain permalinks)
|
|
"""
|
|
url = (site_url or "").strip().rstrip("/")
|
|
if not url:
|
|
return ""
|
|
if "rest_route=" in url or url.endswith("/mcp") or "/wp-json/" in url:
|
|
return url
|
|
return url + MCP_PATH
|
|
|
|
|
|
def _alt_endpoint(endpoint: str) -> str:
|
|
"""Plain-permalink fallback for a /wp-json/ endpoint (404 → ?rest_route=)."""
|
|
if "/wp-json/" in endpoint:
|
|
root = endpoint.split("/wp-json/", 1)[0]
|
|
return f"{root.rstrip('/')}/{MCP_REST_ROUTE}"
|
|
return ""
|
|
|
|
|
|
class Site:
|
|
"""One configured WordPress site."""
|
|
|
|
def __init__(self, name: str, url: str, key: str, timeout: int = DEFAULT_TIMEOUT):
|
|
self.name = name
|
|
self.url = url
|
|
self.key = key
|
|
self.timeout = timeout
|
|
# Set once the ?rest_route= fallback has proven to be the working form.
|
|
self._use_alt = False
|
|
|
|
@property
|
|
def endpoint(self) -> str:
|
|
return _endpoint(self.url)
|
|
|
|
@property
|
|
def active_endpoint(self) -> str:
|
|
if self._use_alt:
|
|
alt = _alt_endpoint(self.endpoint)
|
|
if alt:
|
|
return alt
|
|
return self.endpoint
|
|
|
|
def __repr__(self) -> str: # pragma: no cover - debug helper
|
|
return f"<Site {self.name} {self.active_endpoint}>"
|
|
|
|
|
|
def load_sites() -> list:
|
|
"""Read site config from the environment.
|
|
|
|
Preferred (multi-site):
|
|
WORDVIBE_MCP_SITES='[{"name":"client-a","url":"https://a.com","key":"..."},
|
|
{"name":"client-b","url":"https://b.com","key":"..."}]'
|
|
|
|
Simple (single site):
|
|
WORDVIBE_SITE_URL=https://a.com
|
|
WORDVIBE_MCP_KEY=...
|
|
"""
|
|
sites = []
|
|
|
|
raw = (os.getenv("WORDVIBE_MCP_SITES") or "").strip()
|
|
if raw:
|
|
try:
|
|
data = json.loads(raw)
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
for i, item in enumerate(data or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
url = str(item.get("url") or "").strip()
|
|
if not url:
|
|
continue
|
|
sites.append(
|
|
Site(
|
|
name=str(item.get("name") or f"site{i + 1}").strip(),
|
|
url=url,
|
|
key=str(item.get("key") or item.get("token") or "").strip(),
|
|
timeout=int(item.get("timeout") or DEFAULT_TIMEOUT),
|
|
)
|
|
)
|
|
except Exception: # noqa: BLE001 - malformed JSON must not break the plugin
|
|
pass
|
|
|
|
if not sites:
|
|
url = (os.getenv("WORDVIBE_SITE_URL") or "").strip()
|
|
key = (os.getenv("WORDVIBE_MCP_KEY") or "").strip()
|
|
if url:
|
|
sites.append(Site(name=os.getenv("WORDVIBE_SITE_NAME") or "default", url=url, key=key))
|
|
|
|
return sites
|
|
|
|
|
|
def resolve_site(site: str = "") -> Site:
|
|
"""Resolve a site reference (name or URL) to a configured Site.
|
|
|
|
Falls back to the first configured site when `site` is empty.
|
|
Raises LookupError when nothing matches — callers turn that into JSON.
|
|
"""
|
|
sites = load_sites()
|
|
if not sites:
|
|
raise LookupError(
|
|
"No WordVibe site configured. Set WORDVIBE_SITE_URL + WORDVIBE_MCP_KEY "
|
|
"(or WORDVIBE_MCP_SITES for multiple sites) and restart Hermes."
|
|
)
|
|
|
|
want = (site or "").strip().lower()
|
|
if not want:
|
|
return sites[0]
|
|
|
|
for s in sites:
|
|
if s.name.lower() == want or s.url.lower().rstrip("/") == want.rstrip("/"):
|
|
return s
|
|
for s in sites:
|
|
if want in s.url.lower() or want in s.name.lower():
|
|
return s
|
|
|
|
known = ", ".join(f"{s.name} ({s.url})" for s in sites)
|
|
raise LookupError(f"Unknown site '{site}'. Configured sites: {known}")
|
|
|
|
|
|
def _decode_body(raw: bytes) -> str:
|
|
"""Decode a JSON or SSE-framed response body."""
|
|
text = raw.decode("utf-8", "replace").strip()
|
|
if text.startswith("event:") or text.startswith("data:"):
|
|
for line in text.splitlines():
|
|
if line.startswith("data:"):
|
|
return line[5:].strip()
|
|
return text
|
|
|
|
|
|
class WordVibeMCP:
|
|
"""JSON-RPC client bound to one site."""
|
|
|
|
def __init__(self, site: Site):
|
|
self.site = site
|
|
self._next_id = 1
|
|
|
|
# -- transport ---------------------------------------------------------
|
|
def _post(self, url: str, payload: dict) -> dict:
|
|
"""POST one JSON-RPC request to `url` and return the parsed response."""
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream",
|
|
"User-Agent": "wordvibe-hermes/1.0",
|
|
}
|
|
if self.site.key:
|
|
headers["Authorization"] = f"Bearer {self.site.key}"
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(
|
|
req, timeout=self.site.timeout, context=ssl.create_default_context()
|
|
) as resp:
|
|
status = resp.status
|
|
body = _decode_body(resp.read())
|
|
except urllib.error.HTTPError as exc:
|
|
detail = ""
|
|
try:
|
|
detail = _decode_body(exc.read())[:400]
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
hint = ""
|
|
if exc.code == 403:
|
|
hint = " (MCP Server Mode is probably off: enable it in WordVibe → Settings → MCP)"
|
|
elif exc.code == 401:
|
|
hint = " (check the WORDVIBE_MCP_KEY bearer token)"
|
|
return {
|
|
"error": {
|
|
"code": exc.code,
|
|
"message": f"HTTP {exc.code}{hint} {detail}".strip(),
|
|
"_url": url,
|
|
}
|
|
}
|
|
except Exception as exc: # noqa: BLE001 - network/SSL/timeout
|
|
return {"error": {"code": -1, "message": f"{type(exc).__name__}: {exc}", "_url": url}}
|
|
|
|
if not body:
|
|
return {"error": {"code": status, "message": "Empty response body", "_url": url}}
|
|
|
|
try:
|
|
return json.loads(body)
|
|
except Exception: # noqa: BLE001
|
|
return {
|
|
"error": {"code": status, "message": f"Non-JSON response: {body[:300]}", "_url": url}
|
|
}
|
|
|
|
def rpc(self, method: str, params: dict = None) -> dict:
|
|
"""POST one JSON-RPC request, retrying via ?rest_route= on a 404."""
|
|
if not self.site.endpoint:
|
|
return {"error": {"code": -1, "message": "Site URL is empty"}}
|
|
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": self._next_id,
|
|
"method": method,
|
|
"params": params or {},
|
|
}
|
|
self._next_id += 1
|
|
|
|
result = self._post(self.site.active_endpoint, payload)
|
|
|
|
# Plain-permalink sites 404 on /wp-json/ — retry the rest_route form once,
|
|
# then stick with whichever form worked for the rest of the session.
|
|
if (result.get("error") or {}).get("code") == 404 and not self.site._use_alt:
|
|
alt = _alt_endpoint(self.site.endpoint)
|
|
if alt:
|
|
retry = self._post(alt, payload)
|
|
if (retry.get("error") or {}).get("code") != 404:
|
|
self.site._use_alt = True
|
|
return retry
|
|
return result
|
|
|
|
# -- MCP surface -------------------------------------------------------
|
|
def initialize(self) -> dict:
|
|
return self.rpc(
|
|
"initialize",
|
|
{
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "wordvibe-hermes", "version": "1.0.0"},
|
|
},
|
|
)
|
|
|
|
def list_tools(self) -> dict:
|
|
return self.rpc("tools/list", {})
|
|
|
|
def call_tool(self, name: str, arguments: dict = None) -> dict:
|
|
return self.rpc("tools/call", {"name": name, "arguments": arguments or {}})
|
|
|
|
|
|
def result_text(response: dict) -> str:
|
|
"""Flatten an MCP tools/call response into plain text."""
|
|
if not isinstance(response, dict):
|
|
return str(response)
|
|
|
|
if "error" in response and response.get("error"):
|
|
err = response["error"]
|
|
if isinstance(err, dict):
|
|
return f"ERROR: {err.get('message') or json.dumps(err)}"
|
|
return f"ERROR: {err}"
|
|
|
|
result = response.get("result") or {}
|
|
if isinstance(result, dict) and result.get("isError"):
|
|
parts = result.get("content") or []
|
|
return "ERROR: " + " ".join(
|
|
str(c.get("text", "")) for c in parts if isinstance(c, dict)
|
|
)
|
|
|
|
content = result.get("content") if isinstance(result, dict) else None
|
|
if isinstance(content, list):
|
|
chunks = []
|
|
for c in content:
|
|
if isinstance(c, dict):
|
|
if c.get("type") == "text":
|
|
chunks.append(str(c.get("text", "")))
|
|
else:
|
|
chunks.append(json.dumps(c))
|
|
if chunks:
|
|
return "\n".join(chunks)
|
|
|
|
return json.dumps(result, indent=2, default=str)
|