Merge pull request #123 from blorm-network/fetch_tweet_details

Twitter Stream Depth

Author: rux-eth

agents/example.json

@@ -24,7 +24,12 @@
       "name": "twitter",
       "timeline_read_count": 10,
       "own_tweet_replies_count":2,
-      "tweet_interval": 5400
+      "tweet_interval": 5400,
+      "respond_to_mentions": {
+        "enabled": false,
+        "accounts_mentioned": ["blormmy"],
+        "accounts_to_listen_to": ["0xzerebro"]
+      }
     },
     {
       "name": "farcaster",

src/actions/twitter_actions.py

@@ -98,17 +98,49 @@ def like_tweet(agent, **kwargs):
 @register_action("respond-to-mentions")
 def respond_to_mentions(agent,**kwargs): #REQUIRES TWITTER PREMIUM PLAN
 
-    filter_str = f"@{agent.username} -is:retweet"
+    accounts_to_listen_to = kwargs.get('accounts_to_listen_to', [])
+    accounts_mentioned = kwargs.get('accounts_mentioned', [])
+    
+    # Get user IDs of accounts to listen to
+    accounts_to_listen_to_ids = []
+    user_ids = agent.connection_manager.perform_action(
+        connection_name="twitter",
+        action_name="get-user-details",
+        params = [[], accounts_to_listen_to]
+    )
+    if user_ids:
+        accounts_to_listen_to_ids = [user.get('id') for user in user_ids]
+
+    # Create filter string to get tweets from accounts_to_listen_to_ids mentioning any of accounts_mentioned
+    if accounts_mentioned:
+        mention_str = " OR ".join([f"@{account}" for account in accounts_mentioned])
+        
+        if accounts_to_listen_to_ids:
+            filter_str = " OR ".join([f"from:{user_id}" for user_id in accounts_to_listen_to_ids])
+            filter_str = f"({filter_str}) ({mention_str}) -is:retweet"
+        else:
+            filter_str = f"({mention_str}) -is:retweet"
+    else:
+        filter_str = ""
+    
+
     stream_function = agent.connection_manager.perform_action(
         connection_name="twitter",
         action_name="stream-tweets",
         params=[filter_str]
     )
+
     def process_tweets():
         for tweet_data in stream_function:
-            tweet_id = tweet_data["id"]
-            tweet_text = tweet_data["text"]
-            agent.logger.info(f"Received a mention: {tweet_text}")
+            try:
+                tweet_id = tweet_data.get("id", "")
+                tweet_text = tweet_data.get("text", "")
+                author_id = tweet_data.get("author_id", "")
+                author_username = tweet_data.get("author_username", "")
+                agent.logger.info(f"Received a mention from @{author_username}: {tweet_text}")            
+            except Exception as e:
+                agent.logger.error(f"Error processing tweet in stream: {str(e)}")
+                continue
 
     processing_thread = threading.Thread(target=process_tweets)
     processing_thread.daemon = True

src/agent.py

@@ -40,13 +40,21 @@ def __init__(
             self.use_time_based_weights = agent_dict["use_time_based_weights"]
             self.time_based_multipliers = agent_dict["time_based_multipliers"]
 
-            has_twitter_tasks = any("tweet" in task["name"] for task in agent_dict.get("tasks", []))
+            self.has_twitter_tasks = any("tweet" in task["name"] for task in agent_dict.get("tasks", []))
 
             twitter_config = next((config for config in agent_dict["config"] if config["name"] == "twitter"), None)
 
-            if has_twitter_tasks and twitter_config:
+            if self.has_twitter_tasks and twitter_config:
                 self.tweet_interval = twitter_config.get("tweet_interval", 900)
                 self.own_tweet_replies_count = twitter_config.get("own_tweet_replies_count", 2)
+                respond_to_mentions_config = twitter_config.get("respond_to_mentions", None)
+
+                if (respond_to_mentions_config):
+                    self.respond_to_mentions = respond_to_mentions_config.get("enabled", False)
+                    self.accounts_mentioned = respond_to_mentions_config.get("accounts_mentioned", [])
+                    self.accounts_to_listen_to = respond_to_mentions_config.get("accounts_to_listen_to", [])
+                else:
+                    self.respond_to_mentions = False
 
             # Extract Echochambers config
             echochambers_config = next((config for config in agent_dict["config"] if config["name"] == "echochambers"), None)
@@ -79,7 +87,7 @@ def _setup_llm_provider(self):
         self.model_provider = llm_providers[0]
 
         # Load Twitter username for self-reply detection if Twitter tasks exist
-        if any("tweet" in task["name"] for task in self.tasks):
+        if self.has_twitter_tasks:
             load_dotenv()
             self.username = os.getenv('TWITTER_USERNAME', '').lower()
             if not self.username:
@@ -172,14 +180,20 @@ def loop(self):
             logger.info(f"{i}...")
             time.sleep(1)
 
+        if self.has_twitter_tasks:
+            if self.respond_to_mentions:
+                logger.info("\n👀 Listening for mentions...")
+                if (len(self.accounts_mentioned) == 0):
+                    self.accounts_mentioned = [self.username] # Default to listening to own mentions if no accounts mentioned
+                execute_action(self, "respond-to-mentions", accounts_mentioned=self.accounts_mentioned, accounts_to_listen_to=self.accounts_to_listen_to) 
         try:
             while True:
                 success = False
                 try:
                     # REPLENISH INPUTS
                     # TODO: Add more inputs to complexify agent behavior
                     if "timeline_tweets" not in self.state or self.state["timeline_tweets"] is None or len(self.state["timeline_tweets"]) == 0:
-                        if any("tweet" in task["name"] for task in self.tasks):
+                        if (self.has_twitter_tasks):
                             logger.info("\n👀 READING TIMELINE")
                             self.state["timeline_tweets"] = self.connection_manager.perform_action(
                                 connection_name="twitter",

src/connections/twitter_connection.py

@@ -93,6 +93,21 @@ def register_actions(self) -> None:
                 ],
                 description="Fetch tweet replies"
             ),
+            "get-user-details": Action(
+                name="get-user-details",
+                parameters=[
+                    ActionParameter("user_ids", True, list, "IDs of the user to get details for"),
+                    ActionParameter("usernames", True, list, "Usernames of the user to get details for")
+                ],
+                description="Get details for user(s) by ID or username"
+            ),
+            "get-tweet-details": Action(
+                name="get-tweet-details",
+                parameters=[
+                    ActionParameter("tweet_id", True, str, "ID of the tweet to get details for")
+                ],
+                description="Get details for a specific tweet"
+            ),
             "stream-tweets": Action(
                 name="stream-tweets",
                 parameters=[
@@ -517,6 +532,47 @@ def get_tweet_replies(self, tweet_id: str, count: int = 10, **kwargs) -> List[di
 
         logger.info(f"Retrieved {len(replies)} replies")
         return replies
+
+    def get_user_details(self, user_ids: list, usernames: list, **kwargs) -> dict:
+        """Get details for user(s) by ID or username"""
+        logger.debug(f"Getting details for user_ids: {user_ids} or usernames: {usernames}")
+
+        params = {"user.fields": "username"}
+    
+        if user_ids:
+            params["ids"] = ",".join(user_ids)
+            endpoint = "users"
+        else:
+            params["usernames"] = ",".join(usernames)
+            endpoint = f"users/by/"
+
+        response = self._make_request('get', endpoint, params=params)
+    
+        users = response.get("data", [])
+    
+        logger.info("Retrieved user details")
+        return users
+    
+    def get_tweet_details(self, tweet_id: str, **kwargs) -> dict:
+        """Get details for a specific tweet"""
+        logger.debug(f"Getting details for tweet {tweet_id}")
+        
+        params = {
+            "tweet.fields": "author_id,created_at,text,attachments,referenced_tweets",
+            "expansions": "author_id",
+            "user.fields": "username",
+            "ids": tweet_id
+        }
+        
+        response = self._make_request('get', 'tweets', params=params)
+        tweet = response.get("data", [])[0]
+
+        if "includes" in response and "users" in response["includes"]:
+            author = response["includes"]["users"][0]
+            tweet["author_username"] = author["username"]
+        
+        logger.info("Retrieved tweet details")
+        return tweet
 
     def _bearer_oauth(self,r):
         bearer_token = self._get_credentials().get("TWITTER_BEARER_TOKEN")
@@ -553,17 +609,30 @@ def stream_tweets(self, filter_string:str,**kwargs) ->Iterator[Dict[str, Any]]:
         rules = self._get_rules()
         self._delete_rules(rules)
         self._build_rule(filter_string)
-        logger.info("Starting Twitter stream")
         try:
+            params = {
+                "tweet.fields": "author_id,created_at,text,attachments,referenced_tweets",
+                "user.fields": "username",
+                "expansions": "author_id"
+            }
             response = self._make_request('get', 'tweets/search/stream', 
-                                        use_bearer=True, stream=True)
+                                        use_bearer=True, stream=True, params=params)
 
             if response.status_code != 200:
                 raise TwitterAPIError(f"Stream connection failed with status {response.status_code}: {response.text}")
 
             for line in response.iter_lines():
+
                 if line:
-                    tweet_data = json.loads(line)['data']
+                    tweet_json = json.loads(line)
+                    
+                    tweet_data = tweet_json.get("data", {})
+                    includes = tweet_json.get("includes", {})
+
+                    if includes and "users" in includes:
+                        author = includes["users"][0]
+                        tweet_data["author_username"] = author["username"]
+
                     yield tweet_data
 
         except Exception as e: