#!/usr/bin/env python3 """Small MCP facade for the ReynaBot ESP32/Tactility screen. The board exposes a lightweight JSON-RPC endpoint at /api/mcp but does not complete Hermes native HTTP MCP initialization. This stdio MCP wrapper gives the kids profile stable, child-safe screen/audio tools while proxying to the board. """ from __future__ import annotations import base64 import io import json import math import os import struct import sys import urllib.error import urllib.request import wave from pathlib import Path from typing import Any from mcp.server.fastmcp import FastMCP SCREEN_URL = os.environ.get("REYNABOT_SCREEN_MCP_URL", "http://192.168.68.130/api/mcp") HERMES_REPO = os.environ.get("HERMES_REPO", "/home/adolforeyna/.hermes/hermes-agent") if HERMES_REPO not in sys.path: sys.path.insert(0, HERMES_REPO) mcp = FastMCP("reynabot_screen") def _call_board(tool_name: str, arguments: dict[str, Any], timeout: int = 30) -> str: payload = { "jsonrpc": "2.0", "id": "hermes", "method": "tools/call", "params": {"name": tool_name, "arguments": arguments}, } data = json.dumps(payload).encode("utf-8") req = urllib.request.Request(SCREEN_URL, data=data, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8", "ignore") except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", "ignore") return f"HTTP {exc.code} from board: {body[:500]}" except Exception as exc: return f"Board call failed: {type(exc).__name__}: {exc}" try: parsed = json.loads(body) except Exception: return body[:1000] if parsed.get("error"): return f"Board error: {parsed['error']}" content = ((parsed.get("result") or {}).get("content") or []) texts = [str(item.get("text")) for item in content if isinstance(item, dict) and item.get("text")] return "\n".join(texts) if texts else json.dumps(parsed.get("result"), ensure_ascii=False)[:1000] @mcp.tool() def get_screen_capabilities() -> str: """Get the ReynaBot screen display/audio capabilities.""" return _call_board("get_capabilities", {}, timeout=8) @mcp.tool() def draw_screen_text(text: str, x: int = 8, y: int = 8, size: int = 1, clear: bool = True) -> str: """Draw short text on the ReynaBot screen. Use this for simple visual replies.""" if clear: _call_board("clear_screen", {"color": 0}, timeout=8) return _call_board("draw_text", {"text": text[:900], "x": x, "y": y, "size": 2 if size == 2 else 1}, timeout=8) @mcp.tool() def play_screen_tone(frequency: int = 440, duration_ms: int = 250, volume: int = 35) -> str: """Play a short tone on the ReynaBot speaker.""" frequency = max(80, min(4000, int(frequency))) duration_ms = max(30, min(2000, int(duration_ms))) volume = max(0, min(100, int(volume))) return _call_board("play_tone", {"frequency": frequency, "duration_ms": duration_ms, "volume": volume}, timeout=8) def _image_to_pbm_base64(ref: str) -> str: from PIL import Image, ImageOps ref = ref.strip() if ref.lower().startswith("data:image/"): _, _, encoded = ref.partition(",") raw = base64.b64decode(encoded) elif ref.lower().startswith(("http://", "https://")): with urllib.request.urlopen(ref, timeout=20) as resp: raw = resp.read() else: raw = Path(ref).expanduser().read_bytes() with Image.open(io.BytesIO(raw)) as img: img = ImageOps.exif_transpose(img).convert("RGB") canvas = Image.new("RGB", (320, 240), "white") img.thumbnail((320, 240), Image.Resampling.LANCZOS) canvas.paste(img, ((320 - img.width) // 2, (240 - img.height) // 2)) mono = canvas.convert("1", dither=Image.Dither.FLOYDSTEINBERG) out = io.BytesIO() mono.save(out, format="PPM") # mode=1 writes raw PBM (P4) return base64.b64encode(out.getvalue()).decode("ascii") @mcp.tool() def show_screen_image(image_url_or_path: str) -> str: """Display an image URL, data:image URL, or local image path on the ReynaBot screen. Use this after generating an image. The wrapper converts the image to a small 1-bit PBM so it fits the board's MCP request limits. """ try: pbm_base64 = _image_to_pbm_base64(image_url_or_path) except Exception as exc: return f"Image conversion failed: {type(exc).__name__}: {exc}" return _call_board("draw_image", {"pbm_base64": pbm_base64, "x": 0, "y": 0, "dither": False}, timeout=15) def _text_to_mp3_base64(text: str) -> str: from tools.tts_tool import text_to_speech_tool result = json.loads(text_to_speech_tool(text[:420])) if not result.get("success"): raise RuntimeError(result.get("error") or "TTS failed") path = Path(str(result.get("file_path") or "")) if not path.exists(): raise RuntimeError("TTS file missing") if path.suffix.lower() != ".mp3": # Fall through to board WAV support only for short audio to avoid huge payloads. return "" return base64.b64encode(path.read_bytes()).decode("ascii") def _short_wav_base64(text: str) -> str: # Last-resort tiny tone-like WAV if a TTS provider returns a non-MP3 file. sample_rate = 16000 duration_s = min(0.8, max(0.15, len(text) / 80.0)) buf = io.BytesIO() with wave.open(buf, "wb") as wav: wav.setnchannels(1) wav.setsampwidth(2) wav.setframerate(sample_rate) frames = bytearray() for i in range(int(sample_rate * duration_s)): sample = int(4000 * math.sin(2 * math.pi * 660 * i / sample_rate)) frames += struct.pack(" str: """Speak short text through the ReynaBot speaker using the configured Hermes TTS voice.""" volume = max(0, min(100, int(volume))) try: mp3_base64 = _text_to_mp3_base64(text) if mp3_base64: return _call_board("play_mp3_base64", {"mp3_base64": mp3_base64, "volume": volume}, timeout=75) return _call_board("play_audio_base64", {"wav_base64": _short_wav_base64(text), "volume": volume}, timeout=20) except Exception as exc: return f"Speech failed: {type(exc).__name__}: {exc}" if __name__ == "__main__": mcp.run()