feat: add twitter tools (#47)

* feat: add twitter tools

* chore: revise tweet format

* feat: use agent model if configured

* chore: merge conflict

---------

Co-authored-by: polebug <[email protected]>

.env.example

@@ -7,3 +7,11 @@ GOOGLE_API_KEY=
 # GOOGLE_BASE_URL=
 # OLLAMA_BASE_URL=
 COINGECKO_API_KEY=
+# Twitter API credentials
+TWITTER_BEARER_TOKEN=
+TWITTER_API_KEY=
+TWITTER_API_SECRET=
+TWITTER_CLIENT_ID=
+TWITTER_CLIENT_SECRET=
+TWITTER_ACCESS_TOKEN=
+TWITTER_ACCESS_TOKEN_SECRET=

.gitignore

@@ -5,4 +5,5 @@
 
 .venv
 __pycache__
-.ruff_cache
+.ruff_cache
+uv.lock

openagent/agents/__init__.py

@@ -9,6 +9,7 @@
 
 from .finance import finance_agent
 from .feed import feed_agent
+from .twitter import twitter_agent
 
 
 class UnsupportedModel(Exception):
@@ -42,7 +43,7 @@ def build_model(model: str) -> Model:
 
 def build_agent_team(model: str) -> Agent:
     return Agent(
-        team=[finance_agent, feed_agent],
+        team=[finance_agent, feed_agent, twitter_agent],
         model=build_model(model),
     )
 

openagent/agents/twitter.py

@@ -0,0 +1,10 @@
+from phi.agent import Agent
+from ..tools.twitter.tweet_generator import TweetGeneratorTools
+
+twitter_agent = Agent(
+    name="twitter_agent",
+    description="An agent that generates and posts tweets in different personalities",
+    tools=[TweetGeneratorTools()],
+)
+
+__all__ = ["twitter_agent"]

openagent/tools/__init__.py

@@ -1,12 +1,12 @@
 from .coingecko import CoinGeckoTools
 from typing import Optional
 from .dsl import DSLTools
+from .twitter import TweetGeneratorTools
 from pydantic import BaseModel
 
-__all__ = [CoinGeckoTools, DSLTools]
-
+__all__ = [CoinGeckoTools, DSLTools, TweetGeneratorTools]
 
 class ToolConfig(BaseModel):
     tool_id: int
     model_id: int
-    parameters: Optional[dict] = None
+    parameters: Optional[dict] = None

openagent/tools/tests/test_twitter_tools.py

@@ -0,0 +1,78 @@
+import os
+import unittest
+from dotenv import load_dotenv
+from ..twitter.tweet_generator import TweetGeneratorTools
+
+load_dotenv()
+
+
+class TestTwitterTools(unittest.TestCase):
+    def setUp(self):
+        """Set up test fixtures before each test method."""
+        self.tweet_tools = TweetGeneratorTools()
+        self.required_env_vars = [
+            "OPENAI_API_KEY",
+            "TWITTER_API_KEY",
+            "TWITTER_API_SECRET",
+            "TWITTER_ACCESS_TOKEN",
+            "TWITTER_ACCESS_TOKEN_SECRET",
+        ]
+
+    def test_environment_variables(self):
+        """Test if all required environment variables are present"""
+        print("\ntest_environment_variables")
+        missing_vars = []
+        for var in self.required_env_vars:
+            if not os.getenv(var):
+                missing_vars.append(var)
+
+        if missing_vars:
+            self.fail(
+                f"Missing required environment variables: {', '.join(missing_vars)}"
+            )
+
+    def test_tweet_generation_and_posting(self):
+        """Test tweet generation with personality and posting"""
+        print("\ntest_tweet_generation_and_posting")
+
+        # Test case for tweet generation and posting
+        personality = "tech enthusiast"
+        topic = "AI and machine learning innovations"
+        expected_terms = ["AI", "tech", "machine learning", "innovation"]
+
+        try:
+            # Generate and post tweet
+            print(f"\nGenerating and posting tweet as {personality} about {topic}...")
+            success, tweet_content = self.tweet_tools.generate_tweet(
+                personality=personality, topic=topic
+            )
+
+            # Validate the result
+            self.assertTrue(
+                success, f"Tweet generation/posting failed: {tweet_content}"
+            )
+            print("✓ Tweet generated and posted successfully")
+            print(f"Tweet content: {tweet_content}")
+
+            # Validate tweet content
+            self.assertTrue(
+                "#" in tweet_content, "Tweet should contain at least one hashtag"
+            )
+            found_terms = [
+                term for term in expected_terms if term.lower() in tweet_content.lower()
+            ]
+            self.assertTrue(
+                len(found_terms) > 0,
+                f"Tweet should contain at least one of {expected_terms}. Content: {tweet_content}",
+            )
+
+            print("✓ Tweet content validation passed")
+            print(f"Found terms: {', '.join(found_terms)}")
+
+        except Exception as e:
+            self.fail(f"Tweet generation and posting failed: {str(e)}")
+
+
+if __name__ == "__main__":
+    print("\n=== Running Twitter Tools Tests ===\n")
+    unittest.main()

openagent/tools/twitter/__init__.py

@@ -0,0 +1,4 @@
+from .tweet_generator import TweetGeneratorTools
+from .twitter_handler import TwitterHandler
+
+__all__ = ["TweetGeneratorTools", "TwitterHandler"]

openagent/tools/twitter/tweet_generator.py

@@ -0,0 +1,135 @@
+import logging
+import os
+from typing import Tuple, Optional
+from phi.tools import Toolkit
+from phi.model.openai import OpenAIChat
+from phi.model.base import Model
+from phi.model.message import Message
+from dotenv import load_dotenv
+from .twitter_handler import TwitterHandler
+
+# Configure logging
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+SYSTEM_PROMPT = """You are a creative tweet writer who can adapt to different personalities and styles.
+Your task is to generate engaging tweets that match the given personality and topic.
+ALWAYS include relevant hashtags in your tweets to increase visibility and engagement.
+Format your response exactly like a real human tweet - no quotes, no additional text."""
+
+TWEET_REQUIREMENTS = """
+Requirements for the tweet:
+1. Keep it under 280 characters
+2. Use appropriate tone and style for the given personality
+3. MUST include at least 2-3 relevant hashtags
+4. Make it engaging and shareable
+5. Match the tone to the specified personality
+6. Format exactly like a real human tweet
+7. No quotes or other formatting - just the tweet text
+"""
+
+
+class TweetGeneratorTools(Toolkit):
+    def __init__(self, model: Optional[Model] = None):
+        super().__init__(name="tweet_generator_tools")
+        self.twitter_handler = TwitterHandler()
+
+        # Use provided model (from agent) or create a new one
+        if model:
+            self.model = model
+        else:
+            # Initialize OpenAI model for standalone use
+            openai_api_key = os.getenv("OPENAI_API_KEY")
+            if not openai_api_key:
+                raise ValueError("OPENAI_API_KEY not found in environment variables")
+
+            # Initialize model params
+            model_params = {
+                "id": "gpt-4o",
+                "name": "TweetGenerator",
+                "temperature": 0.7,
+                "max_tokens": 280,
+                "api_key": openai_api_key,
+                "structured_outputs": False,
+            }
+
+            # Add base_url only if it's set
+            openai_base_url = os.getenv("OPENAI_BASE_URL")
+            if openai_base_url and openai_base_url.strip():
+                model_params["base_url"] = openai_base_url
+
+            self.model = OpenAIChat(**model_params)
+
+        # Register only the tweet generation function
+        self.register(self.generate_tweet)
+
+    def generate_tweet(self, personality: str, topic: str = None) -> Tuple[bool, str]:
+        """
+        Generate a tweet using the model based on personality and topic, and post it.
+
+        Args:
+            personality (str): The personality/role to use for tweet generation
+            topic (str, optional): Specific topic to tweet about
+        Returns:
+            tuple: (success: bool, message: str) - Success status and response message
+        """
+        try:
+            # Generate prompt messages
+            user_prompt = f"Generate a tweet as {personality}."
+            if topic:
+                user_prompt += f" The tweet should be about: {topic}."
+            user_prompt += TWEET_REQUIREMENTS
+
+            messages = [
+                Message(role="system", content=SYSTEM_PROMPT),
+                Message(role="user", content=user_prompt),
+            ]
+
+            # Generate tweet using the model
+            response = self.model.invoke(messages=messages)
+            # Extract content from the response
+            if hasattr(response, "choices") and len(response.choices) > 0:
+                tweet_content = response.choices[0].message.content.strip()
+            else:
+                tweet_content = str(response).strip()
+
+            # Validate tweet length and hashtag presence
+            if len(tweet_content) > 280:
+                logger.warning(
+                    f"Generated tweet exceeds 280 characters, length: {len(tweet_content)}"
+                )
+                tweet_content = tweet_content[:277] + "..."
+
+            if "#" not in tweet_content:
+                logger.warning(
+                    "Generated tweet does not contain hashtags, regenerating..."
+                )
+                return self.generate_tweet(personality, topic)
+
+            # Post the generated tweet
+            logger.info(f"Posting generated tweet: {tweet_content}")
+            return self.post_tweet(tweet_content)
+
+        except Exception as error:
+            logger.error(f"Error generating/posting tweet: {error}")
+            return False, str(error)
+
+    def post_tweet(self, tweet_content: str) -> Tuple[bool, str]:
+        """
+        Post a tweet using the Twitter handler.
+
+        Args:
+            tweet_content (str): The content to tweet
+        Returns:
+            tuple: (success: bool, message: str) - Success status and response message
+        """
+        try:
+            success, response = self.twitter_handler.post_tweet(tweet_content)
+            if success:
+                return True, tweet_content
+            return False, response
+        except Exception as error:
+            logger.error(f"Error posting tweet: {error}")
+            return False, str(error)

openagent/tools/twitter/twitter_handler.py

@@ -0,0 +1,43 @@
+import os
+import tweepy
+from typing import Tuple
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class TwitterHandler:
+    def __init__(self):
+        self.client = None
+        self.initialize_client()
+
+    def initialize_client(self):
+        """Initialize Twitter client with credentials from environment variables."""
+        try:
+            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"),
+            )
+        except Exception as e:
+            raise ValueError(f"Failed to initialize Twitter client: {str(e)}")
+
+    def post_tweet(self, content: str) -> Tuple[bool, str]:
+        """
+        Post a tweet using the initialized client.
+
+        Args:
+            content (str): The content to tweet
+        Returns:
+            tuple: (success: bool, message: str)
+        """
+        try:
+            if not self.client:
+                self.initialize_client()
+
+            response = self.client.create_tweet(text=content)
+            return True, f"Tweet posted successfully. Tweet ID: {response.data['id']}"
+        except Exception as e:
+            return False, str(e)