"""MCP server command for CUA CLI. Provides a Model Context Protocol server that exposes CUA functionality to AI assistants like Claude. """ import argparse import json import logging import os import sys from enum import Enum from typing import TYPE_CHECKING, Any, Optional if TYPE_CHECKING: from mcp.server.fastmcp import FastMCP # Set up logging to stderr logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", stream=sys.stderr, ) logger = logging.getLogger("cua-mcp") class Permission(Enum): """MCP permission types.""" # Sandbox management SANDBOX_LIST = "sandbox:list" SANDBOX_CREATE = "sandbox:create" SANDBOX_DELETE = "sandbox:delete" SANDBOX_START = "sandbox:start" SANDBOX_STOP = "sandbox:stop" SANDBOX_RESTART = "sandbox:restart" SANDBOX_SUSPEND = "sandbox:suspend" SANDBOX_GET = "sandbox:get" SANDBOX_VNC = "sandbox:vnc" # Computer control COMPUTER_SCREENSHOT = "computer:screenshot" COMPUTER_CLICK = "computer:click" COMPUTER_TYPE = "computer:type" COMPUTER_KEY = "computer:key" COMPUTER_SCROLL = "computer:scroll" COMPUTER_DRAG = "computer:drag" COMPUTER_HOTKEY = "computer:hotkey" COMPUTER_CLIPBOARD = "computer:clipboard" COMPUTER_FILE = "computer:file" COMPUTER_SHELL = "computer:shell" COMPUTER_WINDOW = "computer:window" # Skills SKILLS_LIST = "skills:list" SKILLS_READ = "skills:read" SKILLS_RECORD = "skills:record" SKILLS_DELETE = "skills:delete" # Permission groups for convenience PERMISSION_GROUPS = { "sandbox:all": [ Permission.SANDBOX_LIST, Permission.SANDBOX_CREATE, Permission.SANDBOX_DELETE, Permission.SANDBOX_START, Permission.SANDBOX_STOP, Permission.SANDBOX_RESTART, Permission.SANDBOX_SUSPEND, Permission.SANDBOX_GET, Permission.SANDBOX_VNC, ], "sandbox:readonly": [ Permission.SANDBOX_LIST, Permission.SANDBOX_GET, ], "computer:all": [ Permission.COMPUTER_SCREENSHOT, Permission.COMPUTER_CLICK, Permission.COMPUTER_TYPE, Permission.COMPUTER_KEY, Permission.COMPUTER_SCROLL, Permission.COMPUTER_DRAG, Permission.COMPUTER_HOTKEY, Permission.COMPUTER_CLIPBOARD, Permission.COMPUTER_FILE, Permission.COMPUTER_SHELL, Permission.COMPUTER_WINDOW, ], "computer:readonly": [ Permission.COMPUTER_SCREENSHOT, ], "skills:all": [ Permission.SKILLS_LIST, Permission.SKILLS_READ, Permission.SKILLS_RECORD, Permission.SKILLS_DELETE, ], "skills:readonly": [ Permission.SKILLS_LIST, Permission.SKILLS_READ, ], "all": list(Permission), } def parse_permissions(permissions_str: str) -> set[Permission]: """Parse permissions from a comma-separated string.""" if not permissions_str: return set() permissions = set() for perm in permissions_str.split(","): perm = perm.strip() if not perm: continue # Check if it's a group if perm in PERMISSION_GROUPS: permissions.update(PERMISSION_GROUPS[perm]) else: # Try to match individual permission try: permissions.add(Permission(perm)) except ValueError: logger.warning(f"Unknown permission: {perm}") return permissions def register_parser(subparsers: argparse._SubParsersAction) -> None: """Register the serve-mcp command.""" mcp_parser = subparsers.add_parser( "serve-mcp", help="Start MCP server for AI assistants", description="Start a Model Context Protocol server that exposes CUA functionality", ) mcp_parser.add_argument( "--permissions", type=str, default="", help="Comma-separated list of permissions (default: from CUA_MCP_PERMISSIONS env var)", ) mcp_parser.add_argument( "--sandbox", type=str, default="", help="Default sandbox name for computer commands (default: from CUA_SANDBOX env var)", ) def execute(args: argparse.Namespace) -> int: """Execute the serve-mcp command.""" try: from mcp.server.fastmcp import FastMCP # noqa: F401 except ImportError: print("MCP support not installed. Run: pip install cua-cli[mcp]", file=sys.stderr) return 1 # Parse permissions from args or env var permissions_str = args.permissions or os.environ.get("CUA_MCP_PERMISSIONS", "") permissions = parse_permissions(permissions_str) if not permissions: # Default to all permissions if none specified logger.info("No permissions specified, granting all permissions") permissions = set(Permission) else: logger.info(f"Enabled permissions: {[p.value for p in permissions]}") # Get default sandbox default_sandbox = args.sandbox or os.environ.get("CUA_SANDBOX", "") # Create and run the MCP server import anyio anyio.run(lambda: _run_mcp_server(permissions, default_sandbox)) return 0 async def _run_mcp_server(permissions: set[Permission], default_sandbox: str) -> None: """Create and run the MCP server.""" from mcp.server.fastmcp import FastMCP server = FastMCP(name="cua") # Register tools based on permissions await _register_sandbox_tools(server, permissions) await _register_computer_tools(server, permissions, default_sandbox) await _register_skills_tools(server, permissions) logger.info("Starting CUA MCP server...") await server.run_stdio_async() async def _register_sandbox_tools(server: "FastMCP", permissions: set[Permission]) -> None: """Register sandbox management tools.""" from mcp.server.fastmcp import Context if Permission.SANDBOX_LIST in permissions: @server.tool() async def sandbox_list(ctx: Context) -> str: """List all cloud sandboxes.""" from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox sandboxes = await Sandbox.list(api_key=get_api_key()) return json.dumps( [ { "name": s.name, "status": s.status, "os_type": s.os_type, "created_at": s.created_at, } for s in sandboxes ], indent=2, ) if Permission.SANDBOX_CREATE in permissions: @server.tool() async def sandbox_create( ctx: Context, os_type: str = "linux", size: str = "medium", region: str = "north-america", ) -> str: """Create a new cloud sandbox. Args: os_type: Operating system (linux, macos, windows) size: VM size (small, medium, large, xlarge) region: Region (north-america, europe, asia) """ from cua_cli.auth.store import get_api_key from cua_sandbox import Image, Sandbox if os_type == "macos": image = Image.macos("26") elif os_type == "windows": image = Image.windows("11") else: image = Image.linux("ubuntu", "24.04") sb = await Sandbox.create(image, api_key=get_api_key(), region=region) name = sb.name await sb.disconnect() return json.dumps( { "name": name, "status": "ready", "os_type": os_type, "message": f"Created sandbox: {name}", }, indent=2, ) if Permission.SANDBOX_GET in permissions: @server.tool() async def sandbox_get(ctx: Context, name: str) -> str: """Get details for a specific sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox try: info = await Sandbox.get_info(name, api_key=get_api_key()) return json.dumps( { "name": info.name, "status": info.status, "os_type": info.os_type, "created_at": info.created_at, }, indent=2, ) except Exception as e: return json.dumps({"error": str(e)}) if Permission.SANDBOX_START in permissions: @server.tool() async def sandbox_start(ctx: Context, name: str) -> str: """Start a stopped sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox await Sandbox.resume(name, api_key=get_api_key()) return json.dumps({"success": True, "message": f"Started sandbox: {name}"}) if Permission.SANDBOX_STOP in permissions: @server.tool() async def sandbox_stop(ctx: Context, name: str) -> str: """Stop a running sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox await Sandbox.suspend(name, api_key=get_api_key()) return json.dumps({"success": True, "message": f"Stopped sandbox: {name}"}) if Permission.SANDBOX_RESTART in permissions: @server.tool() async def sandbox_restart(ctx: Context, name: str) -> str: """Restart a sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox await Sandbox.restart(name, api_key=get_api_key()) return json.dumps({"success": True, "message": f"Restarted sandbox: {name}"}) if Permission.SANDBOX_SUSPEND in permissions: @server.tool() async def sandbox_suspend(ctx: Context, name: str) -> str: """Suspend a running sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox await Sandbox.suspend(name, api_key=get_api_key()) return json.dumps({"success": True, "message": f"Suspended sandbox: {name}"}) if Permission.SANDBOX_DELETE in permissions: @server.tool() async def sandbox_delete(ctx: Context, name: str) -> str: """Delete a sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox await Sandbox.delete(name, api_key=get_api_key()) return json.dumps({"success": True, "message": f"Deleted sandbox: {name}"}) if Permission.SANDBOX_VNC in permissions: @server.tool() async def sandbox_vnc(ctx: Context, name: str) -> str: """Get VNC URL for a sandbox. Args: name: Sandbox name """ from cua_cli.auth.store import get_api_key from cua_sandbox import Sandbox try: sb = await Sandbox.connect(name, api_key=get_api_key()) vnc_url = await sb.get_display_url(share=True) await sb.disconnect() return json.dumps({"vnc_url": vnc_url}) except Exception as e: return json.dumps({"error": str(e)}) async def _register_computer_tools( server: "FastMCP", permissions: set[Permission], default_sandbox: str, ) -> None: """Register computer control tools that proxy to computer-server.""" import aiohttp from cua_cli.auth.store import get_api_key from mcp.server.fastmcp import Context from mcp.server.fastmcp.utilities.types import Image async def _get_server_url(sandbox_name: str) -> Optional[str]: """Get the computer-server URL for a sandbox.""" name = sandbox_name or default_sandbox if not name: raise ValueError("No sandbox specified. Use --sandbox or set CUA_SANDBOX env var") api_key = get_api_key() if not api_key: raise ValueError("Not authenticated. Run 'cua auth login' first") from cua_sandbox.transport.cloud import CloudTransport, cloud_get_vm vm = await cloud_get_vm(name, api_key=api_key) if not vm: raise ValueError(f"Sandbox not found: {name}") return CloudTransport._resolve_endpoint(vm) async def _send_command(sandbox_name: str, command: str, params: dict) -> dict: """Send a command to the computer-server.""" server_url = await _get_server_url(sandbox_name) api_key = get_api_key() from cua_core.http import cua_version_headers headers = { "Content-Type": "application/json", "X-API-Key": api_key, "X-Container-Name": sandbox_name or default_sandbox, **cua_version_headers(), } async with aiohttp.ClientSession() as session: async with session.post( f"{server_url}/cmd", json={"command": command, "params": params}, headers=headers, ) as resp: # Read SSE response text = await resp.text() # Parse SSE data lines for line in text.split("\n"): if line.startswith("data: "): return json.loads(line[6:]) return {"success": False, "error": "No response from server"} if Permission.COMPUTER_SCREENSHOT in permissions: @server.tool() async def computer_screenshot(ctx: Context, sandbox: str = "") -> Any: """Take a screenshot of the sandbox screen. Args: sandbox: Sandbox name (optional, uses default if not specified) """ result = await _send_command(sandbox, "screenshot", {}) if result.get("success") and result.get("image_data"): import base64 return Image(format="png", data=base64.b64decode(result["image_data"])) return json.dumps(result) if Permission.COMPUTER_CLICK in permissions: @server.tool() async def computer_click( ctx: Context, x: int, y: int, button: str = "left", sandbox: str = "", ) -> str: """Click at coordinates on the screen. Args: x: X coordinate y: Y coordinate button: Mouse button (left, right, middle) sandbox: Sandbox name (optional) """ if button == "left": result = await _send_command(sandbox, "left_click", {"x": x, "y": y}) elif button == "right": result = await _send_command(sandbox, "right_click", {"x": x, "y": y}) else: result = await _send_command(sandbox, "left_click", {"x": x, "y": y}) return json.dumps(result) @server.tool() async def computer_double_click( ctx: Context, x: int, y: int, sandbox: str = "", ) -> str: """Double-click at coordinates on the screen. Args: x: X coordinate y: Y coordinate sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "double_click", {"x": x, "y": y}) return json.dumps(result) if Permission.COMPUTER_TYPE in permissions: @server.tool() async def computer_type(ctx: Context, text: str, sandbox: str = "") -> str: """Type text on the keyboard. Args: text: Text to type sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "type_text", {"text": text}) return json.dumps(result) if Permission.COMPUTER_KEY in permissions: @server.tool() async def computer_key(ctx: Context, key: str, sandbox: str = "") -> str: """Press a key on the keyboard. Args: key: Key to press (e.g., "enter", "tab", "escape") sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "press_key", {"key": key}) return json.dumps(result) if Permission.COMPUTER_HOTKEY in permissions: @server.tool() async def computer_hotkey(ctx: Context, keys: str, sandbox: str = "") -> str: """Press a keyboard shortcut. Args: keys: Keys to press (e.g., "cmd+c", "ctrl+shift+s") sandbox: Sandbox name (optional) """ key_list = keys.replace("-", "+").split("+") result = await _send_command(sandbox, "hotkey", {"keys": key_list}) return json.dumps(result) if Permission.COMPUTER_SCROLL in permissions: @server.tool() async def computer_scroll( ctx: Context, direction: str = "down", amount: int = 3, sandbox: str = "", ) -> str: """Scroll the screen. Args: direction: Scroll direction (up, down, left, right) amount: Number of scroll clicks sandbox: Sandbox name (optional) """ result = await _send_command( sandbox, "scroll_direction", {"direction": direction, "clicks": amount} ) return json.dumps(result) if Permission.COMPUTER_DRAG in permissions: @server.tool() async def computer_drag( ctx: Context, start_x: int, start_y: int, end_x: int, end_y: int, sandbox: str = "", ) -> str: """Drag from one point to another. Args: start_x: Starting X coordinate start_y: Starting Y coordinate end_x: Ending X coordinate end_y: Ending Y coordinate sandbox: Sandbox name (optional) """ result = await _send_command( sandbox, "drag_to", {"start_x": start_x, "start_y": start_y, "end_x": end_x, "end_y": end_y}, ) return json.dumps(result) if Permission.COMPUTER_CLIPBOARD in permissions: @server.tool() async def computer_clipboard_get(ctx: Context, sandbox: str = "") -> str: """Get clipboard contents. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "copy_to_clipboard", {}) return json.dumps(result) @server.tool() async def computer_clipboard_set(ctx: Context, text: str, sandbox: str = "") -> str: """Set clipboard contents. Args: text: Text to copy to clipboard sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "set_clipboard", {"text": text}) return json.dumps(result) if Permission.COMPUTER_FILE in permissions: @server.tool() async def computer_file_read(ctx: Context, path: str, sandbox: str = "") -> str: """Read a file from the sandbox. Args: path: Path to the file sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "read_text", {"path": path}) return json.dumps(result) @server.tool() async def computer_file_write( ctx: Context, path: str, content: str, sandbox: str = "" ) -> str: """Write a file to the sandbox. Args: path: Path to the file content: File content sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "write_text", {"path": path, "content": content}) return json.dumps(result) @server.tool() async def computer_file_list(ctx: Context, path: str = ".", sandbox: str = "") -> str: """List files in a directory. Args: path: Directory path sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "list_dir", {"path": path}) return json.dumps(result) if Permission.COMPUTER_SHELL in permissions: @server.tool() async def computer_shell(ctx: Context, command: str, sandbox: str = "") -> str: """Run a shell command in the sandbox. Args: command: Shell command to run sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "run_command", {"command": command}) return json.dumps(result) if Permission.COMPUTER_WINDOW in permissions: @server.tool() async def computer_window_list(ctx: Context, sandbox: str = "") -> str: """List open windows. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "get_application_windows", {}) return json.dumps(result) @server.tool() async def computer_window_open(ctx: Context, path: str, sandbox: str = "") -> str: """Open a file or URL. Args: path: Path to file or URL to open sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "open", {"path": path}) return json.dumps(result) @server.tool() async def computer_window_focus(ctx: Context, window_id: str, sandbox: str = "") -> str: """Bring a window to the foreground and focus it. Args: window_id: Window identifier sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "activate_window", {"window_id": window_id}) return json.dumps(result) @server.tool() async def computer_window_unfocus(ctx: Context, sandbox: str = "") -> str: """Remove focus from the currently focused window. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "deactivate_window", {}) if not result.get("success"): # Fallback: press Escape result = await _send_command(sandbox, "press_key", {"key": "escape"}) return json.dumps(result) @server.tool() async def computer_window_minimize(ctx: Context, window_id: str, sandbox: str = "") -> str: """Minimize a window. Args: window_id: Window identifier sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "minimize_window", {"window_id": window_id}) return json.dumps(result) @server.tool() async def computer_window_maximize(ctx: Context, window_id: str, sandbox: str = "") -> str: """Maximize a window. Args: window_id: Window identifier sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "maximize_window", {"window_id": window_id}) return json.dumps(result) @server.tool() async def computer_window_close(ctx: Context, window_id: str, sandbox: str = "") -> str: """Close a window. Args: window_id: Window identifier sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "close_window", {"window_id": window_id}) return json.dumps(result) @server.tool() async def computer_window_resize( ctx: Context, window_id: str, width: int, height: int, sandbox: str = "", ) -> str: """Resize a window. Args: window_id: Window identifier width: New width in pixels height: New height in pixels sandbox: Sandbox name (optional) """ result = await _send_command( sandbox, "set_window_size", {"window_id": window_id, "width": width, "height": height}, ) return json.dumps(result) @server.tool() async def computer_window_move( ctx: Context, window_id: str, x: int, y: int, sandbox: str = "", ) -> str: """Move a window to a position on screen. Args: window_id: Window identifier x: X coordinate for the window's top-left corner y: Y coordinate for the window's top-left corner sandbox: Sandbox name (optional) """ result = await _send_command( sandbox, "set_window_position", {"window_id": window_id, "x": x, "y": y}, ) return json.dumps(result) @server.tool() async def computer_window_get_info( ctx: Context, window_id: str, sandbox: str = "", ) -> str: """Get a window's title, size, and position. Args: window_id: Window identifier sandbox: Sandbox name (optional) """ name_r = await _send_command(sandbox, "get_window_name", {"window_id": window_id}) size_r = await _send_command(sandbox, "get_window_size", {"window_id": window_id}) pos_r = await _send_command(sandbox, "get_window_position", {"window_id": window_id}) return json.dumps( { "window_id": window_id, "title": name_r.get("name") or name_r.get("data"), "size": size_r.get("size") or size_r.get("data"), "position": pos_r.get("position") or pos_r.get("data"), }, indent=2, ) @server.tool() async def computer_launch( ctx: Context, app: str, args: list[str] | None = None, sandbox: str = "", ) -> str: """Launch an application. Args: app: Application executable or bundle identifier args: Optional list of arguments sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "launch", {"app": app, "args": args or []}) return json.dumps(result) if Permission.COMPUTER_CLICK in permissions: @server.tool() async def computer_move_cursor( ctx: Context, x: int, y: int, sandbox: str = "", ) -> str: """Move the mouse cursor without clicking. Args: x: X coordinate y: Y coordinate sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "move_cursor", {"x": x, "y": y}) return json.dumps(result) @server.tool() async def computer_mouse_down( ctx: Context, x: int, y: int, button: str = "left", sandbox: str = "", ) -> str: """Press and hold a mouse button. Args: x: X coordinate y: Y coordinate button: Mouse button (left, right, middle) sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "mouse_down", {"x": x, "y": y, "button": button}) return json.dumps(result) @server.tool() async def computer_mouse_up( ctx: Context, x: int, y: int, button: str = "left", sandbox: str = "", ) -> str: """Release a mouse button. Args: x: X coordinate y: Y coordinate button: Mouse button (left, right, middle) sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "mouse_up", {"x": x, "y": y, "button": button}) return json.dumps(result) if Permission.COMPUTER_KEY in permissions: @server.tool() async def computer_key_down(ctx: Context, key: str, sandbox: str = "") -> str: """Press and hold a key. Args: key: Key to hold (e.g. "shift", "ctrl", "a") sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "key_down", {"key": key}) return json.dumps(result) @server.tool() async def computer_key_up(ctx: Context, key: str, sandbox: str = "") -> str: """Release a previously held key. Args: key: Key to release sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "key_up", {"key": key}) return json.dumps(result) if Permission.COMPUTER_SCREENSHOT in permissions: @server.tool() async def computer_get_screen_size(ctx: Context, sandbox: str = "") -> str: """Get the screen dimensions. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "get_screen_size", {}) return json.dumps(result) @server.tool() async def computer_get_cursor_position(ctx: Context, sandbox: str = "") -> str: """Get the current cursor position. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "get_cursor_position", {}) return json.dumps(result) @server.tool() async def computer_get_accessibility_tree(ctx: Context, sandbox: str = "") -> str: """Get the accessibility tree of the current screen. Args: sandbox: Sandbox name (optional) """ result = await _send_command(sandbox, "get_accessibility_tree", {}) return json.dumps(result) @server.tool() async def computer_get_current_window(ctx: Context, sandbox: str = "") -> str: """Get the currently focused window ID and title. Args: sandbox: Sandbox name (optional) """ win_r = await _send_command(sandbox, "get_current_window_id", {}) window_id = win_r.get("window_id") or win_r.get("data") if window_id: name_r = await _send_command(sandbox, "get_window_name", {"window_id": window_id}) title = name_r.get("name") or name_r.get("data") or "" else: title = "" return json.dumps({"window_id": window_id, "title": title or "Desktop"}) async def _register_skills_tools(server: "FastMCP", permissions: set[Permission]) -> None: """Register skills management tools.""" from pathlib import Path from mcp.server.fastmcp import Context SKILLS_DIR = Path.home() / ".cua" / "skills" if Permission.SKILLS_LIST in permissions: @server.tool() async def skills_list(ctx: Context) -> str: """List all recorded skills.""" if not SKILLS_DIR.exists(): return json.dumps([]) skills = [] for skill_dir in SKILLS_DIR.iterdir(): if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists(): skill_file = skill_dir / "SKILL.md" content = skill_file.read_text() # Extract title from markdown title = skill_dir.name for line in content.split("\n"): if line.startswith("# "): title = line[2:].strip() break # Count trajectory steps trajectory_dir = skill_dir / "trajectory" step_count = ( len(list(trajectory_dir.glob("step_*.md"))) if trajectory_dir.exists() else 0 ) skills.append( { "name": skill_dir.name, "title": title, "steps": step_count, } ) return json.dumps(skills, indent=2) if Permission.SKILLS_READ in permissions: @server.tool() async def skills_read(ctx: Context, name: str) -> str: """Read a skill's content. Args: name: Skill name """ skill_dir = SKILLS_DIR / name skill_file = skill_dir / "SKILL.md" if not skill_file.exists(): return json.dumps({"error": f"Skill not found: {name}"}) content = skill_file.read_text() # Also include trajectory steps trajectory_dir = skill_dir / "trajectory" steps = [] if trajectory_dir.exists(): for step_file in sorted(trajectory_dir.glob("step_*.md")): steps.append( { "file": step_file.name, "content": step_file.read_text(), } ) return json.dumps( { "name": name, "content": content, "steps": steps, }, indent=2, ) if Permission.SKILLS_DELETE in permissions: @server.tool() async def skills_delete(ctx: Context, name: str) -> str: """Delete a skill. Args: name: Skill name """ import shutil skill_dir = SKILLS_DIR / name if not skill_dir.exists(): return json.dumps({"error": f"Skill not found: {name}"}) shutil.rmtree(skill_dir) return json.dumps({"success": True, "message": f"Deleted skill: {name}"}) if Permission.SKILLS_RECORD in permissions: @server.tool() async def skills_record( ctx: Context, name: str, sandbox: str = "", port: int = 8765, ) -> str: """Start recording a skill. This starts a WebSocket server that receives screen recordings. Use the CUA browser extension or screen recorder to send frames. Args: name: Name for the skill sandbox: Sandbox name (optional) port: WebSocket port (default: 8765) """ return json.dumps( { "message": f"To record skill '{name}', use 'cua skills record {name}' from the terminal", "instructions": [ f"1. Run: cua skills record {name}", "2. Use the CUA browser extension or screen recorder", "3. Perform the actions you want to record", "4. Stop the recording when done", ], }, indent=2, )