Files
integration/Plugins/UEWingman/ue-wingman.py
2026-03-31 16:43:28 -04:00

88 lines
2.3 KiB
Python
Executable File

#!/usr/bin/env python3
"""
Human-friendly MCP test client.
Usage: ue-wingman.py <command> [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("=")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
else:
try:
value = int(value)
except ValueError:
pass
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()