| 1 | from helpers.tool import Tool, Response |
| 2 | from agent import AgentContext |
| 3 | from helpers.notification import NotificationPriority, NotificationType |
| 4 | |
| 5 | class NotifyUserTool(Tool): |
| 6 | |
| 7 | async def execute(self, **kwargs): |
| 8 | |
| 9 | message = self.args.get("message", "") |
| 10 | title = self.args.get("title", "") |
| 11 | detail = self.args.get("detail", "") |
| 12 | notification_type = self.args.get("type", NotificationType.INFO) |
| 13 | priority = self.args.get("priority", NotificationPriority.HIGH) # by default, agents should notify with high priority |
| 14 | timeout = int(self.args.get("timeout", 30)) # agent's notifications should have longer timeouts |
| 15 | |
| 16 | try: |
| 17 | notification_type = NotificationType(notification_type) |
| 18 | except ValueError: |
| 19 | return Response(message=f"Invalid notification type: {notification_type}", break_loop=False) |
| 20 | |
| 21 | try: |
| 22 | priority = NotificationPriority(priority) |
| 23 | except ValueError: |
| 24 | return Response(message=f"Invalid notification priority: {priority}", break_loop=False) |
| 25 | |
| 26 | if not message: |
| 27 | return Response(message="Message is required", break_loop=False) |
| 28 | |
| 29 | AgentContext.get_notification_manager().add_notification( |
| 30 | message=message, |
| 31 | title=title, |
| 32 | detail=detail, |
| 33 | type=notification_type, |
| 34 | priority=priority, |
| 35 | display_time=timeout, |
| 36 | ) |
| 37 | return Response(message=self.agent.read_prompt("fw.notify_user.notification_sent.md"), break_loop=False) |