Computer Use Agents: Building AI That Controls Your Screen

class ComputerUseAgent:
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.model = model
        self.screenshot_provider = ScreenshotProvider()
        self.action_executor = ActionExecutor()
        
    async def execute_task(self, task: str) -> Result:
        # 1. Capture current screen
        screenshot = await self.screenshot_provider.capture()
        
        # 2. Analyze with vision model
        analysis = await self.model.analyze_screen(screenshot, task)
        
        # 3. Generate action plan
        actions = await self.model.plan_actions(analysis, task)
        
        # 4. Execute actions iteratively
        for action in actions:
            await self.action_executor.execute(action)
            
            # 5. Verify result
            result = await self.verify_action(action)
            if not result.success:
                # Retry or adjust
                actions.extend(await self.model.recover(result.error))
                
        return await self.model.summarize_results(actions)

class UIElementDetector:
    def __init__(self):
        self.ocr = TesseractOCR()
        self.element_model = DetectionModel()
    
    async def detect_elements(self, screenshot: Image) -> List[UIElement]:
        # Method 1: OCR for text elements
        text_elements = await self.ocr.detect(screenshot)
        
        # Method 2: Vision model for buttons, inputs
        visual_elements = await self.element_model.detect(screenshot)
        
        # Method 3: Accessibility tree (when available)
        a11y_elements = await self.get_accessibility_tree()
        
        # Merge and deduplicate
        elements = self.merge_elements(text_elements, visual_elements, a11y_elements)
        
        return elements
    
    def merge_elements(self, *sources) -> List[UIElement]:
        # Combine detections, remove duplicates
        # Assign bounding boxes
        # Categorize as button, input, link, etc.
        pass

# Anthropic Computer Use API actions
COMPUTER_ACTIONS = {
    # Mouse actions
    "mouse_move": {"x": int, "y": int},
    "left_click": {"x": int, "y": int},
    "right_click": {"x": int, "y": int},
    "double_click": {"x": int, "y": int},
    "scroll_down": {"x": int, "y": int, "scroll_amount": int},
    "drag": {"from_x": int, "from_y": int, "to_x": int, "to_y": int},
    
    # Keyboard actions
    "type": {"text": str},
    "press_key": {"key": str},  # "enter", "escape", "backspace", etc.
    
    # Screen actions
    "screenshot": {},
    
    # Control
    "wait": {"seconds": int},
}

from anthropic import Anthropic

client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    tools=[
        {
            "name": "computer",
            "description": "Use the computer to perform tasks",
            "input_schema": {
                "type": "object",
                "properties": {
                    "action": {
                        "type": "string",
                        "enum": ["screenshot", "mouse_move", "click", "type", "press_key"]
                    },
                    "coordinate_x": {"type": "integer"},
                    "coordinate_y": {"type": "integer"},
                    "text": {"type": "string"},
                }
            }
        }
    ],
    messages=[
        {"role": "user", "content": "Go to example.com and search for AI"}
    ]
)

# Execute the recommended action
for block in response.content:
    if block.type == "tool_use":
        action = block.input
        execute_computer_action(action)

# Good: Break down complex tasks
async def book_flight():
    # Step 1: Open travel site
    await agent.act("screenshot")
    await agent.act("type", text="kayak.com")
    await agent.act("press_key", key="enter")
    
    # Step 2: Wait for load
    await agent.act("wait", seconds=2)
    await agent.act("screenshot")
    
    # Step 3: Fill form
    # ... continue step by step

# Bad: Too complex at once
await agent.act("Go to kayak.com, search flights to NYC for next Friday")

# Install dependencies
pip install opencv-python pytesseract pillow pyautogui

# Install OCR
sudo apt-get install tesseract-ocr

# For mouse/keyboard control
# macOS: Already available
# Linux: sudo apt-get install xdotool
# Windows: Built-in

import pyautogui
import time
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class Action:
    action_type: str
    x: Optional[int] = None
    y: Optional[int] = None
    text: Optional[str] = None
    key: Optional[str] = None

class ComputerController:
    def __init__(self):
        pyautogui.FAILSAFE = True
        pyautogui.PAUSE = 0.5
        
    def screenshot(self) -> Image:
        return pyautogui.screenshot()
    
    def move_mouse(self, x: int, y: int):
        pyautogui.moveTo(x, y)
        
    def click(self, x: int = None, y: int = None, button: str = "left"):
        if x is not None and y is not None:
            pyautogui.click(x, y, button=button)
        else:
            pyautogui.click(button=button)
    
    def type_text(self, text: str):
        pyautogui.write(text)
        
    def press_key(self, key: str):
        pyautogui.press(key)
        
    def scroll(self, amount: int):
        pyautogui.scroll(amount)

class ScreenAnalyzer:
    def __init__(self, vision_model):
        self.model = vision_model
        
    async def analyze(self, screenshot: Image, task: str) -> dict:
        # Convert to base64
        img_bytes = io.BytesIO()
        screenshot.save(img_bytes, format='PNG')
        img_b64 = base64.b64encode(img_bytes.getvalue()).decode()
        
        # Analyze with vision model
        prompt = f"""
        Analyze this screenshot for: {task}
        
        Identify:
        1. Interactive elements (buttons, inputs, links)
        2. Text content
        3. Layout structure
        
        Return a JSON with:
        - "elements": list of clickable elements with coordinates
        - "relevant_text": text that matches the task
        - "next_action": recommended action to progress
        """
        
        response = await self.model.analyze_image(img_b64, prompt)
        return json.loads(response)

class ComputerUseAgent:
    def __init__(self):
        self.controller = ComputerController()
        self.analyzer = ScreenAnalyzer(VisionModel())
        self.max_iterations = 20
        self.history = []
        
    async def execute(self, task: str) -> dict:
        for iteration in range(self.max_iterations):
            # 1. Capture screen
            screenshot = self.controller.screenshot()
            
            # 2. Analyze
            analysis = await self.analyzer.analyze(screenshot, task)
            
            # 3. Decide action
            action = self.decide_action(analysis, task)
            
            if not action:
                # Task complete
                return {"status": "success", "history": self.history}
                
            # 4. Execute
            self.controller.execute(action)
            self.history.append(action)
            
            # 5. Small delay for UI to update
            time.sleep(1)
            
        return {"status": "timeout", "history": self.history}
    
    def decide_action(self, analysis: dict, task: str) -> Optional[Action]:
        # Use LLM to decide best next action
        # based on analysis and task
        pass

class ElementGrounder:
    def __init__(self):
        self.ocr = EasyOCR()
        self.element_matcher = TemplateMatcher()
    
    async def ground(self, screenshot: Image, predicted_elements: List[dict]) -> List[dict]:
        # Get all OCR text with positions
        ocr_results = await self.ocr.readtext(screenshot)
        
        grounded = []
        for pred in predicted_elements:
            # Find matching OCR element
            match = self.find_match(pred, ocr_results)
            if match:
                grounded.append({
                    **pred,
                    "actual_bbox": match["bbox"],
                    "confidence": match["confidence"]
                })
        
        return grounded
    
    def find_match(self, pred, ocr_results):
        # Match predicted element to OCR result
        # using text similarity and position
        pass

class RecoveryManager:
    def __init__(self, agent):
        self.agent = agent
        self.error_patterns = {
            "not_clicked": self.retry_click,
            "wrong_page": self.navigate_back,
            "timeout": self.wait_and_retry,
            "element_missing": self.scroll_and_find,
        }
    
    async def handle_error(self, error: Exception, context: dict):
        error_type = self.classify_error(error)
        recovery = self.error_patterns.get(error_type, self.generic_recovery)
        return await recovery(error, context)
    
    async def retry_click(self, error, context):
        # Re-capture and try slightly different position
        screenshot = self.agent.controller.screenshot()
        # Try again with adjusted coordinates
        pass
    
    async def scroll_and_find(self, error, context):
        # Scroll to find missing element
        self.agent.controller.scroll(-500)
        await asyncio.sleep(1)
        # Retry

class TabManager:
    def __init__(self, controller):
        self.controller = controller
        self.tabs = []
        
    async def new_tab(self, url: str):
        # Ctrl+T
        self.controller.press_key("ctrl+t")
        await asyncio.sleep(0.5)
        
        # Type URL
        self.controller.type_text(url)
        self.controller.press_key("enter")
        
        self.tabs.append(url)
    
    async def switch_tab(self, index: int):
        # Ctrl+1-9
        self.controller.press_key(f"ctrl+{index}")
        
    async def close_tab(self):
        self.controller.press_key("ctrl+w")
        self.tabs.pop()

# AI-powered E2E testing
async def test_login_flow():
    agent = ComputerUseAgent()
    
    # Navigate to app
    await agent.execute("Open the login page at https://app.example.com")
    
    # Fill credentials
    await agent.execute("Enter '[email protected]' in the email field")
    await agent.execute("Enter 'password123' in the password field")
    
    # Click login
    await agent.execute("Click the login button")
    
    # Verify
    await agent.execute("Confirm we're on the dashboard by checking for the user menu")

# Fill forms from data
async def fill_spreadsheet(data: List[dict]):
    agent = ComputerUseAgent()
    
    # Open spreadsheet
    await agent.execute("Open Google Sheets")
    
    for row in data:
        # Enter each field
        await agent.execute(f"Type '{row['name']}' in column A")
        await agent.execute(f"Type '{row['email']}' in column B")
        await agent.execute("Press tab to move to next row")

# Scrape dynamic content
async def scrape_dynamic_site(url: str):
    agent = ComputerUseAgent()
    
    # Navigate
    await agent.execute(f"Go to {url}")
    
    # Scroll to load all content
    for _ in range(10):
        await agent.execute("Scroll down to load more content")
        await asyncio.sleep(2)
    
    # Extract data
    screenshot = agent.controller.screenshot()
    text = extract_text(screenshot)
    
    return parse_data(text)

# Apply to jobs automatically
async def apply_to_jobs(jobs: List[dict]):
    agent = ComputerUseAgent()
    
    for job in jobs:
        # Navigate to job posting
        await agent.execute(f"Open {job['url']}")
        
        # Click apply
        await agent.execute("Click the Apply button")
        
        # Fill form
        await agent.execute(f"Enter '{job['name']}' in name field")
        await agent.execute(f"Enter '{job['email']}' in email field")
        await agent.execute("Upload resume from ~/resume.pdf")
        
        # Submit
        await agent.execute("Click submit")

# Safety guards
class SafetyGuard:
    def __init__(self):
        self.blocked_domains = ["bankofamerica.com", "chase.com"]
        self.blocked_actions = ["transfer", "send money", "delete account"]
        
    async def check(self, task: str, url: str) -> bool:
        # Block sensitive actions
        if any(blocked in task.lower() for blocked in self.blocked_actions):
            raise SafetyError(f"Blocked action: {task}")
            
        # Block sensitive sites
        if any(blocked in url for blocked in self.blocked_domains):
            raise SafetyError(f"Blocked domain: {url}")
            
        return True
    
    # Rate limiting
    async def check_rate_limit(self, user_id: str) -> bool:
        # Limit actions per minute
        pass