200 lines
8.9 KiB
Python
200 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Comprueba el MCP stdio versionado y guarda una captura real sin dependencias extra."""
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import queue
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import tomllib
|
|
import uuid
|
|
from collections import deque
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
class StdioClient:
|
|
"""Cliente de diagnóstico JSON-RPC; cada petición tiene un plazo acotado."""
|
|
|
|
def __init__(self, command, timeout):
|
|
self.timeout = timeout
|
|
self.sequence = 0
|
|
self.messages = queue.Queue()
|
|
self.errors = deque(maxlen=8)
|
|
self.process = subprocess.Popen(
|
|
command, cwd=ROOT, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, text=True, encoding="utf-8", bufsize=1,
|
|
)
|
|
threading.Thread(target=self._read_stdout, daemon=True).start()
|
|
threading.Thread(target=self._read_stderr, daemon=True).start()
|
|
|
|
def _read_stdout(self):
|
|
for line in self.process.stdout:
|
|
try:
|
|
self.messages.put(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
self.messages.put({"invalid_stdout": True})
|
|
self.messages.put(None)
|
|
|
|
def _read_stderr(self):
|
|
for line in self.process.stderr:
|
|
self.errors.append(line.strip())
|
|
|
|
def send(self, message):
|
|
self.process.stdin.write(json.dumps({"jsonrpc": "2.0", **message}) + "\n")
|
|
self.process.stdin.flush()
|
|
|
|
def request(self, method, params):
|
|
self.sequence += 1
|
|
identity = self.sequence
|
|
self.send({"id": identity, "method": method, "params": params})
|
|
deadline = time.monotonic() + self.timeout
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise RuntimeError(f"Tiempo agotado en {method}.")
|
|
try:
|
|
message = self.messages.get(timeout=remaining)
|
|
except queue.Empty as error:
|
|
raise RuntimeError(f"Tiempo agotado en {method}.") from error
|
|
if message is None:
|
|
raise RuntimeError("El servidor MCP cerró stdout. " + " | ".join(self.errors))
|
|
if message.get("invalid_stdout"):
|
|
raise RuntimeError("El servidor escribió contenido no JSON-RPC en stdout.")
|
|
if "method" in message and "id" in message:
|
|
if message["method"] in ("ping", "roots/list"):
|
|
self.send({"id": message["id"], "result": {"roots": []} if message["method"] == "roots/list" else {}})
|
|
else:
|
|
self.send({"id": message["id"], "error": {"code": -32601, "message": "Method not supported"}})
|
|
continue
|
|
if message.get("id") != identity:
|
|
continue
|
|
if "error" in message:
|
|
raise RuntimeError(f"Error JSON-RPC en {method}: {message['error']}")
|
|
return message["result"]
|
|
|
|
def tool(self, name, arguments):
|
|
result = self.request("tools/call", {"name": name, "arguments": arguments})
|
|
if result.get("isError"):
|
|
detail = " ".join(part.get("text", "") for part in result.get("content", []))
|
|
raise RuntimeError(f"{name}: {detail[:1200]}")
|
|
return result
|
|
|
|
def close(self):
|
|
try:
|
|
self.process.stdin.close()
|
|
self.process.wait(timeout=5)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
self.process.terminate()
|
|
try:
|
|
self.process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
self.process.kill()
|
|
self.process.wait(timeout=5)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("url", help="URL real de Vite, por ejemplo http://localhost:5173")
|
|
parser.add_argument("--timeout", type=int, default=120, help="Plazo por petición MCP en segundos")
|
|
parser.add_argument("--output", type=Path, default=ROOT / ".artifacts/playwright/metis-desk.png")
|
|
options = parser.parse_args()
|
|
url = urlsplit(options.url)
|
|
if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password:
|
|
parser.error("La URL debe ser HTTP(S) y no contener credenciales.")
|
|
if options.timeout <= 0:
|
|
parser.error("El timeout debe ser positivo.")
|
|
with (ROOT / ".codex/config.toml").open("rb") as source:
|
|
config = tomllib.load(source)["mcp_servers"]["playwright"]
|
|
args = config["args"]
|
|
images = [arg for arg in args if re.fullmatch(r"mcr\.microsoft\.com/playwright/mcp:v[\d.]+@sha256:[a-f0-9]{64}", arg)]
|
|
if config["command"] != "docker" or len(images) != 1 or args[0] != "run":
|
|
raise RuntimeError("La configuración debe ejecutar Docker con una única imagen oficial fijada por digest.")
|
|
container = f"metis-desk-playwright-check-{uuid.uuid4().hex[:12]}"
|
|
command = [config["command"], "run", f"--name={container}", *args[1:]]
|
|
client = StdioClient(command, options.timeout)
|
|
try:
|
|
initialized = client.request("initialize", {
|
|
"protocolVersion": "2025-06-18", "capabilities": {},
|
|
"clientInfo": {"name": "metis-desk-playwright-check", "version": "1.0.0"},
|
|
})
|
|
client.send({"method": "notifications/initialized"})
|
|
tools = client.request("tools/list", {})
|
|
names = {tool["name"] for tool in tools["tools"]}
|
|
required = {"browser_navigate", "browser_evaluate", "browser_take_screenshot", "browser_console_messages"}
|
|
if not required <= names:
|
|
raise RuntimeError(f"Faltan herramientas requeridas: {sorted(required - names)}")
|
|
client.tool("browser_navigate", {"url": options.url})
|
|
# Espera una pantalla real de acceso/configuración; no sustituye la API por mocks.
|
|
rendered = client.tool("browser_evaluate", {"function": """async () => {
|
|
await new Promise((resolve, reject) => {
|
|
if (document.querySelector('#root form')) {
|
|
resolve();
|
|
return;
|
|
}
|
|
const observer = new MutationObserver(() => {
|
|
if (document.querySelector('#root form')) {
|
|
observer.disconnect();
|
|
clearTimeout(timeout);
|
|
resolve();
|
|
}
|
|
});
|
|
const timeout = setTimeout(() => {
|
|
observer.disconnect();
|
|
reject(new Error('El formulario de Metis Desk no responde.'));
|
|
}, 30000);
|
|
observer.observe(document.body, { childList: true, subtree: true });
|
|
});
|
|
await document.fonts.ready;
|
|
const title = document.title;
|
|
const userAgent = navigator.userAgent;
|
|
if (!title.includes('Metis Desk') || !userAgent.includes('HeadlessChrome')) {
|
|
throw new Error('No se ha reconocido Metis Desk en Chromium headless.');
|
|
}
|
|
return { title, url: location.href, userAgent };
|
|
}"""})
|
|
screenshot = client.tool("browser_take_screenshot", {"type": "png", "fullPage": True, "scale": "css"})
|
|
console = client.tool("browser_console_messages", {"level": "error"})
|
|
image = next((part for part in screenshot.get("content", []) if part.get("type") == "image" and part.get("mimeType") == "image/png"), None)
|
|
if image is None:
|
|
raise RuntimeError("El MCP no devolvió una imagen PNG en la respuesta.")
|
|
data = base64.b64decode(image["data"], validate=True)
|
|
if len(data) < 33 or data[:8] != b"\x89PNG\r\n\x1a\n" or data[12:16] != b"IHDR":
|
|
raise RuntimeError("La captura no tiene una cabecera PNG válida.")
|
|
width, height = struct.unpack(">II", data[16:24])
|
|
if not width or not height:
|
|
raise RuntimeError("La captura no tiene dimensiones válidas.")
|
|
options.output.parent.mkdir(parents=True, exist_ok=True)
|
|
options.output.write_bytes(data)
|
|
report = {
|
|
"image": images[0], "server": initialized["serverInfo"],
|
|
"protocol": initialized["protocolVersion"], "tools": len(names),
|
|
"url": options.url, "screenshot": str(options.output),
|
|
"width": width, "height": height, "bytes": len(data),
|
|
"sha256": hashlib.sha256(data).hexdigest(),
|
|
"page": [part["text"] for part in rendered.get("content", []) if part.get("type") == "text"],
|
|
"console": [part["text"] for part in console.get("content", []) if part.get("type") == "text"],
|
|
}
|
|
options.output.with_suffix(".json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
|
|
print(json.dumps({key: value for key, value in report.items() if key not in ("page", "console")}, ensure_ascii=False, indent=2))
|
|
finally:
|
|
client.close()
|
|
# Limpia exclusivamente el contenedor de esta comprobación, también si el cliente falla.
|
|
subprocess.run(["docker", "rm", "-f", container], capture_output=True, timeout=15, check=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (RuntimeError, OSError, ValueError, subprocess.SubprocessError) as error:
|
|
print(f"Validación fallida: {error}", file=sys.stderr)
|
|
sys.exit(1)
|