unity_bridge.py

"""Direct TCP client for the MCP-for-Unity editor bridge on port 6400.

Protocol: server sends 'WELCOME UNITY-MCP 1 FRAMING=1\\n' on connect, then
all messages are 8-byte big-endian length-prefixed UTF-8 JSON frames.
Command: {"type": "<tool_name>", "params": {...}}
Response: {"status": "success"|"error", "result": {...}} (framed)

Usage:
  python unity_bridge.py call <tool_name> <json_params>
  python unity_bridge.py callfile <tool_name> <params_json_file>
  python unity_bridge.py batch <batch_json_file>   # [{"type":..., "params":...}, ...]
"""
import json
import socket
import struct
import sys

HOST, PORT = "127.0.0.1", 6400
TIMEOUT = 90


class Bridge:
    def __init__(self):
        self.sock = socket.create_connection((HOST, PORT), timeout=TIMEOUT)
        self.sock.settimeout(TIMEOUT)
        # read handshake line
        line = b""
        while not line.endswith(b"\n"):
            ch = self.sock.recv(1)
            if not ch:
                raise ConnectionError("closed during handshake")
            line += ch
        self.handshake = line.decode("ascii", "replace").strip()

    def _read_exact(self, n):
        buf = b""
        while len(buf) < n:
            chunk = self.sock.recv(n - len(buf))
            if not chunk:
                raise ConnectionError("connection closed mid-frame")
            buf += chunk
        return buf

    def call(self, tool, params):
        payload = json.dumps({"type": tool, "params": params or {}}).encode("utf-8")
        self.sock.sendall(struct.pack(">Q", len(payload)) + payload)
        (length,) = struct.unpack(">Q", self._read_exact(8))
        resp = self._read_exact(length).decode("utf-8", "replace")
        try:
            return json.loads(resp)
        except json.JSONDecodeError:
            return {"raw": resp}


def render(resp, max_len=6000):
    out = json.dumps(resp, indent=2, ensure_ascii=False)
    if len(out) > max_len:
        out = out[:max_len] + f"\n... <truncated, {len(out)} chars total>"
    print(out)


def main():
    cmd = sys.argv[1]
    b = Bridge()
    if cmd == "call":
        params = json.loads(sys.argv[3]) if len(sys.argv) > 3 else {}
        render(b.call(sys.argv[2], params))
    elif cmd == "callfile":
        with open(sys.argv[3], encoding="utf-8") as f:
            params = json.load(f)
        render(b.call(sys.argv[2], params))
    elif cmd == "batch":
        with open(sys.argv[2], encoding="utf-8") as f:
            calls = json.load(f)
        for i, c in enumerate(calls):
            print(f"=== [{i}] {c['type']} ===")
            render(b.call(c["type"], c.get("params", {})), max_len=c.get("max_len", 3000))
    elif cmd == "ping":
        print(b.handshake)
        render(b.call("ping", {}))
    else:
        print(f"unknown command {cmd}")


if __name__ == "__main__":
    main()