add get user details, stream by accounts mentioned and tweet authors

Author: Peteredwardj

agents/example.json

@@ -25,7 +25,11 @@
       "timeline_read_count": 10,
       "own_tweet_replies_count":2,
       "tweet_interval": 5400,
-      "respond_to_mentions": false
+      "respond_to_mentions": {
+        "enabled": true,
+        "accounts_mentioned": ["0xzerebro"],
+        "accounts_to_listen_to": ["0xzerebro"]
+      }
     },
     {
       "name": "farcaster",

src/actions/twitter_actions.py

@@ -98,12 +98,38 @@ 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 = [None, 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:
             try:

src/agent.py

@@ -47,7 +47,14 @@ def __init__(
             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)
-                self.respond_to_mentions = twitter_config.get("respond_to_mentions", False)  
+                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)
@@ -176,15 +183,16 @@ def loop(self):
         if self.has_twitter_tasks:
             if self.respond_to_mentions:
                 logger.info("\n👀 Listening for mentions...")
-                execute_action(self, "respond-to-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 "timeline_tweets" not in self.state or self.state["timeline_tweets"] is None or len(self.state["timeline_tweets"]) == 0:
                         if (self.has_twitter_tasks):
                             logger.info("\n👀 READING TIMELINE")
                             self.state["timeline_tweets"] = self.connection_manager.perform_action(
@@ -200,7 +208,7 @@ def loop(self):
                                 connection_name="echochambers",
                                 action_name="get-room-info",
                                 params={}
-                            )
+                            )'''
 
                     # CHOOSE AN ACTION
                     # TODO: Add agentic action selection

src/connections/twitter_connection.py

@@ -93,6 +93,14 @@ 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=[
@@ -524,6 +532,26 @@ 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"""