Dynamic Node

The Dynamic Node allows you to create custom, purpose-built nodes for your skills by writing Python code directly in the workflow. This powerful feature enables you to define custom inputs and outputs dynamically, extending beyond the standard node library.

Basic Structure

Every Dynamic Node starts with this base template:

from skill_components import *

class MyComponent(CustomComponent):
    metadata = CustomComponentMetadata(
        name='custom_component',
        description='A custom component',
    )

    inputs = [
        # Define your inputs here
    ]

    outputs = [
        # Define your outputs here
    ]

    @sp_output
    def run(self):
        # Implement your logic here
        # Return a dictionary matching output names
        pass

Defining Inputs

You can define various input types to accept data into your node. Each input requires:

name: Unique identifier for the input
label: Display name in the UI
description: Explanation of the input's purpose
is_required: Whether the input must be provided (default: False)
is_list: Whether the input accepts multiple values (default: False)
is_config: Whether the input is located in the "customize" tab as code (default: False)

class BaseInput(BaseModel):
    name: str
    is_required: bool = False
    label: str
    description: str
    type: str
    default_value: Any | None = None
    is_list: bool = False
    is_config: bool = False

    model_config = COMPONENT_PYDANTIC_CONFIG

Available Input Types

StringInput: Text data
NumberInput: Numerical values
BoolInput: True/False values
DataFrameInput: Pandas DataFrames
HtmlInput: HTML content
FilterInput: Filter conditions
DictInput: Dictionary/JSON data
EnumInput: Selection from predefined options
SqlInput: SQL queries
ConnectionInput: Database connections
AnyInput: Flexible type (use sparingly)

Example Input Definitions

inputs = [
    StringInput(
        name="text_input",
        label="Text Input",
        description="Enter your text here",
        is_required=True
        is_config=False
    ),
    NumberInput(
        name="threshold",
        label="Threshold Value",
        description="Enter threshold between 0 and 1",
        default_value=0.5
        is_config=False
    ),
    DataFrameInput(
        name="data",
        label="Input DataFrame",
        description="Data to process",
        is_list=True
        is_config=False
    )
]

Defining Outputs

Outputs define what data your node will return. Each output requires:

name: Unique identifier for the output
label: Display name in the UI
description: Explanation of what the output contains
is_list: Whether the output returns multiple values (default: False)

class ComponentOutput(BaseModel):
    name: str
    label: str
    description: str
    _fn: str = "run"
    type: str
    is_list: bool = False

    model_config = COMPONENT_PYDANTIC_CONFIG

Available Output Types

StringOutput: Text data
DataFrameOutput: Pandas DataFrames
HtmlOutput: HTML content
DictOutput: Dictionary/JSON data
VisualizationOutput: Visual representations
SqlOutput: SQL query results
AnyOutput: Flexible type (use sparingly)

Example Output Definitions

outputs = [
    DataFrameOutput(
        name="processed_data",
        label="Processed Data",
        description="Transformed dataset"
    ),
    DictOutput(
        name="metrics",
        label="Performance Metrics",
        description="Calculated metrics"
    )
]

Implementation

The run() method contains your node's logic. It must return a dictionary where keys match your defined output names.

Example Implementation

@sp_output
def run(self):
    # Access input values
    input_text = self.text_input
    threshold = self.threshold
    dataframes = self.data

    # Your processing logic here
    processed_df = process_data(dataframes, threshold)
    metrics = calculate_metrics(processed_df)

    # Return dictionary matching output names
    return {
        "processed_data": processed_df,
        "metrics": metrics
    }

Best Practices

Input Validation
- Add appropriate error handling
- Validate input types and values
- Provide meaningful error messages
Documentation
- Write clear descriptions for inputs and outputs
- Document any assumptions or requirements
- Include examples in the description when helpful
Performance
- Consider memory usage with large datasets
- Optimize processing for efficiency
- Use appropriate data types
Maintenance
- Use meaningful variable names
- Keep logic modular and readable
- Comment complex operations

Refreshing the Node

After modifying your Dynamic Node's code:

Click the refresh button to update the node
The node will recompile with your changes
Input and output interfaces will update automatically
Any connections to invalid inputs/outputs will need to be reconnected

Complete Example

Here's a complete example of a Dynamic Node that processes text data:

from skill_components import *

class TextProcessingNode(CustomComponent):
    metadata = CustomComponentMetadata(
        name='text_processor',
        description='Processes text data and returns statistics',
    )

    inputs = [
        StringInput(
            name="text",
            label="Input Text",
            description="Text to analyze",
            is_required=True
        ),
        BoolInput(
            name="case_sensitive",
            label="Case Sensitive",
            description="Whether to preserve case",
            default_value=False
        )
    ]

    outputs = [
        DictOutput(
            name="stats",
            label="Text Statistics",
            description="Analysis results"
        ),
        StringOutput(
            name="processed_text",
            label="Processed Text",
            description="Transformed text"
        )
    ]

    @sp_output
    def run(self):
        text = self.text
        case_sensitive = self.case_sensitive

        if not case_sensitive:
            text = text.lower()

        word_count = len(text.split())
        char_count = len(text)
        
        return {
            "stats": {
                "word_count": word_count,
                "char_count": char_count
            },
            "processed_text": text
        }