Twitter Operations

from twitter_agent.custom_twitter_actions import TwitterClient
import tweepy
import os
from dotenv import load_dotenv

class TwitterClient:
    def __init__(self):
        """Initialize Twitter API v2 client with credentials from environment variables."""
        self.client = tweepy.Client(
            bearer_token=os.getenv("TWITTER_BEARER_TOKEN"),
            consumer_key=os.getenv("TWITTER_API_KEY"),
            consumer_secret=os.getenv("TWITTER_API_SECRET"),
            access_token=os.getenv("TWITTER_ACCESS_TOKEN"),
            access_token_secret=os.getenv("TWITTER_ACCESS_TOKEN_SECRET"),
            wait_on_rate_limit=True
        )

# Twitter API Credentials (API v2)
TWITTER_BEARER_TOKEN=your_bearer_token
TWITTER_API_KEY=your_api_key
TWITTER_API_SECRET=your_api_secret
TWITTER_ACCESS_TOKEN=your_access_token
TWITTER_ACCESS_TOKEN_SECRET=your_access_token_secret

# Feature toggles
USE_TWITTER_CORE=true
USE_TWEET_DELETE=true
USE_USER_ID_LOOKUP=true
USE_USER_TWEETS_LOOKUP=true
USE_RETWEET=true

from pydantic import BaseModel

class Tweet(BaseModel):
    id: str
    text: str
    author_id: str
    created_at: str

async def get_user_id(self, username: str) -> Optional[str]:
    """Get user ID from username."""
    try:
        user = self.client.get_user(username=username)
        if user and user.data:
            return str(user.data.id)
        return None
    except Exception as e:
        print(f"Error getting user ID for {username}: {str(e)}")
        return None

def create_get_user_id_tool() -> Tool:
    """Create a tool to get a user's ID from their username."""
    return Tool(
        name="get_user_id",
        description="""Get a Twitter user's ID from their username.
        Input should be the username as a string (without the @ symbol).
        Example: get_user_id("TwitterDev")""",
        func=lambda username: asyncio.run(twitter_client.get_user_id(username))
    )

async def get_user_tweets(self, user_id: str, max_results: int = 10) -> List[Tweet]:
    """Get recent tweets from a user."""
    try:
        tweets = self.client.get_users_tweets(
            id=user_id,
            max_results=max_results,
            tweet_fields=['created_at', 'author_id']
        )
        
        if not tweets.data:
            return []
            
        return [
            Tweet(
                id=str(tweet.id),
                text=tweet.text,
                author_id=str(tweet.author_id),
                created_at=tweet.created_at.isoformat()
            )
            for tweet in tweets.data
        ]
    except Exception as e:
        print(f"Error getting tweets for user {user_id}: {str(e)}")
        return []

def create_get_user_tweets_tool() -> Tool:
    """Create a tool to get a user's recent tweets."""
    return Tool(
        name="get_user_tweets",
        description="""Get recent tweets from a Twitter user using their ID.
        Input should be the user ID as a string.
        Example: get_user_tweets("783214")
        Optionally specify max_results (default 10) as: get_user_tweets("783214", max_results=5)""",
        func=lambda user_id, max_results=10: asyncio.run(twitter_client.get_user_tweets(user_id, max_results))
    )

async def delete_tweet(self, tweet_id: str) -> bool:
    """Delete a tweet."""
    try:
        response = self.client.delete_tweet(id=tweet_id)
        return response.data is not None
    except Exception as e:
        print(f"Error deleting tweet {tweet_id}: {str(e)}")
        return False

def create_delete_tweet_tool() -> Tool:
    """Create a delete tweet tool."""
    return Tool(
        name="delete_tweet",
        description="""Delete a tweet using its ID. You can only delete tweets from your own account.
        Input should be the tweet ID as a string.
        Example: delete_tweet("1234567890")""",
        func=lambda tweet_id: asyncio.run(twitter_client.delete_tweet(tweet_id))
    )

async def retweet(self, tweet_id: str) -> bool:
    """Retweet a tweet."""
    try:
        response = self.client.retweet(tweet_id=tweet_id)
        return response.data is not None
    except Exception as e:
        print(f"Error retweeting {tweet_id}: {str(e)}")
        return False

def create_retweet_tool() -> Tool:
    """Create a retweet tool."""
    return Tool(
        name="retweet",
        description="""Retweet a tweet using its ID. You can only retweet public tweets.
        Input should be the tweet ID as a string.
        Example: retweet("1234567890")""",
        func=lambda tweet_id: asyncio.run(twitter_client.retweet(tweet_id))
    )

if os.getenv("USE_TWITTER_CORE", "true").lower() == "true":
    print_system("Adding custom Twitter tools...")
    twitter_client = TwitterClient()
    
    if os.getenv("USE_TWEET_DELETE", "true").lower() == "true":
        tools.append(create_delete_tweet_tool())
        
    if os.getenv("USE_USER_ID_LOOKUP", "true").lower() == "true":
        tools.append(create_get_user_id_tool())
        
    if os.getenv("USE_USER_TWEETS_LOOKUP", "true").lower() == "true":
        tools.append(create_get_user_tweets_tool())
        
    if os.getenv("USE_RETWEET", "true").lower() == "true":
        tools.append(create_retweet_tool())
        
    print_system("Added custom Twitter tools")

from coinbase_agentkit import twitter_action_provider

agent_kit = AgentKit(AgentKitConfig(
    wallet_provider=wallet_provider,
    action_providers=[
        twitter_action_provider(),  # CDP Twitter operations
    ]
))

self.client = tweepy.Client(
    # ... credentials ...
    wait_on_rate_limit=True  # Automatically wait when rate limited
)

from langchain_core.messages import HumanMessage

response = agent_executor.invoke({
    "messages": [HumanMessage(
        content="Get the recent tweets from user ID 783214"
    )]
})

response = agent_executor.invoke({
    "messages": [HumanMessage(
        content="Get the user ID for username 'elonmusk' and show their latest 5 tweets"
    )]
})

response = agent_executor.invoke({
    "messages": [HumanMessage(
        content="Delete tweet with ID 1234567890"
    )]
})

response = agent_executor.invoke({
    "messages": [HumanMessage(
        content="Retweet the tweet with ID 1234567890"
    )]
})

try:
    response = self.client.get_user(username=username)
    if user and user.data:
        return str(user.data.id)
    return None
except Exception as e:
    print(f"Error getting user ID for {username}: {str(e)}")
    return None

from twitter_agent.twitter_state import TwitterState

twitter_state = TwitterState()

# Check if already replied
if not twitter_state.has_replied_to(tweet_id):
    # Reply to tweet
    twitter_state.add_replied_tweet(tweet_id)

# Check if already reposted
if not twitter_state.has_reposted(tweet_id):
    # Retweet
    twitter_state.add_reposted_tweet(tweet_id)