Try
Customization

Creating Tools

Learn how to create reusable tools that can be used across multiple nodes.

Tools are reusable utility classes that encapsulate specific functionality. They can be used by multiple nodes to avoid code duplication and maintain consistency. Examples include language model tools, web search tools, and knowledge base tools.

Tool Structure

Tools are typically Python classes that provide a clean interface for specific operations. They should:

  • Handle initialization and configuration
  • Provide clear, well-documented methods
  • Return consistent data structures
  • Handle errors gracefully
  • Support multiple service providers when applicable

Step-by-Step Guide

Step 1: Create Tool Directory

Create a new directory for your tool in backend/tools/:

bash
backend/tools/
  └── my_custom_tool/
      ├── __init__.py
      └── my_custom_tool.py

Step 2: Implement the Tool Class

Create your tool class with the desired functionality:

python
"""
My Custom Tool - Description of what this tool does
"""

import sys
import os
from typing import Dict, Any, Optional

# Add parent directory to path if needed
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))


class MyCustomTool:
    """
    My Custom Tool - Provides specific functionality.
    
    This tool can be used by multiple nodes to perform
    a specific operation without code duplication.
    """
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """
        Initialize the tool.
        
        Args:
            config: Optional configuration dictionary
        """
        self.config = config or {}
        self.initialized = False
        self._initialize()
    
    def _initialize(self):
        """Initialize the tool (load resources, connect to services, etc.)"""
        try:
            # Initialize your tool here
            # e.g., connect to API, load models, etc.
            self.initialized = True
        except Exception as e:
            print(f"Warning: Could not initialize tool: {e}")
            self.initialized = False
    
    def perform_operation(
        self,
        input_data: str,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Perform the main operation.
        
        Args:
            input_data: The input data to process
            **kwargs: Additional parameters
            
        Returns:
            Dictionary containing the result and metadata
        """
        if not self.initialized:
            return {
                "success": False,
                "error": "Tool not initialized",
                "result": None
            }
        
        try:
            # Perform your operation here
            result = f"Processed: {input_data}"
            
            return {
                "success": True,
                "result": result,
                "metadata": {
                    "input_length": len(input_data),
                    "operation": "custom_operation"
                }
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "result": None
            }
    
    def get_status(self) -> Dict[str, Any]:
        """
        Get the current status of the tool.
        
        Returns:
            Dictionary containing status information
        """
        return {
            "initialized": self.initialized,
            "config": self.config
        }


# Convenience function for quick usage
def use_custom_tool(input_data: str, **kwargs) -> str:
    """
    Quick function to use the custom tool.
    
    Args:
        input_data: The input data
        **kwargs: Additional parameters
        
    Returns:
        The result as a string, or error message if failed
    """
    tool = MyCustomTool()
    result = tool.perform_operation(input_data, **kwargs)
    
    if result["success"]:
        return result["result"]
    else:
        return f"Error: {result['error']}"

Step 3: Use the Tool in a Node

Import and use your tool in any node:

python
from typing import Dict, Any
import sys
import os

sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from base_node import BaseNode, NodeInput, NodeOutput, NodeParameter

# Import your tool
try:
    from tools.my_custom_tool.my_custom_tool import MyCustomTool
except ImportError:
    MyCustomTool = None


class MyNodeUsingTool(BaseNode):
    # ... define inputs, outputs, parameters ...
    
    def execute(self, inputs: Dict[str, Any], parameters: Dict[str, Any]) -> Dict[str, Any]:
        # Check if tool is available
        if MyCustomTool is None:
            return {
                "result": "",
                "success": False,
                "metadata": {"error": "Tool not available"}
            }
        
        # Initialize the tool
        tool = MyCustomTool()
        
        # Use the tool
        input_data = inputs.get("input_data", "")
        tool_result = tool.perform_operation(input_data)
        
        if tool_result["success"]:
            return {
                "result": tool_result["result"],
                "success": True,
                "metadata": tool_result.get("metadata", {})
            }
        else:
            return {
                "result": "",
                "success": False,
                "metadata": {"error": tool_result.get("error", "Unknown error")}
                }

Example: Language Model Tool

Here's a real example from ConvoFlow - the Language Model Tool that supports multiple providers:

python
class LanguageModelTool:
    """
    Tool for using language models across multiple providers.
    """
    
    def __init__(self):
        """Initialize with all available services."""
        self.services = {}
        
        # Add available services
        if OpenAIService:
            try:
                self.services["openai"] = OpenAIService()
            except Exception as e:
                print(f"Warning: Could not initialize OpenAI: {e}")
        
        # Add other services similarly...
    
    def generate_response(
        self,
        query: str,
        service: str = "openai",
        model: Optional[str] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Generate a response using the specified service.
        
        Returns:
            Dictionary with success, response, and metadata
        """
        # Implementation...
        pass

Best Practices

  • Error Handling: Always return structured error responses
  • Initialization: Handle initialization failures gracefully
  • Consistency: Use consistent return structures across methods
  • Documentation: Document all methods and parameters clearly
  • Reusability: Design tools to be reusable across different nodes
  • Configuration: Support configuration through environment variables or parameters

Tool vs Node

Understanding when to create a tool vs a node:

Create a Tool when:

  • Functionality is reusable across multiple nodes
  • You need to encapsulate complex logic
  • You want to support multiple service providers
  • You need to manage resources (connections, models, etc.)

Create a Node when:

  • You need a workflow component
  • You want user-configurable parameters
  • You need inputs/outputs for workflow connections
  • You want visual representation in the workflow
Adding NodesTool Reference
Was this page helpful?