Fusion - the Fusion 360 Bridge

@@ -3833,6 +3833,396 @@ def _handle_set_auto_update(fusion_info: dict, args: dict) -> dict: COMMAND_HANDLERS["set_auto_update"] = _handle_set_auto_update  +# ── AUTODESK FUSION MCP SERVER PROXY (John, 2026-07-23) ──────────────────────────────────────+# Autodesk + Anthropic shipped a Fusion MCP server: a LOCAL HTTP/JSON-RPC endpoint at+# http://127.0.0.1:27182/mcp that exposes Fusion's own text-to-CAD surface (read geometry,+# execute/update features, read Electronics design data).+#+# WHY THE BRIDGE PROXIES IT: that server binds LOOPBACK on the user's machine. It was designed+# for Claude Desktop running on the same box. An Adom AI runs in a CLOUD container and cannot+# reach the user's 127.0.0.1 at all. This bridge already runs on that machine, so it is exactly+# the right proxy: these verbs give the cloud AI Autodesk's MCP tools without reimplementing them.+#+# The server speaks MCP streamable-HTTP: you must `initialize`, capture the MCP-Session-Id+# response header, send notifications/initialized, and pass that header on every later call.+# Skipping it returns 400 "Missing MCP-Session-Id".+_MCP_URL = "http://127.0.0.1:27182/mcp"+_mcp_session_id = None+_mcp_lock = threading.Lock()+++def _mcp_post(body: dict, sid: str = None, timeout: int = 30) -> dict:+    import urllib.request, urllib.error+    h = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}+    if sid:+        h["MCP-Session-Id"] = sid+    req = urllib.request.Request(_MCP_URL, method="POST",+                                 data=json.dumps(body).encode(), headers=h)+    try:+        with urllib.request.urlopen(req, timeout=timeout) as r:+            return {"status": r.status, "headers": dict(r.headers),+                    "body": r.read().decode("utf-8", "replace")}+    except urllib.error.HTTPError as e:+        return {"httpError": e.code, "headers": dict(e.headers),+                "body": e.read().decode("utf-8", "replace")[:1000]}+    except Exception as e:+        return {"error": "%s: %s" % (type(e).__name__, str(e)[:200])}+++def _mcp_new_session() -> dict:+    """initialize + notifications/initialized. Returns {sessionId, serverInfo} or {error}."""+    global _mcp_session_id+    r = _mcp_post({"jsonrpc": "2.0", "id": 1, "method": "initialize",+                   "params": {"protocolVersion": "2024-11-05", "capabilities": {},+                              "clientInfo": {"name": "adom-desktop-fusion-bridge",+                                             "version": BRIDGE_VERSION}}})+    if r.get("error") or r.get("httpError"):+        return {"error": r.get("error") or ("HTTP %s" % r.get("httpError")), "raw": r.get("body")}+    sid = None+    for k, v in (r.get("headers") or {}).items():+        if k.lower() == "mcp-session-id":+            sid = v+    info = {}+    try:+        info = (json.loads(r.get("body") or "{}").get("result") or {}).get("serverInfo") or {}+    except Exception:+        pass+    if sid:+        _mcp_post({"jsonrpc": "2.0", "method": "notifications/initialized"}, sid)+        _mcp_session_id = sid+    return {"sessionId": sid, "serverInfo": info}+++def _mcp_rpc(method: str, params: dict = None, timeout: int = 60) -> dict:+    """One JSON-RPC call with automatic session (re)establishment."""+    global _mcp_session_id+    with _mcp_lock:+        if not _mcp_session_id:+            s = _mcp_new_session()+            if s.get("error"):+                return {"_mcpError": s["error"]}+        sid = _mcp_session_id+    body = {"jsonrpc": "2.0", "id": 7, "method": method}+    if params is not None:+        body["params"] = params+    r = _mcp_post(body, sid, timeout=timeout)+    # a dropped/rotated session shows up as 400 Missing/Invalid session; re-init once+    if r.get("httpError") in (400, 404) and "ession" in str(r.get("body", "")):+        with _mcp_lock:+            _mcp_session_id = None+            s = _mcp_new_session()+            if s.get("error"):+                return {"_mcpError": s["error"]}+            sid = _mcp_session_id+        r = _mcp_post(body, sid, timeout=timeout)+    if r.get("error") or r.get("httpError"):+        return {"_mcpError": r.get("error") or ("HTTP %s: %s" % (r.get("httpError"), r.get("body", "")[:200]))}+    try:+        return json.loads(r.get("body") or "{}")+    except Exception:+        return {"_mcpError": "unparseable response", "raw": (r.get("body") or "")[:400]}+++_MCP_OFF_HINT = (+    "Fusion's MCP server is not reachable on 127.0.0.1:27182. It is OFF by default. The bridge "+    "can turn it on FOR the user: call fusion_mcp_enable, which opens Preferences, expands "+    "General, selects API, ticks 'Fusion MCP Server (runs locally on this device)' and clicks "+    "Apply/OK. It needs Fusion running and briefly uses the foreground (announced with an "+    "on-screen caption). Manual path if you prefer: profile icon (top right) -> Preferences -> "+    "General -> API -> tick the box. NOTE the setting does NOT survive if Fusion is killed before "+    "Apply is clicked."+)+++def _handle_mcp_status(fusion_info: dict, args: dict) -> dict:+    """Is Autodesk's Fusion MCP server up, and what does it expose?"""+    s = _mcp_new_session()+    if s.get("error"):+        return {"success": True, "enabled": False, "url": _MCP_URL, "error": s.get("error"),+                "_hint": _MCP_OFF_HINT, "statusVerb": "fusion_mcp_status"}+    tools = []+    tl = _mcp_rpc("tools/list")+    if not tl.get("_mcpError"):+        tools = [{"name": t.get("name"),+                  "description": (t.get("description") or "").split("\n")[0][:160]}+                 for t in ((tl.get("result") or {}).get("tools") or [])]+    return {"success": True, "enabled": True, "url": _MCP_URL,+            "sessionId": s.get("sessionId"), "serverInfo": s.get("serverInfo"),+            "toolCount": len(tools), "tools": tools,+            "_hint": ("Autodesk's Fusion MCP server is LIVE. Its tools are Autodesk's own, not this "+                      "bridge's: call them with fusion_mcp_call {tool, arguments}. Use "+                      "fusion_mcp_tools for full input schemas and fusion_mcp_resources for the "+                      "Electronics entity schemas. These COMPLEMENT the fusion_* verbs: prefer a "+                      "native verb when one exists (it is tested and returns richer hints), and "+                      "reach for MCP for Autodesk's text-to-CAD surface."),+            "statusVerb": "fusion_mcp_status"}+++def _handle_mcp_tools(fusion_info: dict, args: dict) -> dict:+    """Full tool list WITH input schemas."""+    tl = _mcp_rpc("tools/list")+    if tl.get("_mcpError"):+        return {"success": False, "enabled": False, "error": tl["_mcpError"], "_hint": _MCP_OFF_HINT}+    tools = (tl.get("result") or {}).get("tools") or []+    return {"success": True, "count": len(tools), "tools": tools,+            "_hint": "Call one with fusion_mcp_call {\"tool\":\"<name>\",\"arguments\":{...}}.",+            "statusVerb": "fusion_mcp_status"}+++def _handle_mcp_call(fusion_info: dict, args: dict) -> dict:+    """Call ANY tool on Autodesk's Fusion MCP server."""+    tool = (args or {}).get("tool")+    if not tool:+        return {"success": False, "error": "Pass {tool, arguments}.",+                "_hint": "fusion_mcp_tools lists the available tools and their input schemas."}+    r = _mcp_rpc("tools/call", {"name": tool, "arguments": (args or {}).get("arguments") or {}},+                 timeout=int((args or {}).get("timeout") or 120))+    if r.get("_mcpError"):+        return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+    res = r.get("result") or {}+    return {"success": not res.get("isError", False), "tool": tool, "result": res,+            "_hint": ("Autodesk MCP tool result. Content is usually a list of {type,text} blocks. "+                      "If it complains about no active document, open one first (fusion_aps_open) "+                      "- MCP acts on the ACTIVE Fusion document."),+            "statusVerb": "fusion_mcp_status"}+++def _handle_mcp_resources(fusion_info: dict, args: dict) -> dict:+    """List, or read, Autodesk's MCP resources (the Electronics entity schemas)."""+    uri = (args or {}).get("uri")+    if uri:+        r = _mcp_rpc("resources/read", {"uri": uri})+        if r.get("_mcpError"):+            return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+        return {"success": True, "uri": uri, "result": r.get("result"),+                "statusVerb": "fusion_mcp_status"}+    r = _mcp_rpc("resources/list")+    if r.get("_mcpError"):+        return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+    res = (r.get("result") or {}).get("resources") or []+    return {"success": True, "count": len(res),+            "resources": [x.get("uri") for x in res],+            "_hint": "Read one with fusion_mcp_resources {\"uri\":\"resource://...\"}.",+            "statusVerb": "fusion_mcp_status"}+++def _handle_mcp_enable(fusion_info: dict, args: dict) -> dict:+    """Turn Autodesk's MCP server ON by driving Fusion's Preferences dialog.++    Autodesk exposes NO API for this toggle (apiPreferences has debuggingPort and+    isDeveloperToolsEnabled but nothing for MCP), so the UI is the only route. Proven live+    2026-07-23. The tree children under General are rendered lazily by Qt and are NOT in the+    UIA tree, so the section click is image-space and DOES take the foreground - announced.+    """+    import time as _t+    if not _unwrap(_handle_fusion_readiness({}, {})).get("running"):+        return {"success": False, "error": "Fusion is not running.",+                "_hint": "fusion_start first, then fusion_mcp_enable."}+    already = _mcp_new_session()+    if not already.get("error"):+        return {"success": True, "enabled": True, "alreadyOn": True,+                "_hint": "Already enabled; nothing to do. fusion_mcp_status shows the tools."}++    _announce_foreground("turning on Fusion's MCP server in preferences")+    steps = []+    r = _unwrap(_ad_call("fusion_execute_text_command", {"command": "Commands.Start PreferencesCommand"}, timeout=45))+    _t.sleep(3)+    dlg = None+    for w in (_unwrap(_ad_call("desktop_list_windows", {}, timeout=40)).get("windows") or []):+        if str(w.get("title", "")).strip() == "Preferences":+            dlg = w.get("hwnd")+    if not dlg:+        return {"success": False, "error": "Preferences dialog did not open.",+                "steps": steps, "_hint": _MCP_OFF_HINT}+    steps.append("opened Preferences")++    def shot():+        s = _unwrap(_ad_call("desktop_screenshot_window", {"hwnd": int(dlg)}, timeout=45))+        cm = s.get("coordMap") or {}+        img = cm.get("image") or {}+        return cm.get("shotId"), (img.get("w") or 1400), (img.get("h") or 836)++    sid_, W, H = shot()+    if not sid_:+        return {"success": False, "error": "could not capture Preferences", "steps": steps}+    # fractional coords, measured on the real dialog (1400x836): expand arrow, then API row+    _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(dlg),+                               "x": int(W * 0.029), "y": int(H * 0.092)}, timeout=30)   # expand General+    steps.append("expanded General")+    _t.sleep(2)+    sid_, W, H = shot()+    _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(dlg),+                               "x": int(W * 0.071), "y": int(H * 0.135)}, timeout=30)   # API row+    steps.append("selected API")+    _t.sleep(2)+    sid_, W, H = shot()+    _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(dlg),+                               "x": int(W * 0.493), "y": int(H * 0.569)}, timeout=30)   # the checkbox+    steps.append("ticked 'Fusion MCP Server'")+    _t.sleep(1)+    _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(dlg),+                               "x": int(W * 0.839), "y": int(H * 0.948)}, timeout=30)   # Apply+    _t.sleep(2)+    _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(dlg),+                               "x": int(W * 0.894), "y": int(H * 0.948)}, timeout=30)   # OK+    steps.append("clicked Apply then OK")+    _t.sleep(4)++    s = _mcp_new_session()+    ok = not s.get("error")+    return {"success": ok, "enabled": ok, "steps": steps,+            "sessionId": s.get("sessionId"), "serverInfo": s.get("serverInfo"),+            "narrate": ("Fusion's MCP server is on. I enabled it for you in Preferences."+                        if ok else "I drove the Preferences dialog but the MCP port is still closed."),+            "_hint": ("MCP server ENABLED - fusion_mcp_status lists its tools. Tell the user it is "+                      "handled; they do not need to touch Preferences."+                      if ok else+                      "The toggle did not take. Ask the user to set it manually: profile icon (top "+                      "right) -> Preferences -> General -> API -> tick 'Fusion MCP Server'. Do NOT "+                      "restart or kill Fusion before they click Apply, that discards the setting."),+            "statusVerb": "fusion_mcp_status"}+++COMMAND_HANDLERS["mcp_status"] = _handle_mcp_status+COMMAND_HANDLERS["mcp_tools"] = _handle_mcp_tools+COMMAND_HANDLERS["mcp_call"] = _handle_mcp_call+COMMAND_HANDLERS["mcp_resources"] = _handle_mcp_resources+COMMAND_HANDLERS["mcp_enable"] = _handle_mcp_enable+++# ── FUSION PREFERENCES, DRIVEN THROUGH THE UI (John, 2026-07-23) ─────────────────────────────+# Fusion's Python API exposes only a THIN slice of preferences (generalPreferences: theme, orbit,+# units; apiPreferences: debuggingPort, isDeveloperToolsEnabled). Everything else - including the+# Fusion MCP Server toggle - is UI-only.+#+# But the Preferences dialog IS reachable programmatically:+#   Commands.Start PreferencesCommand   opens it (no menu hunting, no profile-icon click)+# and its LEFT-HAND SECTION TREE is exposed to UIA, so sections can be found by name.+#+# The catch, learned live: the CHILD sections under General (API, Design, Manufacture,+# Electronics, Render, Drawing, Simulation) are rendered LAZILY by Qt and never appear in the+# UIA tree, even after desktop_ui_expand reports success. So navigating into a child section+# needs an image-space click, which takes the foreground and is therefore announced with a+# caption. Everything up to that point is background.+#+# ⛔ The setting is DISCARDED unless Apply/OK is clicked. Killing or restarting Fusion with the+# dialog still open loses it (that is exactly why an earlier MCP enable silently reverted).+_PREF_SECTIONS = {+    # section -> (parent-to-expand or None, fractional x, fractional y) on the 1400x836 dialog+    "general":     (None,      0.071, 0.092),+    "api":         ("general", 0.071, 0.135),+    "design":      ("general", 0.071, 0.179),+    "manufacture": ("general", 0.071, 0.222),+    "electronics": ("general", 0.071, 0.265),+    "render":      ("general", 0.071, 0.310),+    "drawing":     ("general", 0.071, 0.353),+    "material":    (None,      0.071, 0.483),+    "graphics":    (None,      0.071, 0.527),+    "network":     (None,      0.071, 0.570),+    "preview features": (None, 0.071, 0.744),+}+++def _prefs_dialog_hwnd():+    for w in (_unwrap(_ad_call("desktop_list_windows", {}, timeout=40)).get("windows") or []):+        if str(w.get("title", "")).strip() == "Preferences":+            return w.get("hwnd")+    return None+++def _prefs_shot(hwnd):+    s = _unwrap(_ad_call("desktop_screenshot_window", {"hwnd": int(hwnd)}, timeout=45))+    cm = s.get("coordMap") or {}+    img = cm.get("image") or {}+    return (cm.get("shotId"), img.get("w") or 1400, img.get("h") or 836,+            s.get("localSafePath"))+++def _handle_prefs_open(fusion_info: dict, args: dict) -> dict:+    """Open Fusion Preferences, optionally navigate to a section, and hand back a screenshot the+    AI can click in. This is how you reach ANY preference, not just the API-exposed few."""+    import time as _t+    if not _unwrap(_handle_fusion_readiness({}, {})).get("running"):+        return {"success": False, "error": "Fusion is not running.",+                "_hint": "fusion_start first."}+    section = str((args or {}).get("section") or "").strip().lower()+    steps = []+    hwnd = _prefs_dialog_hwnd()+    if not hwnd:+        _announce_foreground("opening Fusion preferences")+        _ad_call("fusion_execute_text_command",+                 {"command": "Commands.Start PreferencesCommand"}, timeout=45)+        _t.sleep(3)+        hwnd = _prefs_dialog_hwnd()+        steps.append("opened Preferences")+    if not hwnd:+        return {"success": False, "error": "Preferences dialog did not open.", "steps": steps}++    if section:+        spec = _PREF_SECTIONS.get(section)+        if not spec:+            return {"success": False, "error": "Unknown section %r." % section,+                    "knownSections": sorted(_PREF_SECTIONS),+                    "_hint": ("Pass one of knownSections, or omit `section` and click the returned "+                              "shotId yourself with desktop_click {space:'image'}.")}+        parent, fx, fy = spec+        sid_, W, H, _p = _prefs_shot(hwnd)+        if parent:+            pspec = _PREF_SECTIONS[parent]+            # click the expand arrow, left of the parent label+            _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(hwnd),+                                       "x": int(W * 0.029), "y": int(H * pspec[2])}, timeout=30)+            steps.append("expanded %s" % parent)+            _t.sleep(2)+            sid_, W, H, _p = _prefs_shot(hwnd)+        _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(hwnd),+                                   "x": int(W * fx), "y": int(H * fy)}, timeout=30)+        steps.append("selected %s" % section)+        _t.sleep(2)++    sid_, W, H, path = _prefs_shot(hwnd)+    return {"success": True, "hwnd": hwnd, "section": section or "general",+            "shotId": sid_, "imageWidth": W, "imageHeight": H, "screenshot": path,+            "steps": steps, "knownSections": sorted(_PREF_SECTIONS),+            "_hint": ("Preferences is OPEN and showing this section. LOOK at the screenshot, then "+                      "click any control with desktop_click {space:'image', shotId, x, y, hwnd}. "+                      "⛔ Nothing is saved until you call fusion_prefs_close {save:true} (Apply+OK) "+                      "- killing or restarting Fusion first DISCARDS the change. Child sections "+                      "under General are lazily rendered and absent from the UIA tree, which is why "+                      "this verb hands you an image to click rather than control names."),+            "statusVerb": "fusion_get_preferences"}+++def _handle_prefs_close(fusion_info: dict, args: dict) -> dict:+    """Apply+OK (save:true, default) or Cancel the Preferences dialog."""+    import time as _t+    save = (args or {}).get("save", True)+    hwnd = _prefs_dialog_hwnd()+    if not hwnd:+        return {"success": True, "closed": False,+                "_hint": "Preferences was not open; nothing to do."}+    sid_, W, H, _p = _prefs_shot(hwnd)+    if save:+        _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(hwnd),+                                   "x": int(W * 0.839), "y": int(H * 0.948)}, timeout=30)  # Apply+        _t.sleep(2)+        _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(hwnd),+                                   "x": int(W * 0.894), "y": int(H * 0.948)}, timeout=30)  # OK+    else:+        _ad_call("desktop_click", {"space": "image", "shotId": sid_, "hwnd": int(hwnd),+                                   "x": int(W * 0.951), "y": int(H * 0.948)}, timeout=30)  # Cancel+    _t.sleep(2)+    return {"success": True, "closed": _prefs_dialog_hwnd() is None, "saved": bool(save),+            "narrate": ("Saved your Fusion preferences." if save else "Closed preferences without saving."),+            "_hint": ("Applied and closed. Some settings (the MCP server is one) only take effect "+                      "once saved this way." if save else "Cancelled; nothing was changed."),+            "statusVerb": "fusion_get_preferences"}+++COMMAND_HANDLERS["prefs_open"] = _handle_prefs_open+COMMAND_HANDLERS["prefs_close"] = _handle_prefs_close++ def _fusion_signin_cfg_path():     import pathlib     d = pathlib.Path(os.path.expanduser("~")) / ".adom" / "fusion-signin"