#!/usr/bin/env python3 """Hisense TV MCP server (stdio, JSON-RPC 2.0) for Hermes and any MCP client. Zero external dependencies. Spawns the proven bridge.py subprocess and translates MCP tool calls to bridge line-JSON requests. Client config (Hermes config.yaml / Claude / Cursor): mcp_servers: hisense_tv: command: /Users/ahmedmohammed/Documents/rae/.venv/bin/python3 args: ["/Users/ahmedmohammed/Documents/rae/HisenseBar/mcp_tv_server.py"] """ import json import os import subprocess import sys import threading import time from pathlib import Path HERE = Path(__file__).resolve().parent BRIDGE = HERE / "bridge.py" VENV_PY = "/Users/ahmedmohammed/Documents/rae/.venv/bin/python3" STORE = str(Path.home() / "Library/Application Support/HisenseBar/tokens.json") HOST = os.environ.get("HISENSE_TV_IP", "192.168.1.146") MAC = os.environ.get("HISENSE_TV_MAC", "e0:3e:cb:f4:68:be") PROTOCOL_VERSION = "2024-11-05" TOOLS = [ {"name": "tv_get_volume", "description": "Get the current TV volume (0-100) and mute state.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_set_volume", "description": "Set the TV volume (0-100).", "inputSchema": {"type": "object", "properties": {"volume": {"type": "integer", "minimum": 0, "maximum": 100}}, "required": ["volume"], "additionalProperties": False}}, {"name": "tv_mute_toggle", "description": "Toggle TV mute.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_send_key", "description": "Send a remote key to the TV (e.g. KEY_HOME, KEY_OK, KEY_UP, KEY_POWER).", "inputSchema": {"type": "object", "properties": {"key": {"type": "string"}}, "required": ["key"], "additionalProperties": False}}, {"name": "tv_list_apps", "description": "List installed TV apps (name, appId, url).", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_launch_app", "description": "Launch an app by its appId+name+url dict (from tv_list_apps) or plain name.", "inputSchema": {"type": "object", "properties": { "app_id": {"type": "string"}, "name": {"type": "string"}, "url": {"type": "string"}}, "required": ["name"], "additionalProperties": False}}, {"name": "tv_list_sources", "description": "List TV input sources (TV, HDMI1...).", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_set_source", "description": "Switch TV input source, e.g. 'HDMI1' (sourceid from tv_list_sources).", "inputSchema": {"type": "object", "properties": {"source": {"type": "string"}}, "required": ["source"], "additionalProperties": False}}, {"name": "tv_power_on", "description": "Wake the TV with Wake-on-LAN.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_power_off", "description": "Power the TV off.", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, {"name": "tv_get_state", "description": "Get current TV state (on/off, active app/source).", "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False}}, ] class Bridge: """Manages the bridge.py subprocess and id-mapped requests.""" def __init__(self): self.proc = None self.lock = threading.Lock() self.next_id = 1 self.waiting = {} # id -> [event, result] self.reader = None self.connected = False def _spawn(self): self.connected = False self.proc = subprocess.Popen( [VENV_PY, str(BRIDGE), HOST, MAC, STORE], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1, ) self.reader = threading.Thread(target=self._read_loop, daemon=True) self.reader.start() def _read_loop(self): for line in self.proc.stdout: line = line.strip() if not line: continue try: msg = json.loads(line) except json.JSONDecodeError: continue rid = msg.get("id") if rid in self.waiting: event, slot = self.waiting.pop(rid) slot.append(msg) event.set() def call(self, cmd: dict, timeout: float = 30.0) -> dict: with self.lock: if self.proc is None or self.proc.poll() is not None: self._spawn() rid = self.next_id self.next_id += 1 msg = dict(cmd) msg["id"] = rid self.waiting[rid] = [threading.Event(), []] try: self.proc.stdin.write(json.dumps(msg) + "\n") self.proc.stdin.flush() except (BrokenPipeError, ValueError): self.proc = None raise RuntimeError("bridge process died") event, slot = self.waiting[rid] if not event.wait(timeout): self.waiting.pop(rid, None) raise TimeoutError(f"bridge timeout after {timeout}s for {cmd.get('cmd')}") return slot[0] def tv_call(self, cmd: dict, timeout: float = 30.0) -> dict: """Ensure the bridge session is connected, then run the command.""" if not self.connected: r = self.call({"cmd": "connect"}, timeout=45) self.connected = bool(r.get("ok")) return self.call(cmd, timeout=timeout) def close(self): if self.proc and self.proc.poll() is None: try: self.proc.stdin.close() except Exception: pass bridge = Bridge() def text_result(text: str, is_error: bool = False) -> dict: return {"content": [{"type": "text", "text": text}], "isError": is_error} def as_json(obj) -> str: return json.dumps(obj, ensure_ascii=False, indent=2, default=str) def tool_call(name: str, args: dict) -> dict: try: if name == "tv_get_volume": r = bridge.tv_call({"cmd": "volume"}) if r.get("volume") is None: return text_result("TV unreachable or not authenticated (is it on?)", is_error=True) return text_result(f"Volume: {r['volume']}") if name == "tv_set_volume": bridge.tv_call({"cmd": "set_volume", "value": int(args["volume"])}) return text_result(f"Volume set to {args['volume']}") if name == "tv_mute_toggle": bridge.tv_call({"cmd": "mute"}) return text_result("Mute toggled") if name == "tv_send_key": bridge.tv_call({"cmd": "key", "key": args["key"]}) return text_result(f"Key {args['key']} sent") if name == "tv_list_apps": r = bridge.tv_call({"cmd": "apps"}, timeout=45) apps = [{"name": a.get("name"), "appId": a.get("appId"), "url": a.get("url")} for a in (r.get("apps") or [])] if not apps: return text_result("No apps returned — TV screen likely asleep. " "Wake it first: tv_send_key KEY_HOME (or tv_power_on), then retry.", is_error=True) return text_result(as_json(apps)) if name == "tv_launch_app": app: dict = {"name": args["name"]} if args.get("app_id"): app["appId"] = args["app_id"] if args.get("url"): app["url"] = args["url"] bridge.tv_call({"cmd": "launch", "app": app}, timeout=45) return text_result(f"Launched {args['name']}") if name == "tv_list_sources": r = bridge.call({"cmd": "sources"}) return text_result(as_json(r.get("sources") or [])) if name == "tv_set_source": sid = args["source"] # rawName lookup for the required sourcename field r = bridge.tv_call({"cmd": "sources"}) name = next((s.get("sourcename") for s in (r.get("sources") or []) if s.get("sourceid") == sid), sid) bridge.tv_call({"cmd": "set_source", "source": sid, "name": name}) return text_result(f"Switched to {sid}") if name == "tv_power_on": bridge.tv_call({"cmd": "wake"}) return text_result("Wake-on-LAN packet sent (TV boots in ~5-10s)") if name == "tv_power_off": bridge.tv_call({"cmd": "power_off"}) return text_result("TV powered off") if name == "tv_get_state": r = bridge.tv_call({"cmd": "state"}) return text_result(as_json(r.get("state") or {})) return text_result(f"Unknown tool {name}", is_error=True) except Exception as e: return text_result(f"{type(e).__name__}: {e}", is_error=True) def reply(msg_id: int, result: dict): sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result}) + "\n") sys.stdout.flush() def main(): for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except json.JSONDecodeError: continue method = msg.get("method", "") msg_id = msg.get("id") if method == "initialize": reply(msg_id, { "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": {"name": "hisense-tv", "version": "1.0.0"}, }) elif method == "notifications/initialized": pass elif method == "tools/list": reply(msg_id, {"tools": TOOLS}) elif method == "tools/call": params = msg.get("params", {}) result = tool_call(params.get("name", ""), params.get("arguments") or {}) reply(msg_id, result) elif method == "ping": reply(msg_id, {}) elif msg_id is not None: reply(msg_id, {"error": {"code": -32601, "message": f"Method not found: {method}"}}) if __name__ == "__main__": try: main() except KeyboardInterrupt: pass finally: bridge.close()