mirror/computer
1
0
Fork 0
mirror of https://github.com/trycua/computer.git synced 2026-04-24 15:38:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

Run uv run pre-commit run --all-files

Browse Source
This commit is contained in:
Dillon DuPont
2025-12-09 13:24:33 -08:00
parent 6ec0347061
commit 7ee7ecd3b3
12 changed files with 105 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/src/app/layout.tsx
+5 -1
View File
@@ -20,7 +20,11 @@ const geistMono = Geist_Mono({
export default function Layout({ children }: { children: ReactNode }) {
return (
<html lang="en" className={`${geist.variable} ${geistMono.variable} font-sans`} suppressHydrationWarning>
<html
lang="en"
className={`${geist.variable} ${geistMono.variable} font-sans`}
suppressHydrationWarning
>
<head>
<link rel="icon" href="/docs/favicon.ico" sizes="any" />
</head>
libs/python/agent/agent/tools/__init__.py
-1
View File
@@ -3,4 +3,3 @@
from .browser_tool import BrowserTool
__all__ = ["BrowserTool"]
libs/python/bench-ui/bench_ui/__init__.py
+1 -1
View File
@@ -1,3 +1,3 @@
from .api import launch_window, get_element_rect, execute_javascript
from .api import execute_javascript, get_element_rect, launch_window
__all__ = ["launch_window", "get_element_rect", "execute_javascript"]
libs/python/bench-ui/bench_ui/api.py
+6 -3
View File
@@ -5,9 +5,10 @@ import sys
import tempfile
import time
from pathlib import Path
from typing import Optional, Dict, Any
from typing import Any, Dict, Optional
from urllib import request
from urllib.error import HTTPError, URLError
import psutil
# Map child PID -> listening port
@@ -16,7 +17,9 @@ _pid_to_port: Dict[int, int] = {}
def _post_json(url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
data = json.dumps(payload).encode("utf-8")
req = request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
req = request.Request(
url, data=data, headers={"Content-Type": "application/json"}, method="POST"
)
try:
with request.urlopen(req, timeout=5) as resp:
text = resp.read().decode("utf-8")
@@ -26,7 +29,7 @@ def _post_json(url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
body = (e.read() or b"").decode("utf-8", errors="ignore")
return json.loads(body)
except Exception:
return {"error": "http_error", "status": getattr(e, 'code', None)}
return {"error": "http_error", "status": getattr(e, "code", None)}
except URLError as e:
return {"error": "url_error", "reason": str(e.reason)}
libs/python/bench-ui/bench_ui/child.py
+14 -4
View File
@@ -18,7 +18,13 @@ def _get_free_port() -> int:
return s.getsockname()[1]
def _start_http_server(window: webview.Window, port: int, ready_event: threading.Event, html_content: str | None = None, folder_path: str | None = None):
def _start_http_server(
window: webview.Window,
port: int,
ready_event: threading.Event,
html_content: str | None = None,
folder_path: str | None = None,
):
async def rect_handler(request: web.Request):
try:
data = await request.json()
@@ -96,13 +102,13 @@ def _start_http_server(window: webview.Window, port: int, ready_event: threading
return web.Response(text=html_content, content_type="text/html")
app = web.Application()
# If serving a folder, add static file routes
if folder_path:
app.router.add_static("/", folder_path, show_index=True)
else:
app.router.add_get("/", index_handler)
app.router.add_post("/rect", rect_handler)
app.router.add_post("/eval", eval_handler)
@@ -193,12 +199,16 @@ def main():
# Track when the page is loaded so JS execution succeeds
window_ready = threading.Event()
def _on_loaded():
window_ready.set()
window.events.loaded += _on_loaded # type: ignore[attr-defined]
# Start HTTP server for control (and optionally serve inline HTML or static folder)
_start_http_server(window, port, window_ready, html_content=html_for_server, folder_path=folder_for_server)
_start_http_server(
window, port, window_ready, html_content=html_for_server, folder_path=folder_for_server
)
# Print startup info for parent to read
print(json.dumps({"pid": os.getpid(), "port": port}), flush=True)
libs/python/bench-ui/examples/simple_example.py
+7 -4
View File
@@ -1,8 +1,10 @@
from __future__ import annotations
import time
from bench_ui import launch_window, get_element_rect, execute_javascript
from pathlib import Path
import os
import time
from pathlib import Path
from bench_ui import execute_javascript, get_element_rect, launch_window
HTML = """
<!doctype html>
@@ -34,6 +36,7 @@ HTML = """
</html>
"""
def main():
os.environ["CUA_BENCH_UI_DEBUG"] = "1"
@@ -55,7 +58,7 @@ def main():
# Take a screenshot and overlay the bbox
try:
from PIL import ImageGrab, ImageDraw
from PIL import ImageDraw, ImageGrab
img = ImageGrab.grab() # full screen
draw = ImageDraw.Draw(img)
libs/python/bench-ui/tests/test_port_detection.py
+2 -2
View File
@@ -1,8 +1,8 @@
import time
import psutil
import pytest
from bench_ui import launch_window, execute_javascript
from bench_ui import execute_javascript, launch_window
from bench_ui.api import _pid_to_port
HTML = """
libs/python/computer-server/computer_server/main.py
+2 -2
View File
@@ -24,8 +24,8 @@ from fastapi import (
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from .handlers.factory import HandlerFactory
from .browser import get_browser_manager
from .handlers.factory import HandlerFactory
# Authentication session TTL (in seconds). Override via env var CUA_AUTH_TTL_SECONDS. Default: 60s
AUTH_SESSION_TTL_SECONDS: int = int(os.environ.get("CUA_AUTH_TTL_SECONDS", "60"))
@@ -805,7 +805,7 @@ async def playwright_exec_endpoint(
try:
browser_manager = get_browser_manager()
result = await browser_manager.execute_command(command, params)
if result.get("success"):
return JSONResponse(content=result)
else:
libs/python/computer/computer/computer.py
+43 -21
View File
@@ -7,8 +7,21 @@ import platform
import re
import time
import traceback
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Literal, Optional, Union, cast, TypeVar
from functools import wraps
from typing import (
TYPE_CHECKING,
Any,
Awaitable,
Callable,
Dict,
List,
Literal,
Optional,
TypeVar,
Union,
cast,
)
try:
from typing import ParamSpec
except Exception: # pragma: no cover
@@ -135,7 +148,7 @@ class Computer:
self.provider_type = provider_type
self.ephemeral = ephemeral
self.api_key = api_key if self.provider_type == VMProviderType.CLOUD else None
# Set default API port if not specified
if self.api_port is None:
self.api_port = 8443 if self.api_key else 8000
@@ -551,7 +564,7 @@ class Computer:
await self._interface.wait_for_ready(timeout=30)
self.logger.info("Sandbox interface connected successfully")
except TimeoutError as e:
port = getattr(self._interface, '_api_port', 8000) # Default to 8000 if not set
port = getattr(self._interface, "_api_port", 8000) # Default to 8000 if not set
self.logger.error(f"Failed to connect to sandbox interface at {ip_address}:{port}")
raise TimeoutError(
f"Could not connect to sandbox interface at {ip_address}:{port}: {str(e)}"
@@ -1041,7 +1054,7 @@ class Computer:
else "echo No requirements to install"
)
return await self.interface.run_command(install_cmd)
async def pip_install(self, requirements: list[str]):
"""Install packages using the system Python/pip (no venv).
@@ -1059,7 +1072,7 @@ class Computer:
reqs = " ".join(requirements)
install_cmd = f"python -m pip install {reqs}"
return await self.interface.run_command(install_cmd)
async def venv_cmd(self, venv_name: str, command: str):
"""Execute a shell command in a virtual environment.
@@ -1216,7 +1229,7 @@ print(f"<<<VENV_EXEC_START>>>{{output_json}}<<<VENV_EXEC_END>>>")
return output_payload["result"]
else:
import builtins
# Recreate and raise the original exception
error_info = output_payload.get("error", {}) or {}
err_type = error_info.get("type") or "Exception"
@@ -1238,7 +1251,9 @@ print(f"<<<VENV_EXEC_START>>>{{output_json}}<<<VENV_EXEC_END>>>")
f"No output payload found. stdout: {result.stdout}, stderr: {result.stderr}"
)
async def venv_exec_background(self, venv_name: str, python_func, *args, requirements: Optional[List[str]] = None, **kwargs) -> int:
async def venv_exec_background(
self, venv_name: str, python_func, *args, requirements: Optional[List[str]] = None, **kwargs
) -> int:
"""Run the Python function in the venv in the background and return the PID.
Uses a short launcher Python that spawns a detached child and exits immediately.
@@ -1269,7 +1284,8 @@ print(f"<<<VENV_EXEC_START>>>{{output_json}}<<<VENV_EXEC_END>>>")
args_b64 = base64.b64encode(args_json.encode("utf-8")).decode("ascii")
kwargs_b64 = base64.b64encode(kwargs_json.encode("utf-8")).decode("ascii")
payload_code = f'''
payload_code = (
f'''
import json
import traceback
import base64
@@ -1285,7 +1301,9 @@ try:
kwargs = json.loads(base64.b64decode(_kwargs_b64).decode('utf-8'))
# Ensure requirements inside the active venv
for pkg in json.loads(''' + repr(reqs_json) + '''):
for pkg in json.loads('''
+ repr(reqs_json)
+ """):
if pkg:
import subprocess, sys
subprocess.run([sys.executable, '-m', 'pip', 'install', pkg], check=False)
@@ -1293,12 +1311,13 @@ try:
except Exception:
import sys
sys.stderr.write(traceback.format_exc())
'''
"""
)
payload_b64 = base64.b64encode(payload_code.encode("utf-8")).decode("ascii")
if self.os_type == "windows":
# Launcher spawns detached child and prints its PID
launcher_code = f'''
launcher_code = f"""
import base64, subprocess, os, sys
DETACHED_PROCESS = 0x00000008
CREATE_NEW_PROCESS_GROUP = 0x00000200
@@ -1306,13 +1325,13 @@ creationflags = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
code = base64.b64decode("{payload_b64}").decode("utf-8")
p = subprocess.Popen(["python", "-c", code], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=creationflags)
print(p.pid)
'''
"""
launcher_b64 = base64.b64encode(launcher_code.encode("utf-8")).decode("ascii")
venv_path = f"%USERPROFILE%\\.venvs\\{venv_name}"
cmd = (
'cmd /c "'
f'call "{venv_path}\\Scripts\\activate.bat" && '
f'python -c "import base64; exec(base64.b64decode(\'{launcher_b64}\').decode(\'utf-8\'))"'
f"python -c \"import base64; exec(base64.b64decode('{launcher_b64}').decode('utf-8'))\""
'"'
)
result = await self.interface.run_command(cmd)
@@ -1320,18 +1339,18 @@ print(p.pid)
return int(pid_str)
else:
log = f"/tmp/cua_bg_{int(_time.time())}.log"
launcher_code = f'''
launcher_code = f"""
import base64, subprocess, os, sys
code = base64.b64decode("{payload_b64}").decode("utf-8")
with open("{log}", "ab", buffering=0) as f:
p = subprocess.Popen(["python", "-c", code], stdout=f, stderr=subprocess.STDOUT, preexec_fn=getattr(os, "setsid", None))
print(p.pid)
'''
"""
launcher_b64 = base64.b64encode(launcher_code.encode("utf-8")).decode("ascii")
venv_path = f"$HOME/.venvs/{venv_name}"
shell = (
f'. "{venv_path}/bin/activate" && '
f'python -c "import base64; exec(base64.b64decode(\'{launcher_b64}\').decode(\'utf-8\'))"'
f"python -c \"import base64; exec(base64.b64decode('{launcher_b64}').decode('utf-8'))\""
)
result = await self.interface.run_command(shell)
pid_str = (result.stdout or "").strip().splitlines()[-1].strip()
@@ -1438,6 +1457,7 @@ print(f"<<<VENV_EXEC_START>>>{{output_json}}<<<VENV_EXEC_END>>>")
return output_payload["result"]
else:
import builtins
error_info = output_payload.get("error", {}) or {}
err_type = error_info.get("type") or "Exception"
err_msg = error_info.get("message") or ""
@@ -1454,7 +1474,9 @@ print(f"<<<VENV_EXEC_START>>>{{output_json}}<<<VENV_EXEC_END>>>")
f"No output payload found. stdout: {result.stdout}, stderr: {result.stderr}"
)
async def python_exec_background(self, python_func, *args, requirements: Optional[List[str]] = None, **kwargs) -> int:
async def python_exec_background(
self, python_func, *args, requirements: Optional[List[str]] = None, **kwargs
) -> int:
"""Run a Python function with the system interpreter in the background and return PID.
Uses a short launcher Python that spawns a detached child and exits immediately.
@@ -1505,7 +1527,7 @@ except Exception:
payload_b64 = base64.b64encode(payload_code.encode("utf-8")).decode("ascii")
if self.os_type == "windows":
launcher_code = f'''
launcher_code = f"""
import base64, subprocess, os, sys
DETACHED_PROCESS = 0x00000008
CREATE_NEW_PROCESS_GROUP = 0x00000200
@@ -1513,7 +1535,7 @@ creationflags = DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
code = base64.b64decode("{payload_b64}").decode("utf-8")
p = subprocess.Popen(["python", "-c", code], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=creationflags)
print(p.pid)
'''
"""
launcher_b64 = base64.b64encode(launcher_code.encode("utf-8")).decode("ascii")
cmd = f"python -c \"import base64; exec(base64.b64decode('{launcher_b64}').decode('utf-8'))\""
result = await self.interface.run_command(cmd)
@@ -1521,13 +1543,13 @@ print(p.pid)
return int(pid_str)
else:
log = f"/tmp/cua_bg_{int(_time.time())}.log"
launcher_code = f'''
launcher_code = f"""
import base64, subprocess, os, sys
code = base64.b64decode("{payload_b64}").decode("utf-8")
with open("{log}", "ab", buffering=0) as f:
p = subprocess.Popen(["python", "-c", code], stdout=f, stderr=subprocess.STDOUT, preexec_fn=getattr(os, "setsid", None))
print(p.pid)
'''
"""
launcher_b64 = base64.b64encode(launcher_code.encode("utf-8")).decode("ascii")
cmd = f"python -c \"import base64; exec(base64.b64decode('{launcher_b64}').decode('utf-8'))\""
result = await self.interface.run_command(cmd)
libs/python/computer/computer/interface/factory.py
+9 -3
View File
@@ -37,10 +37,16 @@ class InterfaceFactory:
from .windows import WindowsComputerInterface
if os == "macos":
return MacOSComputerInterface(ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port)
return MacOSComputerInterface(
ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port
)
elif os == "linux":
return LinuxComputerInterface(ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port)
return LinuxComputerInterface(
ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port
)
elif os == "windows":
return WindowsComputerInterface(ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port)
return WindowsComputerInterface(
ip_address, api_key=api_key, vm_name=vm_name, api_port=api_port
)
else:
raise ValueError(f"Unsupported OS type: {os}")
libs/python/computer/computer/interface/generic.py
+11 -3
View File
@@ -47,7 +47,7 @@ class GenericComputerInterface(BaseComputerInterface):
# Set logger name for the interface
self.logger = Logger(logger_name, LogLevel.NORMAL)
# Store custom ports
self._api_port = api_port
@@ -75,7 +75,11 @@ class GenericComputerInterface(BaseComputerInterface):
"""
protocol = "wss" if self.api_key else "ws"
# Use custom API port if provided, otherwise use defaults based on API key
port = str(self._api_port) if self._api_port is not None else ("8443" if self.api_key else "8000")
port = (
str(self._api_port)
if self._api_port is not None
else ("8443" if self.api_key else "8000")
)
return f"{protocol}://{self.ip_address}:{port}/ws"
@property
@@ -87,7 +91,11 @@ class GenericComputerInterface(BaseComputerInterface):
"""
protocol = "https" if self.api_key else "http"
# Use custom API port if provided, otherwise use defaults based on API key
port = str(self._api_port) if self._api_port is not None else ("8443" if self.api_key else "8000")
port = (
str(self._api_port)
if self._api_port is not None
else ("8443" if self.api_key else "8000")
)
return f"{protocol}://{self.ip_address}:{port}/cmd"
# Mouse actions
libs/python/computer/computer/providers/factory.py
+5 -1
View File
@@ -65,7 +65,11 @@ class VMProviderFactory:
"Please install it with 'pip install cua-computer[lume]'"
)
return LumeProvider(
provider_port=provider_port, host=host, storage=storage, verbose=verbose, ephemeral=ephemeral
provider_port=provider_port,
host=host,
storage=storage,
verbose=verbose,
ephemeral=ephemeral,
)
except ImportError as e:
logger.error(f"Failed to import LumeProvider: {e}")