Custom nodes

This guide creates and tests a local node. Defining a Python subclass does not automatically install it into a hosted workspace or make it available to other users; distribution and hosted runtime availability must be arranged for your workspace.

Implement the node

Save as uppercase_node.py:

import asyncio

from dagy.nodes import (
    ConnectorDef,
    ConnectorDirection,
    DataType,
    ExecutionContext,
    ExecutionResult,
    FlowNode,
    NodeConfigField,
    NodeMetadata,
)
from dagy.nodes.metadata import ConfigFieldType, NodeAvailability, NodeCategory


class UppercaseNode(FlowNode):
    @classmethod
    def metadata(cls) -> NodeMetadata:
        return NodeMetadata(
            node_type="example_uppercase",
            label="Uppercase text",
            description="Normalize incoming text for a downstream system.",
            category=NodeCategory.TRANSFORM,
            availability=NodeAvailability.PREVIEW,
        )

    @classmethod
    def connectors(cls) -> list[ConnectorDef]:
        return [
            ConnectorDef(
                id="text_in",
                name="Text",
                direction=ConnectorDirection.INBOUND,
                data_types=frozenset({DataType.STRING}),
                required=True,
            ),
            ConnectorDef(
                id="text_out",
                name="Normalized text",
                direction=ConnectorDirection.OUTBOUND,
                data_types=frozenset({DataType.STRING}),
            ),
        ]

    @classmethod
    def config_schema(cls) -> list[NodeConfigField]:
        return [
            NodeConfigField(
                name="trim",
                label="Remove surrounding whitespace",
                field_type=ConfigFieldType.BOOLEAN,
                default=True,
            )
        ]

    async def execute(self, context: ExecutionContext) -> ExecutionResult:
        text = context.get_upstream_data("text_in")
        if not isinstance(text, str):
            return ExecutionResult.from_error("text_in must be a string", "InvalidInput")
        if self.config.get("trim", True):
            text = text.strip()
        context.log_info("Normalizing text")
        return ExecutionResult.from_success({"text_out": text.upper()})


async def main() -> None:
    node = UppercaseNode(config={"trim": True}, instance_id="uppercase-1")
    result = await node.run(ExecutionContext(upstream_results={"text_in": " hello "}))
    assert result.success, result.error
    assert result.outputs == {"text_out": "HELLO"}
    print(result.outputs)


if __name__ == "__main__":
    asyncio.run(main())

python uppercase_node.py

Expected output is {'text_out': 'HELLO'}. Call run(context) for the full lifecycle; directly calling execute() bypasses configuration validation, lifecycle hooks, and node timeout handling.

Define the public contract

A node implements three required methods: metadata(), connectors(), and async execute(context). Override config_schema() when configuration is needed.

Supported config types are string, number, boolean, select, textarea, json, secret_ref, file_path, code, key_value, and connection_ref. Import their enum and category/availability types from dagy.nodes.metadata.

Defaults in the configuration definition are form metadata; constructing a node does not fill every default into self.config. Read optional values with .get(name, default). Provide required values explicitly even if a schema field also declares a default.

Ports can declare any, string, number, boolean, json, dataframe, binary, list, embedding, image, audio, document, event, error, or trigger. These are compatibility declarations, not comprehensive payload validation. Validate the shape and meaning of incoming data in your node. Port cardinality supports one, zero_or_one, zero_or_many, and one_or_many.

Use context and secrets

ExecutionContext provides run and node identifiers, attempt information, run parameters, environment, upstream data, logging methods, and secret resolution. get_upstream_data(port_id) returns None when no input is present. get_all_upstream_data() returns the input mapping.

For a local credential-dependent test:

from dagy.nodes import ExecutionContext
from dagy.nodes.execution import SecretStr

context = ExecutionContext(secrets={"service_token": SecretStr("test-only-value")})
assert context.get_secret("service_token") == "test-only-value"

SecretStr masks ordinary string representation; get_secret() returns the actual value for a client call. It does not make logging the unwrapped value safe. In a workspace, use stored secrets, and use the exact reference field expected by the node. Built-in reference fields generally use secret:NAME.

Logging methods are log_info, log_warning, log_error, and log_debug. Keep logs useful without including authorization headers, customer payloads, or full credential-bearing URLs.

Register and inspect

A subclass becomes available when its module is imported. Synchronize the registry before listing imported types:

from dagy.nodes import node_registry
from uppercase_node import UppercaseNode

node_registry.sync_from_subclasses()
assert node_registry.get("example_uppercase") is UppercaseNode
print(UppercaseNode.to_definition_dict())

For built-ins, call node_registry.discover_builtin() before synchronization. The registry supports get, has, list_types, list_all, list_by_category, search, register, unregister, and create_instance. Registration of a different class under an existing type raises NodeRegistrationError unless explicitly forced. Treat stable type and connector IDs as compatibility contracts for saved workflows.

Lifecycle and error handling

The lifecycle is configuration validation, pre_execute, execute, and post_execute. Optional on_error can return a recovery result; on_cancel exists as an extension hook but is not automatically invoked on every cancellation path.

Use ExecutionResult.from_error(message, error_type) for a controlled failure and from_success({port: value}) for success. Unhandled exceptions are converted to an error result. The default base validator checks required fields, numeric ranges, and configured regexes. Override validate_config() for additional checks, returning a list of messages.

The default execution timeout is 300 seconds. Node configuration _timeout_seconds is clamped between 1 and 3,600 seconds. Async timeout enforcement cannot reliably interrupt blocking third-party calls, so set their network deadlines too. The node wrapper does not automatically retry a failed result; add controlled retry logic for transient external errors where safe.

The runtime forwards outputs only from successful nodes. A failed node's error_out is not automatically delivered to a downstream error handler. When you intentionally route validation failures, return a successful result with distinct valid/invalid ports, as the data-quality node does. Do not use this pattern to conceal execution failure from callers.

Context callback and webhook registration methods collect declarations; the base node lifecycle does not dispatch them. To send an outbound HTTP notification, call a supported integration explicitly or use the Webhook Post node.

Run a node graph locally

For integrations that construct a graph programmatically, dagy.nodes.graph_spec provides NodeInstanceSpec, NodeEdgeSpec, and NodeGraphSpec; dagy.nodes.runtime.NodeRuntimeExecutor executes them asynchronously.

import asyncio
from datetime import datetime, timezone
from dagy.nodes.graph_spec import NodeGraphSpec, NodeInstanceSpec
from dagy.nodes.runtime import NodeRuntimeExecutor
from uppercase_node import UppercaseNode

# A root node that reads run parameters can be executed without an inbound edge.
class ParameterTextNode(UppercaseNode):
    @classmethod
    def metadata(cls):
        from dagy.nodes import NodeMetadata
        return NodeMetadata(node_type="example_parameter_text", label="Parameter text")

    async def execute(self, context):
        from dagy.nodes import ExecutionResult
        text = context.parameters.get("text", "")
        return ExecutionResult.from_success({"text_out": str(text).upper()})


graph = NodeGraphSpec(
    name="parameter_text",
    version="1",
    created_at=datetime.now(timezone.utc).isoformat(),
    nodes=[NodeInstanceSpec(instance_id="text", node_type="example_parameter_text")],
    edges=[],
)
result = asyncio.run(NodeRuntimeExecutor().run(graph, parameters={"text": "hello"}))
assert result.outputs["text"]["text_out"] == "HELLO"

The executor schedules nodes in dependency order, sequentially. Edges map source output port IDs to target input port IDs. A required wired input receiving no value causes a skip; cycles or missing source nodes fail execution. NodeInstanceSpec exposes retry/timeout fields, but this executor does not apply them as a retry engine or override the node's _timeout_seconds. Set the node configuration and validate the behavior in the runtime you plan to use.

Test successful and invalid inputs, missing credentials, network failures, port compatibility, and runtime prerequisites before distributing a custom node. See the built-in catalog for established contracts.