Upload 5 files

aworld/runners/handler/agent.py

@@ -0,0 +1,447 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+from typing import AsyncGenerator, Tuple
+
+from aworld.agents.loop_llm_agent import LoopableAgent
+from aworld.core.agent.base import is_agent, AgentFactory
+from aworld.core.agent.swarm import GraphBuildType
+from aworld.core.common import ActionModel, Observation, TaskItem
+from aworld.core.event.base import Message, Constants, TopicType
+from aworld.logs.util import logger
+from aworld.runners.handler.base import DefaultHandler
+from aworld.runners.handler.tool import DefaultToolHandler
+from aworld.runners.utils import endless_detect
+from aworld.output.base import StepOutput
+
+
+class AgentHandler(DefaultHandler):
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, runner: 'TaskEventRunner'):
+        self.swarm = runner.swarm
+        self.endless_threshold = runner.endless_threshold
+
+        self.agent_calls = []
+
+    @classmethod
+    def name(cls):
+        return "_agents_handler"
+
+
+class DefaultAgentHandler(AgentHandler):
+    async def handle(self, message: Message) -> AsyncGenerator[Message, None]:
+        if message.category != Constants.AGENT:
+            if message.sender in self.swarm.agents and message.sender in AgentFactory:
+                if self.agent_calls:
+                    if self.agent_calls[-1] != message.sender:
+                        self.agent_calls.append(message.sender)
+                else:
+                    self.agent_calls.append(message.sender)
+            return
+
+        headers = {"context": message.context}
+        session_id = message.session_id
+        data = message.payload
+        if not data:
+            # error message, p2p
+            yield Message(
+                category=Constants.OUTPUT,
+                payload=StepOutput.build_failed_output(name=f"{message.caller or self.name()}",
+                                                       step_num=0,
+                                                       data="no data to process."),
+                sender=self.name(),
+                session_id=session_id,
+                headers=headers
+            )
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(msg="no data to process.", data=data, stop=True),
+                sender=self.name(),
+                session_id=session_id,
+                topic=TopicType.ERROR,
+                headers=headers
+            )
+            return
+
+        if isinstance(data, Tuple) and isinstance(data[0], Observation):
+            data = data[0]
+            message.payload = data
+        # data is Observation
+        if isinstance(data, Observation):
+            if not self.swarm:
+                msg = Message(
+                    category=Constants.TASK,
+                    payload=data.content,
+                    sender=data.observer,
+                    session_id=session_id,
+                    topic=TopicType.FINISHED,
+                    headers=headers
+                )
+                logger.info(f"agent handler send finished message: {msg}")
+                yield msg
+                return
+
+            agent = self.swarm.agents.get(message.receiver)
+            # agent + tool completion protocol.
+            if agent and agent.finished and data.info.get('done'):
+                self.swarm.cur_step += 1
+                if agent.id() == self.swarm.communicate_agent.id():
+                    msg = Message(
+                        category=Constants.TASK,
+                        payload=data.content,
+                        sender=agent.id(),
+                        session_id=session_id,
+                        topic=TopicType.FINISHED,
+                        headers=headers
+                    )
+                    logger.info(f"agent handler send finished message: {msg}")
+                    yield msg
+                else:
+                    msg = Message(
+                        category=Constants.AGENT,
+                        payload=Observation(content=data.content),
+                        sender=agent.id(),
+                        session_id=session_id,
+                        receiver=self.swarm.communicate_agent.id(),
+                        headers=headers
+                    )
+                    logger.info(f"agent handler send agent message: {msg}")
+                    yield msg
+            else:
+                if data.info.get('done'):
+                    agent_name = self.agent_calls[-1]
+                    async for event in self._stop_check(ActionModel(agent_name=agent_name, policy_info=data.content),
+                                                        message):
+                        yield event
+                    return
+                logger.info(f"agent handler send observation message: {message}")
+                yield message
+            return
+
+        # data is List[ActionModel]
+        for action in data:
+            if not isinstance(action, ActionModel):
+                # error message, p2p
+                yield Message(
+                    category=Constants.OUTPUT,
+                    payload=StepOutput.build_failed_output(name=f"{message.caller or self.name()}",
+                                                           step_num=0,
+                                                           data="action not a ActionModel."),
+                    sender=self.name(),
+                    session_id=session_id,
+                    headers=headers
+                )
+                msg = Message(
+                    category=Constants.TASK,
+                    payload=TaskItem(msg="action not a ActionModel.", data=data, stop=True),
+                    sender=self.name(),
+                    session_id=session_id,
+                    topic=TopicType.ERROR,
+                    headers=headers
+                )
+                logger.info(f"agent handler send task message: {msg}")
+                yield msg
+                return
+
+        tools = []
+        agents = []
+        for action in data:
+            if is_agent(action):
+                agents.append(action)
+            else:
+                tools.append(action)
+
+        if tools:
+            msg = Message(
+                category=Constants.TOOL,
+                payload=tools,
+                sender=self.name(),
+                session_id=session_id,
+                receiver=DefaultToolHandler.name(),
+                headers=headers
+            )
+            logger.info(f"agent handler send tool message: {msg}")
+            yield msg
+        else:
+            yield Message(
+                category=Constants.OUTPUT,
+                payload=StepOutput.build_finished_output(name=f"{message.caller or self.name()}",
+                                                         step_num=0),
+                sender=self.name(),
+                receiver=agents[0].tool_name,
+                session_id=session_id,
+                headers=headers
+            )
+
+        for agent in agents:
+            async for event in self._agent(agent, message):
+                logger.info(f"agent handler send message: {event}")
+                yield event
+
+    async def _agent(self, action: ActionModel, message: Message):
+        self.agent_calls.append(action.agent_name)
+        agent = self.swarm.agents.get(action.agent_name)
+        # be handoff
+        agent_name = action.tool_name
+        if not agent_name:
+            async for event in self._stop_check(action, message):
+                yield event
+            return
+
+        headers = {"context": message.context}
+        session_id = message.session_id
+        cur_agent = self.swarm.agents.get(agent_name)
+        if not cur_agent or not agent:
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(msg=f"Can not find {agent_name} or {action.agent_name} agent in swarm.",
+                                 data=action,
+                                 stop=True),
+                sender=self.name(),
+                session_id=session_id,
+                topic=TopicType.ERROR,
+                headers=headers
+            )
+            return
+
+        cur_agent._finished = False
+        con = action.policy_info
+        if action.params and 'content' in action.params:
+            con = action.params['content']
+        observation = Observation(content=con, observer=agent.id(), from_agent_name=agent.id())
+
+        if agent.handoffs and agent_name not in agent.handoffs:
+            if message.caller:
+                message.receiver = message.caller
+                message.caller = ''
+                yield message
+            else:
+                yield Message(category=Constants.TASK,
+                              payload=TaskItem(msg=f"Can not handoffs {agent_name} agent ", data=observation),
+                              sender=self.name(),
+                              session_id=session_id,
+                              topic=TopicType.RERUN,
+                              headers=headers)
+            return
+
+        yield Message(
+            category=Constants.AGENT,
+            payload=observation,
+            caller=message.caller,
+            sender=action.agent_name,
+            session_id=session_id,
+            receiver=action.tool_name,
+            headers=headers
+        )
+
+    async def _stop_check(self, action: ActionModel, message: Message) -> AsyncGenerator[Message, None]:
+        if GraphBuildType.WORKFLOW.value != self.swarm.build_type:
+            async for event in self._social_stop_check(action, message):
+                yield event
+        else:
+            if self.swarm.has_cycle:
+                async for event in self._loop_sequence_stop_check(action, message):
+                    yield event
+            else:
+                async for event in self._sequence_stop_check(action, message):
+                    yield event
+
+    async def _sequence_stop_check(self, action: ActionModel, message: Message) -> AsyncGenerator[Message, None]:
+        headers = {"context": message.context}
+        session_id = message.session_id
+        agent = self.swarm.agents.get(action.agent_name)
+        ordered_agents = self.swarm.ordered_agents
+        idx = next((i for i, x in enumerate(ordered_agents) if x == agent), -1)
+        if idx == -1:
+            yield Message(
+                category=Constants.TASK,
+                payload=action,
+                sender=self.name(),
+                session_id=session_id,
+                topic=TopicType.ERROR,
+                headers=headers
+            )
+            return
+
+        # The last agent
+        if idx == len(self.swarm.ordered_agents) - 1:
+            receiver = None
+            # agent loop
+            if isinstance(agent, LoopableAgent):
+                agent.cur_run_times += 1
+                if not agent.finished:
+                    receiver = agent.goto
+
+            if receiver:
+                yield Message(
+                    category=Constants.AGENT,
+                    payload=Observation(content=action.policy_info),
+                    sender=agent.id(),
+                    session_id=session_id,
+                    receiver=receiver,
+                    headers=headers
+                )
+            else:
+                logger.info(f"execute loop {self.swarm.cur_step}.")
+                yield Message(
+                    category=Constants.TASK,
+                    payload=action.policy_info,
+                    sender=agent.id(),
+                    session_id=session_id,
+                    topic=TopicType.FINISHED,
+                    headers=headers
+                )
+            return
+
+            # loop agent type
+        if isinstance(agent, LoopableAgent):
+            agent.cur_run_times += 1
+            if agent.finished:
+                receiver = self.swarm.ordered_agents[idx + 1].id()
+            else:
+                receiver = agent.goto
+        else:
+            # means the loop finished
+            receiver = self.swarm.ordered_agents[idx + 1].id()
+        yield Message(
+            category=Constants.AGENT,
+            payload=Observation(content=action.policy_info),
+            sender=agent.id(),
+            session_id=session_id,
+            receiver=receiver,
+            headers=headers
+        )
+
+    async def _loop_sequence_stop_check(self, action: ActionModel, message: Message) -> AsyncGenerator[Message, None]:
+        headers = {"context": message.context}
+        session_id = message.session_id
+        agent = self.swarm.agents.get(action.agent_name)
+        idx = next((i for i, x in enumerate(self.swarm.ordered_agents) if x == agent), -1)
+        if idx == -1:
+            # unknown agent, means something wrong
+            yield Message(
+                category=Constants.TASK,
+                payload=action,
+                sender=self.name(),
+                session_id=session_id,
+                topic=TopicType.ERROR,
+                headers=headers
+            )
+            return
+        if idx == len(self.swarm.ordered_agents) - 1:
+            # supported sequence loop
+            if self.swarm.cur_step >= self.swarm.max_steps:
+                receiver = None
+                # agent loop
+                if isinstance(agent, LoopableAgent):
+                    agent.cur_run_times += 1
+                    if not agent.finished:
+                        receiver = agent.goto
+
+                if receiver:
+                    yield Message(
+                        category=Constants.AGENT,
+                        payload=Observation(content=action.policy_info),
+                        sender=agent.id(),
+                        session_id=session_id,
+                        receiver=receiver,
+                        headers=headers
+                    )
+                else:
+                    # means the task finished
+                    yield Message(
+                        category=Constants.TASK,
+                        payload=action.policy_info,
+                        sender=agent.id(),
+                        session_id=session_id,
+                        topic=TopicType.FINISHED,
+                        headers=headers
+                    )
+            else:
+                self.swarm.cur_step += 1
+                logger.info(f"execute loop {self.swarm.cur_step}.")
+                yield Message(
+                    category=Constants.TASK,
+                    payload='',
+                    sender=agent.id(),
+                    session_id=session_id,
+                    topic=TopicType.START,
+                    headers=headers
+                )
+            return
+
+        if isinstance(agent, LoopableAgent):
+            agent.cur_run_times += 1
+            if agent.finished:
+                receiver = self.swarm.ordered_agents[idx + 1].id()
+            else:
+                receiver = agent.goto
+        else:
+            # means the loop finished
+            receiver = self.swarm.ordered_agents[idx + 1].id()
+        yield Message(
+            category=Constants.AGENT,
+            payload=Observation(content=action.policy_info),
+            sender=agent.name(),
+            session_id=session_id,
+            receiver=receiver,
+            headers=headers
+        )
+
+    async def _social_stop_check(self, action: ActionModel, message: Message) -> AsyncGenerator[Message, None]:
+        headers = {"context": message.context}
+        agent = self.swarm.agents.get(action.agent_name)
+        caller = message.caller
+        session_id = message.session_id
+        if endless_detect(self.agent_calls,
+                          endless_threshold=self.endless_threshold,
+                          root_agent_name=self.swarm.communicate_agent.id()):
+            yield Message(
+                category=Constants.TASK,
+                payload=action.policy_info,
+                sender=agent.id(),
+                session_id=session_id,
+                topic=TopicType.FINISHED,
+                headers=headers
+            )
+            return
+
+        if not caller or caller == self.swarm.communicate_agent.id():
+            if self.swarm.cur_step >= self.swarm.max_steps or self.swarm.finished:
+                yield Message(
+                    category=Constants.TASK,
+                    payload=action.policy_info,
+                    sender=agent.id(),
+                    session_id=session_id,
+                    topic=TopicType.FINISHED,
+                    headers=headers
+                )
+            else:
+                self.swarm.cur_step += 1
+                logger.info(f"execute loop {self.swarm.cur_step}.")
+                yield Message(
+                    category=Constants.AGENT,
+                    payload=Observation(content=action.policy_info),
+                    sender=agent.id(),
+                    session_id=session_id,
+                    receiver=self.swarm.communicate_agent.id(),
+                    headers=headers
+                )
+        else:
+            idx = 0
+            for idx, name in enumerate(self.agent_calls[::-1]):
+                if name == agent.id():
+                    break
+            idx = len(self.agent_calls) - idx - 1
+            if idx:
+                caller = self.agent_calls[idx - 1]
+
+            yield Message(
+                category=Constants.AGENT,
+                payload=Observation(content=action.policy_info),
+                sender=agent.id(),
+                session_id=session_id,
+                receiver=caller,
+                headers=headers
+            )

aworld/runners/handler/base.py

@@ -0,0 +1,31 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+
+from typing import TypeVar, Generic, AsyncGenerator
+
+from aworld.core.event.base import Message
+
+IN = TypeVar('IN')
+OUT = TypeVar('OUT')
+
+
+class Handler(Generic[IN, OUT]):
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractmethod
+    async def handle(self, data: IN) -> AsyncGenerator[OUT, None]:
+        """Process the data as the expected result.
+
+        Args:
+            data: Data generated while running the task.
+        """
+
+    @classmethod
+    def name(cls):
+        """Handler name."""
+        return cls.__name__
+
+
+class DefaultHandler(Handler[Message, AsyncGenerator[Message, None]]):
+    """Default handler."""

aworld/runners/handler/output.py

@@ -0,0 +1,66 @@
+# aworld/runners/handler/output.py
+import json
+from typing import AsyncGenerator
+from aworld.core.task import TaskResponse
+from aworld.models.model_response import ModelResponse
+from aworld.runners.handler.base import DefaultHandler
+from aworld.output.base import StepOutput, MessageOutput, ToolResultOutput, Output
+from aworld.core.common import TaskItem
+from aworld.core.context.base import Context
+from aworld.core.event.base import Message, Constants, TopicType
+from aworld.logs.util import logger
+
+
+class DefaultOutputHandler(DefaultHandler):
+    def __init__(self, runner):
+        self.runner = runner
+
+    async def handle(self, message):
+        if message.category != Constants.OUTPUT:
+            return
+        # 1. get outputs
+        outputs = self.runner.task.outputs
+        if not outputs:
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(msg="Cannot get outputs.", data=message, stop=True),
+                sender=self.name(),
+                session_id=Context.instance().session_id,
+                topic=TopicType.ERROR,
+                headers={"context": message.context}
+            )
+            return
+        # 2. build Output
+        payload = message.payload
+        mark_complete = False
+        output = None
+        try:
+            if isinstance(payload, Output):
+                output = payload
+            elif isinstance(payload, TaskResponse):
+                logger.info(f"output get task_response with usage: {json.dumps(payload.usage)}")
+                if message.topic == TopicType.FINISHED or message.topic == TopicType.ERROR:
+                    mark_complete = True
+            elif isinstance(payload, ModelResponse) or isinstance(payload, AsyncGenerator):
+                output = MessageOutput(source=payload)
+        except Exception as e:
+            logger.warning(f"Failed to parse output: {e}")
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(msg="Failed to parse output.", data=payload, stop=True),
+                sender=self.name(),
+                session_id=Context.instance().session_id,
+                topic=TopicType.ERROR,
+                headers={"context": message.context}
+            )
+        finally:
+            if output:
+                if not output.metadata:
+                    output.metadata = {}
+                output.metadata['sender'] = message.sender
+                output.metadata['receiver'] = message.receiver
+                await outputs.add_output(output)
+            if mark_complete:
+                await outputs.mark_completed()
+
+        return

aworld/runners/handler/task.py

@@ -0,0 +1,131 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+import time
+
+from typing import AsyncGenerator
+
+from aworld.core.common import TaskItem
+from aworld.core.tool.base import Tool, AsyncTool
+
+from aworld.core.event.base import Message, Constants, TopicType
+from aworld.core.task import TaskResponse
+from aworld.logs.util import logger
+from aworld.output import Output
+from aworld.runners.handler.base import DefaultHandler
+from aworld.runners.hook.hook_factory import HookFactory
+from aworld.runners.hook.hooks import HookPoint
+
+
+class TaskHandler(DefaultHandler):
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, runner: 'TaskEventRunner'):
+        self.runner = runner
+        self.retry_count = 0
+        self.hooks = {}
+        if runner.task.hooks:
+            for k, vals in runner.task.hooks.items():
+                self.hooks[k] = []
+                for v in vals:
+                    cls = HookFactory.get_class(v)
+                    if cls:
+                        self.hooks[k].append(cls)
+
+    @classmethod
+    def name(cls):
+        return "_task_handler"
+
+
+class DefaultTaskHandler(TaskHandler):
+    async def handle(self, message: Message) -> AsyncGenerator[Message, None]:
+        if message.category != Constants.TASK:
+            return
+
+        logger.info(f"task handler receive message: {message}")
+
+        headers = {"context": message.context}
+        topic = message.topic
+        task_item: TaskItem = message.payload
+        if topic == TopicType.SUBSCRIBE_TOOL:
+            new_tools = message.payload.data
+            for name, tool in new_tools.items():
+                if isinstance(tool, Tool) or isinstance(tool, AsyncTool):
+                    await self.runner.event_mng.register(Constants.TOOL, name, tool.step)
+                    logger.info(f"dynamic register {name} tool.")
+                else:
+                    logger.warning(f"Unknown tool instance: {tool}")
+            return
+        elif topic == TopicType.SUBSCRIBE_AGENT:
+            return
+        elif topic == TopicType.ERROR:
+            async for event in self.run_hooks(message, HookPoint.ERROR):
+                yield event
+
+            if task_item.stop:
+                await self.runner.stop()
+                logger.warning(f"task {self.runner.task.id} stop, cause: {task_item.msg}")
+                self.runner._task_response = TaskResponse(msg=task_item.msg,
+                                                          answer='',
+                                                          success=False,
+                                                          id=self.runner.task.id,
+                                                          time_cost=(time.time() - self.runner.start_time),
+                                                          usage=self.runner.context.token_usage)
+                return
+            # restart
+            logger.warning(f"The task {self.runner.task.id} will be restarted due to error: {task_item.msg}.")
+            if self.retry_count >= 3:
+                raise Exception(f"The task {self.runner.task.id} failed, due to error: {task_item.msg}.")
+
+            self.retry_count += 1
+            yield Message(
+                category=Constants.TASK,
+                payload='',
+                sender=self.name(),
+                session_id=self.runner.context.session_id,
+                topic=TopicType.START,
+                headers=headers
+            )
+        elif topic == TopicType.FINISHED:
+            async for event in self.run_hooks(message, HookPoint.FINISHED):
+                yield event
+
+            self.runner._task_response = TaskResponse(answer=str(message.payload),
+                                                      success=True,
+                                                      id=self.runner.task.id,
+                                                      time_cost=(time.time() - self.runner.start_time),
+                                                      usage=self.runner.context.token_usage)
+            await self.runner.stop()
+
+            logger.info(f"{self.runner.task.id} finished.")
+        elif topic == TopicType.START:
+            async for event in self.run_hooks(message, HookPoint.START):
+                yield event
+
+            logger.info(f"task start event: {message}, will send init message.")
+            if message.payload:
+                yield message
+            else:
+                yield self.runner.init_message
+        elif topic == TopicType.OUTPUT:
+            yield message
+        elif topic == TopicType.HUMAN_CONFIRM:
+            logger.warn("=============== Get human confirm, pause execution ===============")
+            if self.runner.task.outputs and message.payload:
+                await self.runner.task.outputs.add_output(Output(data=message.payload))
+            self.runner._task_response = TaskResponse(answer=str(message.payload),
+                                                      success=True,
+                                                      id=self.runner.task.id,
+                                                      time_cost=(time.time() - self.runner.start_time),
+                                                      usage=self.runner.context.token_usage)
+            await self.runner.stop()
+
+    async def run_hooks(self, message: Message, hook_point: str) -> AsyncGenerator[Message, None]:
+        hooks = self.hooks.get(hook_point, [])
+        for hook in hooks:
+            try:
+                msg = hook(message)
+                if msg:
+                    yield msg
+            except:
+                logger.warning(f"{hook.point()} {hook.name()} execute fail.")

aworld/runners/handler/tool.py

@@ -0,0 +1,106 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+from typing import AsyncGenerator
+
+from aworld.core.agent.base import is_agent
+from aworld.core.common import ActionModel, TaskItem
+from aworld.core.event.base import Message, Constants, TopicType
+from aworld.core.tool.base import AsyncTool, Tool, ToolFactory
+from aworld.logs.util import logger
+from aworld.runners.handler.base import DefaultHandler
+
+
+class ToolHandler(DefaultHandler):
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, runner: 'TaskEventRunner'):
+        self.tools = runner.tools
+        self.tools_conf = runner.tools_conf
+
+    @classmethod
+    def name(cls):
+        return "_tool_handler"
+
+
+class DefaultToolHandler(ToolHandler):
+    async def handle(self, message: Message) -> AsyncGenerator[Message, None]:
+        if message.category != Constants.TOOL:
+            return
+
+        headers = {"context": message.context}
+        # data is List[ActionModel]
+        data = message.payload
+        if not data:
+            # error message, p2p
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(msg="no data to process.", data=data, stop=True),
+                sender='agent_handler',
+                session_id=message.session_id,
+                topic=TopicType.ERROR,
+                headers=headers
+            )
+            return
+
+        for action in data:
+            if not isinstance(action, ActionModel):
+                # error message, p2p
+                yield Message(
+                    category=Constants.TASK,
+                    payload=TaskItem(msg="action not a ActionModel.", data=data, stop=True),
+                    sender=self.name(),
+                    session_id=message.session_id,
+                    topic=TopicType.ERROR,
+                    headers=headers
+                )
+                return
+
+        new_tools = dict()
+        tool_mapping = dict()
+        # Directly use or use tools after creation.
+        for act in data:
+            if is_agent(act):
+                logger.warning(f"somethings wrong, {act} is an agent.")
+                continue
+
+            if not self.tools or (self.tools and act.tool_name not in self.tools):
+                # dynamic only use default config in module.
+                conf = self.tools_conf.get(act.tool_name)
+                tool = ToolFactory(act.tool_name, conf=conf, asyn=conf.use_async if conf else False)
+                tool.event_driven = True
+                if isinstance(tool, Tool):
+                    tool.reset()
+                elif isinstance(tool, AsyncTool):
+                    await tool.reset()
+                tool_mapping[act.tool_name] = []
+                self.tools[act.tool_name] = tool
+                new_tools[act.tool_name] = tool
+            if act.tool_name not in tool_mapping:
+                tool_mapping[act.tool_name] = []
+            tool_mapping[act.tool_name].append(act)
+
+        if new_tools:
+            yield Message(
+                category=Constants.TASK,
+                payload=TaskItem(data=new_tools),
+                sender=self.name(),
+                session_id=message.session_id,
+                topic=TopicType.SUBSCRIBE_TOOL,
+                headers=headers
+            )
+
+        for tool_name, actions in tool_mapping.items():
+            if not (isinstance(self.tools[tool_name], Tool) or isinstance(self.tools[tool_name], AsyncTool)):
+                logger.warning(f"Unsupported tool type: {self.tools[tool_name]}")
+                continue
+
+            # send to the tool
+            yield Message(
+                category=Constants.TOOL,
+                payload=actions,
+                sender=actions[0].agent_name if actions else '',
+                session_id=message.session_id,
+                receiver=tool_name,
+                headers=headers
+            )