| 1 | from helpers.errors import RepairableException |
| 2 | from helpers.tool import Tool, Response |
| 3 | |
| 4 | |
| 5 | class ResponseTool(Tool): |
| 6 | |
| 7 | async def execute(self, **kwargs): |
| 8 | for key in ("text", "message"): |
| 9 | message = self.args.get(key) |
| 10 | if isinstance(message, str) and message.strip(): |
| 11 | return Response(message=message, break_loop=True) |
| 12 | raise RepairableException( |
| 13 | "response tool requires a non-empty top-level text or message string argument" |
| 14 | ) |
| 15 | |
| 16 | async def before_execution(self, **kwargs): |
| 17 | # self.log = self.agent.context.log.log(type="response", heading=f"{self.agent.agent_name}: Responding", content=self.args.get("text", "")) |
| 18 | # don't log here anymore, we have the live_response extension now |
| 19 | pass |
| 20 | |
| 21 | async def after_execution(self, response, **kwargs): |
| 22 | # do not add anything to the history or output |
| 23 | |
| 24 | if self.loop_data and "log_item_response" in self.loop_data.params_temporary: |
| 25 | log = self.loop_data.params_temporary["log_item_response"] |
| 26 | log.update(finished=True) # mark the message as finished |