Initial mcqemu: FastMCP server with lifecycle, image, snapshot, and guest-agent tools

.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.pyc
+.venv/
+dist/
+build/
+*.egg-info/
+.pytest_cache/
+.ruff_cache/
+*.qcow2
+*.img
+*.iso
+.env

README.md

@@ -0,0 +1,60 @@
+# mcqemu
+
+An MCP server that lets LLM agents manage QEMU virtual machines: launch and
+stop VMs, inspect them over QMP, manage disk images with qemu-img, take live
+snapshots, and run commands inside guests through qemu-guest-agent.
+
+## Requirements
+
+- Linux with QEMU installed (`qemu-system-*` and `qemu-img` on PATH)
+- `/dev/kvm` access for hardware acceleration (optional — TCG emulation works
+  without it, just slower)
+- Python 3.11+ managed with [uv](https://docs.astral.sh/uv/)
+
+## Install
+
+```bash
+# From this checkout
+uv sync
+
+# Add to Claude Code
+claude mcp add mcqemu -- uv run --directory /path/to/mcqemu mcqemu
+```
+
+## What it can do
+
+| Group | Tools |
+|---|---|
+| 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` |
+| 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` |
+
+VMs are daemonized QEMU processes with QMP control sockets, so they survive
+MCP server restarts. The registry lives in `~/.local/share/mcqemu/`, sockets
+in `$XDG_RUNTIME_DIR/mcqemu/`.
+
+Guest tools (`guest_*`) need `qemu-guest-agent` installed inside the guest OS;
+the host-side virtio-serial channel is wired on every launch, so installing
+the agent in the guest is the only step.
+
+## Quick start
+
+```
+image_create(path="~/vms/test.qcow2", size="10G")
+launch_vm(name="test", disks=["~/vms/test.qcow2"], iso="~/isos/alpine.iso",
+          port_forwards=["2222:22"])
+# ... install the OS via the serial console log ...
+stop_vm(name="test")
+launch_vm(name="test", disks=["~/vms/test.qcow2"])
+guest_exec(name="test", command="uname", args=["-a"])
+```
+
+## Development
+
+```bash
+uv run pytest                  # unit tests (QMP and subprocess mocked)
+uv run pytest -m integration   # boots a real tiny VM (needs QEMU installed)
+uv run ruff check .
+```

pyproject.toml

@@ -0,0 +1,55 @@
+[project]
+name = "mcqemu"
+version = "2026.8.16"
+description = "MCP server for managing QEMU virtual machines"
+readme = "README.md"
+requires-python = ">=3.11"
+authors = [{ name = "Ryan Malloy", email = "ryan@supported.systems" }]
+license = "MIT"
+keywords = ["mcp", "qemu", "qmp", "virtualization", "kvm"]
+dependencies = [
+    "fastmcp>=3.4.7,<4",
+    "qemu.qmp>=0.0.6",
+]
+
+[project.scripts]
+mcqemu = "mcqemu.server:main"
+
+[dependency-groups]
+dev = [
+    "pytest>=8",
+    "pytest-asyncio>=1.0",
+    "ruff>=0.12",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/mcqemu"]
+
+[tool.hatch.build.targets.sdist]
+exclude = [
+    "CLAUDE.md",
+    ".env",
+    ".mcp.json",
+    "tests/",
+    ".pytest_cache/",
+    ".ruff_cache/",
+    "dist/",
+]
+
+[tool.ruff]
+line-length = 100
+src = ["src", "tests"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B", "SIM", "ASYNC"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+markers = [
+    "integration: boots real QEMU (run with '-m integration')",
+]
+addopts = "-m 'not integration'"

src/mcqemu/__init__.py

@@ -0,0 +1,3 @@
+"""mcqemu — MCP server for managing QEMU virtual machines."""
+
+__version__ = "2026.8.16"

src/mcqemu/config.py

@@ -0,0 +1,103 @@
+"""Path resolution, launch defaults, and validation helpers."""
+
+from __future__ import annotations
+
+import os
+import re
+import platform
+from dataclasses import dataclass
+from pathlib import Path
+
+# UEFI firmware per arch. x64 ships split CODE/VARS images sized for pflash;
+# aarch64's QEMU_EFI.fd is a 2 MB image meant for -bios (pflash wants 64 MB pads).
+FIRMWARE_MAP: dict[str, dict[str, str]] = {
+    "x86_64": {
+        "mode": "pflash",
+        "code": "/usr/share/edk2/x64/OVMF_CODE.4m.fd",
+        "vars": "/usr/share/edk2/x64/OVMF_VARS.4m.fd",
+    },
+    "aarch64": {
+        "mode": "bios",
+        "code": "/usr/share/edk2/aarch64/QEMU_EFI.fd",
+    },
+    "riscv64": {
+        "mode": "bios",
+        "code": "/usr/share/edk2/riscv64/RISCV_VIRT_CODE.fd",
+    },
+}
+
+# Default -machine per arch; arches not listed use QEMU's own default.
+DEFAULT_MACHINE: dict[str, str] = {
+    "x86_64": "q35",
+    "i386": "q35",
+    "aarch64": "virt",
+    "riscv64": "virt",
+    "riscv32": "virt",
+}
+
+_VM_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,47}$")
+
+
+def valid_vm_name(name: str) -> bool:
+    """Names become directory and socket path components, so keep them tame."""
+    return bool(_VM_NAME_RE.match(name))
+
+
+@dataclass(frozen=True)
+class Config:
+    state_dir: Path
+    runtime_dir: Path
+
+    @classmethod
+    def from_env(cls) -> Config:
+        state = os.environ.get("MCQEMU_STATE_DIR")
+        if state:
+            state_dir = Path(state)
+        else:
+            xdg_data = os.environ.get("XDG_DATA_HOME", "~/.local/share")
+            state_dir = Path(xdg_data).expanduser() / "mcqemu"
+
+        runtime = os.environ.get("MCQEMU_RUNTIME_DIR")
+        if runtime:
+            runtime_dir = Path(runtime)
+        else:
+            xdg_run = os.environ.get("XDG_RUNTIME_DIR")
+            base = Path(xdg_run) if xdg_run else Path(f"/tmp/mcqemu-{os.getuid()}")
+            runtime_dir = base / "mcqemu" if xdg_run else base
+
+        return cls(state_dir=state_dir, runtime_dir=runtime_dir)
+
+    @property
+    def registry_path(self) -> Path:
+        return self.state_dir / "vms.json"
+
+    def vm_state_dir(self, name: str) -> Path:
+        return self.state_dir / "vms" / name
+
+    def vm_runtime_dir(self, name: str) -> Path:
+        return self.runtime_dir / name
+
+
+def host_arch() -> str:
+    return platform.machine()
+
+
+def kvm_available() -> bool:
+    return os.access("/dev/kvm", os.R_OK | os.W_OK)
+
+
+def resolve_image_path(path: str, must_exist: bool = True) -> Path:
+    """Normalize a disk-image path and reject things that would surprise QEMU.
+
+    Raises ValueError with a human-readable reason; tools wrap it in ToolError.
+    """
+    p = Path(path).expanduser().resolve()
+    if p.exists():
+        if not p.is_file():
+            raise ValueError(f"{p} exists but is not a regular file")
+    else:
+        if must_exist:
+            raise ValueError(f"image not found: {p}")
+        if not p.parent.is_dir():
+            raise ValueError(f"parent directory does not exist: {p.parent}")
+    return p

src/mcqemu/errors.py

@@ -0,0 +1,31 @@
+"""Map QMP / subprocess / registry failures to actionable ToolErrors."""
+
+from __future__ import annotations
+
+from fastmcp.exceptions import ToolError
+
+
+def vm_not_found(name: str, known: list[str]) -> ToolError:
+    hint = f"Known VMs: {', '.join(sorted(known))}" if known else "No VMs are registered."
+    return ToolError(f"No VM named {name!r}. {hint} Use list_vms to see current state.")
+
+
+def qmp_unreachable(name: str, socket_path: str, detail: str, qemu_log: str | None) -> ToolError:
+    log_hint = f" Check the QEMU log at {qemu_log} for crash details." if qemu_log else ""
+    return ToolError(
+        f"Cannot reach QMP socket for VM {name!r} at {socket_path} ({detail}). "
+        f"The VM has likely exited — run list_vms to refresh status.{log_hint}"
+    )
+
+
+def qmp_command_failed(command: str, error_class: str, desc: str) -> ToolError:
+    return ToolError(f"QMP command {command!r} failed [{error_class}]: {desc}")
+
+
+def guest_agent_unavailable(name: str) -> ToolError:
+    return ToolError(
+        f"The guest agent in VM {name!r} is not responding. The guest OS must have "
+        "qemu-guest-agent installed and running (e.g. 'apt install qemu-guest-agent' "
+        "or 'apk add qemu-guest-agent', then start its service). The virtio-serial "
+        "channel is already wired on the host side."
+    )

src/mcqemu/launcher.py

@@ -0,0 +1,193 @@
+"""Build QEMU command lines (pure) and spawn daemonized VMs."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import shutil
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from fastmcp.exceptions import ToolError
+
+from .config import DEFAULT_MACHINE, FIRMWARE_MAP, Config, host_arch, kvm_available
+from .models import VMConfig, VMRecord
+from .qemu_img import probe_format
+
+_FWD_RE = re.compile(r"^(\d{1,5}):(\d{1,5})$")
+
+
+@dataclass(frozen=True)
+class VMPaths:
+    """Everything path-shaped that one VM needs, derived from Config."""
+
+    qmp_socket: Path
+    qga_socket: Path
+    pidfile: Path
+    qemu_log: Path
+    serial_log: Path
+    uefi_vars: Path
+
+    @classmethod
+    def for_vm(cls, config: Config, name: str) -> VMPaths:
+        run = config.vm_runtime_dir(name)
+        state = config.vm_state_dir(name)
+        return cls(
+            qmp_socket=run / "qmp.sock",
+            qga_socket=run / "qga.sock",
+            pidfile=run / "pid",
+            qemu_log=state / "qemu.log",
+            serial_log=state / "serial.log",
+            uefi_vars=state / "uefi-vars.fd",
+        )
+
+
+def pick_accel(arch: str) -> str:
+    """KVM only helps when guest arch matches the host; otherwise TCG emulation."""
+    return "kvm" if arch == host_arch() and kvm_available() else "tcg"
+
+
+def build_cmdline(
+    cfg: VMConfig,
+    paths: VMPaths,
+    *,
+    accel: str,
+    disk_formats: dict[str, str],
+) -> list[str]:
+    """Pure function: (config, paths, probed formats) -> argv for qemu-system-*.
+
+    Deliberately does no I/O so it can be exhaustively table-tested.
+    """
+    args: list[str] = [f"qemu-system-{cfg.arch}", "-name", cfg.name]
+
+    machine = cfg.machine or DEFAULT_MACHINE.get(cfg.arch)
+    if machine:
+        args += ["-machine", f"{machine},accel={accel}"]
+    else:
+        args += ["-accel", accel]
+    args += ["-cpu", "host" if accel == "kvm" else "max"]
+    args += ["-m", str(cfg.memory_mb), "-smp", str(cfg.cpus)]
+
+    if cfg.firmware == "uefi":
+        fw = FIRMWARE_MAP.get(cfg.arch)
+        if fw is None:
+            raise ToolError(
+                f"No UEFI firmware known for arch {cfg.arch!r} "
+                f"(available: {', '.join(FIRMWARE_MAP)}). Use firmware='bios'."
+            )
+        if fw["mode"] == "pflash":
+            args += [
+                "-drive", f"if=pflash,format=raw,readonly=on,file={fw['code']}",
+                "-drive", f"if=pflash,format=raw,file={paths.uefi_vars}",
+            ]
+        else:
+            args += ["-bios", fw["code"]]
+
+    for disk in cfg.disks:
+        fmt = disk_formats[disk]
+        args += ["-drive", f"file={disk},if=virtio,format={fmt}"]
+
+    if cfg.iso:
+        args += ["-drive", f"file={cfg.iso},media=cdrom,readonly=on"]
+        if cfg.disks:
+            # Boot the installer once; subsequent boots hit the disk.
+            args += ["-boot", "once=d"]
+
+    if cfg.no_net:
+        args += ["-nic", "none"]
+    else:
+        netdev = "user,id=net0"
+        for fwd in cfg.port_forwards:
+            m = _FWD_RE.match(fwd)
+            if not m:
+                raise ToolError(
+                    f"Invalid port forward {fwd!r} — expected 'HOSTPORT:GUESTPORT', e.g. '2222:22'."
+                )
+            netdev += f",hostfwd=tcp::{m.group(1)}-:{m.group(2)}"
+        args += ["-netdev", netdev, "-device", "virtio-net-pci,netdev=net0"]
+
+    args += [
+        "-display", "none",
+        "-serial", f"file:{paths.serial_log}",
+        "-qmp", f"unix:{paths.qmp_socket},server=on,wait=off",
+        "-chardev", f"socket,id=qga0,path={paths.qga_socket},server=on,wait=off",
+        "-device", "virtio-serial",
+        "-device", "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0",
+        "-daemonize",
+        "-pidfile", str(paths.pidfile),
+        "-D", str(paths.qemu_log),
+    ]
+    args += cfg.extra_args
+    return args
+
+
+async def spawn_vm(cfg: VMConfig, config: Config) -> VMRecord:
+    """Probe disks, prepare per-VM dirs, launch QEMU daemonized, return a record.
+
+    QEMU's -daemonize semantics: the foreground process exits 0 only after the
+    VM is fully initialized (QMP socket listening), so awaiting it gives us a
+    synchronous verdict with real stderr on failure.
+    """
+    binary = shutil.which(f"qemu-system-{cfg.arch}")
+    if binary is None:
+        raise ToolError(
+            f"qemu-system-{cfg.arch} not found on PATH — is that arch installed? "
+            "(Arch Linux: pacman -S qemu-full)"
+        )
+
+    paths = VMPaths.for_vm(config, cfg.name)
+    config.vm_runtime_dir(cfg.name).mkdir(parents=True, exist_ok=True)
+    config.vm_state_dir(cfg.name).mkdir(parents=True, exist_ok=True)
+
+    # Leftover sockets from a previous life of this name confuse liveness checks.
+    for stale in (paths.qmp_socket, paths.qga_socket, paths.pidfile):
+        stale.unlink(missing_ok=True)
+
+    disk_formats = {disk: await probe_format(disk) for disk in cfg.disks}
+
+    if cfg.firmware == "uefi":
+        fw = FIRMWARE_MAP.get(cfg.arch, {})
+        if fw.get("mode") == "pflash" and not paths.uefi_vars.exists():
+            shutil.copy(fw["vars"], paths.uefi_vars)
+
+    cmdline = build_cmdline(cfg, paths, accel=pick_accel(cfg.arch), disk_formats=disk_formats)
+
+    proc = await asyncio.create_subprocess_exec(
+        binary,
+        *cmdline[1:],
+        stdin=asyncio.subprocess.DEVNULL,
+        stdout=asyncio.subprocess.DEVNULL,
+        stderr=asyncio.subprocess.PIPE,
+        start_new_session=True,
+    )
+    _, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        detail = stderr.decode(errors="replace").strip() or f"exit code {proc.returncode}"
+        raise ToolError(f"QEMU failed to launch VM {cfg.name!r}: {detail}")
+
+    try:
+        pid = int(paths.pidfile.read_text().strip())
+    except (OSError, ValueError) as e:
+        raise ToolError(
+            f"QEMU launched but pidfile {paths.pidfile} is unreadable ({e}) — "
+            "the VM state is unknown; check the QEMU log."
+        ) from e
+
+    accel = pick_accel(cfg.arch)
+    return VMRecord(
+        name=cfg.name,
+        source="spawned",
+        qmp_socket=str(paths.qmp_socket),
+        qga_socket=str(paths.qga_socket),
+        pidfile=str(paths.pidfile),
+        pid=pid,
+        binary=binary,
+        arch=cfg.arch,
+        accel=accel,
+        cmdline=cmdline,
+        config=cfg.__dict__.copy(),
+        qemu_log=str(paths.qemu_log),
+        serial_log=str(paths.serial_log),
+        created_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
+    )

src/mcqemu/models.py

@@ -0,0 +1,51 @@
+"""Dataclasses shared across the server: VM configuration and registry records."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field, asdict
+from typing import Any, Literal
+
+
+@dataclass
+class VMConfig:
+    """User-facing launch parameters, persisted so relaunches are reproducible."""
+
+    name: str
+    arch: str = "x86_64"
+    machine: str | None = None
+    memory_mb: int = 2048
+    cpus: int = 2
+    disks: list[str] = field(default_factory=list)
+    iso: str | None = None
+    firmware: Literal["bios", "uefi"] = "bios"
+    port_forwards: list[str] = field(default_factory=list)
+    no_net: bool = False
+    extra_args: list[str] = field(default_factory=list)
+
+
+@dataclass
+class VMRecord:
+    """A registry entry for one VM — spawned by us or attached externally."""
+
+    name: str
+    source: Literal["spawned", "attached"]
+    qmp_socket: str
+    qga_socket: str | None = None
+    pidfile: str | None = None
+    pid: int | None = None
+    binary: str | None = None
+    arch: str | None = None
+    accel: str | None = None
+    cmdline: list[str] = field(default_factory=list)
+    config: dict[str, Any] = field(default_factory=dict)
+    qemu_log: str | None = None
+    serial_log: str | None = None
+    created_at: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> VMRecord:
+        known = {f for f in cls.__dataclass_fields__}
+        return cls(**{k: v for k, v in data.items() if k in known})

src/mcqemu/prompts.py

@@ -0,0 +1,29 @@
+"""MCP prompts: reusable recipes for common VM workflows."""
+
+from fastmcp import FastMCP
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.prompt(
+        name="provision_test_vm",
+        description="Step-by-step recipe for provisioning a fresh test VM from an installer ISO.",
+    )
+    def provision_test_vm(os_hint: str = "a small Linux distro like Alpine") -> str:
+        return f"""Provision a fresh test VM running {os_hint} using the mcqemu tools:
+
+1. Create a disk: image_create(path="~/vms/test.qcow2", size="10G") — qcow2 grows
+   on demand, so a generous virtual size is free.
+2. Launch with the installer: launch_vm(name="test", disks=["~/vms/test.qcow2"],
+   iso="/path/to/installer.iso", port_forwards=["2222:22"], memory_mb=2048).
+   The VM boots the ISO first, then the disk on later boots.
+3. Watch progress through the serial log (path is in the launch result) — many
+   installers support serial consoles; check vm_info for status.
+4. After installation completes inside the guest, stop_vm(name="test") and
+   launch_vm again WITHOUT the iso parameter to boot from disk.
+5. Install qemu-guest-agent inside the guest, then verify with
+   guest_ping(name="test") — that unlocks guest_exec and guest_file_* tools.
+6. SSH is reachable at host port 2222 (forwarded to guest port 22) if the guest
+   runs sshd.
+
+Before starting: confirm the ISO path exists, and pick VM names that are short
+and filesystem-safe."""

src/mcqemu/qemu_img.py

@@ -0,0 +1,55 @@
+"""Async wrapper around the qemu-img CLI."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+from typing import Any
+
+from fastmcp.exceptions import ToolError
+
+QEMU_IMG = "qemu-img"
+
+
+async def run_qemu_img(*args: str) -> str:
+    """Run qemu-img, returning stdout. Raises ToolError on failure."""
+    binary = shutil.which(QEMU_IMG)
+    if binary is None:
+        raise ToolError("qemu-img not found on PATH — install the qemu-img package.")
+    proc = await asyncio.create_subprocess_exec(
+        binary,
+        *args,
+        stdin=asyncio.subprocess.DEVNULL,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    stdout, stderr = await proc.communicate()
+    if proc.returncode != 0:
+        err = stderr.decode(errors="replace").strip()
+        if "Failed to get" in err and "lock" in err:
+            raise ToolError(
+                f"qemu-img: image is locked — it is attached to a running VM. "
+                f"Stop the VM first. Underlying error: {err.splitlines()[-1]}"
+            )
+        tail = "\n".join(err.splitlines()[-4:]) or f"exit code {proc.returncode}"
+        raise ToolError(f"qemu-img {args[0]} failed: {tail}")
+    return stdout.decode(errors="replace")
+
+
+async def image_info_json(path: str, backing_chain: bool = False) -> Any:
+    args = ["info", "--output=json"]
+    if backing_chain:
+        args.append("--backing-chain")
+    args.append(path)
+    out = await run_qemu_img(*args)
+    return json.loads(out)
+
+
+async def probe_format(path: str) -> str:
+    """Detect an image's format so QEMU never has to guess (raw vs qcow2
+    ambiguity is a classic security/correctness trap)."""
+    info = await image_info_json(path)
+    if isinstance(info, list):  # --backing-chain shape safety
+        info = info[0]
+    return info["format"]

src/mcqemu/qga.py

@@ -0,0 +1,56 @@
+"""Guest-agent sessions.
+
+The QEMU guest agent speaks the QMP wire protocol but sends no greeting and
+does no capability negotiation, so the same qemu.qmp client works with those
+two switches off. A guest-sync handshake (mandatory per the QGA protocol docs)
+doubles as our "is an agent actually alive in there?" probe — its timeout is
+the signal that the guest lacks qemu-guest-agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import secrets
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+from fastmcp.exceptions import ToolError
+from qemu.qmp import QMPClient
+
+from .errors import guest_agent_unavailable
+from .models import VMRecord
+
+SYNC_TIMEOUT = 3.0
+
+
+@asynccontextmanager
+async def qga_session(record: VMRecord) -> AsyncIterator[QMPClient]:
+    if not record.qga_socket:
+        raise ToolError(
+            f"VM {record.name!r} has no guest-agent socket registered. "
+            "For attached VMs, re-attach with qga_socket= pointing at the "
+            "virtio-serial chardev socket."
+        )
+    client = QMPClient(f"{record.name}-qga")
+    client.await_greeting = False
+    client.negotiate = False
+    try:
+        try:
+            await asyncio.wait_for(client.connect(record.qga_socket), timeout=SYNC_TIMEOUT)
+            token = secrets.randbelow(2**31)
+            reply = await asyncio.wait_for(
+                client.execute("guest-sync", {"id": token}), timeout=SYNC_TIMEOUT
+            )
+            if reply != token:
+                raise guest_agent_unavailable(record.name)
+        except (asyncio.TimeoutError, OSError):
+            raise guest_agent_unavailable(record.name) from None
+        except ToolError:
+            raise
+        except Exception:  # qemu.qmp ConnectError et al.
+            raise guest_agent_unavailable(record.name) from None
+        yield client
+    finally:
+        with contextlib.suppress(Exception):
+            await client.disconnect()

src/mcqemu/qmp.py

@@ -0,0 +1,67 @@
+"""Scoped QMP sessions over qemu.qmp's asyncio client.
+
+QEMU's QMP unix socket accepts one client at a time, so every tool call
+opens a short-lived session and disconnects — the socket stays free for
+qmp-shell or any other tooling between calls, and there is no stale
+connection state to reconcile after either side restarts.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from contextlib import asynccontextmanager
+from typing import Any, AsyncIterator
+
+from qemu.qmp import ExecuteError, QMPClient
+
+from .errors import qmp_command_failed, qmp_unreachable
+from .models import VMRecord
+
+CONNECT_TIMEOUT = 5.0
+
+
+@asynccontextmanager
+async def qmp_session(record: VMRecord) -> AsyncIterator[QMPClient]:
+    client = QMPClient(record.name)
+    try:
+        try:
+            await asyncio.wait_for(client.connect(record.qmp_socket), timeout=CONNECT_TIMEOUT)
+        except asyncio.TimeoutError:
+            raise qmp_unreachable(
+                record.name, record.qmp_socket, "connect timed out", record.qemu_log
+            ) from None
+        except OSError as e:
+            raise qmp_unreachable(record.name, record.qmp_socket, str(e), record.qemu_log) from e
+        except Exception as e:  # qemu.qmp wraps failures in its own ConnectError
+            raise qmp_unreachable(record.name, record.qmp_socket, str(e), record.qemu_log) from e
+        yield client
+    finally:
+        # After a 'quit' the peer hangs up first; EOF on teardown is normal.
+        with contextlib.suppress(Exception):
+            await client.disconnect()
+
+
+async def execute(client: QMPClient, command: str, arguments: dict[str, Any] | None = None) -> Any:
+    """Run one QMP command, translating QMP-level errors to ToolError."""
+    try:
+        return await client.execute(command, arguments or {})
+    except ExecuteError as e:
+        error_class = getattr(e, "error_class", None) or "GenericError"
+        raise qmp_command_failed(command, error_class, str(e)) from e
+
+
+async def wait_for_event(client: QMPClient, event_name: str, timeout: float) -> bool:
+    """Drain the session's event stream until event_name arrives or time runs out."""
+
+    async def _drain() -> None:
+        while True:
+            event = await client.events.get()
+            if event.get("event") == event_name:
+                return
+
+    try:
+        await asyncio.wait_for(_drain(), timeout=timeout)
+        return True
+    except asyncio.TimeoutError:
+        return False

src/mcqemu/registry.py

@@ -0,0 +1,91 @@
+"""Persistent VM registry with PID-liveness staleness detection."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+from .config import Config
+from .models import VMRecord
+
+
+def pid_alive(pid: int | None) -> bool:
+    """True iff the PID exists AND is actually a qemu-system process.
+
+    The /proc/<pid>/comm check guards against PID reuse after reboot or
+    long uptimes — a recycled PID belonging to some other process must not
+    make a dead VM look alive.
+    """
+    if not pid or pid <= 0:
+        return False
+    try:
+        comm = Path(f"/proc/{pid}/comm").read_text().strip()
+    except OSError:
+        return False
+    return comm.startswith("qemu-system")
+
+
+class VMRegistry:
+    def __init__(self, config: Config) -> None:
+        self._config = config
+        self._vms: dict[str, VMRecord] = {}
+
+    def load(self) -> None:
+        path = self._config.registry_path
+        if not path.exists():
+            self._vms = {}
+            return
+        data = json.loads(path.read_text())
+        self._vms = {name: VMRecord.from_dict(rec) for name, rec in data.items()}
+
+    def save(self) -> None:
+        path = self._config.registry_path
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(".tmp")
+        tmp.write_text(json.dumps({n: r.to_dict() for n, r in self._vms.items()}, indent=2))
+        os.replace(tmp, path)
+
+    def get(self, name: str) -> VMRecord | None:
+        return self._vms.get(name)
+
+    def add(self, record: VMRecord) -> None:
+        self._vms[record.name] = record
+        self.save()
+
+    def remove(self, name: str) -> None:
+        self._vms.pop(name, None)
+        self.save()
+
+    def all(self) -> list[VMRecord]:
+        return list(self._vms.values())
+
+    def names(self) -> list[str]:
+        return list(self._vms)
+
+    def refresh_pid(self, record: VMRecord) -> int | None:
+        """Re-read the pidfile for a spawned VM; returns the PID or None."""
+        if record.pidfile:
+            try:
+                record.pid = int(Path(record.pidfile).read_text().strip())
+            except (OSError, ValueError):
+                pass
+        return record.pid
+
+    def is_process_alive(self, record: VMRecord) -> bool:
+        """Liveness for spawned VMs via pidfile; attached VMs are judged by
+        their QMP socket instead (we may not know their PID)."""
+        if record.source == "attached" and record.pid is None:
+            return Path(record.qmp_socket).exists()
+        self.refresh_pid(record)
+        return pid_alive(record.pid)
+
+    def disks_in_use(self) -> dict[str, str]:
+        """Map of absolute disk path -> VM name, for every live registered VM."""
+        used: dict[str, str] = {}
+        for rec in self._vms.values():
+            if not self.is_process_alive(rec):
+                continue
+            for disk in rec.config.get("disks", []):
+                used[str(Path(disk).expanduser().resolve())] = rec.name
+        return used

src/mcqemu/resources.py

@@ -0,0 +1,19 @@
+"""MCP resources: browsable views of the VM registry."""
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+
+def register(mcp: FastMCP) -> None:
+    @mcp.resource("mcqemu://vms", description="All registered VMs and their records (JSON).")
+    async def vms_resource(ctx: Context = None) -> list[dict]:
+        registry = ctx.lifespan_context.registry
+        return [r.to_dict() for r in registry.all()]
+
+    @mcp.resource("mcqemu://vm/{name}", description="Full registry record for one VM (JSON).")
+    async def vm_resource(name: str, ctx: Context = None) -> dict:
+        registry = ctx.lifespan_context.registry
+        record = registry.get(name)
+        if record is None:
+            raise ToolError(f"No VM named {name!r}.")
+        return record.to_dict()

src/mcqemu/server.py

@@ -0,0 +1,55 @@
+"""mcqemu server — composition root and entry point."""
+
+import sys
+from contextlib import asynccontextmanager
+
+from fastmcp import FastMCP
+
+from . import __version__, prompts, resources
+from .config import Config
+from .registry import VMRegistry
+from .state import AppContext
+from .tools import register_all
+
+INSTRUCTIONS = """\
+mcqemu manages QEMU virtual machines on this host.
+
+Typical flows:
+- Fresh VM: image_create -> launch_vm(disks=[...], iso=...) -> watch serial log
+  -> stop_vm -> relaunch without iso.
+- Inspect: list_vms / vm_info. VMs survive MCP server restarts (they are
+  daemonized QEMU processes).
+- Inside the guest: guest_exec / guest_file_read / guest_file_write — these
+  need qemu-guest-agent installed in the guest OS (guest_ping to check).
+- Existing QEMU processes started outside this server can be managed after
+  attach_vm(name, qmp_socket=...).
+
+KVM acceleration is automatic when the guest arch matches the host; other
+architectures run under TCG emulation (works, but much slower).
+"""
+
+
+@asynccontextmanager
+async def lifespan(mcp: FastMCP):
+    config = Config.from_env()
+    config.state_dir.mkdir(parents=True, exist_ok=True)
+    config.runtime_dir.mkdir(parents=True, exist_ok=True)
+    registry = VMRegistry(config)
+    registry.load()
+    yield AppContext(config=config, registry=registry)
+    registry.save()  # VMs keep running by design — they are daemonized.
+
+
+mcp = FastMCP("mcqemu", lifespan=lifespan, instructions=INSTRUCTIONS)
+register_all(mcp)
+resources.register(mcp)
+prompts.register(mcp)
+
+
+def main() -> None:
+    print(f"mcqemu v{__version__}", file=sys.stderr)
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()

src/mcqemu/state.py

@@ -0,0 +1,14 @@
+"""Lifespan-scoped application state shared by all tools."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .config import Config
+from .registry import VMRegistry
+
+
+@dataclass
+class AppContext:
+    config: Config
+    registry: VMRegistry

src/mcqemu/tools/__init__.py

@@ -0,0 +1,10 @@
+"""Tool registration."""
+
+from fastmcp import FastMCP
+
+from . import guest, images, lifecycle, query, snapshots
+
+
+def register_all(mcp: FastMCP) -> None:
+    for module in (lifecycle, query, snapshots, images, guest):
+        module.register(mcp)

src/mcqemu/tools/_common.py

@@ -0,0 +1,19 @@
+"""Helpers shared by tool modules."""
+
+from fastmcp import Context
+
+from ..errors import vm_not_found
+from ..models import VMRecord
+from ..state import AppContext
+
+
+def app(ctx: Context) -> AppContext:
+    return ctx.lifespan_context
+
+
+def require_vm(ctx: Context, name: str) -> tuple[AppContext, VMRecord]:
+    state = app(ctx)
+    record = state.registry.get(name)
+    if record is None:
+        raise vm_not_found(name, state.registry.names())
+    return state, record

src/mcqemu/tools/guest.py

@@ -0,0 +1,147 @@
+"""Tools that reach inside the guest via qemu-guest-agent."""
+
+import asyncio
+import base64
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..qga import qga_session
+from ._common import require_vm
+
+_READ_CHUNK = 48 * 1024  # QGA caps single reads at 48 MiB of b64; stay modest.
+
+
+async def guest_ping(name: str, ctx: Context = None) -> dict:
+    """Check whether the qemu-guest-agent inside the VM is alive and
+    responding. A failure means the guest OS doesn't have the agent installed
+    or running — the other guest_* tools won't work until it does."""
+    _, record = require_vm(ctx, name)
+    async with qga_session(record) as client:
+        await asyncio.wait_for(client.execute("guest-ping"), timeout=3.0)
+    return {"name": name, "guest_agent": "responding"}
+
+
+async def guest_info(name: str, ctx: Context = None) -> dict:
+    """Report the guest OS details (name, version, kernel) and the guest
+    agent's version and supported commands."""
+    _, record = require_vm(ctx, name)
+    async with qga_session(record) as client:
+        agent = await client.execute("guest-info")
+        try:
+            osinfo = await client.execute("guest-get-osinfo")
+        except Exception:
+            osinfo = None
+    supported = sorted(
+        c["name"] for c in agent.get("supported_commands", []) if c.get("enabled")
+    )
+    return {
+        "name": name,
+        "agent_version": agent.get("version"),
+        "os": osinfo,
+        "supported_commands": supported,
+    }
+
+
+async def guest_exec(
+    name: str,
+    command: str,
+    args: list[str] | None = None,
+    stdin: str | None = None,
+    timeout: int = 30,
+    ctx: Context = None,
+) -> dict:
+    """Run a command inside the guest OS and return its stdout, stderr, and
+    exit code. `command` is the executable path or name; pass arguments
+    separately in `args` (this is exec, not a shell — for shell features use
+    command="/bin/sh", args=["-c", "your | pipeline"]). Requires
+    qemu-guest-agent in the guest."""
+    _, record = require_vm(ctx, name)
+    exec_args: dict = {"path": command, "capture-output": True}
+    if args:
+        exec_args["arg"] = args
+    if stdin is not None:
+        exec_args["input-data"] = base64.b64encode(stdin.encode()).decode()
+
+    async with qga_session(record) as client:
+        reply = await client.execute("guest-exec", exec_args)
+        pid = reply["pid"]
+        deadline = asyncio.get_event_loop().time() + timeout
+        while True:
+            status = await client.execute("guest-exec-status", {"pid": pid})
+            if status.get("exited"):
+                break
+            if asyncio.get_event_loop().time() > deadline:
+                raise ToolError(
+                    f"Command still running in guest after {timeout}s (guest pid {pid}). "
+                    "It continues in the background; raise timeout for long commands."
+                )
+            await asyncio.sleep(0.2)
+
+    def _decode(field: str) -> str:
+        data = status.get(field)
+        return base64.b64decode(data).decode(errors="replace") if data else ""
+
+    return {
+        "exitcode": status.get("exitcode"),
+        "stdout": _decode("out-data"),
+        "stderr": _decode("err-data"),
+        "stdout_truncated": status.get("out-truncated", False),
+        "stderr_truncated": status.get("err-truncated", False),
+    }
+
+
+async def guest_file_read(
+    name: str, path: str, max_bytes: int = 1_048_576, ctx: Context = None
+) -> dict:
+    """Read a text file from inside the guest (up to max_bytes, default 1 MiB).
+    Requires qemu-guest-agent in the guest."""
+    _, record = require_vm(ctx, name)
+    async with qga_session(record) as client:
+        handle = await client.execute("guest-file-open", {"path": path, "mode": "r"})
+        try:
+            chunks: list[bytes] = []
+            total = 0
+            truncated = False
+            while total < max_bytes:
+                count = min(_READ_CHUNK, max_bytes - total)
+                reply = await client.execute("guest-file-read", {"handle": handle, "count": count})
+                if reply.get("count", 0) > 0:
+                    chunk = base64.b64decode(reply["buf-b64"])
+                    chunks.append(chunk)
+                    total += len(chunk)
+                if reply.get("eof"):
+                    break
+            else:
+                truncated = True
+        finally:
+            await client.execute("guest-file-close", {"handle": handle})
+    return {
+        "path": path,
+        "content": b"".join(chunks).decode(errors="replace"),
+        "bytes_read": total,
+        "truncated": truncated,
+    }
+
+
+async def guest_file_write(
+    name: str, path: str, content: str, append: bool = False, ctx: Context = None
+) -> dict:
+    """Write a text file inside the guest (mode 'w' truncates, append=True
+    appends). Requires qemu-guest-agent in the guest."""
+    _, record = require_vm(ctx, name)
+    payload = base64.b64encode(content.encode()).decode()
+    async with qga_session(record) as client:
+        handle = await client.execute(
+            "guest-file-open", {"path": path, "mode": "a" if append else "w"}
+        )
+        try:
+            reply = await client.execute("guest-file-write", {"handle": handle, "buf-b64": payload})
+        finally:
+            await client.execute("guest-file-close", {"handle": handle})
+    return {"path": path, "bytes_written": reply.get("count", 0), "appended": append}
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (guest_ping, guest_info, guest_exec, guest_file_read, guest_file_write):
+        mcp.tool(fn)

src/mcqemu/tools/images.py

@@ -0,0 +1,169 @@
+"""Offline disk-image tools wrapping qemu-img."""
+
+from pathlib import Path
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..config import resolve_image_path
+from ..qemu_img import image_info_json, probe_format, run_qemu_img
+from ._common import app
+
+
+def _refuse_if_in_use(ctx: Context, path: Path, operation: str) -> None:
+    in_use = app(ctx).registry.disks_in_use()
+    if str(path) in in_use:
+        raise ToolError(
+            f"Refusing to {operation} {path}: it is attached to running VM "
+            f"{in_use[str(path)]!r}. Stop that VM first."
+        )
+
+
+def _resolve(path: str, must_exist: bool = True) -> Path:
+    try:
+        return resolve_image_path(path, must_exist=must_exist)
+    except ValueError as e:
+        raise ToolError(str(e)) from e
+
+
+async def image_create(
+    path: str,
+    size: str,
+    format: str = "qcow2",
+    backing_file: str | None = None,
+    overwrite: bool = False,
+    ctx: Context = None,
+) -> dict:
+    """Create a new disk image. `size` uses qemu-img suffixes, e.g. "20G".
+    qcow2 grows on demand, so a large virtual size costs almost nothing up
+    front. With `backing_file`, the new image becomes a copy-on-write overlay —
+    great for cheap disposable clones of a base image. Refuses to replace an
+    existing file unless overwrite=True."""
+    p = _resolve(path, must_exist=False)
+    if p.exists():
+        if not overwrite:
+            raise ToolError(f"{p} already exists. Pass overwrite=True to replace it.")
+        _refuse_if_in_use(ctx, p, "overwrite")
+
+    args = ["create", "-f", format]
+    if backing_file:
+        backing = _resolve(backing_file)
+        backing_fmt = await probe_format(str(backing))
+        args += ["-b", str(backing), "-F", backing_fmt]
+    args += [str(p), size]
+    await run_qemu_img(*args)
+    info = await image_info_json(str(p))
+    return {
+        "path": str(p),
+        "format": info["format"],
+        "virtual_size_bytes": info["virtual-size"],
+        "backing_file": backing_file,
+    }
+
+
+async def image_info(path: str, backing_chain: bool = False, ctx: Context = None) -> dict | list:
+    """Inspect a disk image: format, virtual and on-disk size, internal
+    snapshots, and (with backing_chain=True) the full copy-on-write chain."""
+    p = _resolve(path)
+    return await image_info_json(str(p), backing_chain=backing_chain)
+
+
+async def image_convert(
+    source: str,
+    dest: str,
+    format: str = "qcow2",
+    compress: bool = False,
+    overwrite: bool = False,
+    ctx: Context = None,
+) -> dict:
+    """Convert a disk image to another format (e.g. raw -> qcow2, vmdk ->
+    qcow2). Conversion flattens any backing chain into a standalone image.
+    compress=True enables qcow2 compression (smaller, slower)."""
+    src = _resolve(source)
+    dst = _resolve(dest, must_exist=False)
+    if dst.exists():
+        if not overwrite:
+            raise ToolError(f"{dst} already exists. Pass overwrite=True to replace it.")
+        _refuse_if_in_use(ctx, dst, "overwrite")
+    _refuse_if_in_use(ctx, src, "convert")
+
+    src_fmt = await probe_format(str(src))
+    args = ["convert", "-f", src_fmt, "-O", format]
+    if compress:
+        if format != "qcow2":
+            raise ToolError("compress=True is only supported for qcow2 output.")
+        args.append("-c")
+    args += [str(src), str(dst)]
+    await run_qemu_img(*args)
+    info = await image_info_json(str(dst))
+    return {"path": str(dst), "format": info["format"], "actual_size_bytes": info["actual-size"]}
+
+
+async def image_resize(path: str, size: str, shrink: bool = False, ctx: Context = None) -> dict:
+    """Resize a disk image's virtual size (e.g. size="30G", or "+10G" to grow
+    relatively). Growing is safe; shrinking DESTROYS data beyond the new size
+    and requires shrink=True as explicit confirmation (shrink the guest
+    filesystem first!)."""
+    p = _resolve(path)
+    _refuse_if_in_use(ctx, p, "resize")
+    args = ["resize"]
+    if shrink:
+        args.append("--shrink")
+    args += [str(p), size]
+    await run_qemu_img(*args)
+    info = await image_info_json(str(p))
+    return {"path": str(p), "virtual_size_bytes": info["virtual-size"]}
+
+
+async def image_snapshot_list(path: str, ctx: Context = None) -> list[dict]:
+    """List internal snapshots stored in a (not currently running) qcow2 image."""
+    p = _resolve(path)
+    info = await image_info_json(str(p))
+    return [
+        {
+            "tag": s.get("name"),
+            "vm_state_size": s.get("vm-state-size"),
+            "date": s.get("date-sec"),
+        }
+        for s in info.get("snapshots", [])
+    ]
+
+
+async def image_snapshot_create(path: str, tag: str, ctx: Context = None) -> dict:
+    """Create an internal disk-only snapshot in an offline qcow2 image. For
+    running VMs use vm_snapshot_create instead (it also captures RAM)."""
+    p = _resolve(path)
+    _refuse_if_in_use(ctx, p, "snapshot")
+    await run_qemu_img("snapshot", "-c", tag, str(p))
+    return {"path": str(p), "snapshot": tag, "created": True}
+
+
+async def image_snapshot_apply(path: str, tag: str, ctx: Context = None) -> dict:
+    """Revert an offline qcow2 image to internal snapshot `tag`. Data written
+    after the snapshot is lost."""
+    p = _resolve(path)
+    _refuse_if_in_use(ctx, p, "revert")
+    await run_qemu_img("snapshot", "-a", tag, str(p))
+    return {"path": str(p), "snapshot": tag, "applied": True}
+
+
+async def image_snapshot_delete(path: str, tag: str, ctx: Context = None) -> dict:
+    """Delete internal snapshot `tag` from an offline qcow2 image."""
+    p = _resolve(path)
+    _refuse_if_in_use(ctx, p, "modify")
+    await run_qemu_img("snapshot", "-d", tag, str(p))
+    return {"path": str(p), "snapshot": tag, "deleted": True}
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (
+        image_create,
+        image_info,
+        image_convert,
+        image_resize,
+        image_snapshot_list,
+        image_snapshot_create,
+        image_snapshot_apply,
+        image_snapshot_delete,
+    ):
+        mcp.tool(fn)

src/mcqemu/tools/lifecycle.py

@@ -0,0 +1,201 @@
+"""VM lifecycle tools: launch, stop, pause/resume, attach, forget."""
+
+import asyncio
+from pathlib import Path
+from typing import Literal
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..config import resolve_image_path, valid_vm_name
+from ..launcher import spawn_vm
+from ..models import VMConfig
+from ..qmp import execute, qmp_session, wait_for_event
+from ..registry import pid_alive
+from ._common import app, require_vm
+
+
+async def launch_vm(
+    name: str,
+    arch: str = "x86_64",
+    disks: list[str] | None = None,
+    iso: str | None = None,
+    memory_mb: int = 2048,
+    cpus: int = 2,
+    machine: str | None = None,
+    firmware: Literal["bios", "uefi"] = "bios",
+    port_forwards: list[str] | None = None,
+    no_net: bool = False,
+    extra_args: list[str] | None = None,
+    ctx: Context = None,
+) -> dict:
+    """Launch a new QEMU virtual machine and register it for management.
+
+    Use `disks` for existing image files (create them first with image_create);
+    use `iso` to boot an installer or live CD. With both, the VM boots the ISO
+    once then the disk afterwards. `port_forwards` entries like "2222:22" map
+    host port 2222 to guest port 22 (user-mode networking). KVM acceleration is
+    used automatically when the guest arch matches the host. The VM keeps
+    running even if this MCP server restarts; stop it with stop_vm.
+    """
+    state = app(ctx)
+    if not valid_vm_name(name):
+        raise ToolError(
+            f"Invalid VM name {name!r}: use letters, digits, '.', '_', '-' (max 48 chars)."
+        )
+
+    existing = state.registry.get(name)
+    if existing is not None:
+        if state.registry.is_process_alive(existing):
+            raise ToolError(
+                f"A VM named {name!r} is already running. Stop it first, or pick another name."
+            )
+        state.registry.remove(name)  # dead leftover — reuse the name
+
+    resolved_disks: list[str] = []
+    in_use = state.registry.disks_in_use()
+    for disk in disks or []:
+        try:
+            p = resolve_image_path(disk)
+        except ValueError as e:
+            raise ToolError(f"Bad disk path: {e}. Create images with image_create.") from e
+        if str(p) in in_use:
+            raise ToolError(f"Disk {p} is already attached to running VM {in_use[str(p)]!r}.")
+        resolved_disks.append(str(p))
+
+    resolved_iso: str | None = None
+    if iso:
+        try:
+            resolved_iso = str(resolve_image_path(iso))
+        except ValueError as e:
+            raise ToolError(f"Bad ISO path: {e}") from e
+
+    cfg = VMConfig(
+        name=name,
+        arch=arch,
+        machine=machine,
+        memory_mb=memory_mb,
+        cpus=cpus,
+        disks=resolved_disks,
+        iso=resolved_iso,
+        firmware=firmware,
+        port_forwards=port_forwards or [],
+        no_net=no_net,
+        extra_args=extra_args or [],
+    )
+    record = await spawn_vm(cfg, state.config)
+    state.registry.add(record)
+    return {
+        "name": name,
+        "status": "running",
+        "accel": record.accel,
+        "pid": record.pid,
+        "qmp_socket": record.qmp_socket,
+        "serial_log": record.serial_log,
+        "qemu_log": record.qemu_log,
+        "port_forwards": cfg.port_forwards,
+        "note": "TCG software emulation in use (slower than KVM)"
+        if record.accel == "tcg"
+        else "KVM hardware acceleration active",
+    }
+
+
+async def stop_vm(name: str, force: bool = False, timeout: int = 30, ctx: Context = None) -> dict:
+    """Stop a VM. By default sends a graceful ACPI power-button press and waits
+    for the guest to shut down; if the guest ignores it (no OS booted, or OS
+    without ACPI handling), the call fails with advice to retry with force=True,
+    which terminates QEMU immediately (like pulling the power cord).
+    """
+    state, record = require_vm(ctx, name)
+    if not state.registry.is_process_alive(record) and not Path(record.qmp_socket).exists():
+        return {"name": name, "status": "stopped", "note": "VM was already stopped."}
+
+    if force:
+        async with qmp_session(record) as client:
+            await execute(client, "quit")
+        deadline = asyncio.get_event_loop().time() + 10
+        while pid_alive(record.pid) and asyncio.get_event_loop().time() < deadline:
+            await asyncio.sleep(0.1)
+        return {"name": name, "status": "stopped", "method": "force quit"}
+
+    async with qmp_session(record) as client:
+        await execute(client, "system_powerdown")
+        if not await wait_for_event(client, "SHUTDOWN", timeout=timeout):
+            raise ToolError(
+                f"VM {name!r} did not shut down within {timeout}s after the ACPI "
+                "power signal — the guest may have no OS or ignores ACPI. "
+                "Retry with force=True to terminate it immediately."
+            )
+    deadline = asyncio.get_event_loop().time() + 10
+    while pid_alive(record.pid) and asyncio.get_event_loop().time() < deadline:
+        await asyncio.sleep(0.1)
+    return {"name": name, "status": "stopped", "method": "graceful ACPI shutdown"}
+
+
+async def pause_vm(name: str, ctx: Context = None) -> dict:
+    """Pause (freeze) a running VM's virtual CPUs. The VM stays in memory;
+    resume it with resume_vm."""
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        await execute(client, "stop")
+    return {"name": name, "status": "paused"}
+
+
+async def resume_vm(name: str, ctx: Context = None) -> dict:
+    """Resume a VM previously frozen with pause_vm."""
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        await execute(client, "cont")
+    return {"name": name, "status": "running"}
+
+
+async def attach_vm(
+    name: str,
+    qmp_socket: str,
+    qga_socket: str | None = None,
+    pid: int | None = None,
+    ctx: Context = None,
+) -> dict:
+    """Register an externally launched QEMU process so the other tools can
+    manage it. Point qmp_socket at its QMP unix socket (the QEMU process must
+    have been started with e.g. -qmp unix:/path,server=on,wait=off). Optionally
+    provide qga_socket for guest-agent tools and pid for liveness tracking.
+    """
+    state = app(ctx)
+    if not valid_vm_name(name):
+        raise ToolError(
+            f"Invalid VM name {name!r}: use letters, digits, '.', '_', '-' (max 48 chars)."
+        )
+    if state.registry.get(name) is not None:
+        raise ToolError(f"A VM named {name!r} is already registered. Use forget_vm first.")
+    if not Path(qmp_socket).exists():
+        raise ToolError(f"No socket at {qmp_socket}. Is the QEMU process running with -qmp?")
+
+    from ..models import VMRecord
+
+    record = VMRecord(
+        name=name, source="attached", qmp_socket=qmp_socket, qga_socket=qga_socket, pid=pid
+    )
+    async with qmp_session(record) as client:
+        status = await execute(client, "query-status")
+    state.registry.add(record)
+    return {"name": name, "source": "attached", "status": status.get("status", "unknown")}
+
+
+async def forget_vm(name: str, force: bool = False, ctx: Context = None) -> dict:
+    """Remove a VM from the registry WITHOUT stopping it — the QEMU process is
+    left untouched. Refuses to forget a running VM this server spawned unless
+    force=True (to avoid orphaning processes by accident)."""
+    state, record = require_vm(ctx, name)
+    if record.source == "spawned" and state.registry.is_process_alive(record) and not force:
+        raise ToolError(
+            f"VM {name!r} is still running (pid {record.pid}). Stop it with stop_vm, "
+            "or pass force=True to deliberately orphan the process."
+        )
+    state.registry.remove(name)
+    return {"name": name, "forgotten": True, "process_left_running": bool(record.pid and pid_alive(record.pid))}
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (launch_vm, stop_vm, pause_vm, resume_vm, attach_vm, forget_vm):
+        mcp.tool(fn)

src/mcqemu/tools/query.py

@@ -0,0 +1,84 @@
+"""Read-only VM inspection tools."""
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..qmp import execute, qmp_session
+from ._common import app, require_vm
+
+
+async def _live_status(state, record) -> str:
+    if not state.registry.is_process_alive(record):
+        return "stopped"
+    try:
+        async with qmp_session(record) as client:
+            reply = await execute(client, "query-status")
+        return reply.get("status", "unknown")
+    except ToolError:
+        return "unreachable"
+
+
+async def list_vms(ctx: Context = None) -> list[dict]:
+    """List every registered VM with its live status (running, paused,
+    shutdown, stopped, or unreachable). Includes both VMs launched by this
+    server and externally attached ones."""
+    state = app(ctx)
+    out = []
+    for record in state.registry.all():
+        status = await _live_status(state, record)
+        out.append(
+            {
+                "name": record.name,
+                "status": status,
+                "source": record.source,
+                "arch": record.arch,
+                "accel": record.accel,
+                "pid": record.pid,
+                "created_at": record.created_at,
+            }
+        )
+    return out
+
+
+async def vm_info(name: str, ctx: Context = None) -> dict:
+    """Detailed information about one VM: its launch configuration, log file
+    paths, and — when running — live QMP state (status, vCPUs, block devices)."""
+    state, record = require_vm(ctx, name)
+    info: dict = {
+        "name": record.name,
+        "source": record.source,
+        "arch": record.arch,
+        "accel": record.accel,
+        "pid": record.pid,
+        "qmp_socket": record.qmp_socket,
+        "qga_socket": record.qga_socket,
+        "qemu_log": record.qemu_log,
+        "serial_log": record.serial_log,
+        "created_at": record.created_at,
+        "config": record.config,
+    }
+    status = await _live_status(state, record)
+    info["status"] = status
+    if status in ("stopped", "unreachable"):
+        return info
+
+    async with qmp_session(record) as client:
+        cpus = await execute(client, "query-cpus-fast")
+        blocks = await execute(client, "query-block")
+    info["vcpus"] = len(cpus)
+    info["block_devices"] = [
+        {
+            "device": b.get("device") or b.get("qdev", ""),
+            "file": (b.get("inserted") or {}).get("file"),
+            "format": (b.get("inserted") or {}).get("drv"),
+            "read_only": (b.get("inserted") or {}).get("ro"),
+        }
+        for b in blocks
+        if b.get("inserted")
+    ]
+    return info
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (list_vms, vm_info):
+        mcp.tool(fn)

src/mcqemu/tools/snapshots.py

@@ -0,0 +1,73 @@
+"""Live (internal) snapshots of running VMs.
+
+v1 drives savevm/loadvm/delvm through QMP's human-monitor-command passthrough;
+the QMP jobs API (snapshot-save/-load) is the eventual upgrade path but needs
+job polling for the same v1 outcome. Internal snapshots require all writable
+disks to be qcow2.
+"""
+
+from fastmcp import Context, FastMCP
+from fastmcp.exceptions import ToolError
+
+from ..qmp import execute, qmp_session
+from ._common import require_vm
+
+
+async def _hmp(client, command_line: str) -> str:
+    return await execute(client, "human-monitor-command", {"command-line": command_line})
+
+
+async def _hmp_expect_silence(client, command_line: str, what: str) -> None:
+    out = await _hmp(client, command_line)
+    if out and out.strip():
+        raise ToolError(f"{what} failed: {out.strip()}")
+
+
+def _check_tag(tag: str) -> None:
+    if not tag or any(c.isspace() for c in tag) or tag.startswith("-"):
+        raise ToolError(f"Invalid snapshot tag {tag!r}: no whitespace, must not start with '-'.")
+
+
+async def vm_snapshot_create(name: str, tag: str, ctx: Context = None) -> dict:
+    """Save a live internal snapshot (RAM + device + disk state) of a running
+    VM under `tag`. Requires the VM's writable disks to be qcow2. The VM pauses
+    briefly while state is written. Restore later with vm_snapshot_restore."""
+    _check_tag(tag)
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        await _hmp_expect_silence(client, f"savevm {tag}", f"savevm {tag!r}")
+    return {"name": name, "snapshot": tag, "created": True}
+
+
+async def vm_snapshot_restore(name: str, tag: str, ctx: Context = None) -> dict:
+    """Roll a running VM back to the internal snapshot `tag` (RAM, devices and
+    disks all revert). Anything that happened after the snapshot is lost."""
+    _check_tag(tag)
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        await _hmp_expect_silence(client, f"loadvm {tag}", f"loadvm {tag!r}")
+    return {"name": name, "snapshot": tag, "restored": True}
+
+
+async def vm_snapshot_delete(name: str, tag: str, ctx: Context = None) -> dict:
+    """Delete the internal snapshot `tag` from a running VM's disks. The VM
+    keeps running; only the saved snapshot is removed."""
+    _check_tag(tag)
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        await _hmp_expect_silence(client, f"delvm {tag}", f"delvm {tag!r}")
+    return {"name": name, "snapshot": tag, "deleted": True}
+
+
+async def vm_snapshot_list(name: str, ctx: Context = None) -> dict:
+    """List internal snapshots visible to a running VM. For stopped VMs use
+    image_snapshot_list on the disk file instead."""
+    _, record = require_vm(ctx, name)
+    async with qmp_session(record) as client:
+        out = await _hmp(client, "info snapshots")
+    return {"name": name, "snapshots_raw": out.strip() or "(none)"}
+
+
+def register(mcp: FastMCP) -> None:
+    for fn in (vm_snapshot_create, vm_snapshot_restore, vm_snapshot_delete, vm_snapshot_list):
+        mcp.tool(fn)