Merge pull request #119 from blorm-network/twitter_stream

Add Twitter Stream

Author: Peteredwardj

.env.example

@@ -14,6 +14,7 @@ TWITTER_CONSUMER_SECRET=
 TWITTER_ACCESS_TOKEN=            
 TWITTER_ACCESS_TOKEN_SECRET=      
 TWITTER_USER_ID=
+TWITTER_BEARER_TOKEN=
 SOLANA_PRIVATE_KEY=
 SONIC_PRIVATE_KEY=    
 GOAT_RPC_PROVIDER_URL=
@@ -22,3 +23,4 @@ SOLANA_PRIVATE_KEY=
 DISCORD_TOKEN=
 XAI_API_KEY=
 TOGETHER_API_KEY=
+

src/actions/twitter_actions.py

@@ -1,4 +1,4 @@
-import time 
+import time,threading
 from src.action_handler import register_action
 from src.helpers import print_h_bar
 from src.prompts import POST_TWEET_PROMPT, REPLY_TWEET_PROMPT
@@ -94,3 +94,22 @@ def like_tweet(agent, **kwargs):
     else:
         agent.logger.info("\n👀 No tweets found to like...")
     return False
+
+@register_action("respond-to-mentions")
+def respond_to_mentions(agent,**kwargs): #REQUIRES TWITTER PREMIUM PLAN
+
+    filter_str = f"@{agent.username} -is:retweet"
+    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}")
+
+    processing_thread = threading.Thread(target=process_tweets)
+    processing_thread.daemon = True
+    processing_thread.start()

src/connections/twitter_connection.py

@@ -1,10 +1,11 @@
 import os
 import logging
-from typing import Dict, Any, List, Tuple
+from typing import Dict, Any, List, Tuple, Iterator
 from requests_oauthlib import OAuth1Session
 from dotenv import set_key, load_dotenv
 from src.connections.base_connection import BaseConnection, Action, ActionParameter
 from src.helpers import print_h_bar
+import json,requests
 
 logger = logging.getLogger("connections.twitter_connection")
 
@@ -91,6 +92,13 @@ def register_actions(self) -> None:
                     ActionParameter("tweet_id", True, str, "ID of the tweet to query for replies")
                 ],
                 description="Fetch tweet replies"
+            ),
+            "stream-tweets": Action(
+                name="stream-tweets",
+                parameters=[
+                    ActionParameter("filter_string", True, str, "Filter string for rules of the stream , e.g @username")
+                ],
+                description="Stream tweets based on filter rule"
             )
         }
 
@@ -107,6 +115,8 @@ def _get_credentials(self) -> Dict[str, str]:
             'TWITTER_USER_ID': 'user ID'
         }
 
+        optional_vars = {'TWITTER_BEARER_TOKEN'} # Bearer Token is used for streaming, Twitter premium plan is required
+
         credentials = {}
         missing = []
 
@@ -119,11 +129,14 @@ def _get_credentials(self) -> Dict[str, str]:
         if missing:
             error_msg = f"Missing Twitter credentials: {', '.join(missing)}"
             raise TwitterConfigurationError(error_msg)
+        
+        for env_var in optional_vars:
+            credentials[env_var] = os.getenv(env_var)
 
         logger.debug("All required credentials found")
         return credentials
 
-    def _make_request(self, method: str, endpoint: str, **kwargs) -> dict:
+    def _make_request(self, method: str, endpoint: str,use_bearer: bool = False, stream: bool = False, **kwargs) -> dict:
         """
         Make a request to the Twitter API with error handling
 
@@ -133,16 +146,25 @@ def _make_request(self, method: str, endpoint: str, **kwargs) -> dict:
             **kwargs: Additional request parameters
 
         Returns:
-            Dict containing the API response
+            Dict containing the API response (or raw response if stream=True)
         """
         logger.debug(f"Making {method.upper()} request to {endpoint}")
         try:
-            oauth = self._get_oauth()
             full_url = f"https://api.twitter.com/2/{endpoint.lstrip('/')}"
 
-            response = getattr(oauth, method.lower())(full_url, **kwargs)
+            if use_bearer:
+                response = requests.request(
+                    method=method.lower(),
+                    url=full_url,
+                    auth=self._bearer_oauth,
+                    stream=stream,
+                    **kwargs
+                )
+            else:
+                oauth = self._get_oauth()
+                response = getattr(oauth, method.lower())(full_url, **kwargs)
 
-            if response.status_code not in [200, 201]:
+            if not stream and response.status_code not in [200, 201]:
                 logger.error(
                     f"Request failed: {response.status_code} - {response.text}"
                 )
@@ -151,6 +173,10 @@ def _make_request(self, method: str, endpoint: str, **kwargs) -> dict:
                 )
 
             logger.debug(f"Request successful: {response.status_code}")
+
+            if stream:
+                return response
+        
             return response.json()
 
         except Exception as e:
@@ -307,6 +333,10 @@ def configure(self) -> None:
                 oauth_tokens.get('oauth_token_secret')
             }
 
+            bearer_token = input("Input Bearer Token for Twitter Streams (optional, hit Enter to skip): ").strip()
+            if bearer_token:
+                env_vars['TWITTER_BEARER_TOKEN'] = bearer_token
+
             for key, value in env_vars.items():
                 set_key('.env', key, value)
                 logger.debug(f"Saved {key} to .env")
@@ -486,4 +516,58 @@ def get_tweet_replies(self, tweet_id: str, count: int = 10, **kwargs) -> List[di
         replies = response.get("data", [])
 
         logger.info(f"Retrieved {len(replies)} replies")
-        return replies
+        return replies
+    
+    def _bearer_oauth(self,r):
+        bearer_token = self._get_credentials().get("TWITTER_BEARER_TOKEN")
+        if not bearer_token:
+            raise TwitterConfigurationError("Bearer token is required for streaming API access")
+        r.headers["Authorization"] = f"Bearer {bearer_token}"
+        r.headers["User-Agent"] = "v2FilteredStreamPython"
+        return r
+    
+
+    def _get_rules(self):
+        """Get stream rules"""
+        logger.debug("Getting stream rules")
+        return self._make_request('get', 'tweets/search/stream/rules', use_bearer=True)
+    
+    def _delete_rules(self,rules) -> None:
+        """Delete stream rules"""
+        if rules is None or "data" not in rules:
+            return None
+
+        ids = list(map(lambda rule: rule["id"], rules["data"]))
+        payload = {"delete": {"ids": ids}}
+        return self._make_request('post', 'tweets/search/stream/rules', use_bearer=True, json=payload)
+
+    
+    def _build_rule(self, filter_string, **kwargs) -> None:
+        """Build a rule for the stream"""
+        rule = [{"value":filter_string }]
+        payload = {"add": rule}
+        return self._make_request('post', 'tweets/search/stream/rules', use_bearer=True, json=payload)
+    
+    def stream_tweets(self, filter_string:str,**kwargs) ->Iterator[Dict[str, Any]]:
+        """Stream tweets. Requires Twitter Premium Plan and Bearer Token"""
+        rules = self._get_rules()
+        self._delete_rules(rules)
+        self._build_rule(filter_string)
+        logger.info("Starting Twitter stream")
+        try:
+            response = self._make_request('get', 'tweets/search/stream', 
+                                        use_bearer=True, stream=True)
+            
+            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']
+                    yield tweet_data
+                
+        except Exception as e:
+            logger.error(f"Error streaming tweets: {str(e)}")
+            raise TwitterAPIError(f"Error streaming tweets: {str(e)}")
+        
+