Skip to main content

Workflow Nodes

Manage workflow node types (the catalog of available node definitions).

Access it as client.workflow_nodes on a Strongly client, or the same path on AsyncStrongly with await. All methods exist on both with identical signatures.

Quick start​

from strongly import Strongly

client = Strongly()

# List (auto-paginates as you iterate)  # filters: search, category, type, is_system
for workflow in client.workflow_nodes.list():
    print(workflow.id)

# Create
created = client.workflow_nodes.create(
    label="...",
    category="...",
    type="...",
    function_definition={...},
)
print(created.id)

# Read it back
fetched = client.workflow_nodes.retrieve(created.id)

# Delete (returns None; raises on failure)
client.workflow_nodes.delete(created.id)

Methods​

Core​

list​

list(*, search: Optional[str] = None, category: Optional[str] = None, type: Optional[str] = None, is_system: Optional[bool] = None, limit: Optional[int] = None) -> SyncPaginator[WorkflowNode]

List workflow node types with pagination and filtering.

Args: search: Search by label, description, or tags. category: Filter by category. type: Filter by node type. is_system: Filter by system (True) vs custom (False) nodes. limit: Maximum number of items to return (default: all matching items).

create​

create(*, label: str, category: str, type: str, function_definition: Mapping[str, Any], description: Optional[str] = None, is_active: Optional[bool] = None, input_definition: Optional[Mapping[str, Any]] = None, output_definition: Optional[Mapping[str, Any]] = None, editor_config: Optional[Mapping[str, Any]] = None, default_data: Optional[Mapping[str, Any]] = None) -> WorkflowNode

Create a custom workflow node type.

Args: label: Display label for the node. category: Node category (triggers, sources, transform, ai, agents, control-flow, destinations, utilities). type: Unique node type identifier. function_definition: The node's executable definition (free-form schema describing how the node runs). description: Optional description of the node. is_active: Whether the node is active (default: True server-side). input_definition, output_definition: Optional I/O schema definitions. editor_config: Optional editor configuration. default_data: Optional default node data.

retrieve​

retrieve(node_id: str) -> WorkflowNode

Get a single workflow node type by ID.

update​

update(node_id: str, *, label: Optional[str] = None, category: Optional[str] = None, type: Optional[str] = None, description: Optional[str] = None, is_active: Optional[bool] = None, input_definition: Optional[Mapping[str, Any]] = None, output_definition: Optional[Mapping[str, Any]] = None, editor_config: Optional[Mapping[str, Any]] = None, default_data: Optional[Mapping[str, Any]] = None) -> WorkflowNode

Update a workflow node type (system nodes cannot be modified).

delete​

delete(node_id: str) -> None

Delete a workflow node type (system nodes cannot be deleted).

Other​

datasource_fields​

datasource_fields(datasource_id: str) -> Dict[str, Any]

List fields exposed by a data source for mapping in node inputs.

schema​

schema(node_id: str) -> Dict[str, Any]

Get the input/output schema for a node type.

services_addons​

services_addons(*, type: Optional[str] = None) -> ServiceAddonsResponse

List available add-ons for use in workflow nodes.

Args: type: Optional addon type filter.

services_datasources​

services_datasources(*, type: Optional[str] = None, category: Optional[str] = None) -> ServiceDataSourcesResponse

List available data sources for use in workflow nodes.

Args: type: Optional data source type filter. category: Optional category filter.

services_models​

services_models(*, provider: Optional[str] = None, type: Optional[str] = None) -> ServiceModelsResponse

List available AI models for use in workflow nodes.

Args: provider: Optional provider filter (e.g. "openai", "anthropic"). type: Optional model type filter.

suggest_mappings​

suggest_mappings(*, source_node_id: str, target_node_id: str, **params) -> Dict[str, Any]

Get auto-suggested input mappings between two nodes.

sync_from_s3​

sync_from_s3() -> Dict[str, Any]

Trigger a sync of the workflow-node catalog from S3 (admin only).