Instagram tool

config_template.yaml

@@ -38,6 +38,7 @@ RESOURCES_OUTPUT_ROOT_DIR: workspace/output/{agent_id}/{agent_execution_id} # Fo
 
 #S3 RELATED DETAILS ONLY WHEN STORAGE_TYPE IS "S3"
 BUCKET_NAME:
+INSTAGRAM_TOOL_BUCKET_NAME:                                   #Public read bucket, Images generated by stable diffusion are put in this bucket and the public url of the same is generated.
 AWS_ACCESS_KEY_ID:
 AWS_SECRET_ACCESS_KEY:
 

superagi/tools/image_generation/stable_diffusion_image_gen.py

@@ -5,13 +5,13 @@
 import requests
 from PIL import Image
 from pydantic import BaseModel, Field
-
+from superagi.helper.resource_helper import ResourceHelper
 from superagi.resource_manager.file_manager import FileManager
 from superagi.tools.base_tool import BaseTool
 
 
 class StableDiffusionImageGenInput(BaseModel):
-    prompt: str = Field(..., description="Prompt for Image Generation to be used by Stable Diffusion.")
+    prompt: str = Field(..., description="Prompt for Image Generation to be used by Stable Diffusion. The prompt should be as descriptive as possible and mention all the details of the image to be generated")
     height: int = Field(..., description="Height of the image to be Generated. default height is 512")
     width: int = Field(..., description="Width of the image to be Generated. default width is 512")
     num: int = Field(..., description="Number of Images to be generated. default num is 2")
@@ -46,7 +46,6 @@ def _execute(self, prompt: str, image_names: list, width: int = 512, height: int
 
         if api_key is None:
             return "Error: Missing Stability API key."
-
         response = self.call_stable_diffusion(api_key, width, height, num, prompt, steps)
 
         if response.status_code != 200:
@@ -59,6 +58,7 @@ def _execute(self, prompt: str, image_names: list, width: int = 512, height: int
         for artifact in artifacts:
             base64_strings.append(artifact['base64'])
 
+        image_paths=[]
         for i in range(num):
             image_base64 = base64_strings[i]
             img_data = base64.b64decode(image_base64)
@@ -69,7 +69,10 @@ def _execute(self, prompt: str, image_names: list, width: int = 512, height: int
 
             self.resource_manager.write_binary_file(image_names[i], img_byte_arr.getvalue())
 
-        return "Images downloaded and saved successfully"
+        for image in image_names:
+            image_paths.append(ResourceHelper.get_resource_path(image))
+
+        return f"Images downloaded and saved successfully at the following locations: {image_paths}"
 
     def call_stable_diffusion(self, api_key, width, height, num, prompt, steps):
         engine_id = self.get_tool_config("ENGINE_ID")

superagi/tools/instagram_tool/README.MD

@@ -0,0 +1,42 @@
+<p align=center>
+<a href="https://superagi.co"><img src=https://superagi.co/wp-content/uploads/2023/05/SuperAGI_icon.png></a>
+</p>
+
+# SuperAGI Instagram Tool
+
+The SuperAGI Instagram Tool works with the stable diffusion tool, generates an image & caption based on the goals defined by the user and posts it on their instagram business account.
+
+## ⚙️ Installation
+
+### 🛠 **Setting Up of SuperAGI**
+Set up the SuperAGI by following the instructions given (https://github.com/TransformerOptimus/SuperAGI/blob/main/README.MD)
+
+If you've put the correct Google API key and Custom Search Engine ID, you'll be able to use the Google Search Tool as well.
+
+### 🔧 **Instagram tool requirements**
+
+Since the tool uses the official instagram graph API's to post media on user accounts, There are a few requirements:
+
+You will need access to the following:
+
+    1. An Instagram Business Account or Instagram Creator Account
+    2. A Facebook Page connected to that account
+    3. A Facebook Developer account that can perform Tasks on that Page
+    4. A registered Facebook App with Basic settings configured
+
+Once everything is set up, add the meta user access token (to be generated from facebook developer account), Facebook page ID (can be found on the facebook page connected to the instagram account under 'Page transparency' in 'About' section of the page ) and the stability API key to the correspponding toolkits.
+
+Follow the steps given in the link to set up meta requirements: (https://developers.facebook.com/docs/instagram-api/getting-started)
+Follow the link to generate stability API key: (https://dreamstudio.com/api/)
+
+### 🔧 **Configuring in SuperAGI Dashboard:**
+
+-You can add your meta user access token and facebook ID to the Instagram Toolkit Page and stability API key to the Image Generation Toolkit Page
+
+## Running SuperAGI Instagram Tool
+
+Once everything has been set up just run/schedule an agent with the goal explaining the media to be published and add instagram tool (which will automatically add stable diffusion tool)
+
+## Warning
+
+It is advised to run the instagram tool in restricted mode since it allows you to validate the photos generated. You can schedule agent runs (recurring runs are supported as well). Also, only one photo will be posted to your account in a run. To post multiple photos use recurring runs.

superagi/tools/instagram_tool/instagram.py

@@ -0,0 +1,226 @@
+import json
+import urllib
+import boto3
+import os
+from superagi.config.config import get_config
+from superagi.helper.resource_helper import ResourceHelper
+from typing import Type, Optional
+from pydantic import BaseModel, Field
+from superagi.helper.token_counter import TokenCounter
+from superagi.llms.base_llm import BaseLlm
+from superagi.tools.base_tool import BaseTool
+import os
+import requests
+from superagi.tools.tool_response_query_manager import ToolResponseQueryManager
+import random
+
+class InstagramSchema(BaseModel):
+    photo_description: str = Field(
+        ...,
+        description="description of the photo",
+    )
+
+class InstagramTool(BaseTool):
+    """
+    Instagram tool
+
+    Attributes:
+        name : The name.
+        description : The description.
+        args_schema : The args schema.
+    """
+    llm: Optional[BaseLlm] = None
+    name = "Instagram tool"
+    description = (
+        "A tool for posting an AI generated photo on Instagram"
+    )
+    args_schema: Type[InstagramSchema] = InstagramSchema
+    tool_response_manager: Optional[ToolResponseQueryManager] = None
+    agent_id:int =None
+    class Config:
+        arbitrary_types_allowed = True
+
+    def _execute(self, photo_description: str) -> str:
+        """
+        Execute the Instagram tool.
+
+        Args:
+            photo_description : description of the photo to be posted
+
+        Returns:
+            Image posted successfully message if image has been posted on instagram or error message.
+        """
+        meta_user_access_token = self.get_tool_config("META_USER_ACCESS_TOKEN")
+        facebook_page_id=self.get_tool_config("FACEBOOK_PAGE_ID")
+
+        if meta_user_access_token is None:
+            return "Error: Missing meta user access token."
+
+        if facebook_page_id is None:
+            return "Error: Missing facebook page id."
+        #create caption for the instagram
+        caption=self.create_caption(photo_description)   
+
+        #get request for fetching the instagram_business_account_id
+        root_api_url="https://graph.facebook.com/v17.0/"
+        response=self.get_req_insta_id(root_api_url,facebook_page_id,meta_user_access_token)
+
+        if response.status_code != 200:
+            return f"Non-200 response: {str(response.text)}"
+
+        data = response.json()
+        insta_business_account_id=data["instagram_business_account"]["id"]
+        file_path=self.get_file_path_from_image_generation_tool()     
+        #handling case where image generation generates multiple images
+        if(file_path=="resources"):
+            return "A photo has already been posted on your instagram account. To post multiple photos use recurring runs."
+
+        image_url,encoded_caption=self.get_img_url_and_encoded_caption(photo_description,file_path)
+        #post request for getting the media container ID
+        response=self.post_media_container_id(root_api_url,insta_business_account_id,image_url,encoded_caption,meta_user_access_token)
+
+        if response.status_code != 200:
+            return f"Non-200 response: {str(response.text)}"
+
+        data = response.json()
+        container_ID=data["id"]
+        #post request to post the media container on instagram account
+        response=self.post_media(root_api_url,insta_business_account_id,container_ID,meta_user_access_token)
+        if response.status_code != 200:
+            return f"Non-200 response: {str(response.text)}"
+        return "Photo posted successfully!"
+
+    def create_caption(self, photo_description: str) -> str:
+        """
+        Create a caption for the instagram post based on the photo description
+
+        Args:
+            photo_description : Description of the photo to be posted
+
+        Returns:
+            Description of the photo to be posted
+        """
+        caption_prompt ="""Generate an instagram post caption for the following text `{photo_description}`
+            Attempt to make it as relevant as possible to the description and should be different and unique everytime. Add relevant emojis and hashtags."""
+
+        caption_prompt = caption_prompt.replace("{photo_description}", str(photo_description))
+
+        messages = [{"role": "system", "content": caption_prompt}]
+        result = self.llm.chat_completion(messages, max_tokens=self.max_token_limit)
+        caption=result["content"]
+
+        encoded_caption=urllib. parse. quote(caption)     
+
+        return encoded_caption
+
+    def get_image_from_s3(self,s3,file_path):
+        """
+        Gets the image from the s3 bucket
+
+        Args:
+            s3: S3 client
+            file_path: path of the image file in s3
+
+        Returns
+            The image file from s3
+        """
+
+        response = s3.get_object(Bucket=get_config("BUCKET_NAME"), Key=file_path)
+        content = response["Body"].read()
+
+        return content
+
+    def get_file_path_from_image_generation_tool(self):
+        """
+        Parses the output of the previous tool (Stable diffusion) and returns the path of the image file
+
+        Args:
+
+        Returns:
+            The path of the image file generated by the image generation toolkit
+        """
+
+        last_tool_response = self.tool_response_manager.get_last_response()
+        file_path="resources"+last_tool_response.partition("['")[2].partition("']")[0]
+
+        if ',' in file_path:
+            # Split the string based on the comma and get the first element (substring before the comma)
+            file_path = file_path.split(',')[0].strip()
+            file_path=file_path[:-1]
+
+        return file_path
+
+    def create_s3_client(self):
+        """
+        Creates an s3 client
+
+        Args:
+
+        Returns:
+            The s3 client
+        """
+
+        s3 = boto3.client(
+            's3',
+            aws_access_key_id=get_config("AWS_ACCESS_KEY_ID"),
+            aws_secret_access_key=get_config("AWS_SECRET_ACCESS_KEY"),
+        )
+
+        return s3
+
+    def get_img_public_url(self,s3,file_path,content):
+        """
+        Puts the image generated by image generation tool in the s3 bucket and returns the public url of the same
+        Args:
+            s3 : S3 bucket
+            file_path: Path of the image file in s3
+            content: Image file
+
+        Returns:
+            The public url of the image put in s3 bucket
+        """
+
+        bucket_name = get_config("INSTAGRAM_TOOL_BUCKET_NAME")
+        object_key=f"instagram_upload_images/{file_path.split('/')[-1]}{random.randint(0, 1000)}"
+        s3.put_object(Bucket=bucket_name, Key=object_key, Body=content)
+
+        image_url = f"https://{bucket_name}.s3.amazonaws.com/{object_key}"
+        return image_url
+
+    def get_img_url_and_encoded_caption(self,photo_description,file_path):
+        #creating an s3 client
+        s3 = self.create_s3_client()    
+
+        #fetching the image from the s3 using the file_path
+        content = self.get_image_from_s3(s3,file_path)      
+
+        #storing the image in a public bucket and getting the image url
+        image_url = self.get_img_public_url(s3,file_path,content)       
+        #encoding the caption with possible emojis and hashtags and removing the starting and ending double quotes 
+        encoded_caption=self.create_caption(photo_description)    
+
+        return image_url,encoded_caption
+
+    def get_req_insta_id(self,root_api_url,facebook_page_id,meta_user_access_token):
+        url_to_get_acc_id=f"{root_api_url}{facebook_page_id}?fields=instagram_business_account&access_token={meta_user_access_token}"
+        response=requests.get(
+            url_to_get_acc_id
+        )
+
+        return response
+
+    def post_media_container_id(self,root_api_url,insta_business_account_id,image_url,encoded_caption,meta_user_access_token):
+        url_to_create_media_container=f"{root_api_url}{insta_business_account_id}/media?image_url={image_url}&caption={encoded_caption}&access_token={meta_user_access_token}"
+        response = requests.post(       
+            url_to_create_media_container
+        )
+
+        return response
+
+    def post_media(self,root_api_url,insta_business_account_id,container_ID,meta_user_access_token):
+        url_to_post_media_container=f"{root_api_url}{insta_business_account_id}/media_publish?creation_id={container_ID}&access_token={meta_user_access_token}"
+        response = requests.post(
+            url_to_post_media_container
+        )
+
+        return response

superagi/tools/instagram_tool/instagram_toolkit.py

@@ -0,0 +1,18 @@
+from abc import ABC
+from typing import List
+from superagi.tools.base_tool import BaseTool, BaseToolkit
+from superagi.tools.instagram_tool.instagram import InstagramTool
+
+class InstagramToolkit(BaseToolkit, ABC):
+    name: str = "Instagram Toolkit"
+    description: str = "Toolkit containing tools for posting AI generated photo on Instagram. Posts only one photo in a run "
+
+    def get_tools(self) -> List[BaseTool]:
+        return [InstagramTool()]
+
+    def get_env_keys(self) -> List[str]:
+        return [
+            "META_USER_ACCESS_TOKEN",
+            "FACEBOOK_PAGE_ID"
+            # Add more config keys specific to your project
+        ]

tests/unit_tests/tools/image_generation/test_stable_diffusion_image_gen.py

@@ -48,8 +48,7 @@ def test_execute(stable_diffusion_tool):
 
 
     result = tool._execute('prompt', ['img1.png', 'img2.png'])
-
-    assert result == 'Images downloaded and saved successfully'
+    assert result.startswith('Images downloaded and saved successfully')
     tool.resource_manager.write_binary_file.assert_called()
 
 def test_call_stable_diffusion(stable_diffusion_tool):