sdk.py

import subprocess
import json
import threading
from typing import Any, Dict, Optional


class NodeSDK:
    def __init__(self, sdk_path):
        self._proc = subprocess.Popen(
            ["node", "--no-warnings", sdk_path],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )

        self._id = 0
        self._lock = threading.Lock()

    def call(self, method: str, params: Optional[Dict[str, Any]] = None) -> Any:
        with self._lock:
            self._id += 1
            req_id = self._id

            request = {
                "jsonrpc": "2.0",
                "id": req_id,
                "method": method,
                "params": params or {},
            }

            try:
                if not self._proc.stdin or not self._proc.stdout:
                    raise RuntimeError("Failed to open Node process pipes")

                self._proc.stdin.write(json.dumps(request) + "\n")
                self._proc.stdin.flush()

                line = self._proc.stdout.readline()
                if not line:
                    raise RuntimeError("Node process terminated unexpectedly")

                response = json.loads(line)

            except Exception as e:
                raise RuntimeError(f"RPC communication failed: {e}") from e

            if "error" in response:
                raise RuntimeError(response["error"])

            if "result" not in response:
                raise RuntimeError(f"Invalid RPC response: {response}")

            return response["result"]

    def __getattr__(self, name: str):
        def method(**kwargs):
            return self.call(name, kwargs)

        return method

    def close(self):
        if self._proc.poll() is None:
            self._proc.terminate()
            self._proc.wait()