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 <noreply@anthropic.com>
This commit is contained in:
ispyisail
2026-10-02 10:40:29 +13:00
parent bf74468c40
commit ae06377f1a
3 changed files with 134 additions and 6 deletions
+8
View File
@@ -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:
+89 -5
View File
@@ -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
+37 -1
View File
@@ -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"}))