warehack.ing/mcqemu
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add see & drive tools: screenshot, keyboard/mouse injection, serial console

Browse Source
This commit is contained in:
Ryan Malloy 2026-08-16 23:47:58 -06:00
parent 08f5e4329e
commit 9d20f28998
9 changed files with 525 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -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` |

160
src/mcqemu/keymap.py Normal file
View File

@ -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)."
)

3
src/mcqemu/launcher.py
View File

@ -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),

4
src/mcqemu/tools/__init__.py
View File

@ -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)

171
src/mcqemu/tools/display.py Normal file
View File

@ -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)

121
tests/test_display.py Normal file
View File

@ -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"

15
tests/test_integration.py
View File

@ -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)

51
tests/test_keymap.py Normal file
View File

@ -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("é")

1
tests/test_launcher.py
View File

@ -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