Socket

Sockets are used to indicate the type of data that can be transferred from one task to another. You can only connect inputs and outputs with the same socket type.

Location

According to its location, there are two kinds of sockets:

• Input

• Output

Each task can have input sockets and output sockets.

from node_graph import Graph

ng = Graph(name="socket_example")
float1 = ng.add_task("node_graph.test_float", name="float1")
float2 = ng.add_task("node_graph.test_float", name="float2")

# connect output of float1 to input of float2
ng.add_link(float1.outputs.result, float2.inputs.value)

TaskLink(from="float1.result", to="float2.value")

Data type

Socket can have different data types:

• SocketFloat

• SocketInt

• SocketBool

• SocketString

• SocketAny

• SocketAnnotated (for annotated Python types without an explicit mapping)

One can extend the socket type by designing a custom socket (see the custom_socket page in the docs).

Annotated types

When a task input/output is annotated with a Python type, the socket identifier is resolved from the type mapping. If the type is not mapped, the socket falls back to node_graph.annotated and records the Python type in metadata.

This keeps links type-safe without requiring a custom socket for every domain type. Links between two node_graph.annotated sockets must match the recorded Python type, otherwise a Socket annotated type mismatch error is raised.

from typing import Optional
from node_graph import task


class CustomType:
    pass


@task()
def takes_custom(x: CustomType) -> CustomType:
    return x


@task()
def takes_custom_optional(x: Optional[CustomType]) -> Optional[CustomType]:
    return x

Property

A socket has a property, which stores the data when there is no connection to the input socket. The property is a node_graph.property.Property object. You can add properties when defining a custom socket or later with add_property for SocketAny.

from node_graph.task import Task


class SymbolTask(Task):
    identifier = "SymbolTask"
    name = "SymbolTask"

    def update_spec(self):
        # create an Any type socket
        inp = self.add_input("node_graph.any", "symbols")
        # add a string property to the socket with default value "H"
        inp.add_property("node_graph.string", "default", default="H")


task = SymbolTask(parent=ng)
print("Custom task with Any socket and property:", task.inputs.symbols.value)

Custom task with Any socket and property: H

Serialization

Sockets support serialization and deserialization, which determine how results are stored and read from the database.

There are two built-in serialization methods:

• None: no serialization, used for base types (Int, Float, Bool, String).

• Json: JSON serialization, used for complex types (e.g. lists, dicts).

Users can define a new socket type with a custom serialization method.

Dynamic sockets

Users can update sockets dynamically based on a property value using an update callback. For details, see the dynamic_socket page in the documentation.