(self, **kwargs)
| 40 | self.progress += content |
| 41 | |
| 42 | async def before_execution(self, **kwargs): |
| 43 | PrintStyle(font_color="#1B4F72", padding=True, background_color="white", bold=True).print(f"{self.agent.agent_name}: Using tool '{self.name}'") |
| 44 | self.log = self.get_log_object() |
| 45 | if self.args and isinstance(self.args, dict): |
| 46 | for key, value in self.args.items(): |
| 47 | ctx = {"content": str(value) if not isinstance(value, str) else value} |
| 48 | await call_extensions_async("tool_output_update", self.agent, ctx=ctx) |
| 49 | display_value = ctx["content"] |
| 50 | PrintStyle(font_color="#85C1E9", bold=True).stream(self.nice_key(key)+": ") |
| 51 | PrintStyle(font_color="#85C1E9", padding=isinstance(value,str) and "\n" in value).stream(display_value) |
| 52 | PrintStyle().print() |
| 53 | |
| 54 | async def after_execution(self, response: Response, **kwargs): |
| 55 | text = sanitize_string(response.message.strip()) |
no test coverage detected