From ae06377f1ab9498262e3bfb44c8b0052b6a6ba38 Mon Sep 17 00:00:00 2001 From: ispyisail Date: Fri, 2 Oct 2026 10:40:29 +1300 Subject: [PATCH] qet-mcp: live commands, folios, undo of the assistant's own step, screenshot qet_live_screenshot returns an MCP image the assistant can look at; --call prints an image part as a data: URI instead of failing on it. Co-Authored-By: Claude Opus 5.5 --- misc/qet-mcp/README.md | 8 +++ misc/qet-mcp/qet_mcp.py | 94 ++++++++++++++++++++++++++++++++++-- misc/qet-mcp/test_qet_mcp.py | 38 ++++++++++++++- 3 files changed, 134 insertions(+), 6 deletions(-) diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index 5504af40d..ae82ded29 100644 --- a/misc/qet-mcp/README.md +++ b/misc/qet-mcp/README.md @@ -296,6 +296,14 @@ QElectroTech, in front of you, so you can watch, stop or undo: | `qet_live_status` | what is on screen: project, folio, selection, last undo step, stored scripts | | `qet_live_run_script` | run script text on the open project: one undo step named "Assistant : …" | | `qet_live_run_stored` | press a stored script's button | +| `qet_live_command` | an editor command from an allow-list that opens no dialog: selection, zoom, rotate, snap, group, reset wires | +| `qet_live_show_folio` | show another folio | +| `qet_live_undo_last` | undo the newest step, only if the assistant made it | +| `qet_live_screenshot` | a picture of the folio on screen, as an MCP image | + +A script the assistant writes on the spot is shown to you first, with +*Exécuter*, *Refuser* or *Toujours pour cette session*; the Assistant +panel lists everything it did. QElectroTech only listens when three things are true: diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index ccbbf488b..bb6bc5908 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -2966,7 +2966,7 @@ def tool_live_status() -> dict: return _live_call({"cmd": "status"}) -def tool_live_run_script(source: str, name: str = "", timeout: int = 60) -> dict: +def tool_live_run_script(source: str, name: str = "", timeout: int = 300) -> dict: _require_script_consent() if not isinstance(source, str) or not source.strip(): raise ValueError("'source' must be the script's text") @@ -2979,6 +2979,32 @@ def tool_live_run_stored(script_id: str, timeout: int = 60) -> dict: return _live_call({"cmd": "run_stored", "script": _script_id(script_id)}, timeout) +def tool_live_command(action: str) -> dict: + _require_script_consent() + if not isinstance(action, str) or not action: + raise ValueError("'action' must be a command id, e.g. diagrameditor.zoom_fit") + return _live_call({"cmd": "command", "action": action}) + + +def tool_live_show_folio(folio: int) -> dict: + if not isinstance(folio, int) or isinstance(folio, bool): + raise ValueError("'folio' must be an index counted from 0") + return _live_call({"cmd": "show_folio", "folio": folio}) + + +def tool_live_undo_last() -> dict: + _require_script_consent() + return _live_call({"cmd": "undo_last"}) + + +def tool_live_screenshot() -> dict: + answer = _live_call({"cmd": "screenshot"}) + data = answer.pop("png_base64", None) + if data: + answer["_image_png_base64"] = data + return answer + + TOOLS = [ { "name": "qet_project_info", @@ -3706,12 +3732,13 @@ TOOLS = [ "properties": { "source": {"type": "string", "description": "the script's text"}, "name": {"type": "string", "description": "what the user sees in the undo step"}, - "timeout": {"type": "integer", "default": 60}, + "timeout": {"type": "integer", "default": 300, + "description": "seconds; the user may be reading the script before saying yes"}, }, "required": ["source"], }, "handler": lambda a: tool_live_run_script(a["source"], a.get("name", ""), - a.get("timeout", 60)), + a.get("timeout", 300)), }, { "name": "qet_live_run_stored", @@ -3727,6 +3754,51 @@ TOOLS = [ }, "handler": lambda a: tool_live_run_stored(a["id"], a.get("timeout", 60)), }, + { + "name": "qet_live_command", + "description": "LIVE MODE. Trigger one editor command in the QElectroTech the " + "user has open, by id. Only commands that open no dialog are " + "allowed: diagrameditor.select_all, select_nothing, " + "select_invert, select_all_conductors, select_all_text_fields, " + "zoom_in, zoom_out, zoom_content, zoom_fit, zoom_reset, " + "rotate_selection, rotate_texts, snap_selection_to_grid, " + "group_selection, ungroup_selection, conductor_reset (all " + "prefixed diagrameditor.). Anything else -- saving, deleting, " + "exporting -- is refused; use a script for edits.", + "inputSchema": { + "type": "object", + "properties": {"action": {"type": "string"}}, + "required": ["action"], + }, + "handler": lambda a: tool_live_command(a["action"]), + }, + { + "name": "qet_live_show_folio", + "description": "LIVE MODE. Show another folio of the open project (index " + "from 0), so qet.currentFolio() and qet_live_screenshot " + "follow it.", + "inputSchema": { + "type": "object", + "properties": {"folio": {"type": "integer"}}, + "required": ["folio"], + }, + "handler": lambda a: tool_live_show_folio(a["folio"]), + }, + { + "name": "qet_live_undo_last", + "description": "LIVE MODE. Undo the newest step in the open project, only if " + "the assistant made it (its name starts \"Assistant :\"); " + "the user's own steps are never undone this way.", + "inputSchema": {"type": "object", "properties": {}}, + "handler": lambda a: tool_live_undo_last(), + }, + { + "name": "qet_live_screenshot", + "description": "LIVE MODE. An image of the folio on screen in the user's " + "QElectroTech, as they see it. Changes nothing.", + "inputSchema": {"type": "object", "properties": {}}, + "handler": lambda a: tool_live_screenshot(), + }, ] _BY_NAME = {t["name"]: t for t in TOOLS} @@ -4049,8 +4121,15 @@ def handle(msg: dict) -> dict | None: arguments = params.get("arguments") or {} enforce_path_policy(name, arguments) result = tool["handler"](arguments) + content = [] + # A tool may return a picture (qet_live_screenshot): sent as an + # MCP image so the assistant can look at it, not as a string. + image = result.pop("_image_png_base64", None) if isinstance(result, dict) else None + if image: + content.append({"type": "image", "data": image, "mimeType": "image/png"}) text = json.dumps(result, indent=2, ensure_ascii=False) - return _ok(mid, {"content": [{"type": "text", "text": text}]}) + content.append({"type": "text", "text": text}) + return _ok(mid, {"content": content}) except Exception as exc: # surfaced to the model, not the transport return _ok(mid, { "isError": True, @@ -4118,7 +4197,12 @@ def call_once(argv: list[str], stdin=sys.stdin, stdout=sys.stdout, return 2 result = reply["result"] for part in result["content"]: - print(part["text"], file=stdout) + if part.get("type") == "image": + # A picture has no text; print it whole, as a data: URI a + # browser or a script can use, rather than drop it. + print(f"data:{part['mimeType']};base64,{part['data']}", file=stdout) + else: + print(part["text"], file=stdout) return 1 if result.get("isError") else 0 diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index 7f0a50862..47238b2d0 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -167,7 +167,8 @@ class ToolRegistry(unittest.TestCase): "qet_continuity", "qet_items", "qet_script_api", "qet_script_test", "qet_script_install", "qet_script_list", "qet_script_read", "qet_script_remove", "qet_live_status", "qet_live_run_script", - "qet_live_run_stored"}) + "qet_live_run_stored", "qet_live_command", "qet_live_show_folio", + "qet_live_undo_last", "qet_live_screenshot"}) class EditValidation(unittest.TestCase): @@ -3073,6 +3074,41 @@ class LiveClient(unittest.TestCase): m.tool_live_run_stored("../x") self.assertEqual(len(self.seen), 1) + def test_screenshot_is_sent_as_an_image(self): + """The picture must reach the assistant as an MCP image, not as a + long base64 string inside the text.""" + self.session() + with mock.patch.object(m, "_live_call", + return_value={"ok": True, "width": 2, "height": 1, + "png_base64": "iVBORw0K"}): + reply = m.handle({"jsonrpc": "2.0", "id": 9, "method": "tools/call", + "params": {"name": "qet_live_screenshot", "arguments": {}}}) + content = reply["result"]["content"] + self.assertEqual(content[0], {"type": "image", "data": "iVBORw0K", + "mimeType": "image/png"}) + self.assertNotIn("iVBORw0K", content[1]["text"]) + + def test_call_once_prints_an_image_as_a_data_uri(self): + import io + self.session() + out = io.StringIO() + with mock.patch.object(m, "_live_call", + return_value={"ok": True, "png_base64": "iVBORw0K"}): + code = m.call_once(["qet_live_screenshot"], stdout=out) + self.assertEqual(code, 0) + self.assertEqual(out.getvalue().splitlines()[0], "data:image/png;base64,iVBORw0K") + + def test_commands_and_folios_are_sent_as_asked(self): + self.session() + m.tool_live_command("diagrameditor.zoom_fit") + m.tool_live_show_folio(2) + m.tool_live_undo_last() + self.assertEqual([(r["cmd"], r.get("action"), r.get("folio")) for r in self.seen], + [("command", "diagrameditor.zoom_fit", None), + ("show_folio", None, 2), ("undo_last", None, None)]) + with self.assertRaises(ValueError): + m.tool_live_show_folio("2") + def test_stale_session_file(self): (self.scripts.parent / "live-session.json").write_text( json.dumps({"socket": self.sock_path + "-gone", "token": "T0K"}))