feat(tools): Add X(twitter) tool

api/core/tools/provider/builtin/x/tools/get_user_timeline.py

@@ -0,0 +1,180 @@
+"""
+Get User Timeline Tool for X (Twitter)
+"""
+
+from datetime import datetime
+from typing import Any, Union
+
+import tweepy
+
+from core.tools.entities.tool_entities import ToolInvokeMessage
+from core.tools.tool.builtin_tool import BuiltinTool
+
+
+class GetUserTimelineTool(BuiltinTool):
+    def _validate_max_results(self, max_results: str) -> int:
+        """
+        Validate and normalize max_results parameter
+        """
+        try:
+            max_results = int(max_results)
+            if max_results < 5:
+                return 5
+            elif max_results > 100:
+                return 100
+            return max_results
+        except (TypeError, ValueError):
+            return 10
+
+    def _convert_tweet_to_dict(self, tweet: tweepy.Tweet) -> dict[str, Any]:
+        """
+        Convert tweet object to dictionary, handling datetime serialization
+        """
+        tweet_dict = {}
+        # Convert Tweet object to dictionary
+        for field in tweet.data:
+            value = tweet.data[field]
+            # Skip None values
+            if value is None:
+                continue
+            # Handle datetime fields
+            if field == "created_at" and isinstance(value, datetime):
+                tweet_dict[field] = value.isoformat()
+            else:
+                tweet_dict[field] = value
+
+        # Handle media attachments
+        if "attachments" in tweet_dict and "media_keys" in tweet_dict["attachments"]:
+            media_keys = tweet_dict["attachments"]["media_keys"]
+            tweet_dict["media"] = []
+
+            for media_key in media_keys:
+                media_info = {"media_key": media_key, "type": None, "url": None}
+                tweet_dict["media"].append(media_info)
+
+        return tweet_dict
+
+    def _invoke(
+        self, user_id: str, tool_parameters: dict[str, Any]
+    ) -> Union[ToolInvokeMessage, list[ToolInvokeMessage]]:
+        """
+        Get user timeline from Twitter
+        """
+        try:
+            username = tool_parameters.get("username", "").strip().lstrip("@")
+            if not username:
+                return ToolInvokeMessage(message="Username is required", status="error")
+
+            max_results = self._validate_max_results(tool_parameters.get("max_results"))
+
+            client = tweepy.Client(
+                bearer_token=self.runtime.credentials["bearer_token"],
+                consumer_key=self.runtime.credentials["consumer_key"],
+                consumer_secret=self.runtime.credentials["consumer_secret"],
+                access_token=self.runtime.credentials["access_token"],
+                access_token_secret=self.runtime.credentials["access_token_secret"],
+            )
+
+            tweet_fields = [
+                "attachments",
+                "author_id",
+                "context_annotations",
+                "conversation_id",
+                "created_at",
+                "entities",
+                "geo",
+                "id",
+                "in_reply_to_user_id",
+                "lang",
+                "public_metrics",
+                "possibly_sensitive",
+                "referenced_tweets",
+                "reply_settings",
+                "source",
+                "text",
+                "withheld",
+            ]
+
+            user_fields = [
+                "created_at",
+                "description",
+                "entities",
+                "id",
+                "location",
+                "name",
+                "pinned_tweet_id",
+                "profile_image_url",
+                "protected",
+                "public_metrics",
+                "url",
+                "username",
+                "verified",
+                "withheld",
+            ]
+
+            media_fields = [
+                "duration_ms",
+                "height",
+                "media_key",
+                "preview_image_url",
+                "type",
+                "url",
+                "width",
+                "public_metrics",
+                "alt_text",
+            ]
+
+            try:
+                # Get user ID from username
+                user_response = client.get_user(
+                    username=username,
+                )
+                if not user_response.data:
+                    return ToolInvokeMessage(message=f"User @{username} not found", status="error")
+
+                user_data = user_response.data
+                # Get user's tweets
+                tweets_response = client.get_users_tweets(
+                    id=user_data.id,
+                    max_results=max_results,
+                    tweet_fields=tweet_fields,
+                    user_fields=user_fields,
+                    media_fields=media_fields,
+                    exclude=["retweets"],  # Exclude retweets to get more original content
+                )
+
+                print(tweets_response.data)
+                if not tweets_response.data:
+                    return ToolInvokeMessage(
+                        message=f"No tweets found for user @{username}",
+                        status="success",
+                    )
+
+                tweets = []
+                for tweet in tweets_response.data:
+                    tweet_dict = self._convert_tweet_to_dict(tweet)
+                    tweets.append(tweet_dict)
+
+                # Convert user data
+                user_dict = {}
+                for field in user_data:
+                    value = user_data[field]
+                    if value is None:
+                        continue
+                    if field == "created_at" and isinstance(value, datetime):
+                        user_dict[field] = value.isoformat()
+                    else:
+                        user_dict[field] = value
+
+                return self.create_json_message(
+                    {
+                        "message": f"Retrieved {len(tweets)} tweets from user @{username}",
+                        "user": user_dict,
+                        "tweets": tweets,
+                    }
+                )
+            except tweepy.TweepyException as te:
+                return ToolInvokeMessage(message=f"Twitter API error: {str(te)}", status="error")
+
+        except Exception as e:
+            return ToolInvokeMessage(message=f"Error retrieving user timeline: {str(e)}", status="error")

api/core/tools/provider/builtin/x/tools/get_user_timeline.yaml

@@ -0,0 +1,41 @@
+identity:
+  name: get_user_timeline
+  author: Yash Parmar
+  label:
+    en_US: Get User Timeline
+    zh_Hans: 获取用户时间线
+  icon: icon.svg
+
+description:
+  human:
+    en_US: Get comprehensive tweet data from a user's timeline including media, polls, metrics, and more
+    zh_Hans: 获取用户时间线上的完整推文数据，包括媒体、投票、指标等
+  llm: A tool for retrieving detailed tweet data from a specified user's timeline on X (Twitter), including comprehensive metadata, media attachments, polls, metrics, and related content.
+
+parameters:
+  - name: username
+    type: string
+    required: true
+    label:
+      en_US: Username
+      zh_Hans: 用户名
+    human_description:
+      en_US: The username of the account to get tweets from (without @ symbol)
+      zh_Hans: 要获取推文的账户用户名（不带@符号）
+    llm_description: The username of the account to get tweets from. Do not include the @ symbol.
+    form: llm
+
+  - name: max_results
+    type: number
+    required: false
+    default: 10
+    minimum: 5
+    maximum: 100
+    label:
+      en_US: Maximum Results
+      zh_Hans: 最大结果数
+    human_description:
+      en_US: Number of tweets to return (5-100, default 10)
+      zh_Hans: 要返回的推文数量（5-100，默认10）
+    llm_description: Number of tweets to retrieve. Must be between 5 and 100. Default is 10.
+    form: llm

api/core/tools/provider/builtin/x/tools/like_tweet.py

@@ -0,0 +1,67 @@
+"""
+Like Tweet Tool for liking/unliking tweets
+"""
+
+from typing import Any, Union
+
+import tweepy
+
+from core.tools.entities.tool_entities import ToolInvokeMessage
+from core.tools.tool.builtin_tool import BuiltinTool
+
+
+class LikeTweetTool(BuiltinTool):
+    def _invoke(
+        self, user_id: str, tool_parameters: dict[str, Any]
+    ) -> Union[ToolInvokeMessage, list[ToolInvokeMessage]]:
+        """
+        Like or unlike a tweet
+
+        Args:
+            user_id: The ID of the user making the request
+            tool_parameters: Dictionary containing tweet_id and action parameters
+
+        Returns:
+            ToolInvokeMessage with the result of the operation
+        """
+        try:
+            # Initialize client
+            client = tweepy.Client(
+                bearer_token=self.runtime.credentials["bearer_token"],
+                consumer_key=self.runtime.credentials["consumer_key"],
+                consumer_secret=self.runtime.credentials["consumer_secret"],
+                access_token=self.runtime.credentials["access_token"],
+                access_token_secret=self.runtime.credentials["access_token_secret"],
+            )
+
+            # Get and validate parameters
+            tweet_id = tool_parameters.get("tweet_id")
+            action = tool_parameters.get("action")
+
+            if not tweet_id:
+                return ToolInvokeMessage(message="Tweet ID is required", status="error")
+
+            if action not in ["like", "unlike"]:
+                return ToolInvokeMessage(
+                    message="Invalid action. Must be either 'like' or 'unlike'",
+                    status="error",
+                )
+
+            try:
+                if action == "like":
+                    response = client.like(tweet_id=tweet_id, user_auth=True)
+                    message = f"Successfully liked tweet {tweet_id}"
+                else:
+                    response = client.unlike(tweet_id=tweet_id, user_auth=True)
+                    message = f"Successfully unliked tweet {tweet_id}"
+
+                return ToolInvokeMessage(
+                    message=message,
+                    data={"tweet_id": tweet_id, "action": action, "success": True},
+                )
+
+            except tweepy.TweepyException as te:
+                return ToolInvokeMessage(message=f"Twitter API error: {str(te)}", status="error")
+
+        except Exception as e:
+            return ToolInvokeMessage(message=f"Error performing {action} action: {str(e)}", status="error")

api/core/tools/provider/builtin/x/tools/like_tweet.yaml

@@ -0,0 +1,47 @@
+identity:
+  name: like_tweet
+  author: Yash Parmar
+  label:
+    en_US: Like/Unlike Tweet
+    zh_Hans: 点赞/取消点赞推文
+  icon: icon.svg
+
+description:
+  human:
+    en_US: Like or unlike a tweet on X (Twitter)
+    zh_Hans: 在 X (Twitter) 上点赞或取消点赞推文
+  llm: A tool for liking or unliking tweets on X (Twitter). Requires tweet ID and action (like/unlike).
+
+parameters:
+  - name: tweet_id
+    type: string
+    required: true
+    label:
+      en_US: Tweet ID
+      zh_Hans: 推文ID
+    human_description:
+      en_US: The ID of the tweet to like or unlike
+      zh_Hans: 要点赞或取消点赞的推文ID
+    llm_description: The unique identifier of the tweet to like or unlike
+    form: llm
+
+  - name: action
+    type: select
+    required: true
+    options:
+      - value: like
+        label:
+          en_US: Like
+          zh_Hans: 点赞
+      - value: unlike
+        label:
+          en_US: Unlike
+          zh_Hans: 取消点赞
+    label:
+      en_US: Action
+      zh_Hans: 操作
+    human_description:
+      en_US: Choose whether to like or unlike the tweet
+      zh_Hans: 选择点赞或取消点赞推文
+    llm_description: Select the action to perform - either like or unlike the specified tweet
+    form: llm