This is a custom Langchain tool to pass to an agent, for building YouTubeShorts with a tool like Canva.

YouTubeShortsAirtableScript.py

import requests
import openai
from pydantic import BaseModel, Field
from langchain.tools import tool

AIRTABLE_BASE_ID = "your_airtable_base_id"
AIRTABLE_API_KEY = "your_airtable_api_key"
TABLE = "your_airtable_table"
MODEL_NAME = "gpt-3.5-turbo-0613"

class CreateRecordInput(BaseModel):
    prompt: str
    title: str
    description: str
    keywords: str
    content: str

class UpdateRecordInput(BaseModel):
    record_id: str
    prompt: str
    title: str
    description: str
    keywords: str
    content: str

@tool(name="create_record", args_schema=CreateRecordInput)
def create_record(prompt: str, title: str, description: str, keywords: str, content: str) -> str:
    formatted_prompt = f"Generate creative content for a YouTube Short video about {prompt}"
    response = openai.Completion.create(
        engine=MODEL_NAME,
        prompt=formatted_prompt,
        max_tokens=100,
        n=1,
        stop=None,
        temperature=0.6,
        top_p=1.0,
        frequency_penalty=0.0,
        presence_penalty=0.0,
        log_level="info"
    )
    generated_content = response["choices"][0]["text"]
    url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}"
    headers = {
        "Authorization": f"Bearer {AIRTABLE_API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "records": [
            {
                "fields": {
                    "Prompt": prompt,
                    "Title": title,
                    "Description": description,
                    "Keywords": keywords,
                    "Content": generated_content
                }
            }
        ]
    }
    response = requests.post(url, headers=headers, json=data)
    return response.json()

@tool(name="get_records")
def get_records() -> str:
    url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}"
    headers = {
        "Authorization": f"Bearer {AIRTABLE_API_KEY}"
    }
    response = requests.get(url, headers=headers)
    return response.json()

@tool(name="update_record", args_schema=UpdateRecordInput)
def update_record(record_id: str, prompt: str, title: str, description: str, keywords: str, content: str) -> str:
    url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}/{record_id}"
    headers = {
        "Authorization": f"Bearer {AIRTABLE_API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "fields": {
            "Prompt": prompt,
            "Title": title,
            "Description": description,
            "Keywords": keywords,
            "Content": content
        }
    }
    response = requests.patch(url, headers=headers, json=data)
    return response.json()

@tool(name="delete_record")
def delete_record(record_id: str) -> str:
    url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}/{record_id}"
    headers = {
        "Authorization": f"Bearer {AIRTABLE_API_KEY}"
    }
    response = requests.delete(url, headers=headers)
    return response.json()

@tool(name="run_conversation")
def run_conversation() -> str:
    # ... [rest of the run_conversation function code]
    # NOTE: This function is quite complex and will need more information 
    #       and adjustments to fully integrate it with the tool decorator.

YouTubeShortsAirtableTool.py

import openai
import requests
from langchain.agents import AgentType, initialize_agent
from langchain.chat_models import ChatOpenAI
from langchain.tools import BaseTool
from typing import Optional, Type
from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)

# Constants
AIRTABLE_BASE_ID = "your_airtable_base_id"
AIRTABLE_API_KEY = "your_airtable_api_key"
TABLE = "your_airtable_table"
MODEL_NAME = "gpt-3.5-turbo-0613"


class YouTubeShortsAirtableTool(BaseTool):
    name = "YouTubeShortsAirtable"
    description = "Manage YouTube Shorts content on Airtable."

    def create_record(self, query: str) -> str:
        prompt, title, description, keywords, content = query.split('|')
        
        formatted_prompt = f"Generate creative content for a YouTube Short video about {prompt}"
        response = openai.Completion.create(
            engine=MODEL_NAME,
            prompt=formatted_prompt,
            max_tokens=100,
            n=1,
            stop=None,
            temperature=0.6,
            top_p=1.0,
            frequency_penalty=0.0,
            presence_penalty=0.0,
            log_level="info"
        )
        generated_content = response["choices"][0]["text"]

        url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}"
        headers = {
            "Authorization": f"Bearer {AIRTABLE_API_KEY}",
            "Content-Type": "application/json"
        }
        data = {
            "records": [
                {
                    "fields": {
                        "Prompt": prompt,
                        "Title": title,
                        "Description": description,
                        "Keywords": keywords,
                        "Content": generated_content
                    }
                }
            ]
        }
        response = requests.post(url, headers=headers, json=data)
        return str(response.json())

    def get_records(self, query: str) -> str:
        url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}"
        headers = {
            "Authorization": f"Bearer {AIRTABLE_API_KEY}"
        }
        response = requests.get(url, headers=headers)
        return str(response.json())

    def update_record(self, query: str) -> str:
        record_id, prompt, title, description, keywords, content = query.split('|')
        
        url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}/{record_id}"
        headers = {
            "Authorization": f"Bearer {AIRTABLE_API_KEY}",
            "Content-Type": "application/json"
        }
        data = {
            "fields": {
                "Prompt": prompt,
                "Title": title,
                "Description": description,
                "Keywords": keywords,
                "Content": content
            }
        }
        response = requests.patch(url, headers=headers, json=data)
        return str(response.json())

    def delete_record(self, query: str) -> str:
        record_id = query
        
        url = f"https://api.airtable.com/v0/{AIRTABLE_BASE_ID}/{TABLE}/{record_id}"
        headers = {
            "Authorization": f"Bearer {AIRTABLE_API_KEY}"
        }
        response = requests.delete(url, headers=headers)
        return str(response.json())

    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
        if "create" in query:
            return self.create_record(query)
        elif "get" in query:
            return self.get_records(query)
        elif "update" in query:
            return self.update_record(query)
        elif "delete" in query:
            return self.delete_record(query)
        else:
            return "Invalid query."

    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
        raise NotImplementedError("This tool does not support async yet.")


tools = [YouTubeShortsAirtableTool()]

agent = initialize_agent(
    tools,
    ChatOpenAI(temperature=0),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Sample run for the agent
print(agent.run("create|This is a prompt|Title|Description|Keywords|Generated Content"))