diff --git a/misc/qet-mcp/README.md b/misc/qet-mcp/README.md index aa913ff84..f22347461 100644 --- a/misc/qet-mcp/README.md +++ b/misc/qet-mcp/README.md @@ -301,6 +301,44 @@ 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 | +| `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: + +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 puts +its name and a random token in the `live` part of `qet-assistant.json`, +and clears it when the channel closes; `qet_about` says whether one is +open but never shows the token. + ## Worked examples **What did that edit change?** diff --git a/misc/qet-mcp/qet_mcp.py b/misc/qet-mcp/qet_mcp.py index 3182894f6..d39cc44e2 100755 --- a/misc/qet-mcp/qet_mcp.py +++ b/misc/qet-mcp/qet_mcp.py @@ -3000,6 +3000,125 @@ 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 puts the socket +# name and token in the "live" part of qet-assistant.json, and clears it +# when the channel closes. Each action is one undo step in front of the user. + +def _live_session() -> dict: + """The live channel QElectroTech advertises in qet-assistant.json.""" + info = assistant_info() + path = assistant_info_file() + if info is None: + raise ValueError( + "QElectroTech has not written qet-assistant.json yet (looked for " + f"{path}): start QElectroTech, a version with live mode, first.") + live = info.get("live") + if not info.get("running") or not live: + features = info.get("features") or {} + if not info.get("running"): + why = "QElectroTech is not running" + elif not features.get("live_mode_setting"): + why = ("live mode is off: in QElectroTech, Configurer QElectroTech > " + "Général > \"Autoriser un assistant IA à agir sur le projet " + "ouvert\", then restart it") + else: + why = ("live mode is on but not open for this session: answer " + "\"Continuer\" in the warning QElectroTech shows at start, or " + "restart it if \"Pas pour cette session\" or Arrêter was chosen") + raise ValueError(f"no QElectroTech is listening for an assistant: {why}.") + return live + + +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 = 300) -> 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) + + +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", @@ -3708,6 +3827,98 @@ 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": 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", 300)), + }, + { + "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)), + }, + { + "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(), + }, { "name": "qet_about", "description": "Start here. What QElectroTech last wrote about itself in " @@ -4043,8 +4254,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, @@ -4112,7 +4330,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 35322cd67..d653e0ebd 100644 --- a/misc/qet-mcp/test_qet_mcp.py +++ b/misc/qet-mcp/test_qet_mcp.py @@ -166,7 +166,9 @@ 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_about"}) + "qet_script_remove", "qet_live_status", "qet_live_run_script", + "qet_live_run_stored", "qet_live_command", "qet_live_show_folio", + "qet_live_undo_last", "qet_live_screenshot", "qet_about"}) class EditValidation(unittest.TestCase): @@ -3000,6 +3002,135 @@ 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 info(self, live, running=True, setting=True): + (self.scripts.parent / "qet-assistant.json").write_text(json.dumps({ + "format": 1, "running": running, + "features": {"live_mode_setting": setting}, "live": live})) + + def session(self): + self.info({"socket": self.sock_path, "token": "T0K", "pid": 1}) + + def test_no_session_says_what_to_switch_on(self): + """Each reason nothing is listening gets its own answer.""" + with self.assertRaisesRegex(ValueError, "start QElectroTech"): + m.tool_live_status() + self.info(None, running=False) + with self.assertRaisesRegex(ValueError, "not running"): + m.tool_live_status() + self.info(None, setting=False) + with self.assertRaisesRegex(ValueError, "live mode is off"): + m.tool_live_status() + self.info(None) + 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_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.info({"socket": self.sock_path + "-gone", "token": "T0K"}) + with self.assertRaisesRegex(ValueError, "could not reach"): + m.tool_live_status() class AssistantInfoFile(unittest.TestCase): """qet-assistant.json: QElectroTech says where things are; the server believes it over its own per-platform guess."""