mirror/computer
1
0
Fork 0
mirror of https://github.com/trycua/computer.git synced 2026-01-04 04:19:57 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into feature/computer/typescript

Browse Source
This commit is contained in:
Morgan Dean
2025-06-26 12:55:34 -07:00
parent 4625b65bab 01199f0943
commit 718000f9c8
14 changed files with 266 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
libs/python/agent/agent/providers/anthropic/tools/bash.py
View File

@@ -50,8 +50,8 @@ class BashTool(BaseBashTool, BaseAnthropicTool):
try:
async with asyncio.timeout(self._timeout):
stdout, stderr = await self.computer.interface.run_command(command)
return CLIResult(output=stdout or "", error=stderr or "")
result = await self.computer.interface.run_command(command)
return CLIResult(output=result.stdout or "", error=result.stderr or "")
except asyncio.TimeoutError as e:
raise ToolError(f"Command timed out after {self._timeout} seconds") from e
except Exception as e:

18
libs/python/agent/agent/providers/anthropic/tools/edit.py
View File

@@ -95,13 +95,13 @@ class EditTool(BaseEditTool, BaseAnthropicTool):
result = await self.computer.interface.run_command(
f'[ -e "{str(path)}" ] && echo "exists" || echo "not exists"'
)
exists = result[0].strip() == "exists"
exists = result.stdout.strip() == "exists"
if exists:
result = await self.computer.interface.run_command(
f'[ -d "{str(path)}" ] && echo "dir" || echo "file"'
)
is_dir = result[0].strip() == "dir"
is_dir = result.stdout.strip() == "dir"
else:
is_dir = False
@@ -126,7 +126,7 @@ class EditTool(BaseEditTool, BaseAnthropicTool):
result = await self.computer.interface.run_command(
f'[ -d "{str(path)}" ] && echo "dir" || echo "file"'
)
is_dir = result[0].strip() == "dir"
is_dir = result.stdout.strip() == "dir"
if is_dir:
if view_range:
@@ -136,7 +136,7 @@ class EditTool(BaseEditTool, BaseAnthropicTool):
# List directory contents using ls
result = await self.computer.interface.run_command(f'ls -la "{str(path)}"')
contents = result[0]
contents = result.stdout
if contents:
stdout = f"Here's the files and directories in {path}:\n{contents}\n"
else:
@@ -272,9 +272,9 @@ class EditTool(BaseEditTool, BaseAnthropicTool):
"""Read the content of a file using cat command."""
try:
result = await self.computer.interface.run_command(f'cat "{str(path)}"')
if result[1]: # If there's stderr output
raise ToolError(f"Error reading file: {result[1]}")
return result[0]
if result.stderr: # If there's stderr output
raise ToolError(f"Error reading file: {result.stderr}")
return result.stdout
except Exception as e:
raise ToolError(f"Failed to read {path}: {str(e)}")
@@ -291,8 +291,8 @@ class EditTool(BaseEditTool, BaseAnthropicTool):
{content}
EOFCUA"""
result = await self.computer.interface.run_command(cmd)
if result[1]: # If there's stderr output
raise ToolError(f"Error writing file: {result[1]}")
if result.stderr: # If there's stderr output
raise ToolError(f"Error writing file: {result.stderr}")
except Exception as e:
raise ToolError(f"Failed to write to {path}: {str(e)}")

2
libs/python/agent/pyproject.toml
View File

@@ -19,7 +19,7 @@ dependencies = [
"pydantic>=2.6.4",
"rich>=13.7.1",
"python-dotenv>=1.0.1",
"cua-computer>=0.2.0,<0.3.0",
"cua-computer>=0.3.0,<0.4.0",
"cua-core>=0.1.0,<0.2.0",
"certifi>=2024.2.2"
]

30
libs/python/computer/computer/interface/base.py
View File

@@ -3,8 +3,7 @@
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any, Tuple, List
from ..logger import Logger, LogLevel
from .models import MouseButton
from .models import MouseButton, CommandResult
class BaseComputerInterface(ABC):
"""Base class for computer control interfaces."""
@@ -234,8 +233,31 @@ class BaseComputerInterface(ABC):
pass
@abstractmethod
async def run_command(self, command: str) -> Tuple[str, str]:
"""Run shell command."""
async def run_command(self, command: str) -> CommandResult:
"""Run shell command and return structured result.
Executes a shell command using subprocess.run with shell=True and check=False.
The command is run in the target environment and captures both stdout and stderr.
Args:
command (str): The shell command to execute
Returns:
CommandResult: A structured result containing:
- stdout (str): Standard output from the command
- stderr (str): Standard error from the command
- returncode (int): Exit code from the command (0 indicates success)
Raises:
RuntimeError: If the command execution fails at the system level
Example:
result = await interface.run_command("ls -la")
if result.returncode == 0:
print(f"Output: {result.stdout}")
else:
print(f"Error: {result.stderr}, Exit code: {result.returncode}")
"""
pass
# Accessibility Actions

11
libs/python/computer/computer/interface/linux.py
View File

@@ -9,8 +9,7 @@ import websockets
from ..logger import Logger, LogLevel
from .base import BaseComputerInterface
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
from .models import Key, KeyType, MouseButton
from .models import Key, KeyType, MouseButton, CommandResult
class LinuxComputerInterface(BaseComputerInterface):
"""Interface for Linux."""
@@ -616,11 +615,15 @@ class LinuxComputerInterface(BaseComputerInterface):
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to delete directory"))
async def run_command(self, command: str) -> Tuple[str, str]:
async def run_command(self, command: str) -> CommandResult:
result = await self._send_command("run_command", {"command": command})
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to run command"))
return result.get("stdout", ""), result.get("stderr", "")
return CommandResult(
stdout=result.get("stdout", ""),
stderr=result.get("stderr", ""),
returncode=result.get("return_code", 0)
)
# Accessibility Actions
async def get_accessibility_tree(self) -> Dict[str, Any]:

11
libs/python/computer/computer/interface/macos.py
View File

@@ -9,8 +9,7 @@ import websockets
from ..logger import Logger, LogLevel
from .base import BaseComputerInterface
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
from .models import Key, KeyType, MouseButton
from .models import Key, KeyType, MouseButton, CommandResult
class MacOSComputerInterface(BaseComputerInterface):
"""Interface for macOS."""
@@ -623,11 +622,15 @@ class MacOSComputerInterface(BaseComputerInterface):
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to delete directory"))
async def run_command(self, command: str) -> Tuple[str, str]:
async def run_command(self, command: str) -> CommandResult:
result = await self._send_command("run_command", {"command": command})
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to run command"))
return result.get("stdout", ""), result.get("stderr", "")
return CommandResult(
stdout=result.get("stdout", ""),
stderr=result.get("stderr", ""),
returncode=result.get("return_code", 0)
)
# Accessibility Actions
async def get_accessibility_tree(self) -> Dict[str, Any]:

12
libs/python/computer/computer/interface/models.py
View File

@@ -1,5 +1,17 @@
from enum import Enum
from typing import Dict, List, Any, TypedDict, Union, Literal
from dataclasses import dataclass
@dataclass
class CommandResult:
stdout: str
stderr: str
returncode: int
def __init__(self, stdout: str, stderr: str, returncode: int):
self.stdout = stdout
self.stderr = stderr
self.returncode = returncode
# Navigation key literals
NavigationKey = Literal['pagedown', 'pageup', 'home', 'end', 'left', 'right', 'up', 'down']

11
libs/python/computer/computer/interface/windows.py
View File

@@ -9,8 +9,7 @@ import websockets
from ..logger import Logger, LogLevel
from .base import BaseComputerInterface
from ..utils import decode_base64_image, encode_base64_image, bytes_to_image, draw_box, resize_image
from .models import Key, KeyType, MouseButton
from .models import Key, KeyType, MouseButton, CommandResult
class WindowsComputerInterface(BaseComputerInterface):
"""Interface for Windows."""
@@ -615,11 +614,15 @@ class WindowsComputerInterface(BaseComputerInterface):
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to delete directory"))
async def run_command(self, command: str) -> Tuple[str, str]:
async def run_command(self, command: str) -> CommandResult:
result = await self._send_command("run_command", {"command": command})
if not result.get("success", False):
raise RuntimeError(result.get("error", "Failed to run command"))
return result.get("stdout", ""), result.get("stderr", "")
return CommandResult(
stdout=result.get("stdout", ""),
stderr=result.get("stderr", ""),
returncode=result.get("return_code", 0)
)
# Accessibility Actions
async def get_accessibility_tree(self) -> Dict[str, Any]:

2
libs/python/computer/pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "pdm.backend"
[project]
name = "cua-computer"
version = "0.2.0"
version = "0.3.0"
description = "Computer-Use Interface (CUI) framework powering Cua"
readme = "README.md"
authors = [