#!/usr/bin/env python3 """ Human-friendly MCP test client. Usage: ue-wingman.py [key=value ...] ue-wingman.py (reads JSON with "command" from stdin) """ import sys import json import socket HOST = "localhost" PORT = 9851 TIMEOUT = 120 def main(): args = sys.argv[1:] if not args: # No arguments: read a complete JSON object from stdin. # The JSON must already contain a "command" key. decoder = json.JSONDecoder() raw = "" msg = None for line in sys.stdin: raw += line stripped = raw.strip() try: msg, _ = decoder.raw_decode(stripped) break except json.JSONDecodeError as e: if e.pos < len(stripped): print(f"Malformed JSON: {e.msg}") sys.exit(1) continue if msg is None: print("Could not parse a complete JSON object from stdin") sys.exit(1) if "command" not in msg: print("JSON object must contain a \"command\" key") sys.exit(1) else: msg = {"command": args[0]} for arg in args[1:]: key, _, value = arg.partition("=") msg[key] = value sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(TIMEOUT) try: sock.connect((HOST, PORT)) except (ConnectionRefusedError, socket.timeout, OSError) as e: print(f"Cannot connect to {HOST}:{PORT} — is the editor running?") sys.exit(1) sock.sendall((json.dumps(msg) + "\n").encode()) result = b"" while True: chunk = sock.recv(65536) if not chunk: break result += chunk if b"\0" in result: break sock.close() result = result[:result.index(b"\0")].decode() if b"\0" in result else result.decode() try: parsed = json.loads(result) print(json.dumps(parsed, indent=2)) except json.JSONDecodeError: print(result) if __name__ == "__main__": main()