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 self.override_timeout = 0 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"] } } ] }, "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 self.override_timeout = time.ticks_ms() + 30000 # 30-sec override color = int(args.get("color", 0)) self.display.clear(color) self.display.show() return "Screen cleared." elif name == "draw_text": self.override_active = True self.override_timeout = time.ticks_ms() + 30000 # 30-sec override 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 self.override_timeout = time.ticks_ms() + 30000 # 30-sec override 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}" else: raise ValueError(f"Unknown tool: {name}")