ahmed
cc2018aae7
- SwiftUI popover: apps grid w/ live search, sources, glass remote, gradient volume slider, animations - Persistent python protocol bridge (vidaa-control, MQTT/TLS :36669) - Wake-on-LAN power-on, one-time PIN pairing with token auto-refresh - MCP server for AI agents (11 tv_* tools) + live test script - Gitea Actions CI workflow
244 lines
10 KiB
Python
244 lines
10 KiB
Python
#!/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()
|