1#!/usr/bin/env python3
2"""MCP server (stdio, newline-delimited JSON-RPC) and CLI for driving the
3Windows 7 box `wayback` through its AutoHotkey exec listener."""
4
5import base64
6import json
7import os
8from pathlib import Path
9import socket
10import subprocess
11import sys
12import urllib.error
13import urllib.request
14
15WAYBACK_BASE = os.environ.get("WIN7", "http://100.104.74.68:8777").rstrip("/")
16WAYBACK_TOKEN = os.environ.get("WIN7_TOKEN")
17DEFAULT_TARGET = os.environ.get("WIN7_TARGET", "wayback")
18TARGETS_PATH = Path(os.environ.get(
19 "WIN7_TARGETS_FILE", "/Volumes/Documents/OneNote VMs/targets.json"
20)).expanduser()
21VM = Path(__file__).with_name("vm.py")
22
23
24class Win7Error(Exception):
25 pass
26
27
28def resolve_target(name):
29 targets = {
30 "wayback": {"base": WAYBACK_BASE, "token": WAYBACK_TOKEN},
31 "local": {"base": "http://127.0.0.1:18777"},
32 }
33 if TARGETS_PATH.exists():
34 try:
35 configured = json.loads(TARGETS_PATH.read_text())
36 except (OSError, ValueError) as e:
37 raise Win7Error("Cannot read %s: %s" % (TARGETS_PATH, e))
38 if not isinstance(configured, dict):
39 raise Win7Error("Windows targets must be a JSON object: %s" % TARGETS_PATH)
40 targets.update(configured)
41 name = name or DEFAULT_TARGET
42 target = targets.get(name)
43 if not isinstance(target, dict) or not target.get("base"):
44 raise Win7Error(
45 "Unknown Windows target %r. Choose: %s"
46 % (name, ", ".join(sorted(targets)))
47 )
48 token = target.get("token")
49 if target.get("token_env"):
50 token = os.environ.get(target["token_env"])
51 return name, target["base"].rstrip("/"), token
52
53
54def request(path, payload, timeout_ms=15000, target=None):
55 """POST json to the listener (GET when payload is None). Raises Win7Error."""
56 name, base, token = resolve_target(target)
57 headers = {"Content-Type": "application/json"}
58 if token:
59 headers["X-Win7-Token"] = token
60 body = None if payload is None else json.dumps(payload).encode("utf-8")
61 req = urllib.request.Request(base + path, data=body, headers=headers)
62 try:
63 # The box owns the deadline; give the socket slack so its own timeout
64 # wins and we get a real stdout/stderr back instead of a dead socket.
65 with urllib.request.urlopen(req, timeout=timeout_ms / 1000.0 + 15) as resp:
66 raw = resp.read()
67 except urllib.error.HTTPError as e:
68 raise Win7Error("%s %s: HTTP %d %s" % (path, base, e.code, e.reason))
69 except (urllib.error.URLError, socket.timeout, OSError) as e:
70 reason = getattr(e, "reason", e)
71 raise Win7Error(
72 "Cannot reach %r at %s (%s). Start it and open the desktop agent."
73 % (name, base, reason)
74 )
75 try:
76 return json.loads(raw.decode("utf-8"))
77 except ValueError:
78 raise Win7Error("%s returned non-JSON: %r" % (path, raw[:200]))
79
80
81def do_health(target=None):
82 return request("/health", None, target=target)
83
84
85def do_shot(target=None):
86 return request("/shot", {}, target=target)
87
88
89def do_exec(script, shot_delay_ms=500, timeout_ms=60000, target=None):
90 return request(
91 "/exec",
92 {"script": script, "shot_delay_ms": shot_delay_ms, "timeout_ms": timeout_ms},
93 timeout_ms,
94 target,
95 )
96
97
98def do_cmd(command, timeout_ms=60000, target=None):
99 return request(
100 "/cmd", {"command": command, "timeout_ms": timeout_ms}, timeout_ms, target
101 )
102
103
104def do_spawn(command, target=None):
105 return request("/spawn", {"command": command}, target=target)
106
107
108def do_ui(target=None):
109 return request("/ui", {}, target=target)
110
111
112# The base64 stays inside this process on both transfers: a tool that took file
113# bytes as an argument would spend the whole file as context tokens.
114def do_put(local, remote, target=None):
115 with open(local, "rb") as f:
116 blob = f.read()
117 resp = request("/put", {"path": remote, "b64": base64.b64encode(blob).decode("ascii")},
118 120000, target)
119 resp.setdefault("bytes", len(blob))
120 return resp
121
122
123def do_get(remote, local, target=None):
124 resp = request("/get", {"path": remote}, 120000, target)
125 if resp.get("b64"):
126 with open(local, "wb") as f:
127 f.write(base64.b64decode(resp["b64"]))
128 return {"bytes": resp.get("bytes"), "path": os.path.abspath(local),
129 "error": resp.get("error")}
130
131
132# Raw string: this text is mostly about backslashes, and rendering it correctly
133# matters more than keeping the source lines joined.
134EXEC_DESCRIPTION = r"""Run an AutoHotkey v2 script on the Windows 7 desktop.
135Returns whatever the script printed, plus a screenshot taken shot_delay_ms
136after the script exits.
137
138Coordinates are screen-absolute and match the returned screenshot
139pixel-for-pixel (CoordMode Screen is already set; do not change it). The only
140way to send text back is FileAppend(text, "*") -- there is no implicit output.
141Put a whole sequence of actions in one script; one call per click is slow and
142blind.
143
144BACKSLASHES. AutoHotkey's escape character is the backtick, NOT the backslash,
145so a backslash inside an AHK string is already literal. You are emitting this
146script as a JSON string, so one literal backslash is written "\\" in the JSON
147and arrives in the script as "\". Never write "\\\\" -- that is what makes an
148app receive A:\\cute.png instead of A:\cute.png. Escape inside AHK with the
149backtick instead: `n newline, `t tab, `" quote.
150
151LITERAL TEXT. Send() reads ^ + ! # { } as Ctrl/Shift/Alt/Win and key groups.
152Use SendText() for anything literal -- paths, passwords, arbitrary content --
153and keep Send() for actual key combinations.
154
155CLEAN UP. When you finish a task, close the applications you opened (WinClose,
156or the app's own quit path). Leaving windows stacked makes later screenshots
157harder to read, and a forgotten modal swallows input from the next script.
158
159Click something, let the UI settle:
160 Click(512, 384)
161 Sleep(300)
162
163Type a literal path into the focused field:
164 SendText("A:\cute.png")
165 Send("{Enter}")
166
167Shortcut, then read the result out of the clipboard:
168 Send("^a^c")
169 ClipWait(1)
170 FileAppend(A_Clipboard, "*")
171
172Launch an app, wait for its window, and close it when done:
173 Run("mspaint.exe")
174 WinWait("Paint", , 10)
175 WinActivate()
176 WinClose("Paint")
177
178Raise timeout_ms when the script itself waits on the UI; raise shot_delay_ms
179when an animation or app launch needs longer to settle before the screenshot."""
180
181TARGET_PROPERTY = {
182 "type": "string",
183 "description": "Target name. Omit to use the configured default.",
184}
185
186TOOLS = [
187 {
188 "name": "win7_exec",
189 "description": EXEC_DESCRIPTION,
190 "inputSchema": {
191 "type": "object",
192 "properties": {
193 "target": TARGET_PROPERTY,
194 "script": {"type": "string", "description": "AutoHotkey v2 source."},
195 "shot_delay_ms": {
196 "type": "integer",
197 "default": 500,
198 "description": "Wait this long after the script ends, then screenshot.",
199 },
200 "timeout_ms": {
201 "type": "integer",
202 "default": 60000,
203 "description": "Kill the script after this long.",
204 },
205 },
206 "required": ["script"],
207 },
208 },
209 {
210 "name": "win7_cmd",
211 "description": (
212 "Run a command through cmd.exe on the Windows 7 box and return its output. "
213 "No screenshot -- use it to inspect files, launch programs and check state "
214 "without spending a screenshot on it."
215 ),
216 "inputSchema": {
217 "type": "object",
218 "properties": {
219 "target": TARGET_PROPERTY,
220 "command": {"type": "string", "description": "Passed to cmd.exe /c."},
221 "timeout_ms": {"type": "integer", "default": 60000},
222 },
223 "required": ["command"],
224 },
225 },
226 {
227 "name": "win7_spawn",
228 "description": (
229 "Start a detached Windows process and return immediately. Its standard "
230 "handles are closed, so a long-lived GUI or capture process cannot wedge "
231 "the control channel."
232 ),
233 "inputSchema": {
234 "type": "object",
235 "properties": {
236 "target": TARGET_PROPERTY,
237 "command": {"type": "string", "description": "Windows command line."},
238 },
239 "required": ["command"],
240 },
241 },
242 {
243 "name": "win7_put",
244 "description": (
245 "Copy a file from this Mac to the Windows 7 box. Give two paths; the bytes "
246 "never pass through the conversation, so file size costs nothing."
247 ),
248 "inputSchema": {
249 "type": "object",
250 "properties": {
251 "target": TARGET_PROPERTY,
252 "local": {"type": "string", "description": "Path on the Mac."},
253 "remote": {
254 "type": "string",
255 "description": "Windows path, e.g. C:\\\\work\\\\a.one.",
256 },
257 },
258 "required": ["local", "remote"],
259 },
260 },
261 {
262 "name": "win7_get",
263 "description": "Copy a file from the Windows 7 box back to this Mac.",
264 "inputSchema": {
265 "type": "object",
266 "properties": {
267 "target": TARGET_PROPERTY,
268 "remote": {"type": "string", "description": "Windows path."},
269 "local": {"type": "string", "description": "Path on the Mac."},
270 },
271 "required": ["remote", "local"],
272 },
273 },
274 {
275 "name": "win7_shot",
276 "description": "Screenshot the Windows 7 desktop without running anything.",
277 "inputSchema": {"type": "object", "properties": {"target": TARGET_PROPERTY}},
278 },
279 {
280 "name": "win7_ui",
281 "description": (
282 "Dump the foreground window's control tree as text -- class name, window "
283 "text and client rect (l,t,w,h) for the window and each child control. "
284 "It reads real Win32 controls, so it is excellent for dialogs, menus and "
285 "standard controls, and near-useless for custom-drawn canvases like "
286 "OneNote's page surface -- reach for a screenshot there instead."
287 ),
288 "inputSchema": {"type": "object", "properties": {"target": TARGET_PROPERTY}},
289 },
290]
291
292TOOLS += [
293 {
294 "name": "win7_vm_up",
295 "description": (
296 "Create a named Windows 7 clone when absent, then boot it. Creation settings "
297 "are ignored for an existing clone. Set wait to return only when its "
298 "authenticated desktop agent reports the expected hostname."
299 ),
300 "inputSchema": {
301 "type": "object",
302 "properties": {
303 "name": {"type": "string",
304 "description": "Unique 1-11 character lowercase VM name."},
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 (("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