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, workflow_type, is_system
for workflow in client.workflow_nodes.list():
print(workflow.id)
Methods
Core
list
list(*, search: str | None = None, category: str | None = None, type: str | None = None, workflow_type: str | None = None, is_system: bool | None = None, limit: int | None = None) -> SyncPaginator[WorkflowNode]
List workflow node types with pagination and filtering.
Parameters
search(str | None, optional): Search by label, description, or tags.category(str | None, optional): Filter by category.type(str | None, optional): Filter by node type.workflow_type(str | None, optional): Filter by execution mode:"batch"(nodes usable in batch workflows) or"streaming"(real-time frame-based nodes). Omit to list nodes of both modes.is_system(bool | None, optional): Filter by system (True) vs custom (False) nodes.limit(int | None, optional): Maximum number of items to return (default: all matching items).
create
create(*, label: str, category: str, type: str, function_definition: Mapping[str, Any], description: str | None = None, is_active: bool | None = None, input_definition: Mapping[str, Any] | None = None, output_definition: Mapping[str, Any] | None = None, editor_config: Mapping[str, Any] | None = None, default_data: Mapping[str, Any] | None = None) -> WorkflowNode
Create a custom workflow node type.
Parameters
label(str): Display label for the node.category(str): Node category (triggers,sources,transform,ai,agents,control-flow,destinations,utilities).type(str): Unique node type identifier.function_definition(Mapping[str, Any]): The node's executable definition (free-form schema describing how the node runs).description(str | None, optional): Optional description of the node.is_active(bool | None, optional): Whether the node is active (default:Trueserver-side).input_definition, output_definition(Mapping[str, Any] | None, optional): Optional I/O schema definitions.editor_config(Mapping[str, Any] | None, optional): Optional editor configuration.default_data(Mapping[str, Any] | None, optional): Optional default node data.
retrieve
retrieve(node_id: str) -> WorkflowNode
Get a single workflow node type by ID.
update
update(node_id: str, *, label: str | None = None, category: str | None = None, type: str | None = None, description: str | None = None, is_active: bool | None = None, input_definition: Mapping[str, Any] | None = None, output_definition: Mapping[str, Any] | None = None, editor_config: Mapping[str, Any] | None = None, default_data: Mapping[str, Any] | None = 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
activate_tool
activate_tool(node_id: str, *, config: Mapping[str, Any] | None = None) -> dict[str, Any]
Register an MCP server node in your organization's MCP gateway (or re-register it).
Parameters
node_id(str): The MCP server's node type ID.config(Mapping[str, Any] | None, optional): Your own configuration values for it; your calls through it use them, and no one else's do.
add_member
add_member(resource_id: str, *, user_id: str, role: str) -> Permissions
Share the resource with a user.
Parameters
resource_id(str): The resource's id.user_id(str): The user to share it with.role(str):"editor"(use and change it) or"user"(use it only; types without a use-only tier, such as knowledge bases, take"editor"only).
datasource_fields
datasource_fields(datasource_id: str) -> dict[str, Any]
List fields exposed by a data source for mapping in node inputs.
deactivate_tool
deactivate_tool(node_id: str) -> dict[str, Any]
Remove an MCP server's registration from your organization's MCP gateway.
Parameters
node_id(str): The MCP server's node type ID.
duplicate
duplicate(node_id: str) -> WorkflowNode
Copy a workflow node type into a new one you own.
Parameters
node_id(str): The node type to copy.
permissions
permissions(resource_id: str) -> Permissions
Return who can reach the resource: its owner, members and their roles, and visibility.
Parameters
resource_id(str): The resource's id.
remove_member
remove_member(resource_id: str, user_id: str) -> None
Stop sharing the resource with a user.
Parameters
resource_id(str): The resource's id.user_id(str): The user to stop sharing it with.
schema
schema(node_id: str) -> dict[str, Any]
Get the input/output schema for a node type.
services_addons
services_addons(*, type: str | None = None) -> ServiceAddonsResponse
List available add-ons for use in workflow nodes.
Parameters
type(str | None, optional): Optional addon type filter.
services_datasources
services_datasources(*, type: str | None = None, category: str | None = None) -> ServiceDataSourcesResponse
List available data sources for use in workflow nodes.
Parameters
type(str | None, optional): Optional data source type filter.category(str | None, optional): Optional category filter.
services_models
services_models(*, provider: str | None = None, type: str | None = None) -> ServiceModelsResponse
List available AI models for use in workflow nodes.
Parameters
provider(str | None, optional): Optional provider filter (e.g."openai","anthropic").type(str | None, optional): Optional model type filter.
set_tool_enabled
set_tool_enabled(node_id: str, *, enabled: bool) -> dict[str, Any]
Switch a registered MCP server's tools on or off, keeping its registration.
Parameters
node_id(str): The MCP server's node type ID.enabled(bool): Whether its tools are on.
set_visibility
set_visibility(resource_id: str, visibility: str) -> Permissions
Make the resource public (every user can find and use it) or private.
Parameters
resource_id(str): The resource's id.visibility(str):"public"or"private".
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).