Add process management and shell MCP tools

ac_launch_process spawns a detached subprocess from an argv list
(no shell expansion); ac_shell runs a command line via shlex.split
and reports exit_code/stdout/stderr; ac_list_processes and
ac_kill_process wrap psutil for inspection and cleanup. Working
directories are validated against os.path.realpath; missing psutil
returns a clear runtime error rather than ImportError.

Author: JE-Chen

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -751,6 +751,57 @@ def trigger_tools() -> List[MCPTool]:
     ]
 
 
+def process_and_shell_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_launch_process",
+            description=("Spawn a subprocess with the given argv list "
+                         "(detached, stdio piped to /dev/null). Returns "
+                         "{pid, argv}. Optional working_directory."),
+            input_schema=schema({
+                "argv": {"type": "array", "items": {"type": "string"}},
+                "working_directory": {"type": "string"},
+            }, required=["argv"]),
+            handler=h.launch_process,
+            annotations=DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_list_processes",
+            description=("List running processes (psutil required). "
+                         "Optionally filter by case-insensitive substring."),
+            input_schema=schema({
+                "name_contains": {"type": "string"},
+            }),
+            handler=h.list_processes,
+            annotations=READ_ONLY,
+        ),
+        MCPTool(
+            name="ac_kill_process",
+            description=("Terminate a PID gracefully, escalating to "
+                         "SIGKILL after ``timeout``. Returns 'terminated' "
+                         "/ 'killed' / 'not-found'. psutil required."),
+            input_schema=schema({
+                "pid": {"type": "integer"},
+                "timeout": {"type": "number"},
+            }, required=["pid"]),
+            handler=h.kill_process,
+            annotations=DESTRUCTIVE,
+        ),
+        MCPTool(
+            name="ac_shell",
+            description=("Run a shell-style command line via shlex.split "
+                         "(NO shell expansion). Returns {exit_code, "
+                         "stdout, stderr}."),
+            input_schema=schema({
+                "command": {"type": "string"},
+                "timeout": {"type": "number"},
+            }, required=["command"]),
+            handler=h.shell_command,
+            annotations=DESTRUCTIVE,
+        ),
+    ]
+
+
 def hotkey_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -802,5 +853,5 @@ def hotkey_tools() -> List[MCPTool]:
     mouse_tools, keyboard_tools, screen_tools, image_and_ocr_tools,
     window_tools, system_tools, recording_tools, drag_and_send_tools,
     semantic_locator_tools, scheduler_tools, trigger_tools, hotkey_tools,
-    screen_record_tools,
+    screen_record_tools, process_and_shell_tools,
 )

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -472,6 +472,96 @@ def window_restore(title_substring: str,
     return _show_command(title_substring, bool(case_sensitive), cmd_show=9)
 
 
+def launch_process(argv: List[str],
+                   working_directory: Optional[str] = None,
+                   ) -> Dict[str, Any]:
+    """Spawn a detached subprocess with a sanitised argv list."""
+    import subprocess  # nosec B404  # reason: required to spawn child processes
+    if not isinstance(argv, list) or not argv:
+        raise ValueError("argv must be a non-empty list")
+    cleaned = [str(part) for part in argv]
+    cwd = None
+    if working_directory is not None:
+        cwd = os.path.realpath(os.fspath(working_directory))
+        if not os.path.isdir(cwd):
+            raise ValueError(f"working_directory does not exist: {cwd}")
+    process = subprocess.Popen(  # nosec B603  # reason: argv list, no shell expansion
+        cleaned, cwd=cwd, stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
+    )
+    return {"pid": int(process.pid), "argv": cleaned}
+
+
+def list_processes(name_contains: Optional[str] = None,
+                    ) -> List[Dict[str, Any]]:
+    """List running processes via ``psutil`` if installed; raise otherwise."""
+    try:
+        import psutil  # type: ignore[import-untyped]
+    except ImportError as error:
+        raise RuntimeError(
+            "ac_list_processes requires psutil — pip install psutil"
+        ) from error
+    needle = name_contains.lower() if name_contains else None
+    out: List[Dict[str, Any]] = []
+    for proc in psutil.process_iter(["pid", "name", "username"]):
+        info = proc.info or {}
+        name = (info.get("name") or "")
+        if needle and needle not in name.lower():
+            continue
+        out.append({
+            "pid": int(info.get("pid") or 0),
+            "name": name,
+            "username": info.get("username") or "",
+        })
+    return out
+
+
+def kill_process(pid: int, timeout: float = 5.0) -> str:
+    """Terminate a PID gracefully, escalating to SIGKILL after ``timeout``."""
+    try:
+        import psutil  # type: ignore[import-untyped]
+    except ImportError as error:
+        raise RuntimeError(
+            "ac_kill_process requires psutil — pip install psutil"
+        ) from error
+    try:
+        proc = psutil.Process(int(pid))
+    except psutil.NoSuchProcess:
+        return "not-found"
+    proc.terminate()
+    try:
+        proc.wait(timeout=float(timeout))
+        return "terminated"
+    except psutil.TimeoutExpired:
+        proc.kill()
+        return "killed"
+
+
+def shell_command(command: str, timeout: float = 30.0
+                  ) -> Dict[str, Any]:
+    """Run a shell-style command line and return stdout/stderr/exit_code.
+
+    Uses argv-list parsing via ``shlex.split`` so we never enable a
+    shell — protects against the parameterised command injection
+    classes Bandit B602 / B605 cover.
+    """
+    import shlex
+    import subprocess  # nosec B404  # reason: required for child execution
+
+    if not command or not command.strip():
+        raise ValueError("command must be a non-empty string")
+    argv = shlex.split(command, posix=False) if os.name == "nt" \
+        else shlex.split(command)
+    proc = subprocess.run(  # nosec B603  # reason: argv from shlex.split, no shell
+        argv, capture_output=True, text=True,
+        timeout=float(timeout), check=False,
+    )
+    return {
+        "exit_code": int(proc.returncode),
+        "stdout": proc.stdout, "stderr": proc.stderr,
+    }
+
+
 def get_clipboard() -> str:
     from je_auto_control.utils.clipboard.clipboard import get_clipboard as _get
     return _get()

test/unit_test/headless/test_mcp_server.py

@@ -1350,6 +1350,60 @@ def fake_show(hwnd, cmd_show):
     assert seen["call"] == (456, 6)
 
 
+def test_process_tools_present_in_default_registry():
+    names = {tool.name for tool in build_default_tool_registry()}
+    assert {"ac_launch_process", "ac_list_processes",
+            "ac_kill_process", "ac_shell"}.issubset(names)
+
+
+def test_shell_command_returns_exit_code_and_stdout():
+    """Run a portable command and verify the shape."""
+    import sys as _sys
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    result = by_name["ac_shell"].invoke({
+        "command": f"{_sys.executable} -V",
+        "timeout": 5.0,
+    })
+    assert result["exit_code"] == 0
+    # Python 3.4+ prints the version to stdout.
+    assert "Python" in (result["stdout"] + result["stderr"])
+
+
+def test_shell_command_rejects_empty():
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    try:
+        by_name["ac_shell"].invoke({"command": "   "})
+    except ValueError:
+        pass
+    else:
+        raise AssertionError("expected ValueError for empty command")
+
+
+def test_launch_process_validates_working_directory(tmp_path):
+    import sys as _sys
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    missing = tmp_path / "ghost"
+    try:
+        by_name["ac_launch_process"].invoke({
+            "argv": [_sys.executable, "-V"],
+            "working_directory": str(missing),
+        })
+    except ValueError as error:
+        assert "does not exist" in str(error)
+    else:
+        raise AssertionError("expected ValueError")
+
+
+def test_launch_process_rejects_empty_argv():
+    by_name = {tool.name: tool for tool in build_default_tool_registry()}
+    try:
+        by_name["ac_launch_process"].invoke({"argv": []})
+    except ValueError:
+        pass
+    else:
+        raise AssertionError("expected ValueError for empty argv")
+
+
 def test_default_registry_lists_core_automation_tools():
     names = {tool.name for tool in build_default_tool_registry()}
     expected = {