examples/tools/browsers/action/executor.py

# coding: utf-8
# Copyright (c) 2025 inclusionAI.

from typing import Tuple, List, Any

from aworld.core.tool.action_factory import ActionFactory
from aworld.core.common import ActionModel, ActionResult, Observation
from aworld.logs.util import logger
from aworld.core.tool.base import Tool, ToolActionExecutor


class BrowserToolActionExecutor(ToolActionExecutor):
    def __init__(self, tool: Tool = None):
        super(BrowserToolActionExecutor, self).__init__(tool)

    def execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[
        List[ActionResult], Any]:
        """Execute the specified browser action sequence by agent policy.

        Args:
            actions: Tool action sequence.

        Returns:
            Browser page and action result list.
        """
        action_results = []
        page = self.tool.page
        for action in actions:
            action_result, page = self._exec(action, **kwargs)
            action_results.append(action_result)
        return action_results, page

    async def async_execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[
        List[ActionResult], Any]:
        """Execute the specified browser action sequence by agent policy.

        Args:
            actions: Tool action sequence.

        Returns:
            Browser page and action result list.
        """
        action_results = []
        page = self.tool.page
        for action in actions:
            action_result, page = await self._async_exec(action, **kwargs)
            action_results.append(action_result)
        return action_results, page

    def _exec(self, action_model: ActionModel, **kwargs):
        action_name = action_model.action_name
        if action_name not in ActionFactory:
            raise ValueError(f'Action {action_name} not found')

        action = ActionFactory(action_name)
        action_result, page = action.act(action_model, page=self.tool.page, browser=self.tool.context, **kwargs)
        logger.info(f"{action_name} execute finished")
        return action_result, page

    async def _async_exec(self, action_model: ActionModel, **kwargs):
        action_name = action_model.action_name
        if action_name not in ActionFactory:
            action_name = action_model.tool_name + action_model.action_name
            if action_name not in ActionFactory:
                raise ValueError(f'Action {action_name} not found')

        action = ActionFactory(action_name)
        action_result, page = await action.async_act(action_model, page=self.tool.page, browser=self.tool.context, **kwargs)
        logger.info(f"{action_name} execute finished")
        return action_result, page