qet-mcp: find the live channel in qet-assistant.json

QElectroTech now advertises its live channel in the "live" part of qet-assistant.json instead of a live-session.json of its own. When nothing is listening, the error now tells apart QElectroTech not running, live mode switched off, and the start-up warning not accepted. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-03 01:04:13 +02:00 · 2026-10-02 13:03:33 +13:00
parent ae06377f1a 15c114edd5
commit a99c58487d
3 changed files with 225 additions and 25 deletions
@@ -41,6 +41,7 @@ here read the model.
 | `qet_element_search` | **find a symbol** in a collection by name (any language), type or terminal count |
 | `qet_check` | **design-rule checks** — duplicate labels, unlabelled masters, unnumbered conductors, empty folios |
 | `qet_query` | **ask the project database** — read-only SQL over the views and tables |
+| `qet_about` | **start here** — where QElectroTech keeps things, what is switched on, the stored scripts, the calls a script can make (from `qet-assistant.json`) |
 | `qet_script_api` | **what a script can call** — every `qet.*` call of this build, and the header that makes a script a button |
 | `qet_script_test` | **try a script** on a copy of a project: what it would change, what it logged, its errors |
 | `qet_script_install` | **make a button** — store a script (and an SVG icon) where QElectroTech shows it in Projet > Scripts and the Scripts toolbar |
@@ -251,6 +252,21 @@ each tool. Importing `qet_mcp` and calling `tool_export()` from your own
 Python is not confined and is not meant to be — that is your code calling a
 library, and you already chose the paths.

+## What QElectroTech tells the server: `qet-assistant.json`
+
+Each time an editor window opens, and whenever its stored scripts,
+settings or live channel change, QElectroTech writes `qet-assistant.json`
+in its standard data folder (`~/.local/share/QElectroTech/QElectroTech/`
+on Linux, `%APPDATA%\QElectroTech\QElectroTech\` on Windows). It names
+every folder actually in use, even when QElectroTech was started with
+`--data-dir`, which features are on, every call a script can make, and the
+stored scripts. The server reads it instead of guessing; `qet_about` shows
+it. Set `QET_MCP_INFO_FILE` to read it from somewhere else.
+
+The server also sends the assistant a short note at first contact: the two
+ways of working (files, or live), the usual order of tools, and to start
+with `qet_about`.
+
 ## Script buttons

 QElectroTech turns every `.js` file in its scripts folder that starts with a
@@ -318,9 +334,10 @@ 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.
+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

@@ -2611,18 +2611,110 @@ def default_scripts_dir(os_name: str, platform: str, env, home) -> PurePath:
    return base / "QElectroTech" / "QElectroTech" / "scripts"


+def assistant_info_file() -> Path:
+    """Where QElectroTech writes qet-assistant.json.
+
+    QET_MCP_INFO_FILE if set; next to a QET_MCP_SCRIPTS_DIR if that is set
+    (one folder up, as in QElectroTech's own layout); else the platform's
+    standard data folder, where QElectroTech always writes it, even when
+    --data-dir moves the rest.
+    """
+    explicit = os.environ.get("QET_MCP_INFO_FILE", "").strip()
+    if explicit:
+        return Path(explicit).expanduser()
+    scripts = os.environ.get("QET_MCP_SCRIPTS_DIR", "").strip()
+    if scripts:
+        return Path(scripts).expanduser().parent / "qet-assistant.json"
+    default = default_scripts_dir(os.name, sys.platform, os.environ, str(Path.home()))
+    return Path(str(default.parent)) / "qet-assistant.json"
+
+
+def assistant_info() -> dict | None:
+    """qet-assistant.json as QElectroTech last wrote it, or None.
+
+    QElectroTech rewrites it whenever an editor opens and whenever its
+    stored scripts, settings or live channel change: its folders, features,
+    the calls a script can make, the stored scripts, and the live channel
+    while one is open.
+    """
+    path = assistant_info_file()
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
 def scripts_dir() -> Path:
    """The folder QElectroTech reads stored scripts from.

-    QET_MCP_SCRIPTS_DIR overrides it, for a QElectroTech started with
-    --data-dir or a test.
+    QET_MCP_SCRIPTS_DIR first (a test, or a deliberate choice); then the
+    folder qet-assistant.json names; else the platform's default.
    """
    env = os.environ.get("QET_MCP_SCRIPTS_DIR", "").strip()
    if env:
        return Path(env).expanduser()
+    info = assistant_info() or {}
+    named = (info.get("folders") or {}).get("scripts")
+    if isinstance(named, str) and named:
+        return Path(named)
    return Path(str(default_scripts_dir(os.name, sys.platform, os.environ, str(Path.home()))))


+SERVER_INSTRUCTIONS = (
+    "This server works with QElectroTech, a free editor for electrical "
+    "diagrams. A project (.qet) holds folios (sheets) of symbols "
+    "(elements) joined by wires (conductors). Folio indexes count from 0.\n"
+    "Start with qet_about: where QElectroTech keeps things, what is "
+    "switched on, and the stored scripts.\n"
+    "Two ways of working. HEADLESS (qet_* and qet_script_*): read, check "
+    "and edit .qet files and store script buttons; nothing the user has "
+    "open is touched. To make a button: qet_script_api for the calls, "
+    "qet_script_test on a copy until the diff is right, then "
+    "qet_script_install with test_project set. LIVE (qet_live_*): act on "
+    "the project open in the user's QElectroTech while they watch; only "
+    "when they switched live mode on and accepted its warning at this "
+    "start, each action one undo step, scripts written on the spot shown "
+    "to them first.\n"
+    "Verify edits by reading the result (qet_diff, qet_elements), not by "
+    "assuming them.")
+
+
+def tool_about() -> dict:
+    """What this server and the QElectroTech it works with look like now."""
+    info = assistant_info()
+    binary = resolve_binary()
+    out = {
+        "server": {
+            "version": SERVER_VERSION,
+            "qelectrotech_binary": str(binary) if binary else None,
+            "workspace": [str(r) for r in workspace_roots()] or "any path (QET_MCP_ALLOW_ANY_PATH=1)",
+            "scripting_allowed_here": os.environ.get("QET_ENABLE_SCRIPTING") == "1",
+        },
+        "info_file": str(assistant_info_file()),
+        "scripts_folder": str(scripts_dir()),
+    }
+    if info is None:
+        out["found"] = False
+        out["note"] = ("QElectroTech writes this file when an editor window opens; "
+                       "start it once (a version with script buttons) to fill it in. "
+                       "Until then folders are this server's own guess.")
+        return out
+    live = info.get("live")
+    out.update({
+        "found": True,
+        "qelectrotech": {k: info.get(k) for k in
+                         ("qelectrotech_version", "program", "running", "written")},
+        "folders": info.get("folders"),
+        "features": info.get("features"),
+        "stored_scripts": info.get("stored_scripts"),
+        "refused_scripts": info.get("refused_scripts"),
+        "script_api": info.get("script_api"),
+        # Never the token: it is for the live tools, not the conversation.
+        "live": {"open": bool(live), "pid": (live or {}).get("pid")},
+    })
+    return out
+
+
 def parse_script_header(text: str, script_id: str) -> dict:
    """The same rules as QElectroTech's ScriptHeader::parse().

@@ -2903,24 +2995,33 @@ def tool_script_remove(script_id: str) -> dict:
 # 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.
+# 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:
-    path = scripts_dir().parent / "live-session.json"
-    if not path.is_file():
+    """The live channel QElectroTech advertises in qet-assistant.json."""
+    info = assistant_info()
+    path = assistant_info_file()
+    if info is None:
        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
+            "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:
@@ -3799,6 +3900,18 @@ TOOLS = [
        "inputSchema": {"type": "object", "properties": {}},
        "handler": lambda a: tool_live_screenshot(),
    },
+    {
+        "name": "qet_about",
+        "description": "Start here. What QElectroTech last wrote about itself in "
+                       "qet-assistant.json: version, every folder (data, settings, "
+                       "scripts, element and title block collections), which "
+                       "features are on (scripting, live mode), every call a script "
+                       "can make, the stored scripts and the ones refused with why, "
+                       "and whether a live session is open; plus this server's own "
+                       "setup. Reads one file; changes nothing.",
+        "inputSchema": {"type": "object", "properties": {}},
+        "handler": lambda a: tool_about(),
+    },
 ]

 _BY_NAME = {t["name"]: t for t in TOOLS}
@@ -4100,6 +4213,7 @@ def handle(msg: dict) -> dict | None:
            "protocolVersion": want or DEFAULT_PROTOCOL,
            "capabilities": {"tools": {}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
+            "instructions": SERVER_INSTRUCTIONS,
        })

    if method in ("notifications/initialized", "initialized"):
@@ -168,7 +168,7 @@ class ToolRegistry(unittest.TestCase):
        "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_command", "qet_live_show_folio",
-        "qet_live_undo_last", "qet_live_screenshot"})
+        "qet_live_undo_last", "qet_live_screenshot", "qet_about"})


 class EditValidation(unittest.TestCase):
@@ -3044,11 +3044,25 @@ class LiveClient(unittest.TestCase):
        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.scripts.parent / "live-session.json").write_text(
-            json.dumps({"version": 1, "socket": self.sock_path, "token": "T0K"}))
+        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()

@@ -3110,10 +3124,65 @@ class LiveClient(unittest.TestCase):
            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"}))
+        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."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+        self.info = self.root / "info" / "qet-assistant.json"
+        self.info.parent.mkdir()
+        self.env = mock.patch.dict(os.environ, {"QET_MCP_INFO_FILE": str(self.info),
+                                                "QET_MCP_SCRIPTS_DIR": ""})
+        self.env.start()
+
+    def tearDown(self):
+        self.env.stop()
+        self.tmp.cleanup()
+
+    def write(self, **extra):
+        data = {"format": 1, "qelectrotech_version": "0.200.1", "running": True,
+                "folders": {"scripts": str(self.root / "moved" / "scripts")},
+                "features": {"scripting_enabled": True, "live_mode_setting": False},
+                "script_api": ["int currentFolio()"], "stored_scripts": [],
+                "refused_scripts": [], "live": None}
+        data.update(extra)
+        self.info.write_text(json.dumps(data))
+
+    def test_scripts_folder_comes_from_the_file(self):
+        """QElectroTech started with --data-dir keeps its scripts elsewhere;
+        only the file knows."""
+        self.write()
+        self.assertEqual(m.scripts_dir(), self.root / "moved" / "scripts")
+        with mock.patch.dict(os.environ, {"QET_MCP_SCRIPTS_DIR": "/explicit"}):
+            self.assertEqual(m.scripts_dir(), Path("/explicit"))
+
+    def test_without_the_file_the_guess_stands(self):
+        about = m.tool_about()
+        self.assertFalse(about["found"])
+        self.assertIn("start it once", about["note"])
+
+    def test_about_never_shows_the_live_token(self):
+        self.write(live={"socket": "s", "token": "SECRET", "pid": 7})
+        about = m.tool_about()
+        self.assertTrue(about["found"])
+        self.assertEqual(about["live"], {"open": True, "pid": 7})
+        self.assertNotIn("SECRET", json.dumps(about))
+        self.assertEqual(about["script_api"], ["int currentFolio()"])
+
+    def test_first_contact_carries_the_instructions(self):
+        reply = m.handle({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}})
+        self.assertIn("qet_about", reply["result"]["instructions"])
+
+    def test_default_location_is_next_to_the_default_scripts_folder(self):
+        with mock.patch.dict(os.environ, {"QET_MCP_INFO_FILE": "", "XDG_DATA_HOME": "/x"}), \
+                mock.patch.object(m.os, "name", "posix"), mock.patch.object(m.sys, "platform", "linux"):
+            self.assertEqual(m.assistant_info_file(),
+                             Path("/x/QElectroTech/QElectroTech/qet-assistant.json"))


 class PathPolicyOverStdio(unittest.TestCase):