kotschak/qelectrotech-source-mirror
0
0
Fork 0
mirror of https://github.com/qelectrotech/qelectrotech-source-mirror.git synced 2026-10-03 01:04:13 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #1225 from ispyisail/feature/mcp-live-tools

Browse Source 
Add MCP live tools to act on the project open in QElectroTech
This commit is contained in:
ispyisail
2026-10-02 15:13:29 +13:00
committed by GitHub
parent 0f909e1246 a99c58487d
commit f6867a0f4e
3 changed files with 395 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
misc/qet-mcp/README.md
+38
View File
@@ -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?**
misc/qet-mcp/qet_mcp.py
+225 -2
View File
@@ -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\<name>.
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
misc/qet-mcp/test_qet_mcp.py
+132 -1
View File
@@ -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."""