Twitter tool

gui/pages/Content/Toolkits/ToolkitWorkspace.js

@@ -1,7 +1,7 @@
 import React, {useEffect, useState} from 'react';
 import Image from 'next/image';
 import {ToastContainer, toast} from 'react-toastify';
-import {updateToolConfig, getToolConfig, authenticateGoogleCred} from "@/pages/api/DashboardService";
+import {updateToolConfig, getToolConfig, authenticateGoogleCred, authenticateTwitterCred} from "@/pages/api/DashboardService";
 import styles from './Tool.module.css';
 import {EventBus} from "@/utils/eventBus";
 
@@ -25,6 +25,13 @@ export default function ToolkitWorkspace({toolkitDetails}){
       window.location.href = `https://accounts.google.com/o/oauth2/v2/auth?client_id=${client_id}&redirect_uri=${redirect_uri}&access_type=offline&response_type=code&scope=${scope}`;
     }
 
+    function getTwitterToken(oauth_data){
+      const oauth_token = oauth_data.oauth_token
+      const oauth_token_secret = oauth_data.oauth_token_secret
+      const authUrl = `https://api.twitter.com/oauth/authenticate?oauth_token=${oauth_token}`
+      window.location.href = authUrl
+    }
+
     useEffect(() => {
       if(toolkitDetails !== null) {
         if (toolkitDetails.tools) {
@@ -36,7 +43,7 @@ export default function ToolkitWorkspace({toolkitDetails}){
             const apiConfigs = response.data || [];
             setApiConfigs(apiConfigs);
           })
-          .catch((error) => {
+          .catch((errPor) => {
             console.log('Error fetching API data:', error);
           })
           .finally(() => {
@@ -71,6 +78,16 @@ export default function ToolkitWorkspace({toolkitDetails}){
         });
     };
 
+    const handleTwitterAuthClick = async () => {
+      authenticateTwitterCred(toolkitDetails.id)
+      .then((response) => {
+        getTwitterToken(response.data);
+      })
+      .catch((error) => {
+        console.error('Error fetching data: ', error);
+      });
+    };
+
     return (<>
         <div className={styles.tools_container}>
           <div style={{display: 'flex',justifyContent:'flex-start',marginBottom:'20px', width:'600px'}}>
@@ -116,6 +133,7 @@ export default function ToolkitWorkspace({toolkitDetails}){
             <div style={{ marginLeft: 'auto', display: 'flex', justifyContent:'space-between'}}>
               <div>
                 {toolkitDetails.name === 'Google Calendar Toolkit' && <button style={{width:'200px'}} className={styles.primary_button} onClick={handleAuthenticateClick}>Authenticate Tool</button>}
+                {toolkitDetails.name === 'Twitter Toolkit' && <button style={{width:'200px'}} className={styles.primary_button} onClick={handleTwitterAuthClick}>Authenticate Tool</button>}
               </div>
               <div style={{ display: 'flex', justifyContent: 'flex-end' }}>
                 <button className={styles.primary_button} onClick={handleUpdateChanges} >Update Changes</button>

gui/pages/api/DashboardService.js

@@ -128,6 +128,10 @@ export const authenticateGoogleCred = (toolKitId) => {
   return api.get(`/google/get_google_creds/toolkit_id/${toolKitId}`);
 }
 
+export const authenticateTwitterCred = (toolKitId) => {
+  return api.get(`/twitter/get_twitter_creds/toolkit_id/${toolKitId}`);
+}
+
 export const fetchToolTemplateList = () => {
   return api.get(`/toolkits/get/list?page=0`);
 }

main.py

@@ -15,6 +15,10 @@
 from sqlalchemy.orm import sessionmaker
 
 import superagi
+import urllib.parse
+import http.client as http_client
+from superagi.helper.twitter_tokens import TwitterTokens
+from datetime import datetime, timedelta
 from superagi.agent.agent_prompt_builder import AgentPromptBuilder
 from superagi.config.config import get_config
 from superagi.controllers.agent import router as agent_router
@@ -320,6 +324,31 @@ async def google_auth_calendar(code: str = Query(...), Authorize: AuthJWT = Depe
     frontend_url = superagi.config.config.get_config("FRONTEND_URL", "http://localhost:3000")
     return RedirectResponse(frontend_url)
 
+@app.get('/oauth-twitter')
+async def twitter_oauth(oauth_token: str = Query(...),oauth_verifier: str = Query(...), Authorize: AuthJWT = Depends()):
+    token_uri = f'https://api.twitter.com/oauth/access_token?oauth_verifier={oauth_verifier}&oauth_token={oauth_token}'
+    conn = http_client.HTTPSConnection("api.twitter.com")
+    conn.request("POST", token_uri, "")
+    res = conn.getresponse()
+    response_data = res.read().decode('utf-8')
+    conn.close()
+    response = dict(urllib.parse.parse_qsl(response_data))
+    root_dir = superagi.config.config.get_config('RESOURCES_OUTPUT_ROOT_DIR')
+    file_name = "twitter_credentials.pickle"
+    final_path = file_name
+    if root_dir is not None:
+        root_dir = root_dir if root_dir.startswith("/") else os.getcwd() + "/" + root_dir
+        root_dir = root_dir if root_dir.endswith("/") else root_dir + "/"
+        final_path = root_dir + file_name
+    else:
+        final_path = os.getcwd() + "/" + file_name
+    try:
+        with open(final_path, mode="wb") as file:
+            pickle.dump(response, file)
+    except Exception as err:
+        return f"Error: {err}"
+    frontend_url = superagi.config.config.get_config("FRONTEND_URL", "http://localhost:3000")
+    return RedirectResponse(frontend_url)
 
 @app.get('/github-login')
 def github_login():
@@ -411,6 +440,16 @@ def get_google_calendar_tool_configs(toolkit_id: int):
         "client_id": google_calendar_config.value
     }
 
+@app.get("/twitter/get_twitter_creds/toolkit_id/{toolkit_id}")
+def get_twitter_tool_configs(toolkit_id: int):
+    twitter_config_key = db.session.query(ToolConfig).filter(ToolConfig.toolkit_id == toolkit_id,ToolConfig.key == "TWITTER_API_KEY").first()
+    twitter_config_secret = db.session.query(ToolConfig).filter(ToolConfig.toolkit_id == toolkit_id,ToolConfig.key == "TWITTER_API_SECRET").first()
+    api_data =  {
+        "api_key": twitter_config_key.value,
+        "api_secret": twitter_config_secret.value
+    }
+    response = TwitterTokens().get_request_token(api_data)
+    return response
 
 @app.get("/validate-open-ai-key/{open_ai_key}")
 async def root(open_ai_key: str, Authorize: AuthJWT = Depends()):

superagi/helper/twitter_tokens.py

@@ -0,0 +1,93 @@
+import os
+import pickle
+import json
+import hmac
+import time
+import random
+import base64
+import hashlib
+import urllib.parse
+import http.client as http_client
+from superagi.config.config import get_config
+from sqlalchemy.orm import sessionmaker
+from superagi.models.db import connect_db
+from superagi.models.tool_config import ToolConfig
+from superagi.resource_manager.manager import ResourceManager
+
+class TwitterTokens:
+
+    def get_request_token(self,api_data):
+        api_key = api_data["api_key"]
+        api_secret_key = api_data["api_secret"]
+        http_method = 'POST'
+        base_url = 'https://api.twitter.com/oauth/request_token'
+
+        params = {
+            'oauth_callback': 'http://localhost:3000/api/oauth-twitter',
+            'oauth_consumer_key': api_key,
+            'oauth_nonce': self.gen_nonce(),
+            'oauth_signature_method': 'HMAC-SHA1',
+            'oauth_timestamp': int(time.time()),
+            'oauth_version': '1.0'
+        }
+
+        params_sorted = sorted(params.items())
+        params_qs = '&'.join([f'{k}={self.percent_encode(str(v))}' for k, v in params_sorted])
+
+        base_string = f'{http_method}&{self.percent_encode(base_url)}&{self.percent_encode(params_qs)}'
+
+        signing_key = f'{self.percent_encode(api_secret_key)}&'
+        signature = hmac.new(signing_key.encode(), base_string.encode(), hashlib.sha1)
+        params['oauth_signature'] = base64.b64encode(signature.digest()).decode()
+
+        auth_header = 'OAuth ' + ', '.join([f'{k}="{self.percent_encode(str(v))}"' for k, v in params.items()])
+
+        headers = {
+            'Content-Type': 'application/x-www-form-urlencoded',
+            'Authorization': auth_header
+        }
+        conn = http_client.HTTPSConnection("api.twitter.com")
+        conn.request("POST", "/oauth/request_token", "", headers)
+        res = conn.getresponse()
+        response_data = res.read().decode('utf-8')
+        conn.close()
+        request_token_resp = dict(urllib.parse.parse_qsl(response_data))
+        return request_token_resp
+
+    def percent_encode(self, val):
+        return urllib.parse.quote(val, safe='')
+
+    def gen_nonce(self):
+        nonce = ''.join([str(random.randint(0, 9)) for i in range(32)])
+        return nonce
+
+    def get_twitter_creds(self, toolkit_id):
+        file_name = "twitter_credentials.pickle"
+        root_dir = get_config('RESOURCES_OUTPUT_ROOT_DIR')
+        file_path = file_name
+        if root_dir is not None:
+            root_dir = root_dir if root_dir.startswith("/") else os.getcwd() + "/" + root_dir
+            root_dir = root_dir if root_dir.endswith("/") else root_dir + "/"
+            file_path = root_dir + file_name
+        else:
+            file_path = os.getcwd() + "/" + file_name
+        if os.path.exists(file_path):
+            with open(file_path,'rb') as file:
+                creds = pickle.load(file)
+            if isinstance(creds, str):
+                creds = json.loads(creds)
+        engine = connect_db()
+        Session = sessionmaker(bind=engine)
+        session = Session()
+        twitter_creds = session.query(ToolConfig).filter(ToolConfig.toolkit_id == toolkit_id).all()
+        api_key = ""
+        api_key_secret = ""
+        for credentials in twitter_creds:
+            credentials = credentials.__dict__
+            if credentials["key"] == "TWITTER_API_KEY":
+                api_key = credentials["value"]
+            if credentials["key"] == "TWITTER_API_SECRET":
+                api_key_secret = credentials["value"]
+        creds["api_key"] = api_key
+        creds["api_key_secret"] = api_key_secret
+        return creds

superagi/tools/twitter/send_tweets.py

@@ -0,0 +1,87 @@
+import os
+import json
+import base64
+import requests
+from typing import Any, Type
+from pydantic import BaseModel, Field
+from superagi.tools.base_tool import BaseTool
+from superagi.helper.twitter_tokens import TwitterTokens
+from requests_oauthlib import OAuth1
+from requests_oauthlib import OAuth1Session
+from superagi.helper.resource_helper import ResourceHelper
+
+class SendTweetsInput(BaseModel):
+    tweet_text: str = Field(..., description="Tweet text to be posted from twitter handle, if no value is given keep the default value as 'None'")
+    is_media: bool = Field(..., description="'True' if there is any media to be posted with Tweet else 'False'.")
+    media_num: int = Field(..., description="Integer value for the number of media files to be uploaded, default value is 0")
+    media_files: list = Field(..., description="Name of the media files to be uploaded.")
+
+class SendTweetsTool(BaseTool):
+    name: str = "Send Tweets Tool"
+    args_schema: Type[BaseModel] = SendTweetsInput
+    description: str = "Send and Schedule Tweets for your Twitter Handle"
+    agent_id: int = None
+
+    def _execute(self, is_media: bool, tweet_text: str = 'None', media_num: int = 0, media_files: list = []):
+        toolkit_id = self.toolkit_config.toolkit_id
+        creds = TwitterTokens().get_twitter_creds(toolkit_id)
+        params = {}
+        if is_media:
+            media_ids = self.get_media_ids(media_files, creds)
+            params["media"] = {"media_ids": media_ids}
+        if tweet_text is not None:
+            params["text"] = tweet_text
+        tweet_response = self.send_tweets(params, creds)
+        if tweet_response.status_code ==  201:
+            return "Tweet posted successfully!!"
+        else:
+            return "Error posting tweet. (Status code: {})".format(tweet_response.status_code)
+
+
+    def get_media_ids(self, media_files, creds):
+        media_ids = []
+        oauth = OAuth1(creds["api_key"],
+                    client_secret=creds["api_key_secret"],
+                    resource_owner_key=creds["oauth_token"],
+                    resource_owner_secret=creds["oauth_token_secret"])
+        for file in media_files:
+            file_path = self.get_file_path(file)
+            image_data = open(file_path, 'rb').read()
+            b64_image = base64.b64encode(image_data)
+            upload_endpoint = 'https://upload.twitter.com/1.1/media/upload.json'
+            headers = {'Authorization': 'application/octet-stream'}
+            response = requests.post(upload_endpoint, headers=headers,
+                             data={'media_data': b64_image},
+                             auth=oauth)
+            ids = json.loads(response.text)['media_id']
+            media_ids.append(str(ids))
+
+        return media_ids
+
+    def get_file_path(self, file_name):
+        output_root_dir = ResourceHelper.get_root_output_dir()
+
+        final_path = ResourceHelper.get_root_input_dir() + file_name
+        if "{agent_id}" in final_path:
+            final_path = final_path.replace("{agent_id}", str(self.agent_id))
+
+        if final_path is None or not os.path.exists(final_path):
+            if output_root_dir is not None:
+                final_path = ResourceHelper.get_root_output_dir() + file_name
+                if "{agent_id}" in final_path:
+                    final_path = final_path.replace("{agent_id}", str(self.agent_id))
+
+        if final_path is None or not os.path.exists(final_path):
+            raise FileNotFoundError(f"File '{file_name}' not found.")
+
+        return final_path
+
+    def send_tweets(self, params, creds):
+        tweet_endpoint = "https://api.twitter.com/2/tweets"
+        oauth = OAuth1Session(creds["api_key"],
+                    client_secret=creds["api_key_secret"],
+                    resource_owner_key=creds["oauth_token"],
+                    resource_owner_secret=creds["oauth_token_secret"])
+
+        response = oauth.post(tweet_endpoint,json=params)
+        return response

superagi/tools/twitter/twitter_toolkit.py

@@ -0,0 +1,15 @@
+from abc import ABC
+from superagi.tools.base_tool import BaseToolkit, BaseTool
+from typing import Type, List
+from superagi.tools.twitter.send_tweets import SendTweetsTool
+
+
+class TwitterToolKit(BaseToolkit, ABC):
+    name: str = "Twitter Toolkit"
+    description: str = "Twitter Tool kit contains all tools related to Twitter"
+
+    def get_tools(self) -> List[BaseTool]:
+        return [SendTweetsTool()]
+
+    def get_env_keys(self) -> List[str]:
+        return ["TWITTER_API_KEY", "TWITTER_API_SECRET"]