Rebase onto upstream (a4d95fd) #12

Closed
wylab wants to merge 144 commits from rebase-onto-upstream into main
3 changed files with 220 additions and 0 deletions
Showing only changes of commit cb7d39b221 - Show all commits
@@ -8,6 +8,7 @@ from nanobot.agent.tools.anthropic.base import (
)
from nanobot.agent.tools.anthropic.bash import BashTool20250124
from nanobot.agent.tools.anthropic.edit import EditTool20250728
from nanobot.agent.tools.anthropic.computer import ComputerTool20251124
__all__ = [
"BaseAnthropicTool",
@@ -16,4 +17,5 @@ __all__ = [
"ToolError",
"BashTool20250124",
"EditTool20250728",
"ComputerTool20251124",
]
+136
View File
@@ -0,0 +1,136 @@
"""Computer control tool for VNC desktop interaction.
Ported from anthropic-quickstarts/computer-use-demo.
"""
import base64
from typing import Literal
from loguru import logger
try:
from vncdotool.client import VNCDoToolClient
except ImportError:
VNCDoToolClient = None
from nanobot.agent.tools.anthropic.base import BaseAnthropicTool, ToolResult
class ComputerTool20251124(BaseAnthropicTool):
"""Computer control via VNC for desktop interaction.
Supports keyboard input, mouse control, and screenshots.
"""
api_type: Literal["computer_20251124"] = "computer_20251124"
name: Literal["computer"] = "computer"
beta_flag: str = "computer-use-2025-11-24"
def __init__(self, vnc_host: str = "172.17.0.1", vnc_port: int = 5900):
"""Initialize computer tool.
Args:
vnc_host: VNC server hostname/IP
vnc_port: VNC server port
"""
self.vnc_host = vnc_host
self.vnc_port = vnc_port
def to_params(self):
"""Return tool definition for API."""
return {
"type": self.api_type,
"name": self.name,
}
async def __call__(
self,
action: Literal[
"key", "type", "mouse_move", "left_click", "right_click",
"double_click", "screenshot", "cursor_position"
] | None = None,
coordinate: list[int] | None = None,
text: str | None = None,
**kwargs,
) -> ToolResult:
"""Execute computer control action.
Args:
action: Action to perform
coordinate: [x, y] coordinates for mouse actions
text: Text to type or key name to press
Returns:
ToolResult with action result or screenshot
"""
if not action:
return ToolResult(error="No action provided")
if VNCDoToolClient is None:
return ToolResult(
error="vncdotool is required for computer tool. "
"Install with: pip install vncdotool"
)
try:
async with VNCDoToolClient.create(self.vnc_host, self.vnc_port) as client:
if action == "screenshot":
return await self._screenshot(client)
elif action == "key":
return await self._key(client, text or "")
elif action == "type":
return await self._type(client, text or "")
elif action == "mouse_move":
return await self._mouse_move(client, coordinate or [0, 0])
elif action == "left_click":
return await self._left_click(client)
elif action == "right_click":
return await self._right_click(client)
elif action == "double_click":
return await self._double_click(client)
elif action == "cursor_position":
return ToolResult(output="Cursor position tracking not implemented")
else:
return ToolResult(error=f"Unknown action: {action}")
except Exception as e:
logger.error(f"Computer tool error: {e}")
return ToolResult(error=str(e))
async def _screenshot(self, client) -> ToolResult:
"""Capture screenshot."""
png_data = await client.captureScreen()
base64_data = base64.b64encode(png_data).decode()
return ToolResult(base64_image=base64_data)
async def _key(self, client, text: str) -> ToolResult:
"""Press a key."""
await client.keyPress(text)
return ToolResult(output=f"Pressed key: {text}")
async def _type(self, client, text: str) -> ToolResult:
"""Type text."""
await client.type(text)
return ToolResult(output=f"Typed: {text}")
async def _mouse_move(self, client, coordinate: list[int]) -> ToolResult:
"""Move mouse to coordinate."""
x, y = coordinate[0], coordinate[1]
await client.mouseMove(x, y)
return ToolResult(output=f"Moved mouse to ({x}, {y})")
async def _left_click(self, client) -> ToolResult:
"""Left click."""
await client.mousePress(1)
return ToolResult(output="Left clicked")
async def _right_click(self, client) -> ToolResult:
"""Right click."""
await client.mousePress(3)
return ToolResult(output="Right clicked")
async def _double_click(self, client) -> ToolResult:
"""Double click."""
await client.mousePress(1)
await client.mousePress(1)
return ToolResult(output="Double clicked")
+82
View File
@@ -0,0 +1,82 @@
"""Tests for ComputerTool20251124."""
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from nanobot.agent.tools.anthropic.computer import ComputerTool20251124
from nanobot.agent.tools.anthropic.base import ToolResult
@pytest.mark.asyncio
async def test_computer_tool_screenshot():
"""Test computer tool can take screenshot."""
tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
# Mock VNC client
with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
mock_client = AsyncMock()
mock_client.captureScreen = AsyncMock(return_value=b"fake_png_data")
# Set up async context manager
mock_context = MagicMock()
mock_context.__aenter__ = AsyncMock(return_value=mock_client)
mock_context.__aexit__ = AsyncMock(return_value=None)
mock_vnc.create = MagicMock(return_value=mock_context)
result = await tool(action="screenshot")
assert isinstance(result, ToolResult)
assert result.base64_image is not None
assert len(result.base64_image) > 0
@pytest.mark.asyncio
async def test_computer_tool_mouse_move():
"""Test computer tool can move mouse."""
tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
mock_client = AsyncMock()
mock_client.mouseMove = AsyncMock()
# Set up async context manager
mock_context = MagicMock()
mock_context.__aenter__ = AsyncMock(return_value=mock_client)
mock_context.__aexit__ = AsyncMock(return_value=None)
mock_vnc.create = MagicMock(return_value=mock_context)
result = await tool(action="mouse_move", coordinate=[100, 200])
assert isinstance(result, ToolResult)
assert result.error is None
mock_client.mouseMove.assert_called_once_with(100, 200)
@pytest.mark.asyncio
async def test_computer_tool_key():
"""Test computer tool can press keys."""
tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
mock_client = AsyncMock()
mock_client.keyPress = AsyncMock()
# Set up async context manager
mock_context = MagicMock()
mock_context.__aenter__ = AsyncMock(return_value=mock_client)
mock_context.__aexit__ = AsyncMock(return_value=None)
mock_vnc.create = MagicMock(return_value=mock_context)
result = await tool(action="key", text="Return")
assert isinstance(result, ToolResult)
assert result.error is None
mock_client.keyPress.assert_called_once_with("Return")
def test_computer_tool_to_params():
"""Test computer tool returns correct params."""
tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
params = tool.to_params()
assert params["type"] == "computer_20251124"
assert params["name"] == "computer"