Add vm_mouse_move: chunked relative PS/2 motion with corner homing and clicks

This commit is contained in:
Ryan Malloy 2026-08-17 10:39:23 -06:00
parent 7020f96fa2
commit 1dbe453731
3 changed files with 165 additions and 2 deletions

View File

@ -29,7 +29,7 @@ claude mcp add mcqemu -- uv run --directory /path/to/mcqemu mcqemu
| Sandboxes | `sandbox_vm` (overlay + launch + wait-for-agent in one call), `sandbox_destroy` | | Sandboxes | `sandbox_vm` (overlay + launch + wait-for-agent in one call), `sandbox_destroy` |
| Inspect | `list_vms`, `vm_info` | | Inspect | `list_vms`, `vm_info` |
| Live snapshots | `vm_snapshot_create` / `restore` / `delete` / `list` | | Live snapshots | `vm_snapshot_create` / `restore` / `delete` / `list` |
| See & drive | `vm_screenshot` (PNG), `vm_send_keys`, `vm_type_text`, `vm_click`, `vm_serial_read` | | See & drive | `vm_screenshot` (PNG), `vm_send_keys`, `vm_type_text`, `vm_click`, `vm_mouse_move` (relative PS/2, for guests without tablet drivers), `vm_serial_read` |
| Disk images | `image_create`, `image_info`, `image_convert`, `image_resize`, `image_snapshot_*` | | Disk images | `image_create`, `image_info`, `image_convert`, `image_resize`, `image_snapshot_*` |
| Guest agent | `guest_ping`, `guest_info`, `guest_exec`, `guest_file_read`, `guest_file_write` | | Guest agent | `guest_ping`, `guest_info`, `guest_exec`, `guest_file_read`, `guest_file_write` |

View File

@ -1,8 +1,10 @@
"""See & drive: screenshots, keyboard/mouse injection, serial console.""" """See & drive: screenshots, keyboard/mouse injection, serial console."""
import asyncio import asyncio
import math
import struct import struct
from pathlib import Path from pathlib import Path
from typing import Literal
from fastmcp import Context, FastMCP from fastmcp import Context, FastMCP
from fastmcp.exceptions import ToolError from fastmcp.exceptions import ToolError
@ -15,6 +17,9 @@ from ._common import require_vm
_ABS_MAX = 32767 # QEMU absolute-pointer coordinate space _ABS_MAX = 32767 # QEMU absolute-pointer coordinate space
# PS/2-style relative buttons for HMP mouse_button (bitmask).
_MOUSE_BUTTONS = {"left": 1, "right": 2, "middle": 4}
def _png_dimensions(data: bytes) -> tuple[int, int]: def _png_dimensions(data: bytes) -> tuple[int, int]:
if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n": if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n":
@ -145,6 +150,87 @@ async def vm_click(
return {"name": name, "clicked": f"{button} at ({x}, {y})", "double": double} return {"name": name, "clicked": f"{button} at ({x}, {y})", "double": double}
def _steps(delta: int, step: int) -> list[int]:
"""Split a delta into chunks no larger than `step` (preserving sign)."""
if delta == 0:
return []
sign = 1 if delta > 0 else -1
full, rest = divmod(abs(delta), step)
return [sign * step] * full + ([sign * rest] if rest else [])
async def vm_mouse_move(
name: str,
dx: int = 0,
dy: int = 0,
home: Literal["top-left", "top-right", "bottom-left", "bottom-right"] | None = None,
click: Literal["left", "right", "middle"] | None = None,
double: bool = False,
step: int = 32,
ctx: Context = None,
) -> dict:
"""Relative mouse control for guests WITHOUT absolute-pointer (tablet)
drivers most pre-2010 OSes. Use this when vm_click has no visible
effect. Motion goes to the emulated PS/2 mouse in small steps (<= `step`
px per packet) because guests often desync or apply acceleration on large
deltas.
Recommended pattern: pass home="bottom-right" (or another corner) to pin
the cursor to a known position first, then dx/dy toward the target, then
vm_screenshot to verify where the cursor actually landed (guest
acceleration may scale motion), correct with further small moves, and
finally click. `click` presses that button after moving; double=True
double-clicks."""
if click is not None and click not in _MOUSE_BUTTONS:
raise ToolError(f"Unsupported button {click!r}: use left, right, or middle.")
if step < 1 or step > 120:
raise ToolError("step must be between 1 and 120 (PS/2 deltas are small signed bytes).")
_, record = require_vm(ctx, name)
homed_steps = 0
if home is not None:
# Screen size bounds how far the corner can be; overshoot a little so
# the cursor pins against the edge regardless of starting position.
png = await _screendump(ctx, name)
width, height = _png_dimensions(png)
sx = -1 if "left" in home else 1
sy = -1 if "top" in home else 1
homed_steps = math.ceil(max(width, height) / step) + 2
async with qmp_session(record) as client:
async def hmp(cmd: str):
return await execute(client, "human-monitor-command", {"command-line": cmd})
for _ in range(homed_steps):
await hmp(f"mouse_move {sx * step} {sy * step}")
await asyncio.sleep(0.02)
for chunk_x in _steps(dx, step):
await hmp(f"mouse_move {chunk_x} 0")
await asyncio.sleep(0.02)
for chunk_y in _steps(dy, step):
await hmp(f"mouse_move 0 {chunk_y}")
await asyncio.sleep(0.02)
if click is not None:
for _ in range(2 if double else 1):
await hmp(f"mouse_button {_MOUSE_BUTTONS[click]}")
await asyncio.sleep(0.08)
await hmp("mouse_button 0")
await asyncio.sleep(0.12)
return {
"name": name,
"homed": home,
"moved": [dx, dy],
"clicked": f"{click}{' double' if double and click else ''}" if click else None,
"note": (
"Motion is relative and the guest may scale it (acceleration). "
"Take a vm_screenshot to verify the cursor position before clicking."
),
}
async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) -> dict: async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) -> dict:
"""Read the last lines of the VM's serial console log. Only useful when """Read the last lines of the VM's serial console log. Only useful when
the guest writes to its serial port (kernel console=ttyS0, or text-mode the guest writes to its serial port (kernel console=ttyS0, or text-mode
@ -167,5 +253,5 @@ async def vm_serial_read(name: str, tail_lines: int = 50, ctx: Context = None) -
def register(mcp: FastMCP) -> None: def register(mcp: FastMCP) -> None:
for fn in (vm_screenshot, vm_send_keys, vm_type_text, vm_click, vm_serial_read): for fn in (vm_screenshot, vm_send_keys, vm_type_text, vm_click, vm_mouse_move, vm_serial_read):
mcp.tool(fn) mcp.tool(fn)

View File

@ -119,3 +119,80 @@ async def test_serial_read_tail(dirs, fake_qmp):
) )
assert data["total_lines"] == 100 assert data["total_lines"] == 100
assert data["tail"] == "line97\nline98\nline99" assert data["tail"] == "line97\nline98\nline99"
def hmp_calls():
return [a["command-line"] for c, a in FakeQMPClient.calls if c == "human-monitor-command"]
async def test_mouse_move_chunks_deltas(dirs, fake_qmp):
FakeQMPClient.responses["human-monitor-command"] = ""
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
data = result_data(
await client.call_tool(
"vm_mouse_move", {"name": "vm1", "dx": 70, "dy": -75, "step": 32}
)
)
assert data["moved"] == [70, -75]
assert hmp_calls() == [
"mouse_move 32 0",
"mouse_move 32 0",
"mouse_move 6 0",
"mouse_move 0 -32",
"mouse_move 0 -32",
"mouse_move 0 -11",
]
async def test_mouse_move_home_pins_to_corner(dirs, fake_qmp):
arm_screendump(width=320, height=200)
FakeQMPClient.responses["human-monitor-command"] = ""
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
data = result_data(
await client.call_tool(
"vm_mouse_move", {"name": "vm1", "home": "bottom-right", "step": 100}
)
)
assert data["homed"] == "bottom-right"
moves = hmp_calls()
# ceil(320/100) + 2 = 6 homing steps, all toward +x/+y
assert moves == ["mouse_move 100 100"] * 6
async def test_mouse_move_home_top_left_direction(dirs, fake_qmp):
arm_screendump(width=100, height=100)
FakeQMPClient.responses["human-monitor-command"] = ""
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
await client.call_tool("vm_mouse_move", {"name": "vm1", "home": "top-left", "step": 50})
assert set(hmp_calls()) == {"mouse_move -50 -50"}
async def test_mouse_move_click_sequence(dirs, fake_qmp):
FakeQMPClient.responses["human-monitor-command"] = ""
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
data = result_data(
await client.call_tool(
"vm_mouse_move", {"name": "vm1", "click": "left", "double": True}
)
)
assert data["clicked"] == "left double"
assert hmp_calls() == ["mouse_button 1", "mouse_button 0"] * 2
async def test_mouse_move_right_click(dirs, fake_qmp):
FakeQMPClient.responses["human-monitor-command"] = ""
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
await client.call_tool("vm_mouse_move", {"name": "vm1", "click": "right"})
assert hmp_calls() == ["mouse_button 2", "mouse_button 0"]
async def test_mouse_move_rejects_bad_step(dirs, fake_qmp):
write_registry(dirs, seeded_record(dirs, "vm1"))
async with Client(mcp) as client:
with pytest.raises(ToolError, match="step"):
await client.call_tool("vm_mouse_move", {"name": "vm1", "dx": 10, "step": 500})