Skip to content

twitter_connector

arai_ai_agents.connectors.twitter_connector

TwitterConnector

A connector for interacting with the Twitter API via Tweepy.

This class handles OAuth 1.0a and OAuth 2.0 authentication for posting tweets and retrieving data from Twitter.

Attributes:

Name Type Description
api_key str

The API key for the Twitter API.
api_secret_key str

The API secret key for the Twitter API.
access_token str

The access token for the Twitter API.
access_token_secret str

The access token secret for the Twitter API.
bearer_token str

The bearer token for the Twitter API.
bot_username str

The username of the bot.
bot_id str

The ID of the bot.
Source code in arai_ai_agents/connectors/twitter_connector.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class TwitterConnector:
    """A connector for interacting with the Twitter API via Tweepy.

    This class handles OAuth 1.0a and OAuth 2.0 authentication for
    posting tweets and retrieving data from Twitter.

    Attributes:
        api_key (str): The API key for the Twitter API.
        api_secret_key (str): The API secret key for the Twitter API.
        access_token (str): The access token for the Twitter API.
        access_token_secret (str): The access token secret for the Twitter API.
        bearer_token (str): The bearer token for the Twitter API.
        bot_username (str): The username of the bot.
        bot_id (str): The ID of the bot.
    """

    def __init__(self):
        """Initializes the Twitter connector.

        Raises:
            ValueError: If Twitter API credentials are not present in the environment variables.
        """

        self.api_key = os.getenv("TWITTER_API_KEY") 
        # Part of the OAuth 1.0a credentials identifying the application (required for user-based authentication).

        self.api_secret_key = os.getenv("TWITTER_API_KEY_SECRET") 
        # Secret counterpart to the API key, used in signing OAuth 1.0a requests.

        self.access_token = os.getenv("TWITTER_ACCESS_TOKEN") 
        # Represents the user’s OAuth 1.0a credentials, required for user-level actions (e.g., posting tweets).

        self.access_token_secret = os.getenv("TWITTER_ACCESS_TOKEN_SECRET") 
        # Secret counterpart to the access token, used in signing user-level requests under OAuth 1.0a.

        self.bearer_token = os.getenv("TWITTER_BEARER_TOKEN") 
        # Used for OAuth 2.0 app-only authentication in Twitter API v2, often for read-only access to public data.

        if not all([self.api_key, self.api_secret_key, self.access_token, self.access_token_secret]):
            raise ValueError("Twitter API credentials are not set in .env file")

        try:
            # Initialize v2 client only
            self.client = tweepy.Client(
                bearer_token=self.bearer_token,
                consumer_key=self.api_key,
                consumer_secret=self.api_secret_key,
                access_token=self.access_token,
                access_token_secret=self.access_token_secret,
                wait_on_rate_limit=True
            )

            # Get the bot's info using v2 API
            me = self.client.get_me()
            self.bot_username = me.data.username.lower()
            self.bot_id = me.data.id
            print(f"Successfully authenticated as @{self.bot_username}")

            self.last_mention_id = None
        except Exception as e:
            print(f"Twitter authentication failed: {str(e)}")
            raise

    def post_tweet(self, message: str) -> str:
        """Posts a tweet to the bot's Twitter account.

        Args:
            message (str): The message to be posted as a tweet. Must be shorter than 280 characters.

        Returns:
            str: A string message indicating success or failure.

        Raises:
            Exception: If there's an error posting the tweet.

        Example:
            >>> twitter_connector.post_tweet("Hello, world!")
        """
        try:
            if not message:                
                return "Error: Tweet message is empty"

            # Truncate if too long
            if len(message) > 280:
                message = message[:277] + "..."

            self.client.create_tweet(text=message)            
            return "Tweeted: " + message
        except Exception as e:            
            return "Error posting tweet: " + str(e)

__init__()

Initializes the Twitter connector.

Raises:

Type Description
ValueError

If Twitter API credentials are not present in the environment variables.
Source code in arai_ai_agents/connectors/twitter_connector.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def __init__(self):
    """Initializes the Twitter connector.

    Raises:
        ValueError: If Twitter API credentials are not present in the environment variables.
    """

    self.api_key = os.getenv("TWITTER_API_KEY") 
    # Part of the OAuth 1.0a credentials identifying the application (required for user-based authentication).

    self.api_secret_key = os.getenv("TWITTER_API_KEY_SECRET") 
    # Secret counterpart to the API key, used in signing OAuth 1.0a requests.

    self.access_token = os.getenv("TWITTER_ACCESS_TOKEN") 
    # Represents the user’s OAuth 1.0a credentials, required for user-level actions (e.g., posting tweets).

    self.access_token_secret = os.getenv("TWITTER_ACCESS_TOKEN_SECRET") 
    # Secret counterpart to the access token, used in signing user-level requests under OAuth 1.0a.

    self.bearer_token = os.getenv("TWITTER_BEARER_TOKEN") 
    # Used for OAuth 2.0 app-only authentication in Twitter API v2, often for read-only access to public data.

    if not all([self.api_key, self.api_secret_key, self.access_token, self.access_token_secret]):
        raise ValueError("Twitter API credentials are not set in .env file")

    try:
        # Initialize v2 client only
        self.client = tweepy.Client(
            bearer_token=self.bearer_token,
            consumer_key=self.api_key,
            consumer_secret=self.api_secret_key,
            access_token=self.access_token,
            access_token_secret=self.access_token_secret,
            wait_on_rate_limit=True
        )

        # Get the bot's info using v2 API
        me = self.client.get_me()
        self.bot_username = me.data.username.lower()
        self.bot_id = me.data.id
        print(f"Successfully authenticated as @{self.bot_username}")

        self.last_mention_id = None
    except Exception as e:
        print(f"Twitter authentication failed: {str(e)}")
        raise

post_tweet(message)

Posts a tweet to the bot's Twitter account.

Parameters:

Name Type Description Default
message str

The message to be posted as a tweet. Must be shorter than 280 characters.

 required

Returns:

Name Type Description
str str

A string message indicating success or failure.

Raises:

Type Description
Exception

If there's an error posting the tweet.
Example

twitter_connector.post_tweet("Hello, world!")

Source code in arai_ai_agents/connectors/twitter_connector.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def post_tweet(self, message: str) -> str:
    """Posts a tweet to the bot's Twitter account.

    Args:
        message (str): The message to be posted as a tweet. Must be shorter than 280 characters.

    Returns:
        str: A string message indicating success or failure.

    Raises:
        Exception: If there's an error posting the tweet.

    Example:
        >>> twitter_connector.post_tweet("Hello, world!")
    """
    try:
        if not message:                
            return "Error: Tweet message is empty"

        # Truncate if too long
        if len(message) > 280:
            message = message[:277] + "..."

        self.client.create_tweet(text=message)            
        return "Tweeted: " + message
    except Exception as e:            
        return "Error posting tweet: " + str(e)