import os import requests from litellm import completion from agents.deals import Opportunity from agents.agent import Agent from free_config import MESSAGING_MODEL, PUSHOVER_CONFIGURED pushover_url = "https://api.pushover.net/1/messages.json" class MessagingAgent(Agent): name = "Messaging Agent" color = Agent.WHITE def __init__(self): """ Set up Pushover notifications (optional) and Groq for message crafting. """ self.log("Messaging Agent is initializing") self.pushover_user = os.getenv("PUSHOVER_USER", "") self.pushover_token = os.getenv("PUSHOVER_TOKEN", "") self.model = MESSAGING_MODEL if PUSHOVER_CONFIGURED: self.log("Messaging Agent has Pushover configured") else: self.log("Messaging Agent: Pushover not configured — alerts will be logged only") def push(self, text): """ Send a Push Notification using the Pushover API, or log if not configured. """ if not PUSHOVER_CONFIGURED: self.log(f"Deal alert (Pushover skipped): {text[:120]}...") return self.log("Messaging Agent is sending a push notification") payload = { "user": self.pushover_user, "token": self.pushover_token, "message": text, "sound": "cashregister", } requests.post(pushover_url, data=payload) def alert(self, opportunity: Opportunity): """ Make an alert about the specified Opportunity """ text = f"Deal Alert! Price=${opportunity.deal.price:.2f}, " text += f"Estimate=${opportunity.estimate:.2f}, " text += f"Discount=${opportunity.discount:.2f} :" text += opportunity.deal.product_description[:10] + "... " text += opportunity.deal.url self.push(text) self.log("Messaging Agent has completed") def craft_message( self, description: str, deal_price: float, estimated_true_value: float ) -> str: user_prompt = "Please summarize this great deal in 2-3 sentences to be sent as an exciting push notification alerting the user about this deal.\n" user_prompt += f"Item Description: {description}\nOffered Price: {deal_price}\nEstimated true value: {estimated_true_value}" user_prompt += "\n\nRespond only with the 2-3 sentence message which will be used to alert & excite the user about this deal" response = completion( model=self.model, messages=[ {"role": "user", "content": user_prompt}, ], ) return response.choices[0].message.content def notify(self, description: str, deal_price: float, estimated_true_value: float, url: str): """ Make an alert about the specified details """ self.log(f"Messaging Agent is using Groq ({self.model}) to craft the message") text = self.craft_message(description, deal_price, estimated_true_value) self.push(text[:200] + "... " + url) self.log("Messaging Agent has completed")