Rename twitter toolkit.

cookbook/examples/agents/shopping_partner.py

@@ -1,7 +1,6 @@
 from agno.agent import Agent
 from agno.models.openai import OpenAIChat
 from agno.tools.exa import ExaTools
-from agno.tools.firecrawl import FirecrawlTools
 
 agent = Agent(
     name="shopping partner",

cookbook/examples/apps/agentic_rag/app.py

@@ -18,6 +18,7 @@
     about_widget,
     add_message,
     display_tool_calls,
+
     export_chat_history,
     rename_session_widget,
     session_selector_widget,

cookbook/examples/apps/agentic_rag/utils.py

@@ -138,7 +138,6 @@ def session_selector_widget(agent: Agent, model_id: str) -> None:
             )
             st.rerun()
 
-
 def about_widget() -> None:
     """Display an about section in the sidebar"""
     st.sidebar.markdown("---")

cookbook/examples/apps/game_generator/README.md

@@ -15,7 +15,7 @@ source ~/.venvs/aienv/bin/activate
 pip install -r cookbook/examples/apps/game_generator/requirements.txt
 ```
 
-### 3. Export `OPENAI_API_KEY` 
+### 3. Export `OPENAI_API_KEY`
 
 ```shell
 export OPENAI_API_KEY=sk-***

cookbook/tools/x_tools.py

@@ -0,0 +1,78 @@
+from agno.agent import Agent
+from agno.tools.x import XTools
+
+"""
+To set up an X developer account and obtain the necessary keys, follow these steps:
+
+1. **Create an X Developer Account:**
+   - Go to the X Developer website: https://developer.x.com/
+   - Sign in with your X account or create a new one if you don't have an account.
+   - Apply for a developer account by providing the required information about your intended use of the X API.
+
+2. **Create a Project and App:**
+   - Once your developer account is approved, log in to the X Developer portal.
+   - Navigate to the "Projects & Apps" section and create a new project.
+   - Within the project, create a new app. This app will be used to generate the necessary API keys and tokens.
+   - You'll get a client id and client secret, but you can ignore them.
+
+3. **Generate API Keys, Tokens, and Client Credentials:**
+   - After creating the app, navigate to the "Keys and tokens" tab.
+   - Generate the following keys, tokens, and client credentials:
+     - **API Key (Consumer Key)**
+     - **API Secret Key (Consumer Secret)**
+     - **Bearer Token**
+     - **Access Token**
+     - **Access Token Secret**
+
+4. **Set Environment Variables:**
+   - Export the generated keys, tokens, and client credentials as environment variables in your system or provide them as arguments to the `XTools` constructor.
+     - `X_CONSUMER_KEY`
+     - `X_CONSUMER_SECRET`
+     - `X_ACCESS_TOKEN`
+     - `X_ACCESS_TOKEN_SECRET`
+     - `X_BEARER_TOKEN`
+"""
+
+
+# Initialize the x toolkit
+x_tools = XTools()
+
+# Create an agent with the X toolkit
+agent = Agent(
+    instructions=[
+        "Use your tools to interact with X (Twitter) as the authorized user @AgnoAgi",
+        "When asked to create a post, generate appropriate content based on the request",
+        "Do not actually post content unless explicitly instructed to do so",
+        "Provide informative responses about the user's timeline and posts",
+        "Respect X's usage policies and rate limits",
+    ],
+    tools=[x_tools],
+    show_tool_calls=True,
+    debug_mode=True,
+)
+
+# Example usage: Get your details
+agent.print_response("Can you return my x profile with my home timeline?", markdown=True)
+
+# # Example usage: Get information about a user
+# agent.print_response(
+#     "Can you retrieve information about this user https://x.com/AgnoAgi ",
+#     markdown=True,
+# )
+
+# # Example usage: Reply To a Post
+# agent.print_response(
+#     "Can you reply to this [post ID] post as a general message as to how great this project is: https://x.com/AgnoAgi",
+#     markdown=True,
+# )
+
+# # Example usage: Send a direct message
+# agent.print_response(
+#     "Send direct message to the user @AgnoAgi telling them I want to learn more about them and a link to their community.",
+#     markdown=True,
+# )
+
+# # Example usage: Create a new post
+# agent.print_response("Create & post content about how 2025 is the year of the AI agent", markdown=True)
+
+

libs/agno/agno/tools/twitter.py → libs/agno/agno/tools/x.py

@@ -11,7 +11,7 @@
     raise ImportError("`tweepy` not installed. Please install using `pip install tweepy`.")
 
 
-class TwitterTools(Toolkit):
+class XTools(Toolkit):
     def __init__(
         self,
         bearer_token: Optional[str] = None,
@@ -21,7 +21,7 @@ def __init__(
         access_token_secret: Optional[str] = None,
     ):
         """
-        Initialize the TwitterTools.
+        Initialize the XTools.
 
         Args:
             bearer_token Optional[str]: The bearer token for Twitter API.
@@ -30,13 +30,13 @@ def __init__(
             access_token Optional[str]: The access token for Twitter API.
             access_token_secret Optional[str]: The access token secret for Twitter API.
         """
-        super().__init__(name="twitter")
+        super().__init__(name="x")
 
-        self.bearer_token = bearer_token or os.getenv("TWITTER_BEARER_TOKEN")
-        self.consumer_key = consumer_key or os.getenv("TWITTER_CONSUMER_KEY")
-        self.consumer_secret = consumer_secret or os.getenv("TWITTER_CONSUMER_SECRET")
-        self.access_token = access_token or os.getenv("TWITTER_ACCESS_TOKEN")
-        self.access_token_secret = access_token_secret or os.getenv("TWITTER_ACCESS_TOKEN_SECRET")
+        self.bearer_token = bearer_token or os.getenv("X_BEARER_TOKEN")
+        self.consumer_key = consumer_key or os.getenv("X_CONSUMER_KEY")
+        self.consumer_secret = consumer_secret or os.getenv("X_CONSUMER_SECRET")
+        self.access_token = access_token or os.getenv("X_ACCESS_TOKEN")
+        self.access_token_secret = access_token_secret or os.getenv("X_ACCESS_TOKEN_SECRET")
 
         self.client = tweepy.Client(
             bearer_token=self.bearer_token,
@@ -45,62 +45,59 @@ def __init__(
             access_token=self.access_token,
             access_token_secret=self.access_token_secret,
         )
-        self.auth = tweepy.OAuth1UserHandler(
-            self.consumer_key, self.consumer_secret, self.access_token, self.access_token_secret
-        )
-        self.api = tweepy.API(self.auth)
-        self.register(self.create_tweet)
-        self.register(self.reply_to_tweet)
+
+        self.register(self.create_post)
+        self.register(self.reply_to_post)
         self.register(self.send_dm)
         self.register(self.get_user_info)
         self.register(self.get_home_timeline)
 
-    def create_tweet(self, text: str) -> str:
+    def create_post(self, text: str) -> str:
         """
-        Create a new tweet.
+        Create a new X post.
 
         Args:
-            text (str): The content of the tweet to create.
+            text (str): The content of the post to create.
 
         Returns:
-            A JSON-formatted string containing the response from Twitter API with the created tweet details,
-            or an error message if the tweet creation fails.
+            A JSON-formatted string containing the response from X API (Twitter API) with the created post details,
+            or an error message if the post creation fails.
         """
-        logger.debug(f"Attempting to create tweet with text: {text}")
+        logger.debug(f"Attempting to create post with text: {text}")
         try:
             response = self.client.create_tweet(text=text)
-            tweet_id = response.data["id"]
+            post_id = response.data["id"]
             user = self.client.get_me().data
-            tweet_url = f"https://twitter.com/{user.username}/status/{tweet_id}"
+            post_url = f"https://x.com/{user.username}/status/{post_id}"
 
-            result = {"message": "Tweet successfully posted!", "url": tweet_url}
+            result = {"message": "Post successfully created!", "url": post_url}
             return json.dumps(result, indent=2)
         except tweepy.TweepyException as e:
-            logger.error(f"Error creating tweet: {e}")
+            logger.error(f"Error creating post: {e}")
             return json.dumps({"error": str(e)})
 
-    def reply_to_tweet(self, tweet_id: str, text: str) -> str:
+    def reply_to_post(self, post_id: str, text: str) -> str:
         """
-        Reply to an existing tweet.
+        Reply to an existing post.
 
         Args:
-            tweet_id (str): The ID of the tweet to reply to.
-            text (str): The content of the reply tweet.
+            post_id (str): The ID of the post to reply to.
+            text (str): The content of the reply post.
 
         Returns:
-            A JSON-formatted string containing the response from Twitter API with the reply tweet details,
+            A JSON-formatted string containing the response from Twitter API with the reply post details,
             or an error message if the reply fails.
         """
-        logger.debug(f"Attempting to reply to {tweet_id} with text {text}")
+        logger.debug(f"Attempting to reply to {post_id} with text {text}")
         try:
-            response = self.client.create_tweet(text=text, in_reply_to_tweet_id=tweet_id)
+            response = self.client.create_tweet(text=text, in_reply_to_tweet_id=post_id)
             reply_id = response.data["id"]
             user = self.client.get_me().data
             reply_url = f"https://twitter.com/{user.username}/status/{reply_id}"
             result = {"message": "Reply successfully posted!", "url": reply_url}
             return json.dumps(result, indent=2)
         except tweepy.TweepyException as e:
-            logger.error(f"Error replying to tweet: {e}")
+            logger.error(f"Error replying to post: {e}")
             return json.dumps({"error": str(e)})
 
     def send_dm(self, recipient: str, text: str) -> str:
@@ -136,7 +133,7 @@ def send_dm(self, recipient: str, text: str) -> str:
             }
             return json.dumps(result, indent=2)
         except tweepy.TweepyException as e:
-            logger.error(f"Error sending DM: {e}")
+            logger.error(f"Error from X while sending DM: {e}")
             error_message = str(e)
             if "User not found" in error_message:
                 error_message = f"User '{recipient}' not found. Please check the username or user ID."

libs/agno/agno/utils/common.py

@@ -51,7 +51,6 @@ def dataclass_to_dict(dataclass_object, exclude: Optional[set[str]] = None, excl
 
 def nested_model_dump(value):
     from pydantic import BaseModel
-
     if isinstance(value, BaseModel):
         return value.model_dump()
     elif isinstance(value, dict):