Try
Customization

Adding New Nodes

Learn how to create custom nodes to extend ConvoFlow's functionality.

Nodes are the building blocks of ConvoFlow workflows. Each node represents a specific operation or functionality. This guide will walk you through creating a custom node from scratch.

Node Structure

Every node in ConvoFlow extends the BaseNode class and implements several key methods:

  • _define_inputs() - Define what inputs the node accepts
  • _define_outputs() - Define what outputs the node produces
  • _define_parameters() - Define configurable parameters
  • execute() - Implement the node's core logic
  • _define_styling() - Customize the node's appearance (optional)

Step-by-Step Guide

Step 1: Create Node Directory

Create a new directory for your node in backend/nodes/:

bash
backend/nodes/
  └── my_custom_node/
      ├── __init__.py
      └── my_custom_node.py

Step 2: Implement the Node Class

Create your node class extending BaseNode:

python
"""
My Custom Node - Description of what this node does
"""

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

# Add parent directory to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from base_node import BaseNode, NodeInput, NodeOutput, NodeParameter, NodeStyling


class MyCustomNode(BaseNode):
    """
    My Custom Node - Handles custom functionality.

    This node does something specific for your use case.
    """

    def _define_category(self) -> str:
        """Define the category for this node"""
        return "Custom"

    def _define_inputs(self) -> List[NodeInput]:
        """Define the input structure"""
        return [
            NodeInput(
                name="input_data",
                type="string",
                description="The input data to process",
                required=True
            )
        ]

    def _define_outputs(self) -> List[NodeOutput]:
        """Define the output structure"""
        return [
            NodeOutput(
                name="result",
                type="string",
                description="The processed result"
            )
        ]

    def _define_parameters(self) -> List[NodeParameter]:
        """Define the parameters"""
        return [
            NodeParameter(
                name="option",
                type="string",
                description="A configuration option",
                required=False,
                default_value="default",
                options=["option1", "option2", "option3"]
            )
        ]

    def _define_styling(self) -> NodeStyling:
        """Define custom styling"""
        return NodeStyling(
            icon="🔧",
            background_color="#3b82f6",
            border_color="#2563eb",
            text_color="#ffffff",
            shape="rounded"
        )

    def execute(self, inputs: Dict[str, Any], parameters: Dict[str, Any]) -> Dict[str, Any]:
        """
        Execute the node logic

        Args:
            inputs: Dictionary containing input values
            parameters: Dictionary containing parameter values

        Returns:
            Dictionary containing output values
        """
        # Get inputs
        input_data = inputs.get("input_data", "")
        option = parameters.get("option", "default")

        # Process the data
        result = f"Processed: {input_data} with option: {option}"

        return {
            "result": result,
            "success": True
        }

Step 3: Register the Node

Register your node in the node registry. Find the registration file (usually in backend/register_nodes.py) and add your node:

python
# In backend/register_nodes.py
from nodes.my_custom_node.my_custom_node import MyCustomNode
from nodes.node_registry import register_node

def register_all_nodes():
    """Register all available nodes in the system"""

    # ... existing node registrations ...

    # Register your custom node
    register_node(MyCustomNode)

    print("[OK] All nodes registered successfully!")

Important: After registering your node, restart the backend server. The node will automatically appear in the node sidebar in the workflow builder.

Step 4: Test Your Node

Restart the backend server and your node should appear in the node sidebar. You can test it by:

  1. Adding it to a workflow
  2. Connecting it to other nodes
  3. Configuring its parameters
  4. Executing the workflow

Using Tools in Nodes

Nodes can use tools to perform complex operations. Here's how to use a tool in your node:

python
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))

# Import the tool
try:
    from tools.language_model_tool.language_model_tool import LanguageModelTool
except ImportError:
    LanguageModelTool = None

class MyCustomNode(BaseNode):
    # ... other methods ...

    def execute(self, inputs: Dict[str, Any], parameters: Dict[str, Any]) -> Dict[str, Any]:
        # Initialize the tool
        if LanguageModelTool is None:
            return {
                "result": "",
                "success": False,
                "metadata": {"error": "Tool not available"}
            }

        tool = LanguageModelTool()

        # Use the tool
        result = tool.generate_response(
            query=inputs.get("query", ""),
            service="openai",
            model="gpt-3.5-turbo"
        )

        return {
            "result": result.get("response", ""),
            "success": result.get("success", False)
        }

Best Practices

  • Error Handling: Always handle errors gracefully and return meaningful error messages
  • Validation: Use the built-in validation methods to check inputs and parameters
  • Documentation: Provide clear docstrings and descriptions for all inputs, outputs, and parameters
  • Styling: Use consistent styling to match the visual design of other nodes
  • Testing: Test your node thoroughly with various inputs and edge cases

Next Steps

Was this page helpful?