Add get_screenshot tool to MCP server and automatic PBM-to-PNG bridge conversion
This commit is contained in:
@@ -34,6 +34,55 @@ def main():
|
||||
|
||||
with urllib.request.urlopen(req, timeout=10.0) as response:
|
||||
resp_data = response.read().decode("utf-8")
|
||||
|
||||
try:
|
||||
resp_json = json.loads(resp_data)
|
||||
result = resp_json.get("result", {})
|
||||
content_list = result.get("content", [])
|
||||
if len(content_list) > 0 and content_list[0].get("type") == "text":
|
||||
text_val = content_list[0].get("text", "")
|
||||
if text_val.startswith("__PBM_BASE64__:"):
|
||||
b64_pbm = text_val.split(":", 1)[1]
|
||||
|
||||
import base64
|
||||
import subprocess
|
||||
import os
|
||||
|
||||
pbm_bytes = base64.b64decode(b64_pbm)
|
||||
with open("temp_mcp.pbm", "wb") as pf:
|
||||
pf.write(pbm_bytes)
|
||||
|
||||
# Convert PBM to PNG on macOS using standard sips tool
|
||||
subprocess.run(["sips", "-s", "format", "png", "temp_mcp.pbm", "--out", "temp_mcp.png"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
|
||||
if os.path.exists("temp_mcp.png"):
|
||||
with open("temp_mcp.png", "rb") as pngf:
|
||||
png_bytes = pngf.read()
|
||||
b64_png = base64.b64encode(png_bytes).decode("utf-8")
|
||||
|
||||
# Replace standard text content with MCP image schema
|
||||
content_list[0] = {
|
||||
"type": "image",
|
||||
"data": b64_png,
|
||||
"mimeType": "image/png"
|
||||
}
|
||||
# Cleanup temp files
|
||||
try:
|
||||
os.remove("temp_mcp.pbm")
|
||||
os.remove("temp_mcp.png")
|
||||
except:
|
||||
pass
|
||||
else:
|
||||
content_list[0] = {
|
||||
"type": "text",
|
||||
"text": "Screenshot captured, but PNG conversion failed on host PC."
|
||||
}
|
||||
resp_data = json.dumps(resp_json)
|
||||
except Exception as ex:
|
||||
sys.stderr.write(f"Bridge image conversion error: {ex}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
# Write the response back to stdio for the LLM client
|
||||
sys.stdout.write(resp_data + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
@@ -164,6 +164,11 @@ class MCPServer:
|
||||
"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": {}}
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -263,5 +268,18 @@ class MCPServer:
|
||||
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}")
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
Reference in New Issue
Block a user