Twitter Elevated Access required

brk275 can you please elaborate or provide a link to the solution that you are suggesting?

started with these three. You also need a callback URL for pkce and tell twitter its a webpage and not a bot. My code isn't perfect but this got things working. Hopefully this can push this in the right direction.

import os
import tweepy

from urllib.parse import urlparse, parse_qs

os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

def get_code_and_state(redirect_response):
    # Parse the URL and extract the query string
    url = urlparse(redirect_response)
    query_string = parse_qs(url.query)

    # The 'code' parameter in the query string is the authorization code
    code = query_string.get('code', [None])[0]
    state = query_string.get('state', [None])[0]

    if not code:
        raise ValueError("No code found in redirect response")

    return code, state

# You get these from your application in Twitter Developer Portal
TW_CLIENT_ID = os.getenv('TW_CLIENT_ID')
TW_CLIENT_SECRET = os.getenv('TW_CLIENT_SECRET')  # This should be confidential
REDIRECT_URI = os.getenv('TW_REDIRECT_URI')  # This should be a localhost address

# Define the required scopes
SCOPES = ['tweet.read', 'tweet.write', 'users.read', 'follows.read', 'follows.write', 'like.read', 'like.write']

# Initialize the handler
oauth2_user_handler = tweepy.OAuth2UserHandler(
    client_id=TW_CLIENT_ID,
    redirect_uri=REDIRECT_URI,
    scope=SCOPES,
    client_secret=TW_CLIENT_SECRET
)

# Get the authorization URL and ask the user to visit it
authorization_url = oauth2_user_handler.get_authorization_url()
print(f'Please go to the following URL: {authorization_url}')

# Get the redirect response from the user
redirect_response = input('Paste the full redirect URL here:')

# Fetch the token using the redirect response URL
token = oauth2_user_handler.fetch_token(redirect_response)

# Save the token for future use
print(f'Token: {token}')

# Save the token to an environment variable
os.environ['TW_ACCESS_TOKEN'] = token['access_token']

# Initialize the client with the access token
client = tweepy.Client(token['access_token'])

from typing import Any, Dict, List, Optional, Tuple, TypedDict, TypeVar
from auto_gpt_plugin_template import AutoGPTPluginTemplate
import os
import tweepy
from datetime import datetime
datetime.now().isoformat()

PromptGenerator = TypeVar("PromptGenerator")

class Message(TypedDict):
    role: str
    content: str

from .oauth2_with_pkce import token

class AutoGPTTwitter(AutoGPTPluginTemplate):
    def __init__(self):
        super().__init__()
        self._name = "autogpt-twitter"
        self._version = "0.1.0"
        self._description = "Twitter API integrations using Tweepy."
        self.tweet_id = []
        self.tweets = []

        self.client = None

        # Read the tokens from the environment
        bearer_token = os.getenv('TW_BEARER_TOKEN')
        consumer_key = os.getenv('TW_CONSUMER_KEY')
        consumer_secret = os.getenv('TW_CONSUMER_SECRET')
        access_token = os.getenv('TW_ACCESS_TOKEN')
        access_token_secret = os.getenv('TW_ACCESS_TOKEN_SECRET')

        if all([bearer_token, consumer_key, consumer_secret, access_token, access_token_secret]):
            self.client = tweepy.Client(
                bearer_token=bearer_token,
                consumer_key=consumer_key,
                consumer_secret=consumer_secret,
                access_token=access_token,
                access_token_secret=access_token_secret
            )
        else:
            print("Some Twitter credentials not found.")

    client = tweepy.Client(os.getenv('TW_AUTH_TOKEN'))

    def can_handle_on_response(self) -> bool:
        """This method is called to check that the plugin can
        handle the on_response method.
        Returns:
            bool: True if the plugin can handle the on_response method."""
        return False

    def on_response(self, response: str, *args, **kwargs) -> str:
        """This method is called when a response is received from the model."""
        pass

    def can_handle_post_prompt(self) -> bool:
        """This method is called to check that the plugin can
        handle the post_prompt method.
        Returns:
            bool: True if the plugin can handle the post_prompt method."""
        return True

    def can_handle_on_planning(self) -> bool:
        """This method is called to check that the plugin can
        handle the on_planning method.
        Returns:
            bool: True if the plugin can handle the on_planning method."""
        return False

    def on_planning(
        self, prompt: PromptGenerator, messages: List[str]
    ) -> Optional[str]:
        """This method is called before the planning chat completeion is done.
        Args:
            prompt (PromptGenerator): The prompt generator.
            messages (List[str]): The list of messages.
        """
        pass

    def can_handle_post_planning(self) -> bool:
        """This method is called to check that the plugin can
        handle the post_planning method.
        Returns:
            bool: True if the plugin can handle the post_planning method."""
        return False

    def post_planning(self, response: str) -> str:
        """This method is called after the planning chat completeion is done.
        Args:
            response (str): The response.
        Returns:
            str: The resulting response.
        """
        pass

    def can_handle_pre_instruction(self) -> bool:
        """This method is called to check that the plugin can
        handle the pre_instruction method.
        Returns:
            bool: True if the plugin can handle the pre_instruction method."""
        return False

    def pre_instruction(self, messages: List[str]) -> List[str]:
        """This method is called before the instruction chat is done.
        Args:
            messages (List[str]): The list of context messages.
        Returns:
            List[str]: The resulting list of messages.
        """
        pass

    def can_handle_on_instruction(self) -> bool:
        """This method is called to check that the plugin can
        handle the on_instruction method.
        Returns:
            bool: True if the plugin can handle the on_instruction method."""
        return False

    def on_instruction(self, messages: List[str]) -> Optional[str]:
        """This method is called when the instruction chat is done.
        Args:
            messages (List[str]): The list of context messages.
        Returns:
            Optional[str]: The resulting message.
        """
        pass

    def can_handle_post_instruction(self) -> bool:
        """This method is called to check that the plugin can
        handle the post_instruction method.
        Returns:
            bool: True if the plugin can handle the post_instruction method."""
        return False

    def post_instruction(self, response: str) -> str:
        """This method is called after the instruction chat is done.
        Args:
            response (str): The response.
        Returns:
            str: The resulting response.
        """
        pass

    def can_handle_pre_command(self) -> bool:
        """This method is called to check that the plugin can
        handle the pre_command method.
        Returns:
            bool: True if the plugin can handle the pre_command method."""
        return False

    def pre_command(
        self, command_name: str, arguments: Dict[str, Any]
    ) -> Tuple[str, Dict[str, Any]]:
        """This method is called before the command is executed.
        Args:
            command_name (str): The command name.
            arguments (Dict[str, Any]): The arguments.
        Returns:
            Tuple[str, Dict[str, Any]]: The command name and the arguments.
        """
        pass

    def can_handle_post_command(self) -> bool:
        """This method is called to check that the plugin can
        handle the post_command method.
        Returns:
            bool: True if the plugin can handle the post_command method."""
        return False

    def post_command(self, command_name: str, response: str) -> str:
        """This method is called after the command is executed.
        Args:
            command_name (str): The command name.
            response (str): The response.
        Returns:
            str: The resulting response.
        """
        pass

    def can_handle_chat_completion(
        self,
        messages: list[Dict[Any, Any]],
        model: str,
        temperature: float,
        max_tokens: int,
    ) -> bool:
        """This method is called to check that the plugin can
        handle the chat_completion method.
        Args:
            messages (Dict[Any, Any]): The messages.
            model (str): The model name.
            temperature (float): The temperature.
            max_tokens (int): The max tokens.
        Returns:
            bool: True if the plugin can handle the chat_completion method."""
        return False

    def handle_chat_completion(
        self,
        messages: list[Dict[Any, Any]],
        model: str,
        temperature: float,
        max_tokens: int,
    ) -> str:
        """This method is called when the chat completion is done.
        Args:
            messages (Dict[Any, Any]): The messages.
            model (str): The model name.
            temperature (float): The temperature.
            max_tokens (int): The max tokens.
        Returns:
            str: The resulting response.
        """
        return None

    def post_prompt(self, prompt: PromptGenerator) -> PromptGenerator:
            """This method is called just after the generate_prompt is called,
                but actually before the prompt is generated.
            Args:
                prompt (PromptGenerator): The prompt generator.
            Returns:
                PromptGenerator: The prompt generator.
            """
            if self.client:
                from .twitter import (
                    get_mentions,
                    post_tweet,
                    search_twitter_user,
                    get_user
                )
                prompt.add_command(
                "get_user",
                "Get Twitter User",
    {
        "username": "<username>",
        "expansions": "<expansions>",
        "tweet_fields": "<tweet_fields>",
        "user_fields": "<user_fields>",
        "user_auth": "<user_auth>",
    },
    get_user,
)
                prompt.add_command(
                    "post_tweet", "Post Tweet", 
                    {"tweet_text": "<tweet_text>",}, post_tweet
                )
                prompt.add_command("get_mentions", "Get Twitter Mentions", {}, get_mentions)
                prompt.add_command(
                    "search_twitter_user",
                    "Search Twitter",
                    {
                        "user_id" : "<user_id>",
                        "end_time" : "<end_time>",
                        "tweet_fields" : "<tweet_fields>",
                        "exclude" : "<exclude>",
                        "max_results" : "<max_results>"
                    },
                    search_twitter_user,
                )

            return prompt


from __future__ import annotations
from . import AutoGPTTwitter
import pandas as pd
import tweepy 
from datetime import datetime
from typing import List, Union
from datetime import datetime
datetime.now().isoformat()

plugin = AutoGPTTwitter()

def post_tweet(
    tweet_text: str = None,
    direct_message_deep_link: str = None,
    for_super_followers_only: bool = None,
    place_id: str = None,
    media_ids: list[int | str] = None,
    media_tagged_user_ids: list[int | str] = None,
    poll_duration_minutes: int = None,
    poll_options: list[str] = None,
    quote_tweet_id: int | str = None,
    exclude_reply_user_ids: list[int | str] = None,
    in_reply_to_tweet_id: int | str = None,
    reply_settings: str = None,
    user_auth: bool = True
) -> str:
    """Posts a tweet to twitter.
    Args:
        tweet_text (str): The tweet to post.
        direct_message_deep_link (str): Tweets a link directly to a Direct Message conversation with an account.
        for_super_followers_only (bool): Allows you to Tweet exclusively for Super Followers.
        place_id (str): Place ID being attached to the Tweet for geo location.
        media_ids (list[int | str]): A list of Media IDs being attached to the Tweet.
        media_tagged_user_ids (list[int | str]): A list of User IDs being tagged in the Tweet with Media.
        poll_duration_minutes (int): Duration of the poll in minutes for a Tweet with a poll.
        poll_options (list[str]): A list of poll options for a Tweet with a poll.
        quote_tweet_id (int | str): Link to the Tweet being quoted.
        exclude_reply_user_ids (list[int | str]): A list of User IDs to be excluded from the reply Tweet.
        in_reply_to_tweet_id (int | str): Tweet ID of the Tweet being replied to.
        reply_settings (str): Settings to indicate who can reply to the Tweet.
        user_auth (bool): Whether or not to use OAuth 1.0a User Context to authenticate. Default is True.
    Returns:
        str: The tweet that was posted.
    """

    tweet = plugin.client.create_tweet(
        text=tweet_text,
        direct_message_deep_link=direct_message_deep_link,
        for_super_followers_only=for_super_followers_only,
        place_id=place_id,
        media_ids=media_ids,
        media_tagged_user_ids=media_tagged_user_ids,
        poll_duration_minutes=poll_duration_minutes,
        poll_options=poll_options,
        quote_tweet_id=quote_tweet_id,
        exclude_reply_user_ids=exclude_reply_user_ids,
        in_reply_to_tweet_id=in_reply_to_tweet_id,
        reply_settings=reply_settings,
        user_auth=user_auth
    )
    if response.ok:  # Check if the request was successful
        tweet = response.json()  # Parse the JSON response
        return f"Success! Tweet: {tweet.get('text', 'No text')}"
    else:
        return f"Error: {response.text}"

def get_mentions(
    user_id: int | str,
    end_time: datetime.datetime | str = None,
    expansions: list[str] | str = None,
    max_results: int = None,
    media_fields: list[str] | str = None,
    pagination_token: str = None,
    place_fields: list[str] | str = None,
    poll_fields: list[str] | str = None,
    since_id: int | str = None,
    start_time: datetime.datetime | str = None,
    tweet_fields: list[str] | str = None,
    until_id: int | str = None,
    user_fields: list[str] | str = None,
    user_auth: bool = False
) -> str | None:
    """Gets the most recent mention.

    Args:
        user_id (int | str): Unique identifier of the user for whom to return Tweets mentioning the user.
        end_time (datetime.datetime | str): The new UTC timestamp from which the Tweets will be provided.
        expansions (list[str] | str): expansions Parameter
        max_results (int): Specifies the number of Tweets to try and retrieve.
        media_fields (list[str] | str): media_fields
        pagination_token (str): This parameter is used to move forwards or backwards through ‘pages’ of results.
        place_fields (list[str] | str): place_fields
        poll_fields (list[str] | str): poll_fields
        since_id (int | str): Returns results with a Tweet ID greater than (that is, more recent than) the specified ‘since’ Tweet ID.
        start_time (datetime.datetime | str): The oldest UTC timestamp from which the Tweets will be provided.
        tweet_fields (list[str] | str): tweet_fields
        until_id (int | str): Returns results with a Tweet ID less less than (that is, older than) the specified ‘until’ Tweet ID.
        user_fields (list[str] | str): user_fields
        user_auth (bool): Whether or not to use OAuth 1.0a User Context to authenticate. Default is False.

    Returns:
        str | None: The most recent mention.
    """

    tweets = plugin.client.get_users_mentions(
        id=user_id,
        end_time=end_time,
        expansions=expansions,
        max_results=max_results,
        media_fields=media_fields,
        pagination_token=pagination_token,
        place_fields=place_fields,
        poll_fields=poll_fields,
        since_id=since_id,
        start_time=start_time,
        tweet_fields=['id', 'text', 'created_at', 'context_annotations'],
        until_id=until_id,
        user_fields=user_fields,
        user_auth=user_auth
    )

    for tweet in tweets:
        return (
            f"@{tweet.user.screen_name} Replied: {tweet.full_text}"
            f" Tweet ID: {tweet.id}"
        )  # Returns most recent mention

def get_user(
    id: Union[int, str, None] = None,
    username: Union[str, None] = None,
    expansions: Union[list[str], str, None] = None,
    tweet_fields: Union[list[str], str, None] = None,
    user_fields: Union[list[str], str, None] = None,
    user_auth: bool = False
) -> dict:
    """Gets information about a single user specified by the requested ID or username.

    Args:
        id (int | str | None): The ID of the user to lookup.
        username (str | None): The Twitter username (handle) of the user.
        expansions (list[str] | str | None): expansions Parameter
        tweet_fields (list[str] | str | None): tweet_fields
        user_fields (list[str] | str | None): user_fields
        user_auth (bool): Whether or not to use OAuth 1.0a User Context to authenticate. Default is False.

    Returns:
        dict: The user's information.
    """

    user = plugin.client.get_user(
        username=username,
        expansions=expansions,
        tweet_fields=tweet_fields,
        user_fields=user_fields,
        user_auth=user_auth
    )

    return user

def search_twitter_user(
    user_id: Union[int, str],
    end_time: Union[datetime, str, None] = None,
    exclude: Union[List[str], str, None] = None,
    expansions: Union[List[str], str, None] = None,
    max_results: int = None,
    media_fields: Union[List[str], str, None] = None,
    pagination_token: str = None,
    place_fields: Union[List[str], str, None] = None,
    poll_fields: Union[List[str], str, None] = None,
    since_id: Union[int, str, None] = None,
    start_time: Union[datetime, str, None] = None,
    tweet_fields: Union[List[str], str, None] = None,
    until_id: Union[int, str, None] = None,
    user_fields: Union[List[str], str, None] = None,
    user_auth: bool = False
) -> str:
    tweets = plugin.client.get_users_tweets(
        id=user_id,
        end_time=end_time,
        exclude=exclude,
        expansions=expansions,
        max_results=max_results,
        media_fields=media_fields,
        pagination_token=pagination_token,
        place_fields=place_fields,
        poll_fields=poll_fields,
        since_id=since_id,
        start_time=start_time,
        tweet_fields=['created_at', 'text', 'context_annotations'],  # corrected 'crested_at' to 'created_at'
        until_id=until_id,
        user_fields=user_fields,
        user_auth=user_auth
    )
    print(f"type of tweets: {type(tweets)}")  # print the type of tweets
    print(f"tweets: {tweets}")  # print the content of tweets

    columns = ["Time", "User", "ID", "Tweet"]
    data = []

    for tweet in tweets.data:
        row = [tweet['created_at'], tweet['author_id'], tweet['id'], tweet['text']]  # accessing fields as dictionary keys
        data.append(row)

    df = pd.DataFrame(data, columns=columns)

    print(df)```

Significant-Gravitas / Auto-GPT-Plugins

Twitter Elevated Access required #157