From bf74468c402712836ce0be07f2678f2924abc07c Mon Sep 17 00:00:00 2001 From: ispyisail Date: Fri, 2 Oct 2026 10:27:25 +1300 Subject: [PATCH 1/2] qet-mcp: live mode tools, to act on the project open in QElectroTech qet_live_status, qet_live_run_script and qet_live_run_stored talk to a running QElectroTech through the channel its LiveServer opens, only when live mode is switched on in its settings and accepted at this start. They find it through live-session.json in QElectroTech's data folder and send its token on every request; with no session, the error says which of the three switches is missing. Running needs QET_ENABLE_SCRIPTING=1 as editing does; asking for the status does not. Co-Authored-By: Claude Opus 5.5 --- misc/qet-mcp/README.md | 29 ++++++++ misc/qet-mcp/qet_mcp.py | 130 +++++++++++++++++++++++++++++++++++ misc/qet-mcp/test_qet_mcp.py | 86 ++++++++++++++++++++++- 3 files changed, 244 insertions(+), 1 deletion(-) diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index 546014f1e..5504af40d 100644 --- a/misc/qet-mcp/README.md +++ b/misc/qet-mcp/README.md @@ -285,6 +285,35 @@ becomes its file name only if it is `a-z`, `0-9`, `-` and `_`. Storing or removing a script needs `QET_ENABLE_SCRIPTING=1` like an edit does: a stored script runs with the user's rights when they click it. +## Live mode: working in the QElectroTech you have open + +Every tool above works on files, with no QElectroTech window involved. The +three `qet_live_*` tools instead act on the project open in **your** +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 | + +QElectroTech only listens when three things are true: + +1. the server has `QET_ENABLE_SCRIPTING=1`, as for editing; +2. in QElectroTech, Configurer > Général > "Autoriser un assistant IA à agir + sur le projet ouvert" is ticked (off by default); +3. at this start, you answered *Continuer* to the warning QElectroTech shows + every time it starts with that setting on. + +While it listens, the status bar says so and shows the assistant's last +action, with an *Arrêter* button that closes the channel for the rest of +the session. Each action is one Ctrl+Z. A script's `qet.showMessage()` is +logged instead of opening a box nobody asked for. + +The channel is a local socket only your user can open. QElectroTech writes +its name and a random token to `live-session.json` in its data folder +(next to `scripts`), and deletes the file when the channel closes. + ## Worked examples **What did that edit change?** diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index 0e994560a..ccbbf488b 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -2895,6 +2895,90 @@ def tool_script_remove(script_id: str) -> dict: return {"removed": removed} +# -------------------------------------------------------------------------- +# Live mode: act on the project open in a running QElectroTech +# -------------------------------------------------------------------------- +# +# Everything above is headless: it reads and writes files and launches its +# own QElectroTech. These tools instead talk to the QElectroTech the user +# has open, which only listens when three things are true: scripting is +# allowed, its "mode direct" setting is on (off by default), and the user +# accepted the warning it shows at every start. It then writes +# live-session.json (socket name and token) in its data folder -- the +# folder the scripts folder is in -- and removes it when the channel +# closes. Each action is one undo step in front of the user. + +def _live_session() -> dict: + path = scripts_dir().parent / "live-session.json" + if not path.is_file(): + raise ValueError( + "no QElectroTech is listening for an assistant. Live mode needs, in " + "QElectroTech: Configurer QElectroTech > Général > \"Autoriser un " + "assistant IA à agir sur le projet ouvert\" ticked, QElectroTech " + "restarted, and \"Continuer\" chosen in the warning it shows at " + f"start. (Looked for {path}.)") + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise ValueError(f"unreadable live session file {path}: {exc}") from None + + +def _live_call(request: dict, timeout: float = 60.0) -> dict: + session = _live_session() + request = dict(request, token=session.get("token", ""), id=1) + line = (json.dumps(request) + "\n").encode("utf-8") + name = session.get("socket", "") + try: + if os.name == "nt": + # QLocalServer is a named pipe on Windows; fullServerName() is + # already \\.\pipe\. + with open(name, "r+b", buffering=0) as pipe: + pipe.write(line) + data = b"" + while not data.endswith(b"\n"): + chunk = pipe.read(1) + if not chunk: + break + data += chunk + else: + import socket + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(timeout) + sock.connect(name) + sock.sendall(line) + data = b"" + while not data.endswith(b"\n"): + chunk = sock.recv(65536) + if not chunk: + break + data += chunk + except OSError as exc: + raise ValueError(f"could not reach QElectroTech's live channel ({exc}); it " + "may have been stopped, or QElectroTech closed") from None + if not data.strip(): + raise ValueError("QElectroTech closed the live channel without answering") + answer = json.loads(data.decode("utf-8")) + answer.pop("id", None) + return answer + + +def tool_live_status() -> dict: + return _live_call({"cmd": "status"}) + + +def tool_live_run_script(source: str, name: str = "", timeout: int = 60) -> dict: + _require_script_consent() + if not isinstance(source, str) or not source.strip(): + raise ValueError("'source' must be the script's text") + return _live_call({"cmd": "run_script", "source": source, "name": name or "script"}, + timeout) + + +def tool_live_run_stored(script_id: str, timeout: int = 60) -> dict: + _require_script_consent() + return _live_call({"cmd": "run_stored", "script": _script_id(script_id)}, timeout) + + TOOLS = [ { "name": "qet_project_info", @@ -3597,6 +3681,52 @@ TOOLS = [ }, "handler": lambda a: tool_script_remove(a["id"]), }, + { + "name": "qet_live_status", + "description": "LIVE MODE. Ask the QElectroTech the user has open what is on " + "screen: the project, the folio shown (index and title), the " + "selected elements, the last undo step and the stored scripts. " + "Works only if the user switched live mode on in QElectroTech " + "and accepted its warning at this start; the error says which " + "step is missing. Changes nothing.", + "inputSchema": {"type": "object", "properties": {}}, + "handler": lambda a: tool_live_status(), + }, + { + "name": "qet_live_run_script", + "description": "LIVE MODE. Run script text on the project the user has open, " + "in front of them, as one undo step named after 'name'. " + "qet.currentFolio() is the folio on screen. Returns what the " + "script logged, its error with the line if it threw, and the " + "undo step (empty if nothing changed). Try a new script with " + "qet_script_test on a copy first when you can. Needs " + "QET_ENABLE_SCRIPTING=1 and live mode on in QElectroTech.", + "inputSchema": { + "type": "object", + "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}, + }, + "required": ["source"], + }, + "handler": lambda a: tool_live_run_script(a["source"], a.get("name", ""), + a.get("timeout", 60)), + }, + { + "name": "qet_live_run_stored", + "description": "LIVE MODE. Press a stored script's button (see " + "qet_script_list) in the QElectroTech the user has open: one " + "undo step. Needs QET_ENABLE_SCRIPTING=1 and live mode on in " + "QElectroTech.", + "inputSchema": { + "type": "object", + "properties": {"id": {"type": "string"}, + "timeout": {"type": "integer", "default": 60}}, + "required": ["id"], + }, + "handler": lambda a: tool_live_run_stored(a["id"], a.get("timeout", 60)), + }, ] _BY_NAME = {t["name"]: t for t in TOOLS} diff --git a/misc/qet-mcp/test_qet_mcp.py b/misc/qet-mcp/test_qet_mcp.py index ebbe49362..7f0a50862 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -166,7 +166,8 @@ class ToolRegistry(unittest.TestCase): "qet_project_new", "qet_element_search", "qet_check", "qet_element_build", "qet_continuity", "qet_items", "qet_script_api", "qet_script_test", "qet_script_install", "qet_script_list", "qet_script_read", - "qet_script_remove"}) + "qet_script_remove", "qet_live_status", "qet_live_run_script", + "qet_live_run_stored"}) class EditValidation(unittest.TestCase): @@ -2996,6 +2997,89 @@ class ScriptToolsIntegration(unittest.TestCase): self.assertTrue((self.dir / "good.js").is_file()) + +@unittest.skipIf(os.name == "nt", "the fake QElectroTech is a Unix socket") +class LiveClient(unittest.TestCase): + """The qet_live_* client against a stand-in for QElectroTech's + LiveServer: one JSON line in, one JSON line out, token on every line.""" + + def setUp(self): + import socket, threading + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.scripts = root / "data" / "scripts" + self.scripts.mkdir(parents=True) + self.env = mock.patch.dict(os.environ, {"QET_MCP_SCRIPTS_DIR": str(self.scripts), + "QET_ENABLE_SCRIPTING": "1"}) + self.env.start() + self.sock_path = str(root / "s") + self.server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.server.bind(self.sock_path) + self.server.listen(4) + self.seen = [] + + def serve(): + while True: + try: + conn, _ = self.server.accept() + except OSError: + return + with conn: + data = b"" + while not data.endswith(b"\n"): + chunk = conn.recv(4096) + if not chunk: + break + data += chunk + req = json.loads(data) + self.seen.append(req) + ok = req.get("token") == "T0K" + conn.sendall((json.dumps({"ok": ok, "id": req.get("id"), + "echo": req.get("cmd")}) + "\n").encode()) + threading.Thread(target=serve, daemon=True).start() + + def tearDown(self): + self.server.close() + self.env.stop() + self.tmp.cleanup() + + def session(self): + (self.scripts.parent / "live-session.json").write_text( + json.dumps({"version": 1, "socket": self.sock_path, "token": "T0K"})) + + def test_no_session_says_what_to_switch_on(self): + with self.assertRaisesRegex(ValueError, "Continuer"): + m.tool_live_status() + + def test_requests_carry_the_token_and_the_script_id(self): + self.session() + self.assertEqual(m.tool_live_status(), {"ok": True, "echo": "status"}) + m.tool_live_run_stored("mark") + m.tool_live_run_script("qet.log(1);", "Note") + self.assertEqual([r["token"] for r in self.seen], ["T0K"] * 3) + # "id" numbers the request; the stored script travels as "script" + # (they once shared "id", and every stored run asked for script "1") + self.assertEqual(self.seen[1]["script"], "mark") + self.assertEqual((self.seen[2]["source"], self.seen[2]["name"]), ("qet.log(1);", "Note")) + + def test_running_needs_consent_status_does_not(self): + self.session() + with mock.patch.dict(os.environ, {"QET_ENABLE_SCRIPTING": ""}): + m.tool_live_status() + with self.assertRaisesRegex(ValueError, "QET_ENABLE_SCRIPTING=1"): + m.tool_live_run_script("qet.log(1);") + with self.assertRaisesRegex(ValueError, "'id' must be"): + os.environ["QET_ENABLE_SCRIPTING"] = "1" + m.tool_live_run_stored("../x") + self.assertEqual(len(self.seen), 1) + + def test_stale_session_file(self): + (self.scripts.parent / "live-session.json").write_text( + json.dumps({"socket": self.sock_path + "-gone", "token": "T0K"})) + with self.assertRaisesRegex(ValueError, "could not reach"): + m.tool_live_status() + + class PathPolicyOverStdio(unittest.TestCase): """Proves the policy is actually wired into the dispatcher. From ae06377f1ab9498262e3bfb44c8b0052b6a6ba38 Mon Sep 17 00:00:00 2001 From: ispyisail Date: Fri, 2 Oct 2026 10:40:29 +1300 Subject: [PATCH 2/2] 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"}))