diff --git a/docs/swarms_tools/twitter.md b/docs/swarms_tools/twitter.md index c6d6b1d6..7e929c9f 100644 --- a/docs/swarms_tools/twitter.md +++ b/docs/swarms_tools/twitter.md @@ -1,353 +1,533 @@ -# TwitterTool Documentation +# TwitterTool Technical Documentation -## Overview +## Class Overview -The TwitterTool is a powerful Python-based interface for interacting with Twitter's API, designed specifically for integration with autonomous agents and AI systems. It provides a streamlined way to perform common Twitter operations while maintaining proper error handling and logging capabilities. +The TwitterTool class provides a comprehensive interface for interacting with Twitter's API. It encapsulates common Twitter operations and provides error handling, logging, and integration capabilities with AI agents. -## Installation - -Before using the TwitterTool, ensure you have the required dependencies installed: +## Installation Requirements ```bash -pip install tweepy swarms-tools +pip install swarms-tools ``` -## Basic Configuration +## Class Definition -The TwitterTool requires Twitter API credentials for authentication. Here's how to set up the basic configuration: +### Constructor Parameters -```python -from swarms_tools.social_media.twitter_tool import TwitterTool +The `options` dictionary accepts the following keys: -import os +| Parameter | Type | Required | Default | Description | +|-----------|------|----------|----------|-------------| +| id | str | No | "twitter_plugin" | Unique identifier for the tool instance | +| name | str | No | "Twitter Plugin" | Display name for the tool | +| description | str | No | Default description | Tool description | +| credentials | Dict[str, str] | Yes | None | Twitter API credentials | -options = { - "id": "your_unique_id", - "name": "your_tool_name", - "description": "Your tool description", - "credentials": { - "apiKey": os.getenv("TWITTER_API_KEY"), - "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), - "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), - "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET") - } -} +The `credentials` dictionary requires: -twitter_tool = TwitterTool(options) +| Credential | Type | Required | Description | +|------------|------|----------|-------------| +| apiKey | str | Yes | Twitter API key | +| apiSecretKey | str | Yes | Twitter API secret key | +| accessToken | str | Yes | Twitter access token | +| accessTokenSecret | str | Yes | Twitter access token secret | + +## Public Methods + +### available_functions + +```python +@property +def available_functions(self) -> List[str] ``` -For security, it's recommended to use environment variables for credentials: +Returns a list of available function names that can be executed by the tool. + +**Returns:** +- List[str]: Names of available functions ['get_metrics', 'reply_tweet', 'post_tweet', 'like_tweet', 'quote_tweet'] +**Example:** ```python -import os -from dotenv import load_dotenv +twitter_tool = TwitterTool(options) +functions = twitter_tool.available_functions +print(f"Available functions: {functions}") +``` -load_dotenv() +### get_function -options = { - "id": "twitter_bot", - "name": "Twitter Bot", - "credentials": { - "apiKey": os.getenv("TWITTER_API_KEY"), - "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), - "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), - "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET") - } -} +```python +def get_function(self, fn_name: str) -> Callable ``` -## Core Functionality +Retrieves a specific function by name for execution. -The TwitterTool provides five main functions: +**Parameters:** +- fn_name (str): Name of the function to retrieve -1. **Posting Tweets**: Create new tweets -2. **Replying to Tweets**: Respond to existing tweets -3. **Quoting Tweets**: Share tweets with additional commentary -4. **Liking Tweets**: Engage with other users' content -5. **Fetching Metrics**: Retrieve account statistics +**Returns:** +- Callable: The requested function -### Basic Usage Examples +**Raises:** +- ValueError: If the function name is not found +**Example:** ```python -# Get a specific function post_tweet = twitter_tool.get_function('post_tweet') -reply_tweet = twitter_tool.get_function('reply_tweet') -quote_tweet = twitter_tool.get_function('quote_tweet') -like_tweet = twitter_tool.get_function('like_tweet') -get_metrics = twitter_tool.get_function('get_metrics') +metrics_fn = twitter_tool.get_function('get_metrics') +``` + +## Protected Methods -# Post a tweet -post_tweet("Hello, Twitter!") +### _get_metrics -# Reply to a tweet -reply_tweet(tweet_id=123456789, reply="Great point!") +```python +def _get_metrics(self) -> Dict[str, int] +``` -# Quote a tweet -quote_tweet(tweet_id=123456789, quote="Interesting perspective!") +Fetches user metrics including followers, following, and tweet counts. -# Like a tweet -like_tweet(tweet_id=123456789) +**Returns:** +- Dict[str, int]: Dictionary containing metrics + - followers: Number of followers + - following: Number of accounts following + - tweets: Total tweet count -# Get account metrics -metrics = get_metrics() +**Example:** +```python +metrics = twitter_tool._get_metrics() print(f"Followers: {metrics['followers']}") +print(f"Following: {metrics['following']}") +print(f"Total tweets: {metrics['tweets']}") ``` -## Integration with Agents +### _reply_tweet -The TwitterTool can be particularly powerful when integrated with AI agents. Here are several examples of agent integrations: +```python +def _reply_tweet(self, tweet_id: int, reply: str) -> None +``` -### 1. Medical Information Bot +Posts a reply to a specific tweet. -This example shows how to create a medical information bot that shares health facts: +**Parameters:** +- tweet_id (int): ID of the tweet to reply to +- reply (str): Text content of the reply +**Example:** ```python -from swarms import Agent -from swarms_models import OpenAIChat - -# Initialize the AI model -model = OpenAIChat( - model_name="gpt-4", - max_tokens=3000, - openai_api_key=os.getenv("OPENAI_API_KEY") +twitter_tool._reply_tweet( + tweet_id=1234567890, + reply="Thank you for sharing this insight!" ) +``` -# Create a medical expert agent -medical_expert = Agent( - agent_name="Medical Expert", - system_prompt=""" - You are a medical expert sharing evidence-based health information. - Your tweets should be: - - Accurate and scientifically sound - - Easy to understand - - Engaging and relevant - - Within Twitter's character limit - """, - llm=model -) +### _post_tweet -# Function to generate and post medical tweets -def post_medical_fact(): - prompt = "Share an interesting medical fact that would be helpful for the general public." - tweet_text = medical_expert.run(prompt) - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) +```python +def _post_tweet(self, tweet: str) -> Dict[str, Any] ``` -### 2. News Summarization Bot +Creates a new tweet. + +**Parameters:** +- tweet (str): Text content of the tweet -This example demonstrates how to create a bot that summarizes news articles: +**Returns:** +- Dict[str, Any]: Response from Twitter API +**Example:** ```python -# Create a news summarization agent -news_agent = Agent( - agent_name="News Summarizer", - system_prompt=""" - You are a skilled news editor who excels at creating concise, - accurate summaries of news articles while maintaining the key points. - Your summaries should be: - - Factual and unbiased - - Clear and concise - - Properly attributed - - Under 280 characters - """, - llm=model +twitter_tool._post_tweet( + tweet="Exploring the fascinating world of AI and machine learning! #AI #ML" ) +``` -def summarize_and_tweet(article_url): - # Generate summary - prompt = f"Summarize this news article in a tweet-length format: {article_url}" - summary = news_agent.run(prompt) - - # Post the summary - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(f"{summary} Source: {article_url}") +### _like_tweet + +```python +def _like_tweet(self, tweet_id: int) -> None ``` -### 3. Interactive Q&A Bot +Likes a specific tweet. -This example shows how to create a bot that responds to user questions: +**Parameters:** +- tweet_id (int): ID of the tweet to like +**Example:** ```python -class TwitterQABot: - def __init__(self): - self.twitter_tool = TwitterTool(options) - self.qa_agent = Agent( - agent_name="Q&A Expert", - system_prompt=""" - You are an expert at providing clear, concise answers to questions. - Your responses should be: - - Accurate and informative - - Conversational in tone - - Limited to 280 characters - - Include relevant hashtags when appropriate - """, - llm=model - ) - - def handle_question(self, tweet_id, question): - # Generate response - response = self.qa_agent.run(f"Answer this question: {question}") - - # Reply to the tweet - reply_tweet = self.twitter_tool.get_function('reply_tweet') - reply_tweet(tweet_id=tweet_id, reply=response) +twitter_tool._like_tweet(tweet_id=1234567890) +``` -qa_bot = TwitterQABot() -qa_bot.handle_question(123456789, "What causes climate change?") +### _quote_tweet + +```python +def _quote_tweet(self, tweet_id: int, quote: str) -> None ``` -## Best Practices +Creates a quote tweet. -When using the TwitterTool, especially with agents, consider these best practices: +**Parameters:** +- tweet_id (int): ID of the tweet to quote +- quote (str): Text content to add to the quote -1. **Rate Limiting**: Implement delays between tweets to comply with Twitter's rate limits: +**Example:** ```python -import time - -def post_with_rate_limit(tweet_text, delay=60): - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) - time.sleep(delay) # Wait 60 seconds between tweets +twitter_tool._quote_tweet( + tweet_id=1234567890, + quote="This is a fascinating perspective on AI development!" +) ``` -2. **Content Tracking**: Maintain a record of posted content to avoid duplicates: -```python -posted_tweets = set() +## Integration Examples -def post_unique_tweet(tweet_text): - if tweet_text not in posted_tweets: - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) - posted_tweets.add(tweet_text) -``` +### Basic Usage with Environment Variables -3. **Error Handling**: Implement robust error handling for API failures: ```python -def safe_tweet(tweet_text): - try: - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) - except Exception as e: - logging.error(f"Failed to post tweet: {e}") - # Implement retry logic or fallback behavior +import os +from dotenv import load_dotenv +from swarms_tools import TwitterTool + +load_dotenv() + +options = { + "id": "twitter_bot", + "name": "AI Twitter Bot", + "description": "An AI-powered Twitter bot", + "credentials": { + "apiKey": os.getenv("TWITTER_API_KEY"), + "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), + "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), + "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET"), + } +} + +twitter_tool = TwitterTool(options) ``` -4. **Content Validation**: Validate content before posting: +### Integration with AI Agent for Content Generation + ```python -def validate_and_post(tweet_text): - if len(tweet_text) > 280: - tweet_text = tweet_text[:277] + "..." - - # Check for prohibited content - prohibited_terms = ["spam", "inappropriate"] - if any(term in tweet_text.lower() for term in prohibited_terms): - return False +from swarms import Agent +from swarms_models import OpenAIChat +from swarms_tools import TwitterTool +import os + +from dotenv import load_dotenv + +load_dotenv() + +# Initialize the model +model = OpenAIChat( + model_name="gpt-4", + max_tokens=3000, + openai_api_key=os.getenv("OPENAI_API_KEY") +) + +# Create content generation agent +content_agent = Agent( + agent_name="Content Creator", + system_prompt=""" + You are an expert content creator for Twitter. + Create engaging, informative tweets that: + 1. Are under 280 characters + 2. Use appropriate hashtags + 3. Maintain a professional tone + 4. Include relevant calls to action + """, + llm=model +) + +class TwitterContentBot: + def __init__(self, twitter_tool: TwitterTool, agent: Agent): + self.twitter_tool = twitter_tool + self.agent = agent + self.post_tweet = twitter_tool.get_function('post_tweet') - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) - return True + def generate_and_post(self, topic: str) -> None: + """ + Generates and posts content about a specific topic. + + Args: + topic (str): The topic to create content about + """ + prompt = f"Create an engaging tweet about {topic}" + tweet_content = self.agent.run(prompt) + self.post_tweet(tweet_content) + +# Usage +bot = TwitterContentBot(twitter_tool, content_agent) +bot.generate_and_post("artificial intelligence") ``` -## Advanced Features +### Automated Engagement System -### Scheduled Posting +```python +import time +from typing import List, Dict -Implement scheduled posting using Python's built-in scheduling capabilities: +class TwitterEngagementSystem: + def __init__(self, twitter_tool: TwitterTool): + self.twitter_tool = twitter_tool + self.like_tweet = twitter_tool.get_function('like_tweet') + self.reply_tweet = twitter_tool.get_function('reply_tweet') + self.get_metrics = twitter_tool.get_function('get_metrics') + + # Track engagement history + self.engagement_history: List[Dict] = [] + + def engage_with_tweet(self, tweet_id: int, should_like: bool = True, + reply_text: Optional[str] = None) -> None: + """ + Engages with a specific tweet through likes and replies. + + Args: + tweet_id (int): The ID of the tweet to engage with + should_like (bool): Whether to like the tweet + reply_text (Optional[str]): Text to reply with, if any + """ + try: + if should_like: + self.like_tweet(tweet_id) + + if reply_text: + self.reply_tweet(tweet_id, reply_text) + + # Record engagement + self.engagement_history.append({ + 'timestamp': time.time(), + 'tweet_id': tweet_id, + 'actions': { + 'liked': should_like, + 'replied': bool(reply_text) + } + }) + + except Exception as e: + print(f"Failed to engage with tweet {tweet_id}: {e}") + +# Usage +engagement_system = TwitterEngagementSystem(twitter_tool) +engagement_system.engage_with_tweet( + tweet_id=1234567890, + should_like=True, + reply_text="Great insights! Thanks for sharing." +) +``` + +### Scheduled Content System ```python -from datetime import datetime import schedule +from datetime import datetime, timedelta -def scheduled_tweet_job(): - twitter_tool = TwitterTool(options) - post_tweet = twitter_tool.get_function('post_tweet') +class ScheduledTwitterBot: + def __init__(self, twitter_tool: TwitterTool, agent: Agent): + self.twitter_tool = twitter_tool + self.agent = agent + self.post_tweet = twitter_tool.get_function('post_tweet') + self.posted_tweets: List[str] = [] - # Generate content using an agent - content = medical_expert.run("Generate a health tip of the day") - post_tweet(content) - -# Schedule tweets for specific times -schedule.every().day.at("10:00").do(scheduled_tweet_job) -schedule.every().day.at("15:00").do(scheduled_tweet_job) - + def generate_daily_content(self) -> None: + """ + Generates and posts daily content based on the current date. + """ + today = datetime.now() + prompt = f""" + Create an engaging tweet for {today.strftime('%A, %B %d')}. + Focus on technology trends and insights. + """ + + tweet_content = self.agent.run(prompt) + + # Avoid duplicate content + if tweet_content not in self.posted_tweets: + self.post_tweet(tweet_content) + self.posted_tweets.append(tweet_content) + + # Keep only last 100 tweets in memory + if len(self.posted_tweets) > 100: + self.posted_tweets.pop(0) + +# Usage +scheduled_bot = ScheduledTwitterBot(twitter_tool, content_agent) + +# Schedule daily posts +schedule.every().day.at("10:00").do(scheduled_bot.generate_daily_content) +schedule.every().day.at("15:00").do(scheduled_bot.generate_daily_content) + +# Run the scheduler while True: schedule.run_pending() time.sleep(60) ``` -### Analytics Integration +## Error Handling -Track the performance of your tweets: +The TwitterTool implements comprehensive error handling through try-except blocks. All methods catch and handle `tweepy.TweepyException` for Twitter API-specific errors. Here's an example of implementing custom error handling: ```python -class TweetAnalytics: - def __init__(self, twitter_tool): - self.twitter_tool = twitter_tool - self.metrics_history = [] - - def record_metrics(self): - get_metrics = self.twitter_tool.get_function('get_metrics') - current_metrics = get_metrics() - self.metrics_history.append({ - 'timestamp': datetime.now(), - 'metrics': current_metrics - }) - - def get_growth_rate(self): - if len(self.metrics_history) < 2: - return None +class TwitterToolWithCustomErrors(TwitterTool): + def _post_tweet(self, tweet: str) -> Dict[str, Any]: + """ + Enhanced tweet posting with custom error handling. + + Args: + tweet (str): The tweet content to post + + Returns: + Dict[str, Any]: Response from Twitter API + + Raises: + ValueError: If tweet exceeds character limit + """ + if len(tweet) > 280: + raise ValueError("Tweet exceeds 280 character limit") + + try: + return super()._post_tweet(tweet) + except tweepy.TweepyException as e: + self.logger.error(f"Twitter API error: {e}") + raise +``` + +## Rate Limiting + +Twitter's API has rate limits that should be respected. Here's an example implementation of rate limiting: + +```python +from time import time, sleep +from collections import deque + +class RateLimitedTwitterTool(TwitterTool): + def __init__(self, options: Dict[str, Any]) -> None: + super().__init__(options) + self.tweet_timestamps = deque(maxlen=300) # Track last 300 tweets + self.max_tweets_per_15min = 300 - latest = self.metrics_history[-1]['metrics'] - previous = self.metrics_history[-2]['metrics'] + def _check_rate_limit(self) -> None: + """ + Checks if we're within rate limits and waits if necessary. + """ + now = time() - return { - 'followers_growth': latest['followers'] - previous['followers'], - 'tweets_growth': latest['tweets'] - previous['tweets'] - } + # Remove timestamps older than 15 minutes + while self.tweet_timestamps and self.tweet_timestamps[0] < now - 900: + self.tweet_timestamps.popleft() + + if len(self.tweet_timestamps) >= self.max_tweets_per_15min: + sleep_time = 900 - (now - self.tweet_timestamps[0]) + if sleep_time > 0: + sleep(sleep_time) + + self.tweet_timestamps.append(now) + + def _post_tweet(self, tweet: str) -> Dict[str, Any]: + self._check_rate_limit() + return super()._post_tweet(tweet) ``` -## Troubleshooting +## Best Practices -Common issues and their solutions: +1. Always use environment variables for credentials: +```python +from dotenv import load_dotenv +load_dotenv() -1. **Authentication Errors**: Double-check your API credentials and ensure they're properly loaded from environment variables. +options = { + "credentials": { + "apiKey": os.getenv("TWITTER_API_KEY"), + "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), + "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), + "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET") + } +} +``` -2. **Rate Limiting**: If you encounter rate limit errors, implement exponential backoff: +2. Implement proper logging: ```python -import time -from random import uniform +import logging -def exponential_backoff(attempt): - wait_time = min(300, (2 ** attempt) + uniform(0, 1)) - time.sleep(wait_time) +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) +``` -def retry_post(tweet_text, max_attempts=5): - for attempt in range(max_attempts): +3. Use type hints consistently: +```python +from typing import Optional, Dict, Any, List + +def process_tweet( + tweet_id: int, + action: str, + content: Optional[str] = None +) -> Dict[str, Any]: + pass +``` + +4. Handle API rate limits gracefully: +```python +from time import sleep + +def retry_with_backoff(func, max_retries: int = 3): + for attempt in range(max_retries): try: - post_tweet = twitter_tool.get_function('post_tweet') - post_tweet(tweet_text) - return True - except Exception as e: - if attempt < max_attempts - 1: - exponential_backoff(attempt) - else: - raise e + return func() + except tweepy.TweepyException as e: + if attempt == max_retries - 1: + raise + sleep_time = (2 ** attempt) + random.uniform(0, 1) + sleep(sleep_time) +``` + +5. Validate input data: +```python +def validate_tweet(tweet: str) -> bool: + if not tweet or len(tweet) > 280: + return False + return True ``` -3. **Content Length Issues**: Implement automatic content truncation: +## Testing + +Example of a basic test suite for the TwitterTool: + ```python -def truncate_tweet(text, max_length=280): - if len(text) <= max_length: - return text +from swarms_tools import TwitterTool +import os +from dotenv import load_dotenv + +load_dotenv() + +def test_twitter_tool(): + # Test configuration + options = { + "id": "test_bot", + "credentials": { + "apiKey": os.getenv("TWITTER_API_KEY"), + "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), + "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), + "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET") + } + } + + # Initialize tool + tool = TwitterTool(options) + + # Test available functions + assert 'post_tweet' in tool.available_functions + assert 'get_metrics' in tool.available_functions - # Try to break at last space before limit - truncated = text[:max_length-3] - last_space = truncated.rfind(' ') - if last_space > 0: - truncated = truncated[:last_space] + # Test function retrieval + post_tweet = tool.get_function('post_tweet') + assert callable(post_tweet) - return truncated + "..." + # Test metrics + metrics = tool._get_metrics() + assert isinstance(metrics, dict) + assert 'followers' in metrics + assert 'following' in metrics + assert 'tweets' in metrics ``` -Remember to regularly check Twitter's API documentation for any updates or changes to rate limits and functionality. The TwitterTool is designed to be extensible, so you can add new features as needed for your specific use case. \ No newline at end of file +Remember to always test your code thoroughly before deploying it in a production environment. The TwitterTool is designed to be extensible, so you can add new features and customizations as needed for your specific use case. \ No newline at end of file