Building a Computer-Use Agent in Python - Part 2: Multi-Agent Architecture

class BossAgent(BaseAgent):
    def __init__(self, ...):
        super().__init__(agent_name="boss", ...)
        self.subagents: Dict[str, Any] = {}

    def get_system_prompt(self) -> str:
        return """# Identity

You are a computer-use agent. You coordinate tasks by delegating to specialized agents:

## Available Agents

1. **BrowserBossAgent**: Browser automation and web interactions
   - Use for: navigating websites, filling forms, clicking web elements

2. **GUIAgent**: Mouse and keyboard interactions with GUI
   - Use for: clicking, typing, hotkeys, scrolling

3. **ShellAgent**: Executes shell commands
   - Use for: running terminal commands, file operations

4. **ResearchAgent**: Researches information using Tavily
   - Use for: searching the web, gathering information

## Your Responsibilities

1. Analyze the user's request and the current screenshot
2. For multi-step tasks: create a plan and get user approval first
3. Delegate manageable subtasks to sub-agents
4. Handle agent responses and coordinate multi-step workflows
5. Report results back to the user

## Response Format

{
  "thought": "Your reasoning about what to do next",
  "action": {
    "type": "delegate",
    "agent": "BrowserBossAgent",
    "message": "Open https://example.com and fill the login form"
  }
}
"""

def get_or_create_agent(self, agent_type: str) -> Any:
    """Get existing agent or create a new one"""
    if agent_type in self.subagents:
        return self.subagents[agent_type]

    if agent_type == "BrowserActionAgent":
        from agents.browser_action_agent import BrowserActionAgent
        agent = BrowserActionAgent(
            websocket_callback=self.websocket_callback,
            config_dict=self.config_dict
        )
    elif agent_type == "GUIAgent":
        from agents.gui_agent import GUIAgent
        agent = GUIAgent(
            websocket_callback=self.websocket_callback,
            config_dict=self.config_dict
        )
    # ... other agent types

    self.subagents[agent_type] = agent
    return agent

from playwright.async_api import async_playwright, Browser, BrowserContext, Page

class BrowserActionAgent(BaseAgent):
    def __init__(self, ...):
        super().__init__(agent_name="browser", ...)
        self.playwright = None
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None

    async def _launch(self, name: str = "chromium") -> Dict[str, Any]:
        """Launch a new browser"""
        if self.playwright is None:
            self.playwright = await async_playwright().start()

        self.browser = await self.playwright.chromium.launch(
            headless=False,
            args=[
                '--no-sandbox',
                '--disable-setuid-sandbox',
                '--disable-dev-shm-usage'
            ]
        )
        self.context = await self.browser.new_context(no_viewport=True)
        self.page = await self.context.new_page()

        return {"success": True, "message": "Browser launched"}

async def _click(self, xpath: str, button: str = "left", click_count: int = 1) -> Dict[str, Any]:
    """Click on an element by XPath"""
    if not self.page:
        return {"success": False, "error": "No active browser page"}

    element = await self.page.query_selector(f"xpath={xpath}")
    if not element:
        return {"success": False, "error": f"No element found: {xpath}"}

    await element.click(button=button, click_count=click_count)
    return {"success": True, "message": f"Clicked: {xpath}"}

async def _fill(self, xpath: str, text: str) -> Dict[str, Any]:
    """Fill an element with text"""
    element = await self.page.query_selector(f"xpath={xpath}")
    if not element:
        return {"success": False, "error": f"No element found: {xpath}"}

    await element.fill(text)
    return {"success": True, "message": f"Filled: {xpath}"}

def get_system_prompt(self) -> str:
    return """# Identity

You are a Browser Action Agent that automates browser interactions using Playwright.

# Available Tools

- launch(name): Launch browser ("chromium" or "firefox")
- navigate(url): Navigate to URL
- click(xpath): Click element by XPath
- fill(xpath, text): Fill element with text
- input_text(xpath, text): Type text character by character
- press_key(xpath, key): Press a key ("Enter", "Tab", "Escape")
- hover(xpath): Hover over element
- get_text(xpath): Get text content
- wait_for_element(xpath, timeout): Wait for element to appear
- scroll_into_view(xpath): Scroll element into view
- wait(seconds): Wait
- exit(summary, exitCode): Exit when finished

# Response Format

{
  "thought": "Your reasoning about what to do next",
  "action": {
    "tool": "click",
    "args": {"xpath": "//button[@id='submit']"}
  }
}
"""

async def execute_action(self, action: Dict[str, Any]) -> Dict[str, Any]:
    """Execute a browser action"""
    tool = action.get("tool")
    args = action.get("args", {})

    if tool == "launch":
        return await self._launch(**args)
    elif tool == "navigate":
        return await self._navigate(**args)
    elif tool == "click":
        return await self._click(**args)
    elif tool == "fill":
        return await self._fill(**args)
    elif tool == "press_key":
        return await self._press_key(**args)
    elif tool == "wait_for_element":
        return await self._wait_for_element(**args)
    elif tool == "exit":
        return self._exit(**args)
    else:
        return {"success": False, "error": f"Unknown tool: {tool}"}

# BossAgent builds context for worker
context_parts = []
if self.compacted_context:
    context_parts.append(f"Previous Context:\n{self.compacted_context}")
context_parts.append(f"User Request: {message.get('content', '')}")

# Include agent responses from previous steps
for agent_resp in message.get('agent_responses', []):
    agent_type = agent_resp.get("agent_type")
    response = agent_resp.get("response", {})
    context_parts.append(f"{agent_type} response: {response}")

# Check if compaction is needed
context_text = str(self.history)
word_count = count_words(context_text)

if self.step_count >= trigger_steps or word_count >= trigger_words:
    self.compacted_context = compact_context(
        self.history,
        task,
        self.config_dict,
        self.websocket_callback
    )
    self.history = []  # Clear after compaction
    self.step_count = 0

def _exit(self, summary: str = None, exitCode: int = 0) -> Dict[str, Any]:
    """Exit the agent"""
    return {
        "success": True,
        "exit": True,
        "summary": summary or "Agent completed",
        "exitCode": exitCode
    }

def send_llm_update(self, event_type: str, data: Dict[str, Any]):
    """Send update via WebSocket"""
    if self.websocket_callback:
        self.websocket_callback({
            "type": event_type,
            "agent_id": self.agent_id,
            "agent_type": self.__class__.__name__,
            "data": data
        })