Upload 7 files

aworld/core/tool/action.py

@@ -0,0 +1,54 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+from enum import Enum, EnumMeta
+from typing import Tuple, Any
+
+from aworld.core.common import ToolActionInfo, ActionModel, ActionResult
+
+
+class ExecutableAction(object):
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    def act(self, action: ActionModel, **kwargs) -> Tuple[ActionResult, Any]:
+        """Execute the action."""
+
+    @abc.abstractmethod
+    async def async_act(self, action: ActionModel, **kwargs) -> Tuple[ActionResult, Any]:
+        """Execute the action."""
+
+
+class DynamicEnumMeta(EnumMeta):
+    def __new__(metacls, cls, bases, classdict):
+        for name, value in classdict.items():
+            if isinstance(value, tuple) and len(value) == 2:
+                classdict[name] = (value[0], value[1])
+        return super().__new__(metacls, cls, bases, classdict)
+
+
+class ToolAction(Enum, metaclass=DynamicEnumMeta):
+    @classmethod
+    def get_value_by_name(cls, name: str) -> ToolActionInfo | None:
+        members = cls.members()
+        name = name.upper()
+        if name in members:
+            if hasattr(members[name], 'value'):
+                return members[name].value
+            else:
+                return members[name]
+        return None
+
+    @classmethod
+    def members(cls):
+        return dict(filter(lambda item: not item[0].startswith("_"), cls.__dict__.items()))
+
+
+TOOL_ACTION = """
+from aworld.config.tool_action import ToolAction
+from aworld.core.common import ToolActionInfo, ParamInfo
+
+class {name}Action(ToolAction):
+    '''{name} action enum.'''
+    # ERROR = ToolActionInfo(name="error", desc="action error")
+"""

aworld/core/tool/action_factory.py

@@ -0,0 +1,110 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+import logging
+import sys
+from typing import Dict, List
+
+from aworld.core.factory import Factory
+from aworld.logs.util import logger
+
+
+class ActionManager(Factory):
+    def __init__(self, type_name: str = None):
+        super(ActionManager, self).__init__(type_name)
+        self._tool_action_mapping: Dict[str, str] = {}
+        self._tool_action_cache: Dict[str, List[str]] = {}
+
+    def __call__(self, name: str, tool_name: str = None, **kwargs):
+        if name is None:
+            raise ValueError("action name is None")
+        if tool_name is None:
+            tool_name = ""
+
+        final_name = tool_name + name
+        try:
+            if final_name in self._cls:
+                act = self._cls[final_name](**kwargs)
+            else:
+                raise RuntimeError("The action was not registered.\nPlease confirm the package has been imported.")
+        except Exception:
+            err = sys.exc_info()
+            logging.warning("Failed to create action with name '%s':\n%s" % (name, err[1]), exc_info=err)
+            act = None
+
+        if act is None:
+            act = UnknownAction(name=name, **kwargs)
+            act.name = name
+        return act
+
+    def register(self, name: str, desc: str, **kwargs):
+        def func(cls):
+            tool_name = kwargs.get("tool_name")
+            if not tool_name:
+                logger.warning("tool name is empty")
+                tool_name = ""
+            if name in self._cls and self._tool_action_mapping.get(name) == tool_name:
+                logger.warning(f"{tool_name} tool {name} action already in {self._type} factory, will override it.")
+
+            self._tool_action_mapping[name] = tool_name
+            final_name = tool_name + name
+            self._cls[final_name] = cls
+            self._desc[final_name] = desc
+            self._ext_info[final_name] = kwargs
+            return cls
+
+        return func
+
+    def get_actions_by_tool(self, tool_name: str):
+        if tool_name in self._tool_action_cache:
+            return self._tool_action_cache[tool_name]
+
+        actions = []
+        for action_name, v in self._tool_action_mapping.items():
+            if tool_name == v:
+                actions.append(action_name)
+        self._tool_action_cache[tool_name] = actions
+        return actions
+
+
+ActionFactory = ActionManager("action_type")
+
+
+class UnknownAction(object):
+    def __init__(self, name: str, *args, **kwargs):
+        self.options = {}
+        self._args = args
+        self._kwargs = kwargs
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, t, v, traceback):
+        pass
+
+    def act(self, *args, **kwds):
+        """Perform optimization and return an SolverResults object."""
+        self._solver_error('act')
+
+    async def async_act(self, *args, **kwds):
+        self._solver_error('async_act')
+
+    def reset(self):
+        """Reset the state of an optimizer"""
+        self._solver_error('reset')
+
+    def set_options(self, istr):
+        """Set the options in the optimizer from a string."""
+        self._solver_error('set_options')
+
+    def __bool__(self):
+        return self.available()
+
+    def __getattr__(self, attr):
+        self._solver_error(attr)
+
+    def _action_error(self, method_name):
+        raise RuntimeError(
+            f"""Attempting to use an unavailable action. The ActionFactory was unable to create the 
+action "{self.type}" and returned an UnknownAction object. This error is raised at the point where 
+the UnknownAction object was used as if it were valid (by calling method "{method_name}").""")

aworld/core/tool/action_template.py

@@ -0,0 +1,45 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+# need action's name, desc, tool_name and func/async_func
+
+ACTION_TEMPLATE = """
+import traceback
+from typing import Tuple, Any, List, Dict
+
+from aworld.core.envs.action_factory import ActionFactory
+from aworld.core.common import ActionModel, ActionResult
+from aworld.logs.util import logger
+from aworld.utils.async_func import async_func
+from aworld.virtual_environments.action import ExecutableAction
+
+
+@ActionFactory.register(name="{name}",
+                        desc="{desc}",
+                        tool_name="{tool_name}")
+class {name}(ExecutableAction):
+    # only for function to tool.
+    def act(self, action: ActionModel, **kwargs) -> Tuple[ActionResult, Any]:
+        {func_import}import {func}
+        try:
+            res = {func}(**action.params)
+            if not res:
+                raise ValueError(f"{func} no result return.")
+            return ActionResult(content=res, success=True), None
+        except Exception as e:
+            logger.error(traceback.format_exc())
+            return ActionResult(content=str(e), error=str(e)), None
+        
+
+    async def async_act(self, action: ActionModel, **kwargs) -> Tuple[ActionResult, Any]:
+        {func_import}import {func}
+        
+        try:
+            res = await {call_func}(**action.params)
+            if not res:
+                raise ValueError(f"{func} no result return.")
+            return ActionResult(content=res, success=True), None
+        except Exception as e:
+            logger.error(traceback.format_exc())
+            return ActionResult(content=str(e), error=str(e)), None
+"""

aworld/core/tool/base.py

@@ -0,0 +1,539 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+import abc
+import traceback
+from typing import Dict, Tuple, Any, TypeVar, Generic, List, Union
+
+from pydantic import BaseModel
+
+from aworld.config.conf import ToolConfig, load_config, ConfigDict
+from aworld.core.event import eventbus
+from aworld.core.tool.action import ToolAction
+from aworld.core.tool.action_factory import ActionFactory
+from aworld.core.common import Observation, ActionModel, ActionResult
+from aworld.core.context.base import Context
+from aworld.core.event.base import Message, ToolMessage, AgentMessage, Constants
+from aworld.core.factory import Factory
+from aworld.events.util import send_message
+from aworld.logs.util import logger
+from aworld.models.model_response import ToolCall
+from aworld.output import ToolResultOutput
+from aworld.output.base import StepOutput
+from aworld.utils.common import convert_to_snake, sync_exec
+
+AgentInput = TypeVar("AgentInput")
+ToolInput = TypeVar("ToolInput")
+
+
+class BaseTool(Generic[AgentInput, ToolInput]):
+    """The basic generic classes of tools in the environment, with two parameterized types: AgentInput and ToolInput.
+
+    We follow the gym/gymnasium protocol to be compatible with gym games, can also build special env tool in the framework.
+    """
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, conf: Union[Dict[str, Any], ConfigDict, ToolConfig], **kwargs) -> None:
+        self.conf = conf
+        if isinstance(conf, ConfigDict):
+            pass
+        elif isinstance(conf, Dict):
+            self.conf = ConfigDict(conf)
+        elif isinstance(conf, ToolConfig):
+            # To add flexibility
+            self.conf = ConfigDict(conf.model_dump())
+        else:
+            logger.warning(f"Unknown conf type: {type(conf)}")
+        self._finished = False
+
+        self._name = kwargs.pop('name', self.conf.get("name", convert_to_snake(self.__class__.__name__)))
+        action_executor.register(name=self.name(), tool=self)
+        self.action_executor = action_executor
+        self.event_driven = kwargs.pop('event_driven', self.conf.get('event_driven', False))
+        self.handler = kwargs.get('handler', self.conf.get('handler', None))
+
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+    def _init_context(self, context: Context):
+        self.context = context
+
+    def name(self):
+        """Tool unique name."""
+        return self._name
+
+    def pre_step(self, action: ToolInput, **kwargs):
+        pass
+
+    def post_step(self,
+                  step_res: Tuple[AgentInput, float, bool, bool, Dict[str, Any]],
+                  action: ToolInput,
+                  **kwargs) -> Message:
+        pass
+
+    def step(self, message: Message, **kwargs) -> Message:
+        self._init_context(message.context)
+        action = message.payload
+        self.pre_step(action, **kwargs)
+        res = self.do_step(action, **kwargs)
+        final_res = self.post_step(res, action, **kwargs)
+        return final_res
+
+    @abc.abstractmethod
+    def reset(self, *, seed: int | None = None, options: Dict[str, str] | None = None) -> Tuple[
+        AgentInput, dict[str, Any]]:
+        """Resets the initial internal state, returning an initial state and extended info."""
+
+    @abc.abstractmethod
+    def do_step(self, action: ToolInput, **kwargs) -> Tuple[AgentInput, float, bool, bool, Dict[str, Any]]:
+        """Run one step of the tool's in env using the actions.
+
+        Args:
+            action(ToolInput): Actions provided by the agent to update the observation.
+        Return:
+            Quintuple，key information: AgentInput and extended info dict.
+        """
+
+    @property
+    def finished(self) -> bool:
+        """The final execution status of the task from agent instructions."""
+        return self._finished
+
+    @abc.abstractmethod
+    def close(self) -> None:
+        """Close the tool resources in the environment."""
+
+    def render(self):
+        """For interface compatibility."""
+        pass
+
+
+class AsyncBaseTool(Generic[AgentInput, ToolInput]):
+    """The basic generic classes of tools in the environment, with two parameterized types: AgentInput and ToolInput.
+
+    We follow the gym/gymnasium protocol to be compatible with gym games, can also build special env tool in the framework.
+    """
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, conf: Union[Dict[str, Any], ConfigDict, ToolConfig], **kwargs) -> None:
+        self.conf = conf
+        if isinstance(conf, ConfigDict):
+            pass
+        elif isinstance(conf, Dict):
+            self.conf = ConfigDict(conf)
+        elif isinstance(conf, ToolConfig):
+            # To add flexibility
+            self.conf = ConfigDict(conf.model_dump())
+        else:
+            logger.warning(f"Unknown conf type: {type(conf)}")
+        self._finished = False
+
+        self._name = kwargs.pop('name', self.conf.get("name", convert_to_snake(self.__class__.__name__)))
+        action_executor.register(name=self.name(), tool=self)
+        self.action_executor = action_executor
+        self.event_driven = kwargs.pop('event_driven', self.conf.get('event_driven', False))
+        self.handler = kwargs.get('handler', self.conf.get('handler', None))
+
+        for k, v in kwargs.items():
+            setattr(self, k, v)
+
+    def _init_context(self, context: Context):
+        self.context = context
+
+    def name(self):
+        """Tool unique name."""
+        return self._name
+
+    async def pre_step(self, action: ToolInput, **kwargs):
+        pass
+
+    async def post_step(self,
+                        step_res: Tuple[AgentInput, float, bool, bool, Dict[str, Any]],
+                        action: ToolInput,
+                        **kwargs) -> Message:
+        pass
+
+    async def step(self, message: Message, **kwargs) -> Message:
+        self._init_context(message.context)
+        action = message.payload
+        await self.pre_step(action, **kwargs)
+        res = await self.do_step(action, **kwargs)
+        final_res = await self.post_step(res, action, **kwargs)
+        return final_res
+
+    @abc.abstractmethod
+    async def reset(self, *, seed: int | None = None, options: Dict[str, str] | None = None) -> Tuple[
+        AgentInput, dict[str, Any]]:
+        """Resets the initial internal state, returning an initial state and extended info."""
+
+    @abc.abstractmethod
+    async def do_step(self, action: ToolInput, **kwargs) -> Tuple[AgentInput, float, bool, bool, Dict[str, Any]]:
+        """Run one step of the tool's in env using the actions.
+
+        Args:
+            action(ToolInput): Actions provided by the agent to update the observation.
+        Return:
+            Quintuple，key information: AgentInput and extended info dict.
+        """
+
+    @property
+    def finished(self) -> bool:
+        """The final execution status of the task from agent instructions."""
+        return self._finished
+
+    @abc.abstractmethod
+    async def close(self) -> None:
+        """Close the tool resources in the environment."""
+
+    async def render(self):
+        """For interface compatibility."""
+        pass
+
+
+class Tool(BaseTool[Observation, List[ActionModel]]):
+    def _internal_process(self, step_res: Tuple[AgentInput, float, bool, bool, Dict[str, Any]],
+                          action: ToolInput,
+                          **kwargs):
+        if not step_res or not action:
+            return
+        for idx, act in enumerate(action):
+            if eventbus is not None:
+                tool_output = ToolResultOutput(
+                    tool_type=kwargs.get("tool_id_mapping", {}).get(act.tool_id) or self.name(),
+                    tool_name=act.tool_name,
+                    data=step_res[0].content,
+                    origin_tool_call=ToolCall.from_dict({
+                        "function": {
+                            "name": act.action_name,
+                            "arguments": act.params,
+                        }
+                    }),
+                    metadata=step_res[0].action_result[idx].metadata
+                )
+                tool_output_message = Message(
+                    category=Constants.OUTPUT,
+                    payload=tool_output,
+                    sender=self.name(),
+                    session_id=self.context.session_id if self.context else "",
+                    headers={"context": self.context}
+                )
+                sync_exec(send_message, tool_output_message)
+
+    def step(self, message: Message, **kwargs) -> Message:
+        self._init_context(message.context)
+        action = message.payload
+        tool_id_mapping = {}
+        for act in action:
+            tool_id = act.tool_id
+            tool_name = act.tool_name
+            tool_id_mapping[tool_id] = tool_name
+        self.pre_step(action, **kwargs)
+        res = self.do_step(action, **kwargs)
+        final_res = self.post_step(res, action, **kwargs)
+        self._internal_process(res, action, tool_id_mapping=tool_id_mapping, **kwargs)
+        return final_res
+
+    def post_step(self,
+                  step_res: Tuple[Observation, float, bool, bool, Dict[str, Any]],
+                  action: List[ActionModel],
+                  **kwargs) -> Tuple[Observation, float, bool, bool, Dict[str, Any]] | Message:
+        if not step_res:
+            raise Exception(f'{self.name()} no observation has been made.')
+
+        step_res[0].from_agent_name = action[0].agent_name
+        for idx, act in enumerate(action):
+            step_res[0].action_result[idx].tool_id = act.tool_id
+
+        agent = self.context.swarm.agents.get(action[0].agent_name)
+        feedback_tool_result = agent.feedback_tool_result if agent else False
+        if feedback_tool_result:
+            return AgentMessage(payload=step_res,
+                                caller=action[0].agent_name,
+                                sender=self.name(),
+                                receiver=action[0].agent_name,
+                                session_id=self.context.session_id,
+                                headers={"context": self.context})
+        else:
+            return AgentMessage(payload=step_res,
+                                sender=action[0].agent_name,
+                                session_id=self.context.session_id,
+                                headers={"context": self.context})
+
+
+class AsyncTool(AsyncBaseTool[Observation, List[ActionModel]]):
+    async def _internal_process(self, step_res: Tuple[AgentInput, float, bool, bool, Dict[str, Any]],
+                                action: ToolInput,
+                                **kwargs):
+        for idx, act in enumerate(action):
+            # send tool results output
+            if eventbus is not None:
+                tool_output = ToolResultOutput(
+                    tool_type=kwargs.get("tool_id_mapping", {}).get(act.tool_id) or self.name(),
+                    tool_name=act.tool_name,
+                    data=step_res[0].content,
+                    origin_tool_call=ToolCall.from_dict({
+                        "function": {
+                            "name": act.action_name,
+                            "arguments": act.params,
+                        }
+                    }),
+                    metadata=step_res[0].action_result[idx].metadata
+                )
+                tool_output_message = Message(
+                    category=Constants.OUTPUT,
+                    payload=tool_output,
+                    sender=self.name(),
+                    session_id=self.context.session_id if self.context else "",
+                    headers={"context": self.context}
+                )
+                await send_message(tool_output_message)
+
+        await send_message(Message(
+            category=Constants.OUTPUT,
+            payload=StepOutput.build_finished_output(name=f"{action[0].agent_name if action else ''}",
+                                                     step_num=0),
+            sender=self.name(),
+            receiver=action[0].agent_name,
+            session_id=self.context.session_id if self.context else "",
+            headers={"context": self.context}
+        ))
+
+    async def step(self, message: Message, **kwargs) -> Message:
+        self._init_context(message.context)
+        action = message.payload
+        tool_id_mapping = {}
+        for act in action:
+            tool_id = act.tool_id
+            tool_name = act.tool_name
+            tool_id_mapping[tool_id] = tool_name
+        await self.pre_step(action, **kwargs)
+        res = await self.do_step(action, **kwargs)
+        final_res = await self.post_step(res, action, **kwargs)
+        await self._internal_process(res, action, tool_id_mapping=tool_id_mapping, **kwargs)
+        return final_res
+
+    async def post_step(self,
+                        step_res: Tuple[Observation, float, bool, bool, Dict[str, Any]],
+                        action: List[ActionModel],
+                        **kwargs) -> Tuple[Observation, float, bool, bool, Dict[str, Any]] | Message:
+        if not step_res:
+            raise Exception(f'{self.name()} no observation has been made.')
+
+        step_res[0].from_agent_name = action[0].agent_name
+        for idx, act in enumerate(action):
+            step_res[0].action_result[idx].tool_id = act.tool_id
+
+        agent = self.context.swarm.agents.get(action[0].agent_name)
+        feedback_tool_result = agent.feedback_tool_result if agent else False
+        if feedback_tool_result:
+            return AgentMessage(payload=step_res,
+                                caller=action[0].agent_name,
+                                sender=self.name(),
+                                receiver=action[0].agent_name,
+                                session_id=Context.instance().session_id,
+                                headers={"context": self.context})
+        else:
+            return AgentMessage(payload=step_res,
+                                sender=action[0].agent_name,
+                                session_id=Context.instance().session_id,
+                                headers={"context": self.context})
+
+
+class ToolsManager(Factory):
+    def __init__(self, type_name: str = None):
+        super(ToolsManager, self).__init__(type_name)
+        self._tool_with_action = {}
+        self._tool_conf = {}
+        self._tool_instance = {}
+
+    def __iter__(self):
+        for name in self._cls:
+            name = "async_" + name if self._asyn.get(name, False) else name
+            yield name
+
+    def __contains__(self, name: str) -> bool:
+        """Whether the name in the factory."""
+        name = "async_" + name if self._asyn.get(name, False) else name
+        return name in self._cls
+
+    def __call__(self, name: str = None, *args, **kwargs):
+        if name is None:
+            return self
+
+        asyn = kwargs.pop("asyn", False)
+        name = "async_" + name if asyn else name
+
+        conf = self._tool_conf.get(name)
+        if not conf:
+            logger.warning(f"{name} not find conf in tool factory")
+            conf = dict()
+        elif isinstance(conf, BaseModel):
+            conf = conf.model_dump()
+
+        user_conf = kwargs.pop('conf', None)
+        if user_conf:
+            if isinstance(user_conf, BaseModel):
+                conf.update(user_conf.model_dump())
+            elif isinstance(user_conf, dict):
+                conf.update(user_conf)
+            else:
+                logger.warning(f"Unknown conf type: {type(user_conf)}, ignored!")
+        self._tool_conf[name] = conf
+
+        # must is a dict
+        conf['name'] = name
+        conf = ConfigDict(conf)
+
+        if kwargs.get("reuse", conf.get('reuse', False)) is True and name in self._tool_instance:
+            return self._tool_instance[name]
+
+        if name in self._cls:
+            tool = self._cls[name](conf=conf, **kwargs)
+            self._tool_instance[name] = tool
+        else:
+            raise RuntimeError(f"can not find {name} tool in the ToolFactory, register it first.")
+
+        action_executor.register(name, tool)
+        return tool
+
+    def get_tool_action(self, tool: str, asyn: bool = False):
+        if asyn:
+            tool = "async_" + tool
+        return self._tool_with_action.get(tool)
+
+    def register(self, name: str, desc: str, supported_action: ToolAction = None, conf_file_name: str = None, **kwargs):
+        """Register a tool to tool factory.
+
+        Args:
+            name: Tool name
+            desc: Tool description
+            supported_action: Tool abilities
+            conf_file_name: Default tool config
+        """
+        res = super(ToolsManager, self).register(name, desc, **kwargs)
+        asyn = kwargs.pop("asyn", False)
+        prefix = "async_" if asyn else ""
+        conf_file_name = conf_file_name if conf_file_name else f"{name}_tool.yaml"
+        conf = load_config(conf_file_name, kwargs.get("dir"))
+        if not conf:
+            logger.debug(f"can not load conf from {conf_file_name}")
+            # use general tool config
+            conf = ToolConfig().model_dump()
+        name = prefix + name
+        self._tool_with_action[name] = supported_action
+        self._tool_conf[name] = conf
+        logger.debug(f"{name} register to the tool factory.")
+        return res
+
+
+ToolFactory = ToolsManager("env_tool_type")
+
+
+class ToolActionExecutor(object):
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, tool: Tool = None):
+        self.tool = tool
+        self.tools: Dict[str, Tool] = {}
+
+    def register(
+            self,
+            name: str,
+            tool: Union[Tool, AsyncTool]):
+        self.tools[name] = tool
+
+    @abc.abstractmethod
+    def execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[List[ActionResult], Any]:
+        """"""
+        return self.execute_env_action(actions, self.tool, **kwargs)
+
+    @abc.abstractmethod
+    async def async_execute_action(self, actions: List[ActionModel], **kwargs) -> Tuple[List[ActionResult], Any]:
+        """"""
+        return await self.async_execute_env_action(actions, self.tool, **kwargs)
+
+    @abc.abstractmethod
+    def execute_env_action(self,
+                           actions: List[ActionModel],
+                           tool: Tool,
+                           **kwargs) -> Tuple[List[ActionResult], Any]:
+        """"""
+        action_results = []
+        ctx = None
+        for action in actions:
+            if action is None:
+                logger.warning("empty action, ignore it.")
+                continue
+
+            if tool is None:
+                tool_name = action.tool_name
+                tool = self.tools.get(tool_name)
+                if tool is None:
+                    tool = ToolFactory(tool_name, conf=kwargs.get("conf", ToolConfig()))
+                    self.tools[tool_name] = tool
+
+            try:
+                action_result, ctx = self.do_act(action, tool, **kwargs)
+            except:
+                logger.warning(traceback.format_exc())
+                action_result = ActionResult(error=traceback.format_exc(), success=False)
+            action_result.action_name = action.action_name
+            action_result.tool_name = action.tool_name
+            action_results.append(action_result)
+        return action_results, ctx
+
+    async def async_execute_env_action(self,
+                                       actions: List[ActionModel],
+                                       tool: Tool,
+                                       **kwargs) -> Tuple[List[ActionResult], Any]:
+        """"""
+        action_results = []
+        ctx = None
+        for action in actions:
+            if action is None:
+                logger.warning("empty action, ignore it.")
+                continue
+
+            if tool is None:
+                tool_name = "async_" + action.tool_name
+                tool = self.tools.get(tool_name)
+                if tool is None:
+                    tool = ToolFactory(tool_name, conf=kwargs.get("conf", ToolConfig()))
+                    self.tools[tool_name] = tool
+            try:
+                action_result, ctx = await self.async_do_act(action, tool, **kwargs)
+            except:
+                logger.warning(traceback.format_exc())
+                action_result = ActionResult(error=traceback.format_exc(), success=False)
+            action_result.action_name = action.action_name
+            action_result.tool_name = action.tool_name
+            action_results.append(action_result)
+        return action_results, ctx
+
+    def do_act(self, action_model: ActionModel, tool: Tool, **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_model.action_name} not found in ActionFactory')
+
+        action = ActionFactory(action_name)
+        action_result, page = action.act(action_model, tool=tool, **kwargs)
+        logger.info(f"{tool.name()}-{action_model.action_name} execute finished")
+        return action_result, page
+
+    async def async_do_act(self, action_model: ActionModel, tool: Tool,
+                           **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_model.action_name} not found in ActionFactory')
+
+        action = ActionFactory(action_name)
+        action_result, page = await action.async_act(action_model, tool=tool, **kwargs)
+        logger.info(f"{tool.name()}-{action_model.action_name} execute finished")
+        return action_result, page
+
+
+action_executor = ToolActionExecutor()

aworld/core/tool/func_to_tool.py

@@ -0,0 +1,270 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import importlib
+import inspect
+import os
+import sys
+
+from typing import Callable, Any, get_type_hints, get_origin, get_args
+
+from pydantic import create_model, Field, BaseModel
+from pydantic.fields import FieldInfo
+
+from aworld.core.common import ToolActionInfo, ParamInfo
+from aworld.core.tool.action import TOOL_ACTION
+from aworld.core.tool.action_factory import ActionFactory
+from aworld.core.tool.action_template import ACTION_TEMPLATE
+from aworld.core.tool.base import ToolFactory
+from aworld.core.tool.tool_template import TOOL_TEMPLATE
+from aworld.logs.util import logger
+
+
+def be_tool(
+        tool_name: str = None,
+        tool_desc: str = None,
+        name: str = None,
+        desc: str = None, **kwargs
+) -> Callable[..., Any]:
+    """Decorate a function to be a tool, auto register the tool and action with the parameters with the factory.
+
+    Example:
+        >>> @be_tool()
+        >>> def example():
+        >>>     return "example"
+
+        # write name and description
+        >>> @be_tool(name="param", desc="example param func")
+        >>> def example_param(param: str):
+        >>>     return param
+
+        >>> @be_tool(tool_name='field_param_tool')
+        >>> def example_param_field(param: str = Field(..., description="param")):
+        >>>     return param
+
+        >>> @be_tool(tool_name='field_param_tool')
+        >>> def example_param_2(param: str = Field(..., description="param")):
+        >>>     return param
+
+    Args:
+        tool_name: Optional name for the tool.
+        tool_desc: Optional description for the tool.
+        name: Optional name for the function.
+        desc: Optional description of function.
+    """
+
+    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
+        # converts python function into a tool with associated actions
+        function_to_tool(
+            func,
+            tool_name=tool_name,
+            tool_desc=tool_desc,
+            name=name,
+            desc=desc,
+            **kwargs
+        )
+        return func
+
+    return decorator
+
+
+def function_to_tool(
+        func: Callable[..., Any],
+        tool_name: str = None,
+        tool_desc: str = None,
+        name: str = None,
+        desc: str = None,
+        **kwargs
+) -> None:
+    """Transform a python function into a tool, and register to the factory.
+
+    Generates necessary code files dynamically and manages tool and action registration.
+
+    Args：
+        func: An executable function.
+        tool_name: The name of the tool that is transformed by the function.
+        tool_desc: The description of the tool what it can do, default the same as tool_name.
+        name: Alias name of function.
+        desc: The description of the function what it can do, default the same as name.
+    """
+    tool_name = tool_name or name or func.__name__
+    action_name = name or func.__name__
+
+    if tool_name == "<lambda>" or action_name == "<lambda>":
+        raise ValueError("You must provide a name for lambda functions")
+
+    # async func, will use AsyncTool
+    is_async = inspect.iscoroutinefunction(func)
+
+    name = action_name
+    if not inspect.iscoroutinefunction(func):
+        name = f"async_func({action_name})"
+
+    # build action
+    if action_name not in ActionFactory:
+        func_import = func.__module__
+        if func_import == '__main__':
+            path = inspect.getsourcefile(func)
+            package = path.replace(os.getcwd(), '').replace('.py', '')
+            if package[0] == '/':
+                package = package[1:]
+            func_import = f"from {package} "
+        else:
+            func_import = f"from {func_import} "
+        con = ACTION_TEMPLATE.format(name=action_name,
+                                     desc=desc if desc else action_name,
+                                     tool_name=tool_name,
+                                     func_import=func_import,
+                                     func=func.__name__,
+                                     call_func=name)
+        with open(f"{action_name}.py", 'w+') as write:
+            write.writelines(con)
+        module = importlib.import_module(action_name)
+        getattr(module, action_name)
+
+        if not kwargs.get('keep_file', False):
+            os.remove(f"{action_name}.py")
+    else:
+        logger.warning(f"{action_name} already register to the tool.")
+        raise ValueError(f"{action_name} already register to a tool.")
+
+    # build params info
+    parameters = func_params(func)
+
+    module_name = f'{tool_name}_action'
+    if module_name not in sys.modules:
+        # ToolAction process
+        with open(f"{module_name}.py", 'w+') as write:
+            write.writelines(TOOL_ACTION.format(name=tool_name))
+        module = importlib.import_module(module_name)
+        if not kwargs.get('keep_file', False):
+            os.remove(f"{module_name}.py")
+        tool_action_cls = getattr(module, f"{tool_name}Action")
+    else:
+        logger.info(f"{module_name} already exists in modules, reuse the tool action.")
+        tool_action_cls = getattr(sys.modules[module_name], f"{tool_name}Action")
+
+    params = {}
+    if parameters:
+        for k, v in parameters['properties'].items():
+            params[k] = ParamInfo(name=k,
+                                  type=v.get('type', 'string'),
+                                  required=False if v.get('default') else True,
+                                  default_value=v.get('default'),
+                                  desc=v.get('description', k))
+
+    setattr(tool_action_cls,
+            action_name.upper(),
+            ToolActionInfo(
+                name=action_name,
+                desc=desc if desc else action_name,
+                input_params=params
+            ))
+
+    # build tool
+    if tool_name not in ToolFactory:
+        con = TOOL_TEMPLATE.format(name=tool_name,
+                                   desc=tool_desc if tool_desc else tool_name,
+                                   action=f"{tool_name}Action",
+                                   action_import=f"from {tool_action_cls.__module__} import {tool_name}Action",
+                                   cls='AsyncTool' if is_async else 'Tool',
+                                   async_flag='async ' if is_async else '',
+                                   async_underline='async_' if is_async else '',
+                                   await_flag='await ' if is_async else '')
+
+        if tool_name == action_name:
+            tool_name += '_tool'
+
+        with open(f"{tool_name}.py", 'w+') as write:
+            write.writelines(con)
+        importlib.import_module(tool_name)
+
+        if not kwargs.get('keep_file', False):
+            os.remove(f"{tool_name}.py")
+
+
+def func_params(func: Callable[..., Any]):
+    """Extracts parameter information from the function.
+
+    Args:
+        func: An executable function.
+
+    Returns:
+        JSON schema of the function input parameters.
+    """
+    sig = inspect.signature(func)
+    type_hints = get_type_hints(func)
+    filtered_params = []
+
+    # The function must have a return value
+    if sig.return_annotation == inspect.Parameter.empty:
+        raise RuntimeError(f"{func} no return value, preferably a string.")
+
+    for name, param in sig.parameters.items():
+        filtered_params.append((name, param))
+
+    fields: dict[str, Any] = {}
+    param_descs = {}
+
+    for name, param in filtered_params:
+        ann = type_hints.get(name, param.annotation)
+        default = param.default
+        def_desc = None
+        if hasattr(default, 'description'):
+            def_desc = default.description
+        field_description = param_descs.get(name, def_desc)
+        
+        if isinstance(default, FieldInfo):
+            default = default.default
+
+        # If there's no type hint, assume `Any`
+        if ann == inspect.Parameter.empty:
+            ann = Any
+
+        # Handle different parameter kinds
+        if param.kind == param.VAR_POSITIONAL:
+            # e.g. *args: extend positional args
+            if get_origin(ann) is tuple:
+                # e.g. def foo(*args: tuple[int, ...]) -> treat as List[int]
+                args_of_tuple = get_args(ann)
+                if len(args_of_tuple) == 2 and args_of_tuple[1] is Ellipsis:
+                    ann = list[args_of_tuple[0]]  # type: ignore
+                else:
+                    ann = list[Any]
+            else:
+                # If user wrote *args: int, treat as List[int]
+                ann = list[ann]  # type: ignore
+
+            # Default factory to empty list
+            fields[name] = (
+                ann,
+                Field(default_factory=list, description=field_description),  # type: ignore
+            )
+        elif param.kind == param.VAR_KEYWORD:
+            # **kwargs handling
+            if get_origin(ann) is dict:
+                # e.g. def foo(**kwargs: dict[str, int])
+                dict_args = get_args(ann)
+                if len(dict_args) == 2:
+                    ann = dict[dict_args[0], dict_args[1]]  # type: ignore
+                else:
+                    ann = dict[str, Any]
+            else:
+                # e.g. def foo(**kwargs: int) -> Dict[str, int]
+                ann = dict[str, ann]
+
+            fields[name] = (
+                ann,
+                Field(default_factory=dict, description=field_description),  # type: ignore
+            )
+        else:
+            if default == inspect.Parameter.empty:
+                # Required field
+                fields[name] = (ann, Field(..., description=field_description))
+            else:
+                # Parameter with a default value
+                fields[name] = (ann, Field(default=default, description=field_description))
+
+    dynamic_model = create_model(f"{func.__name__}".upper(), __base__=BaseModel, **fields)
+    json_schema = dynamic_model.model_json_schema()
+    logger.debug(f"{func} parameters schema: {json_schema}")
+    return json_schema

aworld/core/tool/tool_desc.py

@@ -0,0 +1,121 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+from typing import List, Dict
+
+from aworld.core.tool.action_factory import ActionFactory
+
+from aworld.core.tool.base import ToolFactory
+from aworld.logs.util import logger
+
+
+def tool_action_desc():
+    """Utility method of generate description of tools and their actions.
+
+    The standard protocol can be transformed based on the API of different llm.
+    Define as follows:
+    ```
+    {
+        "tool_name": {
+            "desc": "A toolkit description.",
+            "actions": [
+                {
+                    "name": "action name",
+                    "desc": "action description.",
+                    "params": {
+                        "param_name": {
+                            "desc": "param description.",
+                            "type": "param type, such as int, str, etc.",
+                            "required": True | False
+                        }
+                    }
+                }
+            ]
+        }
+    }
+    ```
+    """
+
+    def process(action_info):
+        action_dict = dict()
+        action_dict["name"] = action_info.name
+        action_dict["desc"] = action_info.desc
+        action_dict["params"] = dict()
+
+        for k, v in action_info.input_params.items():
+            params_dict = v.model_dump()
+            params_dict.pop("name")
+            action_dict["params"][k] = params_dict
+        return action_dict
+
+    descs = dict()
+    for tool in ToolFactory:
+        tool_val_dict = dict()
+        descs[tool] = tool_val_dict
+
+        tool_val_dict["desc"] = ToolFactory.desc(tool)
+        tool_action = ToolFactory.get_tool_action(tool)
+        actions = []
+        action_names = ActionFactory.get_actions_by_tool(tool_name=tool)
+        if action_names:
+            for action_name in action_names:
+                info = tool_action.get_value_by_name(action_name)
+                if not info:
+                    logger.warning(f"{action_name} can not find in {tool}, please check it.")
+                    continue
+                try:
+                    action_dict = process(info)
+                except:
+                    logger.warning(f"{action_name} process fail.")
+                    action_dict = dict()
+                actions.append(action_dict)
+        elif tool_action:
+            for k, info in tool_action.__members__.items():
+                action_dict = process(info.value)
+                actions.append(action_dict)
+        else:
+            logger.debug(f"{tool} no action!")
+        tool_val_dict["actions"] = actions
+    return descs
+
+
+def get_actions() -> List[str]:
+    res = []
+    for _, tool_info in tool_action_desc().items():
+        actions = tool_info.get("actions")
+        if not actions:
+            continue
+
+        for action in actions:
+            res.append(action['name'])
+    return res
+
+
+def get_actions_by_tools(tool_names: Dict = None) -> List[str]:
+    if not tool_names:
+        return get_actions()
+
+    res = []
+    for tool_name, tool_info in tool_action_desc().items():
+        if tool_name not in tool_names:
+            continue
+
+        actions = tool_info.get("actions")
+        if not actions:
+            continue
+
+        for action in actions:
+            res.append(action['name'])
+    return res
+
+
+def get_tool_desc():
+    return tool_action_desc()
+
+
+def get_tool_desc_by_name(name: str):
+    return tool_action_desc().get(name, None)
+
+
+def is_tool_by_name(name: str) -> bool:
+    return name in ToolFactory

aworld/core/tool/tool_template.py

@@ -0,0 +1,82 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+# need tool's name, desc and cls params
+TOOL_TEMPLATE = """
+import json
+from typing import List, Tuple, Dict, Any, Union
+
+from aworld.config.conf import ToolConfig, ConfigDict
+from aworld.core.envs.tool import Tool, AsyncTool, ToolFactory
+from aworld.core.common import Observation, ActionModel, ActionResult
+from aworld.logs.util import logger
+from aworld.virtual_environments.utils import build_observation
+{action_import}
+
+
+@ToolFactory.register(name="{name}", desc="{desc}", supported_action={action})
+class {name}Tool({cls}[Observation, List[ActionModel]]):
+    def __init__(self, conf: Union[Dict[str, Any], ConfigDict, ToolConfig], **kwargs) -> None:
+        super().__init__(conf, **kwargs)
+
+    {async_flag}def reset(self, *, seed: int | None = None, options: Dict[str, str] | None = None) -> Tuple[
+        Observation, dict[str, Any]]:
+        # from options obtain user query
+        return build_observation(observer=self.name(),
+                                 ability=''), dict()
+
+    {async_flag}def step(self, action: List[ActionModel], **kwargs) -> Tuple[Observation, float, bool, bool, Dict[str, Any]]:
+        reward = 0
+        fail_error = ""
+        action_result = None
+
+        invalid_acts: List[int] = []
+        for i, act in enumerate(action):
+            if act.tool_name != "{name}":
+                invalid_acts.append(i)
+
+        if invalid_acts:
+            for i in invalid_acts:
+                action[i] = None
+
+        resp = ""
+        try:
+            action_result, resp = {await_flag}self.action_executor.{async_underline}execute_action(action, **kwargs)
+            reward = 1
+        except Exception as e:
+            fail_error = str(e)
+
+        terminated = kwargs.get("terminated", False)
+        if action_result:
+            for res in action_result:
+                if res.is_done:
+                    terminated = res.is_done
+
+        info = dict()
+        info['exception'] = fail_error
+        info.update(kwargs)
+        if resp:
+            resp = json.dumps(resp)
+        else:
+            resp = action_result[0].content
+
+        action_result = [ActionResult(content=resp, keep=True, is_done=True)]
+        observation = build_observation(observer=self.name(),
+                                        ability=action[-1].action_name,
+                                        content=resp,
+                                        info=info)
+        observation.action_result = action_result
+        self._finished = True
+        return (observation,
+                reward,
+                terminated,
+                kwargs.get("truncated", False),
+                dict())
+
+    {async_flag}def close(self) -> None:
+        pass
+
+    {async_flag}def finished(self) -> bool:
+        # one time
+        return True
+"""