| 1 | from helpers.tool import Tool, Response |
| 2 | from plugins._code_execution.tools.code_execution_tool import CodeExecution |
| 3 | |
| 4 | |
| 5 | class Input(Tool): |
| 6 | |
| 7 | async def execute(self, keyboard="", **kwargs): |
| 8 | # normalize keyboard input |
| 9 | keyboard = keyboard.rstrip() |
| 10 | # keyboard += "\n" # no need to, code_exec does that |
| 11 | |
| 12 | # terminal session number |
| 13 | session = int(self.args.get("session", 0)) |
| 14 | |
| 15 | # forward keyboard input to code execution tool |
| 16 | args = {"runtime": "terminal", "code": keyboard, "session": session, "allow_running": True} |
| 17 | cet = CodeExecution(self.agent, "code_execution_tool", "", args, self.message, self.loop_data) |
| 18 | cet.log = self.log |
| 19 | return await cet.execute(**args) |
| 20 | |
| 21 | def get_log_object(self): |
| 22 | import uuid |
| 23 | return self.agent.context.log.log(type="code_exe", heading=f"icon://keyboard {self.agent.agent_name}: Using tool '{self.name}'", content="", kvps=self.args, id=str(uuid.uuid4())) |
| 24 | |
| 25 | async def after_execution(self, response, **kwargs): |
| 26 | self.agent.hist_add_tool_result(self.name, response.message, id=self.log.id if self.log else "", **(response.additional or {})) |