"""JARVIS Tool System — extensible tool registry.""" import json import logging import platform from typing import Callable _log = logging.getLogger("jarvis.tools") TOOL_REGISTRY: dict[str, dict] = {} # Tools that require macOS (osascript/subprocess) and should be delegated # to a connected device when the server runs on Linux (HF Space / Render). _MACOS_ONLY_TOOLS: set[str] = set() IS_MACOS = platform.system() == "Darwin" def tool(name: str, description: str, parameters: dict = None, macos_only: bool = False): """Decorator to register a tool. Args: macos_only: If True, this tool requires macOS (osascript, AppleScript, etc.) and will be delegated to a connected device when running on Linux. """ def decorator(func: Callable): TOOL_REGISTRY[name] = { "name": name, "description": description, "parameters": parameters or {}, "function": func, } if macos_only: _MACOS_ONLY_TOOLS.add(name) return func return decorator async def _delegate_to_device(name: str, args: dict) -> str: """Delegate a macOS-only tool call to a connected device via WebSocket push.""" try: from user_device_registry import list_devices, send_command_to_device except ImportError: return f"Error: {name} requires macOS but this server runs on Linux. No device delegation available." # Find an online device to delegate to devices = await list_devices("default") online_devices = [d for d in devices if d.get("status") == "online"] if not online_devices: return ( f"Error: '{name}' requires macOS but this server runs in the cloud. " f"No connected Mac device found to execute this command. " f"Make sure the JARVIS listener is running on your Mac." ) target = online_devices[0] device_id = target.get("device_id", "") alias = target.get("alias", "device") # Try WebSocket push first (instant) try: # Import at call time to avoid circular imports import importlib server_mod = importlib.import_module("server") push_fn = getattr(server_mod, "push_command_to_device", None) if push_fn: pushed = await push_fn(device_id, { "action": "execute_tool", "tool": name, "args": args, }) if pushed: return f"Command '{name}' sent to {alias} for execution." except Exception: pass # Fallback: queue command via REST result = await send_command_to_device( target_alias=alias, command=json.dumps({"tool": name, "args": args}), user_id="default", ) if "error" in result: return f"Error delegating '{name}' to {alias}: {result['error']}" return f"Command '{name}' queued for execution on {alias}." async def execute_tool(name: str, args: dict) -> str: if name not in TOOL_REGISTRY: return f"Error: Unknown tool '{name}'" # If running on Linux (HF Space) and this tool needs macOS, delegate it if not IS_MACOS and name in _MACOS_ONLY_TOOLS: return await _delegate_to_device(name, args) # Validate required parameters and basic types before calling schema = TOOL_REGISTRY[name].get("parameters", {}) properties = schema.get("properties", {}) required = schema.get("required", []) for req_param in required: if req_param not in args: return f"Error: {name} requires parameter '{req_param}'" # Basic type coercion for common mismatches from LLM output for param_name, param_value in list(args.items()): if param_name in properties: expected_type = properties[param_name].get("type") if expected_type == "integer" and isinstance(param_value, str): try: args[param_name] = int(param_value) except ValueError: return f"Error: {name} parameter '{param_name}' must be an integer" elif expected_type == "boolean" and isinstance(param_value, str): args[param_name] = param_value.lower() in ("true", "1", "yes") try: result = TOOL_REGISTRY[name]["function"](**args) if hasattr(result, "__await__"): result = await result return str(result) except TypeError as e: return f"Error executing {name}: invalid arguments" except Exception as e: _log.error(f"Tool {name} failed: {type(e).__name__}: {e}") return f"Error executing {name}: an unexpected error occurred" def register_macos_tools(): """Register all macOS-only tool names. Called after all tool modules are imported so TOOL_REGISTRY is populated. These tools use osascript, AppleScript, or macOS-specific CLI commands. On Linux (HF Space / Render), they get automatically delegated to a connected Mac device via WebSocket push. """ macos_tools = { # system_control.py — all 22 tools "set_volume", "get_volume", "set_brightness", "toggle_dark_mode", "wifi_control", "bluetooth_control", "do_not_disturb", "screenshot", "media_control", "clipboard", "send_notification", "list_running_apps", "quit_app", "set_timer", "lock_screen", "sleep_display", "empty_trash", "announce", "open_url", "search_files", "create_reminder", "get_events", # app_automation.py — all 36 tools "spotify_play", "spotify_play_uri", "spotify_search", "spotify_queue", "spotify_status", "spotify_control", "notes_create", "notes_append", "notes_list", "notes_search", "notes_read", "notes_delete", "reminders_add", "reminders_list", "reminders_complete", "calendar_create_event", "calendar_today", "send_imessage", "read_messages", "mail_compose", "mail_unread", "browser_open", "browser_tabs", "browser_read_page", "spotlight_search", "finder_open", "finder_move", "finder_copy", "trash_file", "contacts_search", "maps_directions", "app_keystroke", "app_menu_click", "app_window_manage", "textedit_create", "photos_recent", "photos_search", "photos_albums", "photos_open", "focus_status", "focus_set", # builtin.py — macOS-specific subset "open_app", "open_terminal", "run_in_terminal", # vscode_tools.py — all 7 tools "vscode_open", "vscode_open_terminal", "vscode_run_command", "copilot_chat", "copilot_inline", "vscode_list_extensions", "vscode_diff", # device_control.py — macOS-specific subset "bluetooth_scan", "bluetooth_pair", "bluetooth_connect", "bluetooth_disconnect", "bluetooth_info", "network_scan", "bonjour_discover", "homekit_control", "homekit_list_shortcuts", "find_my_devices", "siri_command", "airdrop_file", } _MACOS_ONLY_TOOLS.update(macos_tools) def get_tool_definitions() -> list[dict]: """Get tool definitions formatted for LLM function calling.""" tools = [] for name, info in TOOL_REGISTRY.items(): tools.append({ "name": info["name"], "description": info["description"], "parameters": info["parameters"], }) return tools def get_tools_prompt() -> str: """Generate a tools description for the system prompt.""" lines = ["You have access to the following tools:\n"] for name, info in TOOL_REGISTRY.items(): params = info["parameters"] param_str = ", ".join( f"{p}: {params['properties'][p].get('description', '')}" for p in params.get("properties", {}) ) lines.append(f"- **{name}**({param_str}): {info['description']}") lines.append( "\n## TOOL CALL FORMAT (MANDATORY)\n" "To use a tool, you MUST respond with a JSON block inside ```tool fences.\n" "The format is EXACTLY:\n" '```tool\n{"tool": "tool_name", "args": {"param": "value"}}\n```\n\n' "## TOOL CALL EXAMPLES\n" "User: \"Open Safari\"\n" "Response: Opening Safari for you, sir.\n" '```tool\n{"tool": "open_app", "args": {"app_name": "Safari"}}\n```\n\n' "User: \"Open VS Code\"\n" "Response: Launching Visual Studio Code now.\n" '```tool\n{"tool": "open_app", "args": {"app_name": "Visual Studio Code"}}\n```\n\n' "User: \"What's the weather in London?\"\n" "Response: Let me check London's weather.\n" '```tool\n{"tool": "get_weather", "args": {"location": "London"}}\n```\n\n' "User: \"Turn on dark mode\"\n" "Response: Switching to dark mode.\n" '```tool\n{"tool": "toggle_dark_mode", "args": {"enable": true}}\n```\n\n' "User: \"Set volume to 50\"\n" "Response: Setting volume to 50%.\n" '```tool\n{"tool": "set_volume", "args": {"level": 50}}\n```\n\n' "## CRITICAL RULES\n" "- ALWAYS emit a ```tool block when the user wants you to DO something (open app, play music, set volume, search, etc.)\n" "- NEVER just describe what you would do — actually call the tool.\n" "- You can include speech text BEFORE the tool block.\n" "- You can chain multiple ```tool blocks in one response.\n" "- After tool results come back, synthesize them into a natural response.\n" "- If a command maps to ANY registered tool, USE IT. Do not respond with just text.\n" ) return "\n".join(lines)