diff --git a/tools/hwlab-node.py b/tools/hwlab-node.py
index f6ce01ea..562a4cb9 100644
--- a/tools/hwlab-node.py
+++ b/tools/hwlab-node.py
@@ -305,6 +305,8 @@ class NodeOpsExecutor:
     def __init__(self, config: dict, logger: logging.Logger):
         self.config = config
         self.logger = logger
+        self.uart = None
+        self.uart_lock = threading.Lock()
 
     @property
     def node_id(self) -> str:
@@ -801,6 +803,8 @@ class NodeOpsExecutor:
         return ["bun", "scripts/serial-monitor-cli.ts"]
 
     def _uart_control(self, args: dict, action: str) -> dict:
+        if str(args.get("uartBackend") or "").lower() == "pyserial":
+            return self._pyserial_control(args, action)
         serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
         command_base = self._serial_command_base(args)
         timeout_ms = safe_int(args.get("timeoutMs"), 10000)
@@ -831,6 +835,8 @@ class NodeOpsExecutor:
         data = str(args.get("data") or "")
         if not data:
             return {"ok": False, "blockerCode": "hwpod_uart_data_required", "stderr": "UART data is required", "exitCode": None}
+        if str(args.get("uartBackend") or "").lower() == "pyserial":
+            return self._pyserial_write(args, data)
         serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
         command = [*self._serial_command_base(args), "send", data]
         output = self._spawn_output(command, serial_dir, safe_int(args.get("timeoutMs"), 10000))
@@ -838,6 +844,8 @@ class NodeOpsExecutor:
         return {**output, "ok": output.get("ok") and body.get("success") is True, "bindingSource": "serial-monitor-cli", "bytes": len(data.encode("utf-8")), "response": body or None}
 
     def _uart_read(self, args: dict) -> dict:
+        if str(args.get("uartBackend") or "").lower() == "pyserial":
+            return self._pyserial_read(args)
         serial_dir = Path(str(args.get("serialMonitorDir") or Path.home() / ".agents" / "skills" / "serial-monitor"))
         command_base = self._serial_command_base(args)
         timeout_ms = safe_int(args.get("timeoutMs"), 10000)
@@ -891,6 +899,41 @@ class NodeOpsExecutor:
             observation["rawArtifactRef"] = source_file
         return {"ok": True, "workspacePath": str(self._workspace_root(args)), "bindingSource": "serial-monitor-cli", "serialMonitorDir": str(serial_dir), "requestedPort": port or None, "resolvedPort": resolved_port, "baudRate": resolved_baud, "command": [*command_base, *fetch_args], "data": rows, "count": fetch_body.get("count", len(rows)), "totalCount": fetch_body.get("totalCount"), "hasMore": fetch_body.get("hasMore", False), "text": truncated_text, "truncated": len(text) > max_bytes or fetch_body.get("truncated") is True, "monitorStatus": status_body, "monitorStarted": not matches, "monitorStart": monitor_start, "sourceFile": source_file or None, "observation": observation}
 
+    def _pyserial_control(self, args: dict, action: str) -> dict:
+        with self.uart_lock:
+            if action == "stop":
+                if self.uart is not None:
+                    self.uart.close()
+                    self.uart = None
+                return {"ok": True, "bindingSource": "pyserial", "status": "closed"}
+            import serial
+            port = str(args.get("port") or "")
+            baud = safe_int(args.get("baudRate") or args.get("baudrate"), 115200)
+            if self.uart is not None and (self.uart.port != port or self.uart.baudrate != baud):
+                self.uart.close()
+                self.uart = None
+            if self.uart is None or not self.uart.is_open:
+                self.uart = serial.Serial(port=port, baudrate=baud, timeout=0)
+            return {"ok": True, "bindingSource": "pyserial", "status": "open", "port": port, "baudRate": baud}
+
+    def _pyserial_write(self, args: dict, data: str) -> dict:
+        with self.uart_lock:
+            if self.uart is None or not self.uart.is_open:
+                return {"ok": False, "blockerCode": "hwpod_uart_not_open", "stderr": "UART must be opened before write", "exitCode": None}
+            payload = data.encode(str(args.get("encoding") or "utf-8"))
+            written = self.uart.write(payload)
+            return {"ok": written == len(payload), "bindingSource": "pyserial", "bytes": written, "port": self.uart.port, "baudRate": self.uart.baudrate}
+
+    def _pyserial_read(self, args: dict) -> dict:
+        with self.uart_lock:
+            if self.uart is None or not self.uart.is_open:
+                return {"ok": False, "blockerCode": "hwpod_uart_not_open", "summary": "UART must be opened before read"}
+            max_bytes = max(1, min(safe_int(args.get("maxBytes"), 16384), 1048576))
+            available = min(safe_int(getattr(self.uart, "in_waiting", 0), 0), max_bytes)
+            payload = self.uart.read(available) if available else b""
+            text = payload.decode(str(args.get("encoding") or "utf-8"), errors="replace")
+            return {"ok": True, "bindingSource": "pyserial", "port": self.uart.port, "baudRate": self.uart.baudrate, "bytes": len(payload), "text": text, "dataBase64": base64.b64encode(payload).decode("ascii"), "truncated": safe_int(getattr(self.uart, "in_waiting", 0), 0) > 0}
+
     def _parse_json(self, text: object) -> dict:
         try:
             value = json.loads(str(text or ""))
diff --git a/tools/hwpod-harness.test.ts b/tools/hwpod-harness.test.ts
index 4b48b81b..b400c420 100644
--- a/tools/hwpod-harness.test.ts
+++ b/tools/hwpod-harness.test.ts
@@ -267,9 +267,18 @@ test("hwpod-cli dry-run invokes hwpod-compiler-cli and exposes hwpod-node-ops pl
     assert.equal(uartSessionOnly.payload.plan.ops[0].args.sessionOnly, true);
     assert.equal(uartSessionOnly.payload.plan.ops[0].args.commandBinding.sessionOnly, true);
 
+    await runHwpodCtl([
+      "spec",
+      "set",
+      "spec.ioProbe.uart.backend",
+      "pyserial",
+      "--spec",
+      specPath
+    ], { now: () => NOW });
     const uartOpen = await runHwpodCli(["uart", "open", "--spec", specPath, "--port", "uart1", "--dry-run"], { now: () => NOW });
     assert.equal(uartOpen.payload.plan.ops[0].op, "io.uart.open");
     assert.equal(uartOpen.payload.plan.ops[0].args.port, "COM9");
+    assert.equal(uartOpen.payload.plan.ops[0].args.uartBackend, "pyserial");
     const uartWrite = await runHwpodCli(["uart", "write", "--spec", specPath, "--port", "uart1", "--data", "status", "--dry-run"], { now: () => NOW });
     assert.equal(uartWrite.payload.plan.ops[0].op, "io.uart.write");
     assert.equal(uartWrite.payload.plan.ops[0].args.data, "status");
diff --git a/tools/hwpod-node.test.ts b/tools/hwpod-node.test.ts
index d154e95b..7979fb4b 100644
--- a/tools/hwpod-node.test.ts
+++ b/tools/hwpod-node.test.ts
@@ -430,6 +430,49 @@ print(json.dumps({"result": result, "commands": executor.commands}))
   ]);
 });
 
+test("python hwpod-node keeps a pyserial connection across UART operations", () => {
+  const script = `
+import importlib.util, json, logging, sys, types
+spec = importlib.util.spec_from_file_location("hwlab_node", "tools/hwlab-node.py")
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+class Port:
+    def __init__(self, port, baudrate, timeout):
+        self.port = port
+        self.baudrate = baudrate
+        self.is_open = True
+        self.buffer = bytearray(b"ready")
+        self.sent = bytearray()
+    @property
+    def in_waiting(self): return len(self.buffer)
+    def read(self, size):
+        value = bytes(self.buffer[:size])
+        del self.buffer[:size]
+        return value
+    def write(self, value):
+        self.sent.extend(value)
+        return len(value)
+    def close(self): self.is_open = False
+sys.modules["serial"] = types.SimpleNamespace(Serial=Port)
+executor = module.NodeOpsExecutor({"nodeId": "node-test"}, logging.getLogger("test"))
+common = {"workspacePath": ".", "port": "COM4", "baudRate": 921600, "uartBackend": "pyserial"}
+results = [
+  executor._execute_op({"opId": "open", "op": "io.uart.open", "args": common}),
+  executor._execute_op({"opId": "write", "op": "io.uart.write", "args": {**common, "data": "status"}}),
+  executor._execute_op({"opId": "read", "op": "io.uart.read", "args": {**common, "maxBytes": 16}}),
+  executor._execute_op({"opId": "close", "op": "io.uart.close", "args": common}),
+]
+print(json.dumps(results))
+`;
+  const completed = spawnSync("python3", ["-c", script], { cwd: process.cwd(), encoding: "utf8" });
+  assert.equal(completed.status, 0, completed.stderr);
+  const results = JSON.parse(completed.stdout);
+  assert.deepEqual(results.map((item: any) => item.status), ["completed", "completed", "completed", "completed"]);
+  assert.equal(results[1].output.bytes, 6);
+  assert.equal(results[2].output.text, "ready");
+  assert.equal(results[2].output.bindingSource, "pyserial");
+});
+
 test("hwpod-node resolves Git for Windows when service PATH omits git", async () => {
   const result = await resolveHwpodNodeCommand("git", {
     platform: "win32",
diff --git a/tools/src/hwpod-harness-lib.ts b/tools/src/hwpod-harness-lib.ts
index 199663ad..08920d85 100644
--- a/tools/src/hwpod-harness-lib.ts
+++ b/tools/src/hwpod-harness-lib.ts
@@ -894,6 +894,7 @@ function uartReadOps(common: any, args: any, document: any) {
       limit: numberValue(args.limit),
       since: text(args.since),
       sessionOnly: args.sessionOnly === true,
+      uartBackend: generated.backend,
       serialMonitorDir: generated.serialMonitorDir,
       serialMonitorCommand: generated.commandBase,
       commandBinding: generated.binding,
@@ -907,7 +908,7 @@ function uartControlOps(common: any, intent: "io.uart.open" | "io.uart.close", a
   const generated = serialMonitorBinding(args, document);
   return [{
     op: intent,
-    args: clean({ ...common, port: generated.physicalPort, baudRate: generated.baudRate, serialMonitorDir: generated.serialMonitorDir, serialMonitorCommand: generated.commandBase, timeoutMs: generated.timeoutMs, reason: text(args.reason) })
+    args: clean({ ...common, port: generated.physicalPort, baudRate: generated.baudRate, uartBackend: generated.backend, serialMonitorDir: generated.serialMonitorDir, serialMonitorCommand: generated.commandBase, timeoutMs: generated.timeoutMs, reason: text(args.reason) })
   }];
 }
 
@@ -915,7 +916,7 @@ function uartWriteOps(common: any, args: any, document: any) {
   const generated = serialMonitorBinding(args, document);
   return [{
     op: "io.uart.write",
-    args: clean({ ...common, port: generated.physicalPort, baudRate: generated.baudRate, data: requiredText(args.data, "data"), serialMonitorDir: generated.serialMonitorDir, serialMonitorCommand: generated.commandBase, timeoutMs: generated.timeoutMs, reason: text(args.reason) })
+    args: clean({ ...common, port: generated.physicalPort, baudRate: generated.baudRate, data: requiredText(args.data, "data"), uartBackend: generated.backend, serialMonitorDir: generated.serialMonitorDir, serialMonitorCommand: generated.commandBase, timeoutMs: generated.timeoutMs, reason: text(args.reason) })
   }];
 }
 
@@ -1043,6 +1044,7 @@ function serialMonitorBinding(args: any, document: any) {
   const requestedPort = text(args.port) || text(uart.id) || "uart1";
   return {
     requestedPort,
+    backend: text(args.uartBackend) || text(uart.backend) || "serial-monitor",
     physicalPort: text(args.physicalPort) || text(uart.port) || text(uart.path) || text(ioProbe.port) || requestedPort,
     baudRate: numberValue(args.baudRate ?? args.baudrate ?? uart.baudRate ?? uart.baudrate ?? ioProbe.baudRate ?? ioProbe.baudrate) ?? 115200,
     serialMonitorDir: text(args.serialMonitorDir) || text(tooling.serialMonitorDir) || text(ioProbe.serialMonitorDir) || "C:\\Users\\liang\\.agents\\skills\\serial-monitor",