NodeFlowWidget

The main widget for creating and managing interactive node-based workflows in Jupyter notebooks.

NodeFlowWidget(height='600px', **kwargs)

Bases: AnyWidget

A Jupyter widget wrapping ReactFlow for interactive node graph visualization.

This widget can be initialized with a list of node classes that implement the NodeFactory protocol. Node types will be automatically registered and made available in the visual editor.

Examples:

>>> from pynodewidget import NodeFlowWidget
>>> 
>>> class MyNode(JsonSchemaNodeWidget):
...     label = "My Node"
...     parameters = MyParams
>>> 
>>> flow = NodeFlowWidget(
...     nodes=[MyNode],
...     height="800px"
... )

Initialize the NodeFlowWidget.

Parameters:

Name	Type	Description	Default
height	str	Height of the widget canvas (default: "600px")	'600px'
**kwargs	Any	Additional widget configuration options	{}

Source code in pynodewidget/widget.py

def __init__(self, height: str = "600px", **kwargs: Any) -> None:
    """Initialize the NodeFlowWidget.

    Args:
        height: Height of the widget canvas (default: "600px")
        **kwargs: Additional widget configuration options
    """
    super().__init__(**kwargs)
    self.height = height
    self._export_id = 0
    self._pending_exports = {}  # Maps export_id -> filename

    # Set up persistent observer for image data
    self.observe(self._on_image_data_received, names=['_export_image_data'])

export_json(filename='flow.json')

Export the current flow to a JSON file.

Parameters:

Name	Type	Description	Default
filename	str	Output filename	'flow.json'

Source code in pynodewidget/widget.py

def export_json(self, filename: str = "flow.json") -> str:
    """Export the current flow to a JSON file.

    Args:
        filename: Output filename
    """
    data = {
        "nodes": self.nodes,
        "edges": self.edges,
        "viewport": self.viewport,
        "node_templates": self.node_templates,
        "node_values": dict(self.node_values)
    }

    with open(filename, 'w') as f:
        json.dump(data, f, indent=2)

    print(f"✓ Flow exported to {filename}")
    return filename

clear()

Clear all nodes and edges.

Source code in pynodewidget/widget.py

def clear(self) -> None:
    """Clear all nodes and edges."""
    self.nodes = {}  # Empty dict instead of list
    self.edges = []
    self.node_values = {}  # Clear values too
    return self

get_flow_data()

Get the current flow data as a dictionary.

Returns:

Type	Description
Dict[str, Any]	Dictionary with nodes and edges

Source code in pynodewidget/widget.py

def get_flow_data(self) -> Dict[str, Any]:
    """Get the current flow data as a dictionary.

    Returns:
        Dictionary with nodes and edges
    """
    return {
        "nodes": self.nodes,
        "edges": self.edges
    }

add_node_type_from_schema(json_schema, type_name, label, description='', icon='', grid_layout=None, style=None, _default_values_override=None)

Add a node type from a JSON schema with grid layout support.

Parameters:

Name	Type	Description	Default
json_schema	Dict[str, Any]	JSON Schema definition (can be from Pydantic model_json_schema())	required
type_name	str	Unique type identifier	required
label	str	Display label for the node	required
description	str	Description shown in the panel	''
icon	str	Unicode emoji or symbol (e.g., "🔧", "⚙️", "📊")	''
grid_layout	Optional[Dict[str, Any]]	Grid layout configuration (use helpers from grid_layouts module). Can be a dict or a NodeGrid Pydantic model. If not provided, defaults to vertical layout with JSON schema fields.	None
style	Optional[Dict[str, Any]]	Style configuration dict with 'minWidth', 'maxWidth', 'shadow', etc.	None
_default_values_override	Optional[Dict[str, Any]]	Internal parameter to override default values extraction	None

Example

from pynodewidget.grid_layouts import create_three_column_grid from pynodewidget.models import BaseHandle, TextField

widget.add_node_type_from_schema( ... json_schema={"type": "object", "properties": {...}}, ... type_name="processor", ... label="Data Processor", ... icon="⚙️", ... grid_layout=create_three_column_grid( ... left_components=[BaseHandle(id="in1", label="Input", handle_type="input")], ... center_components=[TextField(id="name", label="Name")], ... right_components=[BaseHandle(id="out1", label="Output", handle_type="output")] ... ) ... )

Source code in pynodewidget/widget.py

def add_node_type_from_schema(
    self, 
    json_schema: Dict[str, Any],
    type_name: str,
    label: str,
    description: str = "",
    icon: str = "",
    grid_layout: Optional[Dict[str, Any]] = None,
    style: Optional[Dict[str, Any]] = None,
    _default_values_override: Optional[Dict[str, Any]] = None
):
    """Add a node type from a JSON schema with grid layout support.

    Args:
        json_schema: JSON Schema definition (can be from Pydantic model_json_schema())
        type_name: Unique type identifier
        label: Display label for the node
        description: Description shown in the panel
        icon: Unicode emoji or symbol (e.g., "🔧", "⚙️", "📊")
        grid_layout: Grid layout configuration (use helpers from grid_layouts module).
                    Can be a dict or a NodeGrid Pydantic model.
                    If not provided, defaults to vertical layout with JSON schema fields.
        style: Style configuration dict with 'minWidth', 'maxWidth', 'shadow', etc.
        _default_values_override: Internal parameter to override default values extraction

    Example:
        >>> from pynodewidget.grid_layouts import create_three_column_grid
        >>> from pynodewidget.models import BaseHandle, TextField
        >>> 
        >>> widget.add_node_type_from_schema(
        ...     json_schema={"type": "object", "properties": {...}},
        ...     type_name="processor",
        ...     label="Data Processor",
        ...     icon="⚙️",
        ...     grid_layout=create_three_column_grid(
        ...         left_components=[BaseHandle(id="in1", label="Input", handle_type="input")],
        ...         center_components=[TextField(id="name", label="Name")],
        ...         right_components=[BaseHandle(id="out1", label="Output", handle_type="output")]
        ...     )
        ... )
    """
    from .grid_layouts import create_vertical_stack_grid, json_schema_to_components
    from .models import NodeDefinition, NodeTemplate, NodeGrid

    # Initialize default values from schema or override
    if _default_values_override is not None:
        default_values = _default_values_override
    else:
        default_values = {}
        if json_schema and "properties" in json_schema:
            for key, prop in json_schema["properties"].items():
                if "default" in prop:
                    default_values[key] = prop["default"]

    # Use vertical stack grid with JSON schema fields as default if none provided
    if grid_layout is None:
        field_components = json_schema_to_components(json_schema, default_values)
        grid_layout = create_vertical_stack_grid(middle_components=field_components)

    # Convert NodeGrid model to dict if needed
    if isinstance(grid_layout, NodeGrid):
        grid_layout = grid_layout.model_dump()

    # Validate that all component IDs are unique
    try:
        _validate_unique_component_ids(grid_layout)
    except ValueError as e:
        raise ValueError(f"Invalid grid layout for node type '{type_name}': {e}")

    # Build NodeDefinition (visual structure only)
    definition_dict = {
        "grid": grid_layout
    }

    # Add optional style configuration
    if style is not None:
        definition_dict["style"] = style

    # Validate using Pydantic models
    try:
        definition = NodeDefinition(**definition_dict)
        template_dict = {
            "type": type_name,
            "label": label,
            "description": description,
            "icon": icon,
            "definition": definition.model_dump(),
            "defaultValues": default_values
        }
        template = NodeTemplate(**template_dict)

        # Add validated template
        self.node_templates = self.node_templates + [template.model_dump()]
    except Exception as e:
        raise ValueError(f"Failed to create valid node template: {e}")

    return self

Overview

NodeFlowWidget is the entry point for creating node-based UIs in Jupyter. It manages:

• Node types: Registry of available node classes
• Graph state: Nodes, edges, and their positions
• Values: Current parameter values for all nodes
• Viewport: Canvas position and zoom level

Basic Usage

Creating a Widget

from pynodewidget import NodeFlowWidget

# Empty widget
flow = NodeFlowWidget()

# With registered node types
flow = NodeFlowWidget(nodes=[MyNode1, MyNode2])

# Custom height
flow = NodeFlowWidget(height="800px")

Registering Node Types

# Register during initialization
flow = NodeFlowWidget(nodes=[ProcessorNode, SourceNode])

# Register after creation
flow.register_node_type(SinkNode)

# Register with custom type name
flow.register_node_type(MyNode, type_name="custom_processor")

Working with Values

Getting Values

# Get all values for a node
values = flow.get_node_values("node-1")
# Returns: {"threshold": 0.5, "enabled": True}

# Get single value with default
threshold = flow.get_node_value("node-1", "threshold", default=0.5)

Setting Values

# Update single value
flow.update_node_value("node-1", "threshold", 0.8)

# Update multiple values
flow.set_node_values("node-1", {
    "threshold": 0.8,
    "enabled": False,
    "mode": "advanced"
})

Value Synchronization

The node_values trait uses ObservableDict for automatic synchronization:

# This triggers an update to the JavaScript UI
flow.node_values["node-1"]["threshold"] = 0.8

# Changes in the UI automatically update this dict
print(flow.node_values["node-1"]["threshold"])

Managing Graph Structure

Accessing Nodes and Edges

# Get all nodes
nodes = flow.nodes
# [{"id": "node-1", "type": "ProcessorNode", "position": {...}, ...}]

# Get all edges
edges = flow.edges
# [{"id": "e1-2", "source": "node-1", "target": "node-2", ...}]

# Get specific node data
node_data = flow.get_node_data("node-1")

Modifying Graph

# Add a node programmatically
flow.nodes = [
    *flow.nodes,
    {
        "id": "new-node",
        "type": "ProcessorNode",
        "position": {"x": 100, "y": 100},
        "data": {}
    }
]

# Add an edge
flow.edges = [
    *flow.edges,
    {
        "id": "e-new",
        "source": "node-1",
        "target": "new-node",
        "sourceHandle": "out",
        "targetHandle": "in"
    }
]

# Clear everything
flow.clear()

Import and Export

Export to File

# Export complete workflow
flow.export_json("workflow.json")

# Export returns the filename
filename = flow.export_json("my_workflow.json")
print(f"Saved to {filename}")

The exported JSON contains:

{
  "nodes": [...],
  "edges": [...],
  "viewport": {"x": 0, "y": 0, "zoom": 1},
  "node_templates": [...]
}

Import from File

# Load workflow
flow.load_json("workflow.json")

# Method chaining
flow.clear().load_json("workflow.json")

Node Types Must Be Registered

Before loading a workflow, ensure all node types used in the workflow are registered, or they won't render correctly.

Export as Dictionary

# Get flow data as dict
flow_data = flow.get_flow_data()
# Returns: {"nodes": [...], "edges": [...]}

# Full export including templates and viewport
import json
full_export = {
    "nodes": flow.nodes,
    "edges": flow.edges,
    "node_templates": flow.node_templates,
    "viewport": flow.viewport,
    "node_values": dict(flow.node_values)
}

Traits (Synchronized Attributes)

These attributes automatically sync between Python and JavaScript:

nodes: List[Dict]

List of node objects in the graph.

flow.nodes = [
    {
        "id": "node-1",
        "type": "ProcessorNode",
        "position": {"x": 100, "y": 50},
        "data": {...}
    }
]

edges: List[Dict]

List of edge objects connecting nodes.

flow.edges = [
    {
        "id": "e1-2",
        "source": "node-1",
        "target": "node-2",
        "sourceHandle": "out",
        "targetHandle": "in"
    }
]

node_templates: List[Dict]

Registered node type definitions. Populated by register_node_type().

node_values: ObservableDict

Current parameter values for all nodes, keyed by node ID.

flow.node_values = {
    "node-1": {"threshold": 0.5, "enabled": True},
    "node-2": {"count": 10}
}

viewport: Dict

Current viewport position and zoom.

flow.viewport = {"x": 100, "y": 50, "zoom": 1.5}

height: str

Widget height (CSS value).

flow.height = "800px"
flow.height = "100vh"

Legacy Methods

These methods are provided for backward compatibility:

add_node_type_from_schema()

Register a node type from a raw JSON schema.

flow.add_node_type_from_schema(
    json_schema={"type": "object", "properties": {...}},
    type_name="processor",
    label="Processor",
    icon="⚙️",
    inputs=[{"id": "in", "label": "Input"}],
    outputs=[{"id": "out", "label": "Output"}]
)

Prefer register_node_type()

For new code, use register_node_type() with a JsonSchemaNodeWidget subclass instead.

add_node_type_from_pydantic()

Register a node type from a Pydantic model.

from pydantic import BaseModel

class ProcessorParams(BaseModel):
    threshold: float = 0.5

flow.add_node_type_from_pydantic(
    model_class=ProcessorParams,
    type_name="processor",
    label="Processor",
    icon="⚙️"
)

Examples

Complete Workflow Example

from pydantic import BaseModel, Field
from pynodewidget import NodeFlowWidget, JsonSchemaNodeWidget

# Define parameters
class FilterParams(BaseModel):
    threshold: float = Field(default=0.5, ge=0, le=1)
    enabled: bool = True

# Define node
class FilterNode(JsonSchemaNodeWidget):
    label = "Filter"
    parameters = FilterParams
    icon = "🔍"
    inputs = [{"id": "in", "label": "Data"}]
    outputs = [{"id": "out", "label": "Filtered"}]

    def execute(self, inputs):
        config = self.get_values()
        if not config["enabled"]:
            return {"out": inputs["in"]}

        data = inputs["in"]
        threshold = config["threshold"]
        return {"out": [x for x in data if x >= threshold]}

# Create widget
flow = NodeFlowWidget(nodes=[FilterNode], height="600px")

# Later: Update values from Python
flow.update_node_value("filter-1", "threshold", 0.8)

# Execute workflow (custom logic)
def run_workflow(flow):
    # Your execution logic here
    pass

# Export for later use
flow.export_json("filter_workflow.json")

Multiple Node Types

class SourceNode(JsonSchemaNodeWidget):
    label = "Data Source"
    parameters = SourceParams
    outputs = [{"id": "data", "label": "Data"}]

class ProcessorNode(JsonSchemaNodeWidget):
    label = "Processor"
    parameters = ProcessorParams
    inputs = [{"id": "in", "label": "Input"}]
    outputs = [{"id": "out", "label": "Output"}]

class SinkNode(JsonSchemaNodeWidget):
    label = "Data Sink"
    parameters = SinkParams
    inputs = [{"id": "data", "label": "Data"}]

# Create comprehensive workflow
flow = NodeFlowWidget(
    nodes=[SourceNode, ProcessorNode, SinkNode],
    height="800px"
)

Programmatic Graph Construction

# Create widget
flow = NodeFlowWidget(nodes=[MyNode])

# Add nodes programmatically
flow.nodes = [
    {
        "id": "source",
        "type": "my_node",
        "position": {"x": 0, "y": 0},
        "data": {}
    },
    {
        "id": "processor",
        "type": "my_node",
        "position": {"x": 200, "y": 0},
        "data": {}
    }
]

# Connect them
flow.edges = [
    {
        "id": "e1",
        "source": "source",
        "target": "processor",
        "sourceHandle": "out",
        "targetHandle": "in"
    }
]

# Set initial values
flow.set_node_values("processor", {"threshold": 0.7})

See Also

• JsonSchemaNodeWidget: Create custom nodes
• ObservableDict: Auto-syncing dictionary
• Protocols: Extension protocols