1#!/usr/bin/env python3
2"""MCP server (stdio, newline-delimited JSON-RPC) and CLI for driving the
3Windows 7 QEMU clones through their AutoHotkey exec listeners."""
4
5import base64
6import json
7import os
8from pathlib import Path
9import socket
10import subprocess
11import sys
12import urllib.error
13import urllib.request
14
15from env import ROOT, setting
16from vm import BASES
17
18DEFAULT_TARGET = os.environ.get("WIN7_TARGET", "local")
19TARGETS_PATH = Path(setting("WIN7_TARGETS_FILE") or Path(setting("ONE_VM_HOME") or ROOT / "lab-unset") / "targets.json").expanduser()
20VM = Path(__file__).with_name("vm.py")
21
22
23class Win7Error(Exception):
24 pass
25
26
27def resolve_target(name):
28 targets = {"local": {"base": "http://127.0.0.1:18777"}}
29 targets.update({base + "-build": {"base": "http://127.0.0.1:%d" % port}
30 for base, (_stem, port) in BASES.items()})
31 if TARGETS_PATH.exists():
32 try:
33 configured = json.loads(TARGETS_PATH.read_text())
34 except (OSError, ValueError) as e:
35 raise Win7Error("Cannot read %s: %s" % (TARGETS_PATH, e))
36 if not isinstance(configured, dict):
37 raise Win7Error("Windows targets must be a JSON object: %s" % TARGETS_PATH)
38 targets.update(configured)
39 name = name or DEFAULT_TARGET
40 target = targets.get(name)
41 if not isinstance(target, dict) or not target.get("base"):
42 raise Win7Error(
43 "Unknown Windows target %r. Choose: %s"
44 % (name, ", ".join(sorted(targets)))
45 )
46 token = target.get("token")
47 if target.get("token_env"):
48 token = os.environ.get(target["token_env"])
49 return name, target["base"].rstrip("/"), token
50
51
52def request(path, payload, timeout_ms=15000, target=None):
53 """POST json to the listener (GET when payload is None). Raises Win7Error."""
54 name, base, token = resolve_target(target)
55 headers = {"Content-Type": "application/json"}
56 if token:
57 headers["X-Win7-Token"] = token
58 body = None if payload is None else json.dumps(payload).encode("utf-8")
59 req = urllib.request.Request(base + path, data=body, headers=headers)
60 try:
61 # The box owns the deadline; give the socket slack so its own timeout
62 # wins and we get a real stdout/stderr back instead of a dead socket.
63 with urllib.request.urlopen(req, timeout=timeout_ms / 1000.0 + 15) as resp:
64 raw = resp.read()
65 except urllib.error.HTTPError as e:
66 raise Win7Error("%s %s: HTTP %d %s" % (path, base, e.code, e.reason))
67 except (urllib.error.URLError, socket.timeout, OSError) as e:
68 reason = getattr(e, "reason", e)
69 raise Win7Error(
70 "Cannot reach %r at %s (%s). Start it and open the desktop agent."
71 % (name, base, reason)
72 )
73 try:
74 return json.loads(raw.decode("utf-8"))
75 except ValueError:
76 raise Win7Error("%s returned non-JSON: %r" % (path, raw[:200]))
77
78
79def do_health(target=None):
80 return request("/health", None, target=target)
81
82
83def do_shot(target=None):
84 return request("/shot", {}, target=target)
85
86
87def do_exec(script, shot_delay_ms=500, timeout_ms=60000, target=None):
88 return request(
89 "/exec",
90 {"script": script, "shot_delay_ms": shot_delay_ms, "timeout_ms": timeout_ms},
91 timeout_ms,
92 target,
93 )
94
95
96def do_cmd(command, timeout_ms=60000, target=None):
97 return request(
98 "/cmd", {"command": command, "timeout_ms": timeout_ms}, timeout_ms, target
99 )
100
101
102def do_spawn(command, target=None):
103 return request("/spawn", {"command": command}, target=target)
104
105
106def do_ui(target=None):
107 return request("/ui", {}, target=target)
108
109
110# The base64 stays inside this process on both transfers: a tool that took file
111# bytes as an argument would spend the whole file as context tokens.
112def do_put(local, remote, target=None):
113 with open(local, "rb") as f:
114 blob = f.read()
115 resp = request("/put", {"path": remote, "b64": base64.b64encode(blob).decode("ascii")},
116 120000, target)
117 resp.setdefault("bytes", len(blob))
118 return resp
119
120
121def do_get(remote, local, target=None):
122 resp = request("/get", {"path": remote}, 120000, target)
123 if resp.get("b64"):
124 with open(local, "wb") as f:
125 f.write(base64.b64decode(resp["b64"]))
126 return {"bytes": resp.get("bytes"), "path": os.path.abspath(local),
127 "error": resp.get("error")}
128
129
130# Raw string: this text is mostly about backslashes, and rendering it correctly
131# matters more than keeping the source lines joined.
132EXEC_DESCRIPTION = r"""Run an AutoHotkey v2 script on the Windows 7 desktop.
133Returns whatever the script printed, plus a screenshot taken shot_delay_ms
134after the script exits.
135
136Coordinates are screen-absolute and match the returned screenshot
137pixel-for-pixel (CoordMode Screen is already set; do not change it). The only
138way to send text back is FileAppend(text, "*") -- there is no implicit output.
139Put a whole sequence of actions in one script; one call per click is slow and
140blind.
141
142BACKSLASHES. AutoHotkey's escape character is the backtick, NOT the backslash,
143so a backslash inside an AHK string is already literal. You are emitting this
144script as a JSON string, so one literal backslash is written "\\" in the JSON
145and arrives in the script as "\". Never write "\\\\" -- that is what makes an
146app receive A:\\cute.png instead of A:\cute.png. Escape inside AHK with the
147backtick instead: `n newline, `t tab, `" quote.
148
149LITERAL TEXT. Send() reads ^ + ! # { } as Ctrl/Shift/Alt/Win and key groups.
150Use SendText() for anything literal -- paths, passwords, arbitrary content --
151and keep Send() for actual key combinations.
152
153CLEAN UP. When you finish a task, close the applications you opened (WinClose,
154or the app's own quit path). Leaving windows stacked makes later screenshots
155harder to read, and a forgotten modal swallows input from the next script.
156
157Click something, let the UI settle:
158 Click(512, 384)
159 Sleep(300)
160
161Type a literal path into the focused field:
162 SendText("A:\cute.png")
163 Send("{Enter}")
164
165Shortcut, then read the result out of the clipboard:
166 Send("^a^c")
167 ClipWait(1)
168 FileAppend(A_Clipboard, "*")
169
170Launch an app, wait for its window, and close it when done:
171 Run("mspaint.exe")
172 WinWait("Paint", , 10)
173 WinActivate()
174 WinClose("Paint")
175
176Raise timeout_ms when the script itself waits on the UI; raise shot_delay_ms
177when an animation or app launch needs longer to settle before the screenshot."""
178
179TARGET_PROPERTY = {
180 "type": "string",
181 "description": "Target name. Omit to use the configured default.",
182}
183
184TOOLS = [
185 {
186 "name": "win7_exec",
187 "description": EXEC_DESCRIPTION,
188 "inputSchema": {
189 "type": "object",
190 "properties": {
191 "target": TARGET_PROPERTY,
192 "script": {"type": "string", "description": "AutoHotkey v2 source."},
193 "shot_delay_ms": {
194 "type": "integer",
195 "default": 500,
196 "description": "Wait this long after the script ends, then screenshot.",
197 },
198 "timeout_ms": {
199 "type": "integer",
200 "default": 60000,
201 "description": "Kill the script after this long.",
202 },
203 },
204 "required": ["script"],
205 },
206 },
207 {
208 "name": "win7_cmd",
209 "description": (
210 "Run a command through cmd.exe on the Windows 7 box and return its output. "
211 "No screenshot -- use it to inspect files, launch programs and check state "
212 "without spending a screenshot on it."
213 ),
214 "inputSchema": {
215 "type": "object",
216 "properties": {
217 "target": TARGET_PROPERTY,
218 "command": {"type": "string", "description": "Passed to cmd.exe /c."},
219 "timeout_ms": {"type": "integer", "default": 60000},
220 },
221 "required": ["command"],
222 },
223 },
224 {
225 "name": "win7_spawn",
226 "description": (
227 "Start a detached Windows process and return immediately. Its standard "
228 "handles are closed, so a long-lived GUI or capture process cannot wedge "
229 "the control channel."
230 ),
231 "inputSchema": {
232 "type": "object",
233 "properties": {
234 "target": TARGET_PROPERTY,
235 "command": {"type": "string", "description": "Windows command line."},
236 },
237 "required": ["command"],
238 },
239 },
240 {
241 "name": "win7_put",
242 "description": (
243 "Copy a file from this Mac to the Windows 7 box. Give two paths; the bytes "
244 "never pass through the conversation, so file size costs nothing."
245 ),
246 "inputSchema": {
247 "type": "object",
248 "properties": {
249 "target": TARGET_PROPERTY,
250 "local": {"type": "string", "description": "Path on the Mac."},
251 "remote": {
252 "type": "string",
253 "description": "Windows path, e.g. C:\\\\work\\\\a.one.",
254 },
255 },
256 "required": ["local", "remote"],
257 },
258 },
259 {
260 "name": "win7_get",
261 "description": "Copy a file from the Windows 7 box back to this Mac.",
262 "inputSchema": {
263 "type": "object",
264 "properties": {
265 "target": TARGET_PROPERTY,
266 "remote": {"type": "string", "description": "Windows path."},
267 "local": {"type": "string", "description": "Path on the Mac."},
268 },
269 "required": ["remote", "local"],
270 },
271 },
272 {
273 "name": "win7_shot",
274 "description": "Screenshot the Windows 7 desktop without running anything.",
275 "inputSchema": {"type": "object", "properties": {"target": TARGET_PROPERTY}},
276 },
277 {
278 "name": "win7_ui",
279 "description": (
280 "Dump the foreground window's control tree as text -- class name, window "
281 "text and client rect (l,t,w,h) for the window and each child control. "
282 "It reads real Win32 controls, so it is excellent for dialogs, menus and "
283 "standard controls, and near-useless for custom-drawn canvases like "
284 "OneNote's page surface -- reach for a screenshot there instead."
285 ),
286 "inputSchema": {"type": "object", "properties": {"target": TARGET_PROPERTY}},
287 },
288]
289
290TOOLS += [
291 {
292 "name": "win7_vm_up",
293 "description": (
294 "Create a named Windows clone when absent, then boot it. base picks Windows 7 "
295 "(x64, OneNote 2010), 10 (x64, emulated and slow) or 11 (arm64, native speed). "
296 "Creation settings are ignored for an existing clone. Set wait to return only "
297 "when its authenticated desktop agent reports the expected hostname."
298 ),
299 "inputSchema": {
300 "type": "object",
301 "properties": {
302 "name": {"type": "string",
303 "description": "Unique 1-11 character lowercase VM name."},
304 "base": {"type": "string", "enum": sorted(BASES), "default": "win7"},
305 "hostname": {"type": "string", "description": "Optional Windows hostname."},
306 "cpus": {"type": "integer", "default": 2, "minimum": 1, "maximum": 16},
307 "memory_mb": {"type": "integer", "default": 4096,
308 "minimum": 1024, "maximum": 65536},
309 "port": {"type": "integer", "minimum": 1024, "maximum": 65535},
310 "display": {"type": "boolean", "default": False},
311 "wait": {"type": "boolean", "default": False},
312 "timeout": {"type": "integer", "default": 300, "minimum": 1,
313 "maximum": 900},
314 },
315 "required": ["name"],
316 },
317 },
318 {
319 "name": "win7_vm_down",
320 "description": (
321 "Cleanly stop one clone and delete its overlay and metadata. Set "
322 "preserve_machine to keep the stopped clone for reproduction or reuse."
323 ),
324 "inputSchema": {
325 "type": "object",
326 "properties": {
327 "name": {"type": "string"},
328 "timeout": {"type": "integer", "default": 60, "minimum": 1,
329 "maximum": 110},
330 "preserve_machine": {"type": "boolean", "default": False},
331 },
332 "required": ["name"],
333 },
334 },
335 {
336 "name": "win7_vm_status",
337 "description": "List every clone, or report whether one named clone is absent, stopped, or running.",
338 "inputSchema": {
339 "type": "object",
340 "properties": {"name": {"type": "string"}},
341 },
342 },
343]
344
345
346def win_line(resp):
347 win = resp.get("win")
348 if not win or not (win.get("title") or win.get("class")):
349 return ""
350 tag = win.get("class") or ""
351 if win.get("dialog"):
352 tag = (tag + " dialog").strip()
353 return 'window: "%s" (%s)' % (win.get("title", ""), tag)
354
355
356def shot_blocks(resp, text_prefix=""):
357 text = text_prefix + "screen: %sx%s" % (resp.get("w"), resp.get("h"))
358 wl = win_line(resp)
359 if wl:
360 text += "\n" + wl
361 blocks = [{"type": "text", "text": text}]
362 png = resp.get("png_b64")
363 if png:
364 blocks.append({"type": "image", "data": png, "mimeType": "image/png"})
365 return blocks
366
367
368def text_result(resp):
369 lines = ["exit=%s" % resp.get("exit")]
370 for key in ("stdout", "stderr", "error"):
371 val = resp.get(key)
372 if val:
373 lines.append("%s:\n%s" % (key, val))
374 return "\n".join(lines)
375
376
377def vm_tool(name, args):
378 verb = name.removeprefix("win7_vm_")
379 if verb == "status":
380 argv = ["status"] + ([args["name"]] if args.get("name") else [])
381 elif verb == "up":
382 argv = ["up", args["name"]]
383 for key, option in (("base", "--base"), ("hostname", "--hostname"), ("cpus", "--cpus"),
384 ("memory_mb", "--memory"), ("port", "--port")):
385 if args.get(key) is not None:
386 argv += [option, str(args[key])]
387 if args.get("display"):
388 argv.append("--display")
389 if args.get("wait"):
390 argv += ["--wait", "--timeout", str(args.get("timeout", 300))]
391 elif verb == "down":
392 argv = ["down", args["name"], "--timeout", str(args.get("timeout", 60))]
393 if args.get("preserve_machine"):
394 argv.append("--preserve-machine")
395 process = subprocess.run(
396 [sys.executable, str(VM)] + argv,
397 capture_output=True,
398 text=True,
399 timeout=args.get("timeout", 300) + 15 if verb == "up" and args.get("wait")
400 else 120,
401 )
402 output = (process.stdout + process.stderr).strip()
403 return [{"type": "text", "text": output or "ok"}], process.returncode != 0
404
405
406def call_tool(name, args):
407 if name.startswith("win7_vm_"):
408 return vm_tool(name, args)
409 target = args.get("target")
410 if name == "win7_shot":
411 return shot_blocks(do_shot(target)), False
412 if name == "win7_ui":
413 resp = do_ui(target)
414 wl = win_line(resp)
415 parts = [p for p in (wl, resp.get("controls"), resp.get("error") and
416 "error: %s" % resp["error"]) if p]
417 return [{"type": "text", "text": "\n".join(parts)}], False
418 if name == "win7_put":
419 resp = do_put(args["local"], args["remote"], target)
420 text = "wrote %s bytes to %s" % (resp.get("bytes"), resp.get("path"))
421 return [{"type": "text", "text": text}], False
422 if name == "win7_get":
423 resp = do_get(args["remote"], args["local"], target)
424 text = "read %s bytes to %s" % (resp.get("bytes"), resp.get("path"))
425 return [{"type": "text", "text": text}], False
426 if name == "win7_cmd":
427 command = args.get("command")
428 if not isinstance(command, str) or not command.strip():
429 return [{"type": "text", "text": "command is required"}], True
430 resp = do_cmd(command, args.get("timeout_ms", 60000), target)
431 return [{"type": "text", "text": text_result(resp)}], False
432 if name == "win7_spawn":
433 command = args.get("command")
434 if not isinstance(command, str) or not command.strip():
435 return [{"type": "text", "text": "command is required"}], True
436 resp = do_spawn(command, target)
437 return [{"type": "text", "text": "pid=%s" % resp.get("pid")}], False
438 if name == "win7_exec":
439 script = args.get("script")
440 if not isinstance(script, str) or not script.strip():
441 return [{"type": "text", "text": "script is required"}], True
442 resp = do_exec(
443 script,
444 args.get("shot_delay_ms", 500),
445 args.get("timeout_ms", 60000),
446 target,
447 )
448 return shot_blocks(resp, text_result(resp) + "\n"), False
449 raise Win7Error("unknown tool %r" % (name,))
450
451
452def handle(method, params):
453 if method == "initialize":
454 return {
455 "protocolVersion": "2025-06-18",
456 "capabilities": {"tools": {}},
457 "serverInfo": {"name": "win7", "version": "1.0.0"},
458 }
459 if method == "ping":
460 return {}
461 if method == "tools/list":
462 return {"tools": TOOLS}
463 if method == "tools/call":
464 try:
465 content, is_error = call_tool(params.get("name"), params.get("arguments") or {})
466 except Win7Error as e:
467 content, is_error = [{"type": "text", "text": str(e)}], True
468 # No structuredContent key, ever: Codex drops content[] outright when it
469 # is present (openai/codex#10334), which silently discards the screenshot.
470 result = {"content": content, "_meta": {"codex/imageDetail": "original"}}
471 if is_error:
472 result["isError"] = True
473 return result
474 return None
475
476
477def serve():
478 out = sys.stdout
479 for line in sys.stdin:
480 line = line.strip()
481 if not line:
482 continue
483 try:
484 msg = json.loads(line)
485 except ValueError:
486 print("win7: dropping unparseable line: %r" % line[:200], file=sys.stderr)
487 continue
488 mid = msg.get("id")
489 if mid is None:
490 continue # notification: a reply would itself be a protocol error
491 try:
492 result = handle(msg.get("method"), msg.get("params") or {})
493 except Exception as e: # a crash here would wedge the client forever
494 reply = {
495 "jsonrpc": "2.0",
496 "id": mid,
497 "error": {"code": -32603, "message": "%s: %s" % (type(e).__name__, e)},
498 }
499 else:
500 if result is None:
501 reply = {
502 "jsonrpc": "2.0",
503 "id": mid,
504 "error": {"code": -32601, "message": "unknown method %r" % msg.get("method")},
505 }
506 else:
507 reply = {"jsonrpc": "2.0", "id": mid, "result": result}
508 out.write(json.dumps(reply) + "\n")
509 out.flush()
510
511
512SHOT_PATH = "screenshot.png"
513
514
515def write_png(resp):
516 with open(SHOT_PATH, "wb") as f:
517 f.write(base64.b64decode(resp["png_b64"]))
518 print("%sx%s -> %s" % (resp.get("w"), resp.get("h"), SHOT_PATH))
519
520
521def cli(argv):
522 target = None
523 if argv[:1] == ["--target"]:
524 if len(argv) < 3:
525 raise Win7Error("usage: mcp_win7.py --target <name> <command>")
526 target, argv = argv[1], argv[2:]
527 verb = argv[0]
528 if verb == "health":
529 print(json.dumps(do_health(target), indent=2))
530 elif verb == "shot":
531 write_png(do_shot(target))
532 elif verb == "ui":
533 resp = do_ui(target)
534 wl = win_line(resp)
535 if wl:
536 print(wl)
537 print(resp.get("controls") or "")
538 if resp.get("error"):
539 print("error: %s" % resp["error"], file=sys.stderr)
540 elif verb in ("put", "get"):
541 if len(argv) < 3:
542 raise Win7Error("usage: mcp_win7.py put <local> <remote> | get <remote> <local>")
543 resp = (do_put(argv[1], argv[2], target) if verb == "put"
544 else do_get(argv[1], argv[2], target))
545 print("%s bytes -> %s" % (resp.get("bytes"), resp.get("path")))
546 elif verb in ("exec", "cmd", "spawn"):
547 if len(argv) < 2:
548 raise Win7Error("usage: mcp_win7.py %s '<text>'" % verb)
549 if verb == "exec":
550 resp = do_exec(argv[1], target=target)
551 elif verb == "cmd":
552 resp = do_cmd(argv[1], target=target)
553 else:
554 resp = do_spawn(argv[1], target=target)
555 print("pid=%s" % resp.get("pid"))
556 return
557 for stream, text in ((sys.stdout, resp.get("stdout")), (sys.stderr, resp.get("stderr"))):
558 if text:
559 stream.write(text if text.endswith("\n") else text + "\n")
560 if resp.get("error"):
561 print("error: %s" % resp["error"], file=sys.stderr)
562 print("exit=%s" % resp.get("exit"), file=sys.stderr)
563 if verb == "exec":
564 write_png(resp)
565 else:
566 raise Win7Error(
567 "usage: mcp_win7.py [--target <name>] "
568 "health|shot|ui|exec <ahk>|cmd <command>|spawn <command>|put <local> <remote>"
569 "|get <remote> <local>"
570 )
571
572
573if __name__ == "__main__":
574 if len(sys.argv) > 1:
575 try:
576 cli(sys.argv[1:])
577 except Win7Error as e:
578 print("win7: %s" % e, file=sys.stderr)
579 sys.exit(1)
580 else:
581 try:
582 serve()
583 except KeyboardInterrupt:
584 pass