Expose USB passthrough over REST API and first-class MCP tools

Extract the passthrough commands into a single source of truth
(passthrough/commands.py) and drive every surface from it:

- REST: /usb/passthrough/{status,enable}, /usb/acl{,/add,/remove,/default},
  /usb/loopback/{devices,open}, /usb/remote/{devices,open} (bearer-gated;
  ACL export/import deliberately omitted to avoid server-side file paths).
  OpenAPI spec updated.
- MCP: first-class ac_usb_* tools with JSON Schemas and read-only/non-
  destructive annotations, so an agent calls them directly instead of via
  ac_execute_actions.
- The AC_usb_* executor adapters now delegate to the same module.

Tests cover REST (auth, 400/500 paths) and MCP (registration, schema,
read-only filtering). Docs updated; import je_auto_control stays Qt-free.

Author: JE-Chen

docs/source/Eng/doc/operations_layer/usb_passthrough_operator_guide.rst

@@ -301,6 +301,15 @@ Example JSON action::
       ["AC_usb_loopback_open", {"vendor_id": "1050", "product_id": "0407"}]
     ]
 
+The same operations are exposed over two more surfaces:
+
+* **REST API** — ``GET/POST /usb/passthrough/...``, ``/usb/acl...``,
+  ``/usb/loopback/...``, ``/usb/remote/...`` (bearer-token gated; see
+  ``/openapi.json``). ACL export/import are intentionally *not* on REST
+  (server-side file paths).
+* **MCP** — first-class ``ac_usb_*`` tools (``ac_usb_loopback_open`` …)
+  with JSON Schemas, so an agent can call them directly.
+
 
 What is *not* shipped yet
 =========================

docs/source/Zh/doc/operations_layer/usb_passthrough_operator_guide.rst

@@ -281,6 +281,15 @@ JSON action 範例::
       ["AC_usb_loopback_open", {"vendor_id": "1050", "product_id": "0407"}]
     ]
 
+同樣的操作另外提供兩個介面：
+
+* **REST API** — ``GET/POST /usb/passthrough/...``、``/usb/acl...``、
+  ``/usb/loopback/...``、``/usb/remote/...``\ （需 bearer token；見
+  ``/openapi.json``\ ）。ACL 匯入/匯出刻意 **不** 開 REST（伺服器端
+  檔案路徑風險）。
+* **MCP** — 一級 ``ac_usb_*`` 工具（``ac_usb_loopback_open`` …），帶
+  JSON Schema，agent 可直接呼叫。
+
 
 尚未發布的部分
 ==============

je_auto_control/utils/executor/action_executor.py

@@ -957,134 +957,74 @@ def _usb_recent_events(since: int = 0,
     )
 
 
-# --- USB passthrough (Phase 2) ------------------------------------------
-
+# --- USB passthrough (Phase 2) — delegate to the shared command module --
 
 def _usb_passthrough_enable(enabled: bool = True) -> Dict[str, Any]:
-    """Toggle the USB passthrough feature flag (default off)."""
-    from je_auto_control.utils.usb.passthrough import (
-        enable_usb_passthrough, is_usb_passthrough_enabled,
-    )
-    enable_usb_passthrough(bool(enabled))
-    return {"enabled": is_usb_passthrough_enabled()}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.passthrough_enable(enabled)
 
 
 def _usb_passthrough_status() -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import is_usb_passthrough_enabled
-    return {"enabled": is_usb_passthrough_enabled()}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.passthrough_status()
 
 
 def _usb_acl_list() -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import UsbAcl
-    acl = UsbAcl()
-    return {
-        "default": acl.default_policy,
-        "integrity_ok": acl.integrity_ok,
-        "rules": [r.to_dict() for r in acl.list_rules()],
-    }
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_list()
 
 
 def _usb_acl_add(vendor_id: str, product_id: str,
                  serial: Optional[str] = None, allow: bool = True,
                  prompt_on_open: bool = False, label: str = "") -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import AclRule, UsbAcl
-    acl = UsbAcl()
-    acl.add_rule(AclRule(
-        vendor_id=str(vendor_id), product_id=str(product_id),
-        serial=(None if serial is None else str(serial)),
-        label=str(label), allow=bool(allow),
-        prompt_on_open=bool(prompt_on_open),
-    ))
-    return {"added": True, "rules": len(acl.list_rules())}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_add(
+        vendor_id, product_id, serial=serial, allow=allow,
+        prompt_on_open=prompt_on_open, label=label,
+    )
 
 
 def _usb_acl_remove(vendor_id: str, product_id: str,
                     serial: Optional[str] = None) -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import UsbAcl
-    removed = UsbAcl().remove_rule(
-        vendor_id=str(vendor_id), product_id=str(product_id),
-        serial=(None if serial is None else str(serial)),
-    )
-    return {"removed": removed}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_remove(vendor_id, product_id, serial=serial)
 
 
 def _usb_acl_set_default(policy: str) -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import UsbAcl
-    acl = UsbAcl()
-    acl.set_default_policy(str(policy))
-    return {"default": acl.default_policy}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_set_default(policy)
 
 
 def _usb_acl_export(path: str) -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import UsbAcl, export_acl_to_file
-    from pathlib import Path
-    export_acl_to_file(UsbAcl(), Path(path))
-    return {"exported": True, "path": str(path)}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_export(path)
 
 
 def _usb_acl_import(path: str, replace: bool = False) -> Dict[str, Any]:
-    from je_auto_control.utils.usb.passthrough import UsbAcl, import_acl_from_file
-    from pathlib import Path
-    count = import_acl_from_file(UsbAcl(), Path(path), replace=bool(replace))
-    return {"imported": count}
-
-
-def _usb_descriptor_probe(client: Any, vendor_id: str, product_id: str,
-                          serial: Optional[str]) -> Dict[str, Any]:
-    """Open a claim, read the device descriptor, close — return a summary."""
-    from je_auto_control.utils.usb.passthrough import describe_descriptor
-    handle = client.open(
-        vendor_id=str(vendor_id), product_id=str(product_id),
-        serial=(None if serial is None else str(serial)),
-    )
-    try:
-        descriptor = handle.control_transfer(
-            bm_request_type=0x80, b_request=0x06, w_value=0x0100, length=18,
-        )
-    finally:
-        handle.close()
-    return {
-        "ok": True, "vendor_id": str(vendor_id), "product_id": str(product_id),
-        "descriptor_hex": descriptor.hex(),
-        "descriptor": describe_descriptor(descriptor),
-    }
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_import(path, replace=replace)
 
 
 def _usb_loopback_list() -> Dict[str, Any]:
-    """List ACL-visible devices over the in-process loopback channel."""
-    from je_auto_control.utils.usb.passthrough import UsbAcl, UsbLoopback
-    with UsbLoopback(acl=UsbAcl(), viewer_id="executor") as loop:
-        return {"devices": loop.list_devices()}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.loopback_list()
 
 
 def _usb_loopback_open(vendor_id: str, product_id: str,
                        serial: Optional[str] = None) -> Dict[str, Any]:
-    """Claim a local device over loopback and read its descriptor."""
-    from je_auto_control.utils.usb.passthrough import UsbAcl, UsbLoopback
-    with UsbLoopback(acl=UsbAcl(), viewer_id="executor") as loop:
-        return _usb_descriptor_probe(loop, vendor_id, product_id, serial)
-
-
-def _usb_remote_client() -> Any:
-    """Return the live WebRTC viewer's USB client or raise a clear error."""
-    from je_auto_control.utils.remote_desktop.registry import registry
-    client = registry.webrtc_usb_client()
-    if client is None:
-        raise RuntimeError("no live WebRTC viewer with a usb channel")
-    return client
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.loopback_open(vendor_id, product_id, serial=serial)
 
 
 def _usb_remote_list() -> Dict[str, Any]:
-    """List the remote host's devices over the live WebRTC usb channel."""
-    return {"devices": _usb_remote_client().list_devices()}
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.remote_list()
 
 
 def _usb_remote_open(vendor_id: str, product_id: str,
                      serial: Optional[str] = None) -> Dict[str, Any]:
-    """Claim a remote device over the live WebRTC usb channel + probe."""
-    return _usb_descriptor_probe(
-        _usb_remote_client(), vendor_id, product_id, serial,
-    )
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.remote_open(vendor_id, product_id, serial=serial)
 
 
 def _ac_web_run(action: Optional[Dict[str, Any]] = None,

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -1779,6 +1779,113 @@ def gamepad_tools() -> List[MCPTool]:
     ]
 
 
+_VID_PID = {
+    "vendor_id": {"type": "string"},
+    "product_id": {"type": "string"},
+    "serial": {"type": "string"},
+}
+
+
+def usb_passthrough_tools() -> List[MCPTool]:
+    """First-class MCP tools for USB passthrough (default-off feature)."""
+    return [
+        MCPTool(
+            name="ac_usb_passthrough_enable",
+            description=("Toggle the USB passthrough feature flag. Default "
+                         "off; must be enabled before any usb channel is "
+                         "honoured."),
+            input_schema=schema({"enabled": {"type": "boolean"}}),
+            handler=h.usb_passthrough_enable,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_usb_passthrough_status",
+            description="Report whether USB passthrough is enabled.",
+            input_schema=schema({}),
+            handler=h.usb_passthrough_status,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_usb_acl_list",
+            description=("List USB ACL rules plus the default policy and the "
+                         "HMAC integrity state."),
+            input_schema=schema({}),
+            handler=h.usb_acl_list,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_usb_acl_add",
+            description=("Add a per-device USB ACL rule (allow/deny, optional "
+                         "prompt-on-open). vendor_id/product_id are 4 hex "
+                         "digits, e.g. 1050/0407."),
+            input_schema=schema({
+                **_VID_PID,
+                "allow": {"type": "boolean"},
+                "prompt_on_open": {"type": "boolean"},
+                "label": {"type": "string"},
+            }, required=["vendor_id", "product_id"]),
+            handler=h.usb_acl_add,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_usb_acl_remove",
+            description="Remove a per-device USB ACL rule.",
+            input_schema=schema(
+                dict(_VID_PID), required=["vendor_id", "product_id"],
+            ),
+            handler=h.usb_acl_remove,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_usb_acl_set_default",
+            description="Set the USB ACL default policy (allow | deny).",
+            input_schema=schema({
+                "policy": {"type": "string", "enum": ["allow", "deny"]},
+            }, required=["policy"]),
+            handler=h.usb_acl_set_default,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_usb_loopback_list",
+            description=("List ACL-visible USB devices on this machine over "
+                         "the in-process loopback channel."),
+            input_schema=schema({}),
+            handler=h.usb_loopback_list,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_usb_loopback_open",
+            description=("Claim a local USB device over loopback and read its "
+                         "device descriptor (a full protocol-stack probe). "
+                         "Fails closed if the ACL denies it."),
+            input_schema=schema(
+                dict(_VID_PID), required=["vendor_id", "product_id"],
+            ),
+            handler=h.usb_loopback_open,
+            annotations=NON_DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_usb_remote_list",
+            description=("List the remote host's USB devices over the live "
+                         "WebRTC usb channel. Requires a connected WebRTC "
+                         "viewer."),
+            input_schema=schema({}),
+            handler=h.usb_remote_list,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_usb_remote_open",
+            description=("Claim a remote USB device over the live WebRTC usb "
+                         "channel and read its descriptor."),
+            input_schema=schema(
+                dict(_VID_PID), required=["vendor_id", "product_id"],
+            ),
+            handler=h.usb_remote_open,
+            annotations=NON_DESTRUCTIVE,
+        ),
+    ]
+
+
 ALL_FACTORIES = (
     mouse_tools, keyboard_tools, screen_tools, image_and_ocr_tools,
     window_tools, system_tools, recording_tools, drag_and_send_tools,
@@ -1789,4 +1896,5 @@ def gamepad_tools() -> List[MCPTool]:
     redaction_tools, android_widget_tools, ios_tools, webrunner_tools,
     scheduler_tools, trigger_tools, hotkey_tools, screen_record_tools,
     process_and_shell_tools, remote_desktop_tools, gamepad_tools,
+    usb_passthrough_tools,
 )

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -1535,3 +1535,64 @@ def gamepad_reset() -> Dict[str, Any]:
     from je_auto_control.utils.gamepad import default_gamepad
     default_gamepad().reset()
     return {"reset": True}
+
+
+# --- USB passthrough — delegate to the shared command module -------------
+
+
+def usb_passthrough_enable(enabled: bool = True) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.passthrough_enable(enabled)
+
+
+def usb_passthrough_status() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.passthrough_status()
+
+
+def usb_acl_list() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_list()
+
+
+def usb_acl_add(vendor_id: str, product_id: str,
+                serial: Optional[str] = None, allow: bool = True,
+                prompt_on_open: bool = False, label: str = "") -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_add(
+        vendor_id, product_id, serial=serial, allow=allow,
+        prompt_on_open=prompt_on_open, label=label,
+    )
+
+
+def usb_acl_remove(vendor_id: str, product_id: str,
+                   serial: Optional[str] = None) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_remove(vendor_id, product_id, serial=serial)
+
+
+def usb_acl_set_default(policy: str) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.acl_set_default(policy)
+
+
+def usb_loopback_list() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.loopback_list()
+
+
+def usb_loopback_open(vendor_id: str, product_id: str,
+                      serial: Optional[str] = None) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.loopback_open(vendor_id, product_id, serial=serial)
+
+
+def usb_remote_list() -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.remote_list()
+
+
+def usb_remote_open(vendor_id: str, product_id: str,
+                    serial: Optional[str] = None) -> Dict[str, Any]:
+    from je_auto_control.utils.usb.passthrough import commands
+    return commands.remote_open(vendor_id, product_id, serial=serial)