Creating Custom Tools

import asyncio
from langchain.tools import Tool
from twitter_agent.custom_twitter_actions import TwitterClient

# Create client instance
twitter_client = TwitterClient()

def create_delete_tweet_tool() -> Tool:
    """Create a delete tweet tool."""
    return Tool(
        name="delete_tweet",
        description="""Delete a tweet using its ID. You can only delete tweets from your own account.
        Input should be the tweet ID as a string.
        Example: delete_tweet("1234567890")""",
        func=lambda tweet_id: asyncio.run(twitter_client.delete_tweet(tweet_id))
    )

def create_retweet_tool() -> Tool:
    """Create a retweet tool."""
    return Tool(
        name="retweet",
        description="""Retweet a tweet using its ID. You can only retweet public tweets.
        Input should be the tweet ID as a string.
        Example: retweet("1234567890")""",
        func=lambda tweet_id: asyncio.run(twitter_client.retweet(tweet_id))
    )

from pydantic import BaseModel, Field
from hyperbolic_agentkit_core.actions.hyperbolic_action import HyperbolicAction
import requests
import json

class RentComputeInput(BaseModel):
    """Input argument schema for compute rental action."""

    cluster_name: str = Field(
        ..., description="The cluster name that the user wants to rent from")
    node_name: str = Field(
        ..., description="The node ID that the user wants to rent")
    gpu_count: str = Field(
        ..., description="The amount of GPUs that the user wants to rent from the node")

def rent_compute(cluster_name: str, node_name: str, gpu_count: str) -> str:
    """Creates a marketplace instance using the Hyperbolic API."""
    if not cluster_name or not node_name or not gpu_count:
        raise ValueError("cluster_name, node_name, and gpu_count are required")

    api_key = os.getenv("HYPERBOLIC_API_KEY")
    endpoint = f"https://api.hyperbolic.xyz/v1/marketplace/instances/create"
    
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    payload = {
        "cluster_name": cluster_name,
        "node_name": node_name,
        "gpu_count": gpu_count
    }

    response = requests.post(endpoint, headers=headers, json=payload)
    response.raise_for_status()
    
    return json.dumps(response.json(), indent=2)

class RentComputeAction(HyperbolicAction):
    """Rent compute action."""
    name: str = "rent_compute"
    description: str = """This tool will allow you to rent a GPU machine on Hyperbolic platform.
    
    It takes the following inputs:
    - cluster_name: Which cluster the node is on
    - node_name: Which node the user wants to rent
    - gpu_count: How many GPUs the user wants to rent
    """
    args_schema: type[BaseModel] | None = RentComputeInput
    func: Callable[..., str] = rent_compute

from pydantic.v1 import BaseModel, Field
from langchain_core.tools import BaseTool
from typing import Optional, List

class WritingInput(BaseModel):
    """Input for the writing tool."""
    query: str = Field(description="The topic or request for content creation")
    reference_files: Optional[List[str]] = Field(
        description="Optional list of reference files for style analysis", 
        default=None
    )
    target_length: Optional[int] = Field(
        description="Target length of the article in words", 
        default=1500
    )
    output_file: Optional[str] = Field(
        description="Optional path to save the output text",
        default=None
    )

class WritingTool(BaseTool):
    """Tool for autonomous article writing with research capabilities."""
    
    name: str = "writing_agent"
    description: str = """A powerful tool for autonomous article writing with research capabilities.
    
    This tool can:
    - Analyze and mimic the writing style of reference documents
    - Research the web for up-to-date information
    - Create well-structured, informative articles
    - Generate citations and properly attribute sources
    """
    
    args_schema: type[BaseModel] = WritingInput
    
    async def _arun(self, query: str, reference_files: Optional[List[str]] = None, 
                    target_length: Optional[int] = 1500,
                    output_file: Optional[str] = None) -> str:
        """Run the writing agent asynchronously."""
        from writing_agent.writing_agent import WritingAgent
        
        agent = WritingAgent(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        
        if reference_files:
            await agent.load_reference_materials(reference_files)
        
        result = await agent.create_content(query, target_length, output_file)
        return result

def create_agent_tools(llm, knowledge_base, podcast_knowledge_base, agent_kit, config):
    """Create and return a list of tools for the agent to use."""
    tools = []

    # Browser tools - controlled by environment variable
    if os.getenv("USE_BROWSER_TOOLS", "true").lower() == "true":
        browser_toolkit = BrowserToolkit.from_llm(llm)
        tools.extend(browser_toolkit.get_tools())

    # Writing Agent - controlled by environment variable
    if os.getenv("USE_WRITING_AGENT", "true").lower() == "true":
        writing_tool = WritingTool(llm=llm)
        tools.append(writing_tool)

    # Twitter tools - controlled by environment variable
    if os.getenv("USE_TWITTER_CORE", "true").lower() == "true":
        twitter_client = TwitterClient()
        
        if os.getenv("USE_TWEET_DELETE", "true").lower() == "true":
            tools.append(create_delete_tweet_tool())
        
        if os.getenv("USE_RETWEET", "true").lower() == "true":
            tools.append(create_retweet_tool())

    return tools

# Enable/disable custom tools
USE_MY_CUSTOM_TOOL=true
USE_ADVANCED_FEATURES=false

from langchain_core.tools import BaseToolkit
from typing import List
from langchain_core.tools import BaseTool
from my_agent.tool1 import Tool1
from my_agent.tool2 import Tool2
from my_agent.tool3 import Tool3

class MyToolkit(BaseToolkit):
    """Collection of tools for my agent."""
    
    @classmethod
    def from_llm(cls, llm):
        """Create toolkit with LLM instance."""
        return cls(llm=llm)
    
    def __init__(self, llm=None):
        self.llm = llm
    
    def get_tools(self) -> List[BaseTool]:
        """Return all tools in this toolkit."""
        return [
            Tool1(llm=self.llm),
            Tool2(llm=self.llm),
            Tool3(),
        ]

from my_agent.my_toolkit import MyToolkit

def create_agent_tools(llm, knowledge_base, podcast_knowledge_base, agent_kit, config):
    tools = []
    
    if os.getenv("USE_MY_TOOLKIT", "true").lower() == "true":
        my_toolkit = MyToolkit.from_llm(llm)
        tools.extend(my_toolkit.get_tools())
    
    return tools