Add Wi-Fi auto-boot connection, MCP JSON-RPC Server, and host bridge
This commit is contained in:
Adolfo Reyna
+267
@@ -0,0 +1,267 @@
|
||||
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)"}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"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)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
Reference in New Issue
Block a user