diff --git a/README.md b/README.md index a1fe04b..a975f18 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ claude mcp add mcqemu -- uv run --directory /path/to/mcqemu mcqemu | Lifecycle | `launch_vm`, `stop_vm`, `pause_vm`, `resume_vm`, `attach_vm`, `forget_vm` | | Inspect | `list_vms`, `vm_info` | | Live snapshots | `vm_snapshot_create` / `restore` / `delete` / `list` | +| See & drive | `vm_screenshot` (PNG), `vm_send_keys`, `vm_type_text`, `vm_click`, `vm_serial_read` | | 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` | diff --git a/src/mcqemu/keymap.py b/src/mcqemu/keymap.py new file mode 100644 index 0000000..e58209b --- /dev/null +++ b/src/mcqemu/keymap.py @@ -0,0 +1,160 @@ +"""US-layout translation from characters and key names to QEMU qcodes. + +QMP's send-key command speaks "qcodes" (QKeyCode enum values). Typing text +means mapping each character to a qcode plus an optional shift modifier; +key chords like "ctrl-alt-f2" map each part through the alias table. +""" + +from __future__ import annotations + +# Friendly names -> canonical qcodes. +KEY_ALIASES: dict[str, str] = { + "enter": "ret", + "return": "ret", + "space": "spc", + "escape": "esc", + "del": "delete", + "ins": "insert", + "pageup": "pgup", + "pagedown": "pgdn", + "win": "meta_l", + "super": "meta_l", + "meta": "meta_l", + "control": "ctrl", + "lctrl": "ctrl", + "rctrl": "ctrl_r", + "lalt": "alt", + "ralt": "alt_r", + "lshift": "shift", + "rshift": "shift_r", + "capslock": "caps_lock", + "printscreen": "print", + "hyphen": "minus", + "dash": "minus", +} + +# Keys that are already valid qcodes (subset we advertise; QEMU knows more). +KNOWN_QCODES = { + "ret", + "esc", + "spc", + "tab", + "backspace", + "delete", + "insert", + "home", + "end", + "pgup", + "pgdn", + "up", + "down", + "left", + "right", + "ctrl", + "ctrl_r", + "alt", + "alt_r", + "shift", + "shift_r", + "meta_l", + "meta_r", + "caps_lock", + "num_lock", + "scroll_lock", + "print", + "pause", + "minus", + "equal", + "bracket_left", + "bracket_right", + "backslash", + "semicolon", + "apostrophe", + "comma", + "dot", + "slash", + "grave_accent", + *[f"f{i}" for i in range(1, 13)], + *[chr(c) for c in range(ord("a"), ord("z") + 1)], + *[str(d) for d in range(10)], +} + +# char -> (qcode, needs_shift), US layout. +_UNSHIFTED = { + " ": "spc", + "\t": "tab", + "\n": "ret", + "-": "minus", + "=": "equal", + "[": "bracket_left", + "]": "bracket_right", + "\\": "backslash", + ";": "semicolon", + "'": "apostrophe", + ",": "comma", + ".": "dot", + "/": "slash", + "`": "grave_accent", +} +_SHIFTED = { + "!": "1", + "@": "2", + "#": "3", + "$": "4", + "%": "5", + "^": "6", + "&": "7", + "*": "8", + "(": "9", + ")": "0", + "_": "minus", + "+": "equal", + "{": "bracket_left", + "}": "bracket_right", + "|": "backslash", + ":": "semicolon", + '"': "apostrophe", + "<": "comma", + ">": "dot", + "?": "slash", + "~": "grave_accent", +} + +CHAR_QCODES: dict[str, tuple[str, bool]] = {} +for _c in "abcdefghijklmnopqrstuvwxyz0123456789": + CHAR_QCODES[_c] = (_c, False) +for _c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ": + CHAR_QCODES[_c] = (_c.lower(), True) +for _c, _q in _UNSHIFTED.items(): + CHAR_QCODES[_c] = (_q, False) +for _c, _q in _SHIFTED.items(): + CHAR_QCODES[_c] = (_q, True) + + +def resolve_key(name: str) -> str: + """One key name -> qcode. Raises ValueError for unknown keys.""" + key = name.strip().lower() + key = KEY_ALIASES.get(key, key) + if key in KNOWN_QCODES: + return key + raise ValueError( + f"Unknown key {name!r}. Use qcode names like 'ret', 'esc', 'tab', 'f1', " + "'ctrl', 'alt', letters, digits, or aliases like 'enter'/'space'." + ) + + +def parse_chord(chord: str) -> list[str]: + """'ctrl-alt-f2' -> ['ctrl', 'alt', 'f2']. A bare '-' means the minus key.""" + if chord == "-": + return ["minus"] + return [resolve_key(part) for part in chord.split("-") if part] + + +def char_to_keys(char: str) -> tuple[str, bool]: + """One character -> (qcode, needs_shift). Raises ValueError if untypeable.""" + if char in CHAR_QCODES: + return CHAR_QCODES[char] + raise ValueError( + f"Cannot type character {char!r} with the US-layout key map " + "(only printable ASCII, tab and newline are supported)." + ) diff --git a/src/mcqemu/launcher.py b/src/mcqemu/launcher.py index 66b1fad..7ada7a3 100644 --- a/src/mcqemu/launcher.py +++ b/src/mcqemu/launcher.py @@ -122,6 +122,9 @@ def build_cmdline( "virtio-serial", "-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0", + # Absolute-coordinate pointer so vm_click can target exact pixels. + "-device", + "virtio-tablet-pci", "-daemonize", "-pidfile", str(paths.pidfile), diff --git a/src/mcqemu/tools/__init__.py b/src/mcqemu/tools/__init__.py index fa0ffdc..ad3b91b 100644 --- a/src/mcqemu/tools/__init__.py +++ b/src/mcqemu/tools/__init__.py @@ -2,9 +2,9 @@ from fastmcp import FastMCP -from . import guest, images, lifecycle, query, snapshots +from . import display, guest, images, lifecycle, query, snapshots def register_all(mcp: FastMCP) -> None: - for module in (lifecycle, query, snapshots, images, guest): + for module in (lifecycle, query, snapshots, images, guest, display): module.register(mcp) diff --git a/src/mcqemu/tools/display.py b/src/mcqemu/tools/display.py new file mode 100644 index 0000000..2abc739 --- /dev/null +++ b/src/mcqemu/tools/display.py @@ -0,0 +1,171 @@ +"""See & drive: screenshots, keyboard/mouse injection, serial console.""" + +import asyncio +import struct +from pathlib import Path + +from fastmcp import Context, FastMCP +from fastmcp.exceptions import ToolError +from fastmcp.utilities.types import Image + +from ..config import Config +from ..keymap import char_to_keys, parse_chord +from ..qmp import execute, qmp_session +from ._common import require_vm + +_ABS_MAX = 32767 # QEMU absolute-pointer coordinate space + + +def _png_dimensions(data: bytes) -> tuple[int, int]: + if len(data) < 24 or data[:8] != b"\x89PNG\r\n\x1a\n": + raise ToolError("screendump did not produce a valid PNG — is a display device present?") + width, height = struct.unpack(">II", data[16:24]) + return width, height + + +async def _screendump(ctx: Context, name: str) -> bytes: + state, record = require_vm(ctx, name) + config: Config = state.config + dest = config.vm_state_dir(name) / "screenshot.png" + dest.parent.mkdir(parents=True, exist_ok=True) + async with qmp_session(record) as client: + await execute(client, "screendump", {"filename": str(dest), "format": "png"}) + try: + return dest.read_bytes() + except OSError as e: + raise ToolError(f"screendump reported success but {dest} is unreadable: {e}") from e + + +async def vm_screenshot(name: str, ctx: Context = None) -> Image: + """Capture the VM's current display as a PNG image. Works on any running + VM with a display device (the default for launched x86 VMs) — no guest + software needed. Use this to watch installers, read console output, or + verify GUI state before sending keys with vm_send_keys / vm_type_text.""" + data = await _screendump(ctx, name) + return Image(data=data, format="png") + + +async def vm_send_keys( + name: str, keys: list[str], hold_ms: int = 100, delay_ms: int = 50, ctx: Context = None +) -> dict: + """Press keys or key combos in the VM. Each list entry is one press: a + single key ("ret", "esc", "f2", "a") or a chord pressed together + ("ctrl-alt-f2", "ctrl-c"). Entries are sent in order with delay_ms between + them. Aliases like "enter", "space", "escape" work. To type prose, use + vm_type_text instead.""" + _, record = require_vm(ctx, name) + chords = [] + for entry in keys: + try: + chords.append(parse_chord(entry)) + except ValueError as e: + raise ToolError(str(e)) from e + + async with qmp_session(record) as client: + for chord in chords: + await execute( + client, + "send-key", + { + "keys": [{"type": "qcode", "data": q} for q in chord], + "hold-time": hold_ms, + }, + ) + await asyncio.sleep(delay_ms / 1000) + return {"name": name, "keys_sent": len(chords)} + + +async def vm_type_text( + name: str, text: str, enter: bool = False, delay_ms: int = 30, ctx: Context = None +) -> dict: + """Type a string into the VM, character by character (US keyboard layout; + printable ASCII plus tab and newline). Set enter=True to press Enter at + the end — handy for shell commands at a console login or terminal.""" + _, record = require_vm(ctx, name) + presses: list[list[str]] = [] + for char in text: + try: + qcode, shifted = char_to_keys(char) + except ValueError as e: + raise ToolError(str(e)) from e + presses.append(["shift", qcode] if shifted else [qcode]) + if enter: + presses.append(["ret"]) + + async with qmp_session(record) as client: + for chord in presses: + await execute( + client, + "send-key", + {"keys": [{"type": "qcode", "data": q} for q in chord], "hold-time": 60}, + ) + await asyncio.sleep(delay_ms / 1000) + return {"name": name, "characters_typed": len(presses)} + + +async def vm_click( + name: str, x: int, y: int, button: str = "left", double: bool = False, ctx: Context = None +) -> dict: + """Click at pixel coordinates (x, y) on the VM display — coordinates match + what vm_screenshot shows. Requires the VM's tablet device for absolute + positioning (present on VMs launched by this server). button: left, + right, or middle.""" + if button not in ("left", "right", "middle"): + raise ToolError(f"Unsupported button {button!r}: use left, right, or middle.") + # Screenshot first: it proves a display exists and gives us the resolution + # to scale pixel coords into QEMU's 0-32767 absolute space. + data = await _screendump(ctx, name) + width, height = _png_dimensions(data) + if not (0 <= x < width and 0 <= y < height): + raise ToolError(f"({x}, {y}) is outside the {width}x{height} display.") + _, record = require_vm(ctx, name) + abs_x = int(x * _ABS_MAX / max(width - 1, 1)) + abs_y = int(y * _ABS_MAX / max(height - 1, 1)) + + async with qmp_session(record) as client: + await execute( + client, + "input-send-event", + { + "events": [ + {"type": "abs", "data": {"axis": "x", "value": abs_x}}, + {"type": "abs", "data": {"axis": "y", "value": abs_y}}, + ] + }, + ) + clicks = 2 if double else 1 + for _ in range(clicks): + for down in (True, False): + await execute( + client, + "input-send-event", + {"events": [{"type": "btn", "data": {"down": down, "button": button}}]}, + ) + await asyncio.sleep(0.05) + return {"name": name, "clicked": f"{button} at ({x}, {y})", "double": double} + + +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 + the guest writes to its serial port (kernel console=ttyS0, or text-mode + installers); GUI-only guests log nothing here — use vm_screenshot for + those.""" + _, record = require_vm(ctx, name) + if not record.serial_log: + raise ToolError( + f"VM {name!r} has no serial log registered (attached VMs manage their own serial)." + ) + path = Path(record.serial_log) + if not path.exists(): + raise ToolError(f"Serial log {path} does not exist yet.") + lines = path.read_text(errors="replace").splitlines() + return { + "name": name, + "total_lines": len(lines), + "tail": "\n".join(lines[-tail_lines:]) if lines else "(serial log is empty)", + } + + +def register(mcp: FastMCP) -> None: + for fn in (vm_screenshot, vm_send_keys, vm_type_text, vm_click, vm_serial_read): + mcp.tool(fn) diff --git a/tests/test_display.py b/tests/test_display.py new file mode 100644 index 0000000..c41db93 --- /dev/null +++ b/tests/test_display.py @@ -0,0 +1,121 @@ +"""See & drive tools with fake QMP; screendump writes a synthetic PNG.""" + +import struct + +import pytest +from fastmcp import Client +from fastmcp.exceptions import ToolError + +from conftest import FakeQMPClient, result_data, write_registry +from mcqemu.server import mcp +from test_lifecycle import seeded_record + + +def fake_png(width: int, height: int) -> bytes: + """Just enough PNG for signature + IHDR dimension parsing.""" + return ( + b"\x89PNG\r\n\x1a\n" + + b"\x00\x00\x00\x0dIHDR" + + struct.pack(">II", width, height) + + b"\x08\x06\x00\x00\x00" + + b"\x00" * 16 + ) + + +def arm_screendump(width=640, height=480): + def handler(args): + with open(args["filename"], "wb") as f: + f.write(fake_png(width, height)) + return {} + + FakeQMPClient.responses["screendump"] = handler + + +async def test_screenshot_returns_png_image(dirs, fake_qmp): + arm_screendump() + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + result = await client.call_tool("vm_screenshot", {"name": "vm1"}) + block = result.content[0] + assert block.type == "image" + assert block.mimeType == "image/png" + call = next(a for c, a in FakeQMPClient.calls if c == "screendump") + assert call["format"] == "png" + + +async def test_send_keys_chords(dirs, fake_qmp): + FakeQMPClient.responses["send-key"] = {} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "vm_send_keys", {"name": "vm1", "keys": ["ctrl-alt-f2", "enter"], "delay_ms": 0} + ) + ) + assert data["keys_sent"] == 2 + sent = [a["keys"] for c, a in FakeQMPClient.calls if c == "send-key"] + assert sent[0] == [ + {"type": "qcode", "data": "ctrl"}, + {"type": "qcode", "data": "alt"}, + {"type": "qcode", "data": "f2"}, + ] + assert sent[1] == [{"type": "qcode", "data": "ret"}] + + +async def test_send_keys_unknown_key(dirs, fake_qmp): + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="Unknown key"): + await client.call_tool("vm_send_keys", {"name": "vm1", "keys": ["warpdrive"]}) + + +async def test_type_text_with_shift_and_enter(dirs, fake_qmp): + FakeQMPClient.responses["send-key"] = {} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data( + await client.call_tool( + "vm_type_text", {"name": "vm1", "text": "Hi!", "enter": True, "delay_ms": 0} + ) + ) + assert data["characters_typed"] == 4 # H, i, !, Enter + sent = [a["keys"] for c, a in FakeQMPClient.calls if c == "send-key"] + assert sent[0] == [{"type": "qcode", "data": "shift"}, {"type": "qcode", "data": "h"}] + assert sent[1] == [{"type": "qcode", "data": "i"}] + assert sent[2] == [{"type": "qcode", "data": "shift"}, {"type": "qcode", "data": "1"}] + assert sent[3] == [{"type": "qcode", "data": "ret"}] + + +async def test_click_scales_to_abs_space(dirs, fake_qmp): + arm_screendump(width=640, height=480) + FakeQMPClient.responses["input-send-event"] = {} + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + data = result_data(await client.call_tool("vm_click", {"name": "vm1", "x": 320, "y": 240})) + assert "left at (320, 240)" in data["clicked"] + events = [a["events"] for c, a in FakeQMPClient.calls if c == "input-send-event"] + move = events[0] + assert move[0]["data"] == {"axis": "x", "value": int(320 * 32767 / 639)} + assert move[1]["data"] == {"axis": "y", "value": int(240 * 32767 / 479)} + assert events[1][0]["data"] == {"down": True, "button": "left"} + assert events[2][0]["data"] == {"down": False, "button": "left"} + + +async def test_click_outside_display_rejected(dirs, fake_qmp): + arm_screendump(width=640, height=480) + write_registry(dirs, seeded_record(dirs, "vm1")) + async with Client(mcp) as client: + with pytest.raises(ToolError, match="outside"): + await client.call_tool("vm_click", {"name": "vm1", "x": 9999, "y": 10}) + + +async def test_serial_read_tail(dirs, fake_qmp): + record = seeded_record(dirs, "vm1", serial_log=str(dirs.state / "serial.log")) + (dirs.state / "serial.log").write_text("\n".join(f"line{i}" for i in range(100))) + write_registry(dirs, record) + async with Client(mcp) as client: + data = result_data( + await client.call_tool("vm_serial_read", {"name": "vm1", "tail_lines": 3}) + ) + assert data["total_lines"] == 100 + assert data["tail"] == "line97\nline98\nline99" diff --git a/tests/test_integration.py b/tests/test_integration.py index a4d6c40..3f24902 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -3,6 +3,7 @@ Run with: uv run pytest -m integration """ +import base64 import shutil import pytest @@ -46,6 +47,20 @@ async def test_real_vm_lifecycle(dirs): await client.call_tool("resume_vm", {"name": "itest"}) + # See & drive against the live SeaBIOS display. + result = await client.call_tool("vm_screenshot", {"name": "itest"}) + block = result.content[0] + assert block.type == "image" + png = base64.b64decode(block.data) + assert png[:8] == b"\x89PNG\r\n\x1a\n" + + await client.call_tool( + "vm_send_keys", {"name": "itest", "keys": ["f2", "enter"], "delay_ms": 0} + ) + await client.call_tool("vm_click", {"name": "itest", "x": 10, "y": 10}) + data = result_data(await client.call_tool("vm_serial_read", {"name": "itest"})) + assert "tail" in data + data = result_data(await client.call_tool("stop_vm", {"name": "itest", "force": True})) assert data["status"] == "stopped" assert not pid_alive(pid) diff --git a/tests/test_keymap.py b/tests/test_keymap.py new file mode 100644 index 0000000..2097749 --- /dev/null +++ b/tests/test_keymap.py @@ -0,0 +1,51 @@ +"""Key/character translation tables.""" + +import pytest + +from mcqemu.keymap import char_to_keys, parse_chord, resolve_key + + +def test_aliases(): + assert resolve_key("enter") == "ret" + assert resolve_key("Escape") == "esc" + assert resolve_key("space") == "spc" + assert resolve_key("win") == "meta_l" + + +def test_plain_qcodes_pass_through(): + for key in ("ret", "f11", "a", "9", "ctrl", "grave_accent"): + assert resolve_key(key) == key + + +def test_unknown_key_raises(): + with pytest.raises(ValueError, match="Unknown key"): + resolve_key("hyperspace") + + +def test_chords(): + assert parse_chord("ctrl-alt-f2") == ["ctrl", "alt", "f2"] + assert parse_chord("ctrl-c") == ["ctrl", "c"] + assert parse_chord("-") == ["minus"] + + +def test_char_lowercase(): + assert char_to_keys("a") == ("a", False) + assert char_to_keys("5") == ("5", False) + + +def test_char_shifted(): + assert char_to_keys("A") == ("a", True) + assert char_to_keys("!") == ("1", True) + assert char_to_keys("_") == ("minus", True) + assert char_to_keys('"') == ("apostrophe", True) + + +def test_char_specials(): + assert char_to_keys(" ") == ("spc", False) + assert char_to_keys("\n") == ("ret", False) + assert char_to_keys("/") == ("slash", False) + + +def test_untypeable_char_raises(): + with pytest.raises(ValueError, match="Cannot type"): + char_to_keys("é") diff --git a/tests/test_launcher.py b/tests/test_launcher.py index f4db9d7..2dfca45 100644 --- a/tests/test_launcher.py +++ b/tests/test_launcher.py @@ -36,6 +36,7 @@ def test_defaults_kvm(): assert "-daemonize" in s assert "-qmp unix:/run/t/qmp.sock,server=on,wait=off" in s assert "org.qemu.guest_agent.0" in s + assert "-device virtio-tablet-pci" in s assert "-netdev user,id=net0 -device virtio-net-pci,netdev=net0" in s