import socket import json import time class MCPServer: """A lightweight JSON-RPC HTTP server implementing Model Context Protocol (MCP) endpoints.""" def __init__(self, display, led, battery, sensor, rtc, ble): self.display = display self.led = led self.battery = battery self.sensor = sensor self.rtc = rtc self.ble = ble self.sock = None self.active_led_mode = "off" # static, breath, rainbow, off self.override_active = False def start(self, port=80): """Starts the TCP server non-blockingly.""" self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind(('', port)) self.sock.listen(3) # Set socket to non-blocking so the main loop can run animations concurrently self.sock.setblocking(False) print(f"MCP server listening on port {port}...") def update(self): """Check for incoming HTTP requests and handle them non-blockingly.""" if self.sock is None: return try: client, addr = self.sock.accept() except OSError: # No incoming connection return try: # Handle incoming connection client.settimeout(2.0) req = client.recv(2048).decode('utf-8') # Simple HTTP parser lines = req.split('\r\n') if len(lines) == 0: client.close() return first_line = lines[0] parts = first_line.split(' ') if len(parts) < 3: client.close() return method, path = parts[0], parts[1] if method == 'POST' and path == '/api/mcp': # Read content length content_length = 0 for line in lines: if line.lower().startswith('content-length:'): content_length = int(line.split(':')[1].strip()) break # Locate start of JSON body body = "" if '\r\n\r\n' in req: body = req.split('\r\n\r\n', 1)[1] # Read remaining body if not fully received while len(body) < content_length: body += client.recv(1024).decode('utf-8') # Parse JSON-RPC 2.0 Request rpc_req = json.loads(body) rpc_resp = self._handle_rpc(rpc_req) # Send HTTP Response resp_body = json.dumps(rpc_resp) resp = "HTTP/1.1 200 OK\r\n" resp += "Content-Type: application/json\r\n" resp += f"Content-Length: {len(resp_body)}\r\n" resp += "Connection: close\r\n\r\n" resp += resp_body client.send(resp.encode('utf-8')) else: # Return 404 resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" client.send(resp.encode('utf-8')) except Exception as e: print("Error handling client:", e) finally: client.close() def _handle_rpc(self, req): rpc_id = req.get('id') method = req.get('method') params = req.get('params', {}) if method == 'tools/list': return { "jsonrpc": "2.0", "result": { "tools": [ { "name": "clear_screen", "description": "Clear the 400x300 screen to white (0) or black (1).", "inputSchema": { "type": "object", "properties": { "color": {"type": "integer", "enum": [0, 1], "description": "0 = White, 1 = Black"} }, "required": ["color"] } }, { "name": "draw_text", "description": "Draw text on the screen at specified (x,y) coordinates.", "inputSchema": { "type": "object", "properties": { "text": {"type": "string", "description": "The message to display"}, "x": {"type": "integer", "description": "X coordinate (0-390)"}, "y": {"type": "integer", "description": "Y coordinate (0-290)"}, "size": {"type": "integer", "enum": [1, 2], "description": "Text scale (1=normal, 2=large)"} }, "required": ["text", "x", "y"] } }, { "name": "set_led", "description": "Control the onboard WS2812 NeoPixel RGB LED.", "inputSchema": { "type": "object", "properties": { "r": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Red channel (0-255)"}, "g": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Green channel (0-255)"}, "b": {"type": "integer", "minimum": 0, "maximum": 255, "description": "Blue channel (0-255)"}, "mode": {"type": "string", "enum": ["static", "breath", "rainbow", "off"], "description": "LED mode"} }, "required": ["r", "g", "b", "mode"] } }, { "name": "get_battery", "description": "Read the current battery voltage and estimated capacity percentage.", "inputSchema": {"type": "object", "properties": {}} }, { "name": "get_sensors", "description": "Read onboard SHTC3 temperature and relative humidity.", "inputSchema": {"type": "object", "properties": {}} }, { "name": "scan_ble", "description": "Scan for nearby Bluetooth Low Energy devices / tracking tags.", "inputSchema": { "type": "object", "properties": { "duration_ms": {"type": "integer", "description": "Scan duration in milliseconds (default 3000)"} } } }, { "name": "get_screenshot", "description": "Capture the current reflective LCD screen rendering as a PNG image.", "inputSchema": {"type": "object", "properties": {}} }, { "name": "draw_image", "description": "Draw an image (PNG, JPEG, GIF, BMP, etc.) on the screen. The image will be converted to 1-bit monochrome and fit to display boundaries.", "inputSchema": { "type": "object", "properties": { "image_base64": {"type": "string", "description": "Base64 encoded string of the source image file"}, "x": {"type": "integer", "description": "X coordinate to place the image (default 0)", "default": 0}, "y": {"type": "integer", "description": "Y coordinate to place the image (default 0)", "default": 0}, "dither": {"type": "boolean", "description": "Whether to use Floyd-Steinberg dithering (default true)", "default": True} }, "required": ["image_base64"] } }, { "name": "write_file", "description": "Write a file (e.g., python script or config) to the board's flash storage.", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "The destination file path (e.g. 'my_script.py')"}, "content": {"type": "string", "description": "The text content of the file"} }, "required": ["path", "content"] } }, { "name": "read_file", "description": "Read a text file from the board's flash storage.", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "The file path to read (e.g. 'boot.py')"} }, "required": ["path"] } }, { "name": "execute_python", "description": "Execute arbitrary Python code dynamically on the board. Standard output (prints) and errors will be captured and returned.", "inputSchema": { "type": "object", "properties": { "code": {"type": "string", "description": "The Python code to execute"} }, "required": ["code"] } }, { "name": "play_tone", "description": "Play a pure sine wave tone/beep on the board speaker.", "inputSchema": { "type": "object", "properties": { "frequency": {"type": "integer", "description": "Frequency of the tone in Hz (e.g. 440 for A4, default 440)", "default": 440}, "duration_ms": {"type": "integer", "description": "Duration of the tone in milliseconds (default 1000)", "default": 1000}, "volume": {"type": "integer", "description": "Volume of the tone from 0 to 100 (default 50)", "default": 50} } } }, { "name": "play_audio", "description": "Play a standard WAV audio file on the speaker.", "inputSchema": { "type": "object", "properties": { "filename": {"type": "string", "description": "WAV file path to play (e.g. 'response.wav')"}, "volume": {"type": "integer", "description": "Volume from 0 to 100 (default 50)", "default": 50} }, "required": ["filename"] } }, { "name": "record_voice", "description": "Record voice command from the microphone to a raw stereo PCM file.", "inputSchema": { "type": "object", "properties": { "duration_sec": {"type": "integer", "description": "Duration in seconds (default 4)", "default": 4}, "filename": {"type": "string", "description": "Destination file path (e.g. 'recording.pcm')", "default": "recording.pcm"} } } }, { "name": "play_audio_base64", "description": "Decode a base64-encoded WAV file and play it directly on the speaker.", "inputSchema": { "type": "object", "properties": { "wav_base64": {"type": "string", "description": "Base64 encoded string of the WAV audio file"}, "volume": {"type": "integer", "description": "Volume from 0 to 100 (default 50)", "default": 50} }, "required": ["wav_base64"] } }, { "name": "download_file", "description": "Download a file from a URL over Wi-Fi directly to the board storage or microSD card.", "inputSchema": { "type": "object", "properties": { "url": {"type": "string", "description": "The URL of the file to download"}, "filename": {"type": "string", "description": "The destination filename (e.g. 'podcast1.wav')"}, "use_sd": {"type": "boolean", "description": "Save to microSD card if true, or local flash if false (default true)", "default": True} }, "required": ["url", "filename"] } } ] }, "id": rpc_id } elif method == 'tools/call': tool_name = params.get('name') args = params.get('arguments', {}) try: content = self._call_tool(tool_name, args) return { "jsonrpc": "2.0", "result": { "content": [ { "type": "text", "text": content } ] }, "id": rpc_id } except Exception as e: return { "jsonrpc": "2.0", "error": { "code": -32000, "message": str(e) }, "id": rpc_id } return { "jsonrpc": "2.0", "error": { "code": -32601, "message": f"Method {method} not found" }, "id": rpc_id } def _call_tool(self, name, args): """Executes hardware actions depending on the called tool name.""" if name == "clear_screen": self.override_active = True color = int(args.get("color", 0)) self.display.clear(color) self.display.show() return "Screen cleared." elif name == "draw_text": self.override_active = True text = str(args.get("text", "")) x = int(args.get("x", 10)) y = int(args.get("y", 10)) size = int(args.get("size", 1)) if size == 2: self.display.text_large(text, x, y, scale=2, c=1) else: self.display.text(text, x, y, 1) self.display.show() return f"Successfully drew text '{text}' at ({x}, {y}) with size {size}." elif name == "set_led": r = int(args.get("r", 0)) g = int(args.get("g", 0)) b = int(args.get("b", 0)) mode = str(args.get("mode", "static")) self.active_led_mode = mode if mode == "static": self.led.set_color(r, g, b) elif mode == "off": self.led.off() else: # Store base color for animations self.led.base_color = (r, g, b) return f"LED set to mode '{mode}' with base color ({r}, {g}, {b})." elif name == "get_battery": v = self.battery.read_voltage() p = self.battery.read_percentage() return json.dumps({"voltage_v": v, "percentage_pct": p}) elif name == "get_sensors": t, h = self.sensor.read_sensor() return json.dumps({"temperature_c": t, "humidity_pct": h}) elif name == "scan_ble": dur = int(args.get("duration_ms", 3000)) # Scan returns dictionary: {mac: {rssi: rssi, name: name}} devices = self.ble.scan(dur) return json.dumps(devices) elif name == "get_screenshot": import binascii filename = 'screenshot_mcp.pbm' try: self.display.save_screenshot(filename) with open(filename, 'rb') as f: data = f.read() # Encode to base64 b64 = binascii.b2a_base64(data).decode('utf-8').strip() return f"__PBM_BASE64__:{b64}" except Exception as e: raise RuntimeError(f"Screenshot capture failed: {e}") elif name == "draw_image": self.override_active = True pbm_b64 = args.get("pbm_base64") x = int(args.get("x", 0)) y = int(args.get("y", 0)) if not pbm_b64: raise ValueError("Missing pbm_base64 parameter (preprocessed by bridge)") import binascii pbm_bytes = binascii.a2b_base64(pbm_b64) filename = 'temp_recv.pbm' with open(filename, 'wb') as f: f.write(pbm_bytes) try: self.display.draw_pbm(filename, x, y, scale=1) self.display.show() finally: import os try: os.remove(filename) except: pass return "Image displayed successfully." elif name == "write_file": path = str(args.get("path")) content = str(args.get("content")) with open(path, 'w') as f: f.write(content) return f"Successfully wrote {len(content)} characters to '{path}'." elif name == "read_file": path = str(args.get("path")) with open(path, 'r') as f: content = f.read() return content elif name == "execute_python": code = str(args.get("code")) import builtins import sys import io output_buffer = [] old_print = builtins.print def custom_print(*args, **kwargs): sep = kwargs.get('sep', ' ') end = kwargs.get('end', '\n') str_args = [str(arg) for arg in args] msg = sep.join(str_args) + end output_buffer.append(msg) builtins.print = custom_print error = None try: exec(code, globals()) except Exception as e: tb_file = io.StringIO() sys.print_exception(e, tb_file) error_msg = tb_file.getvalue() output_buffer.append(f"\nExecution Failed:\n{error_msg}") error = e finally: builtins.print = old_print output = "".join(output_buffer) if error: return output return f"Execution Succeeded. Console output:\n{output}" elif name == "play_tone": freq = int(args.get("frequency", 440)) duration = int(args.get("duration_ms", 1000)) vol = int(args.get("volume", 50)) # Limit duration to prevent blocking the main loop for too long duration = max(50, min(5000, duration)) freq = max(50, min(10000, freq)) vol = max(0, min(100, vol)) from audio_util import play_tone play_tone(frequency=freq, duration_ms=duration, volume=vol) return f"Played tone of {freq}Hz for {duration}ms at volume {vol}." elif name == "play_audio": filename = str(args.get("filename")) vol = int(args.get("volume", 50)) vol = max(0, min(100, vol)) from audio_util import play_wav success = play_wav(filename, volume=vol) if success: return f"Successfully played audio file '{filename}'." else: raise RuntimeError(f"Failed to play audio file '{filename}'. Check format (16kHz 16-bit PCM WAV recommended).") elif name == "record_voice": duration = int(args.get("duration_sec", 4)) filename = str(args.get("filename", "recording.pcm")) # Clamp duration to a reasonable range duration = max(1, min(15, duration)) from audio_util import record_audio success = record_audio(duration_seconds=duration, filename=filename) if success: return f"Successfully recorded {duration} seconds of audio to '{filename}'." else: raise RuntimeError("Audio recording failed.") elif name == "play_audio_base64": b64_data = args.get("wav_base64") vol = int(args.get("volume", 50)) vol = max(0, min(100, vol)) if not b64_data: raise ValueError("Missing wav_base64 parameter.") import binascii try: audio_bytes = binascii.a2b_base64(b64_data) except Exception as e: raise ValueError(f"Failed to decode base64 audio: {e}") filename = "temp_play.wav" with open(filename, "wb") as f: f.write(audio_bytes) try: from audio_util import play_wav success = play_wav(filename, volume=vol) finally: import os try: os.remove(filename) except: pass if success: return "Successfully played base64 audio stream." else: raise RuntimeError("Failed to play decoded audio stream. Check format (16kHz 16-bit PCM WAV recommended).") elif name == "download_file": url = str(args.get("url")) filename = str(args.get("filename")) use_sd = bool(args.get("use_sd", True)) from download_util import download_file saved_path = download_file(url, filename, use_sd=use_sd) if saved_path: return f"Successfully downloaded file to '{saved_path}'." else: raise RuntimeError(f"Failed to download file from '{url}'.") else: raise ValueError(f"Unknown tool: {name}")