"""Dependency-free stdio MCP tool server for local or run-bound Hermes use.""" from __future__ import annotations import json import sys from typing import Any, Callable, Mapping from uuid import uuid4 from .hermes_tool_client import request_node_action _TOOL_POLICY = { "omni_screen_capture": ("desktop.screenshot", "desktop.observe"), "omni_computer_control": ("desktop.control", "desktop.control"), "omni_file_read": ("files.read", "files.read"), "omni_file_write": ("files.write", "files.write"), "omni_webcam_start": ("webcam.start", "webcam.read"), "omni_webcam_frame": ("webcam.frame", "webcam.read"), "omni_webcam_stop": ("webcam.stop", "webcam.read"), } class OmniMcpServer: def __init__( self, *, requester: Callable[..., tuple[int, dict[str, Any]]] = request_node_action, ) -> None: self._requester = requester def handle(self, message: Mapping[str, Any]) -> dict[str, Any] | None: request_id = message.get("id") method = str(message.get("method") or "") if request_id is None: return None try: if method == "initialize": params = message.get("params") if isinstance(message.get("params"), Mapping) else {} version = str(params.get("protocolVersion") or "2025-11-25") result: dict[str, Any] = { "protocolVersion": version, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "omni-windows-node", "version": "1.0"}, } elif method == "ping": result = {} elif method == "tools/list": result = {"tools": _tools()} elif method == "tools/call": result = self._call_tool(message.get("params")) else: return _error(request_id, -32601, "Method not found") return {"jsonrpc": "2.0", "id": request_id, "result": result} except ValueError as exc: return _error(request_id, -32602, str(exc)) except Exception as exc: return _error(request_id, -32603, f"Tool failed: {type(exc).__name__}") def _call_tool(self, params_value: object) -> dict[str, Any]: if not isinstance(params_value, Mapping): raise ValueError("tools/call params must be an object") name = str(params_value.get("name") or "") policy = _TOOL_POLICY.get(name) if policy is None: raise ValueError("unknown OMNI tool") arguments = params_value.get("arguments", {}) if not isinstance(arguments, Mapping): raise ValueError("tool arguments must be an object") action_type, permission = policy run_ticket = str(arguments.get("run_ticket") or "") if not run_ticket: raise ValueError("run_ticket from OMNI_CONTEXT_V1 is required") reserved = {"run_ticket", "operation_id", "trace_id", "session_id"} action_arguments = {key: value for key, value in arguments.items() if key not in reserved} payload = { "run_ticket": run_ticket, "operation_id": str(arguments.get("operation_id") or f"op_mcp_{uuid4().hex}"), "trace_id": str(arguments.get("trace_id") or f"trace_mcp_{uuid4().hex}"), "action": { "action_id": f"action_mcp_{uuid4().hex}", "type": action_type, "permission": permission, "arguments": action_arguments, }, } status, value = self._requester(payload) is_error = not 200 <= status < 300 text = json.dumps(value, ensure_ascii=False, separators=(",", ":")) return { "content": [{"type": "text", "text": text}], "structuredContent": value, "isError": is_error, } def _tools() -> list[dict[str, Any]]: common = { "run_ticket": { "type": "string", "description": "Required run ticket injected by the authenticated OMNI gateway", }, "operation_id": {"type": "string"}, "trace_id": {"type": "string"}, } def schema(properties: Mapping[str, Any], required: list[str]) -> dict[str, Any]: return { "type": "object", "properties": {**common, **properties}, "required": list(dict.fromkeys(["run_ticket", *required])), "additionalProperties": False, } return [ { "name": "omni_screen_capture", "description": "Capture a granted screen as a local artifact reference.", "inputSchema": schema( { "screen_id": {"type": "string", "default": "primary"}, "all_screens": {"type": "boolean", "default": False}, }, [], ), }, { "name": "omni_computer_control", "description": "Perform one visible, grant-gated mouse/keyboard/clipboard action.", "inputSchema": schema( { "action": { "type": "string", "enum": [ "move", "click", "double_click", "right_click", "drag", "scroll", "type", "hotkey", "clipboard.read", "clipboard.write", ], }, "x": {"type": "integer"}, "y": {"type": "integer"}, "to_x": {"type": "integer"}, "to_y": {"type": "integer"}, "button": {"type": "string"}, "delta": {"type": "integer"}, "text": {"type": "string"}, "keys": {"oneOf": [{"type": "string"}, {"type": "array", "items": {"type": "string"}}]}, }, ["action"], ), }, { "name": "omni_file_read", "description": "Read a bounded chunk from one exact granted path.", "inputSchema": schema( {"path": {"type": "string"}, "offset": {"type": "integer", "minimum": 0}, "max_bytes": {"type": "integer", "minimum": 1}}, ["path"], ), }, { "name": "omni_file_write", "description": "Write content to one exact granted path.", "inputSchema": schema( { "path": {"type": "string"}, "content": {"type": "string"}, "content_base64": {"type": "string"}, "mode": {"type": "string", "enum": ["overwrite", "append"]}, "create_parents": {"type": "boolean"}, }, ["path"], ), }, *[ { "name": name, "description": description, "inputSchema": schema( {"device_id": {"type": "string"}, "device_name": {"type": "string"}}, ["device_id"], ), } for name, description in ( ("omni_webcam_start", "Start one explicitly granted webcam with a visible indicator."), ("omni_webcam_frame", "Capture a frame from the active granted webcam."), ("omni_webcam_stop", "Stop the granted webcam and clear its indicator."), ) ], ] def _error(request_id: object, code: int, message: str) -> dict[str, Any]: return { "jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message[:240]}, } def main() -> int: server = OmniMcpServer() for line in sys.stdin: try: message = json.loads(line) if not isinstance(message, dict): continue response = server.handle(message) if response is not None: sys.stdout.write(json.dumps(response, ensure_ascii=False, separators=(",", ":")) + "\n") sys.stdout.flush() except (UnicodeError, json.JSONDecodeError): continue return 0 if __name__ == "__main__": raise SystemExit(main())