Skip to main content

Workflows

Workflows are node-graph pipelines you build, version, deploy, and run. Use this resource to author a workflow, manage its nodes and connections, run it (synchronously or fire-and-forget), and share it.

Access it as client.workflows 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: status, search, tag
for workflow in client.workflows.list():
    print(workflow.id)

# Create
created = client.workflows.create(name="...")
print(created.id)

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

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

Methods​

Core​

list​

list(*, status: Optional[str] = None, search: Optional[str] = None, tag: Optional[str] = None, limit: Optional[int] = None) -> SyncPaginator[Workflow]

List workflows with pagination and filtering.

Args: status: Filter by workflow status ("draft", "active", "paused", "archived"). search: Search by name or description. tag: Filter by tag. limit: Maximum number of items to return (default: all matching items).

create​

create(*, name: str, description: Optional[str] = None, status: Optional[str] = None, workflow_type: Optional[str] = None, tags: Optional[Sequence[str]] = None, nodes: Optional[Sequence[Mapping[str, Any]]] = None, connections: Optional[Sequence[Mapping[str, Any]]] = None, settings: Optional[Mapping[str, Any]] = None) -> Workflow

Create a new workflow.

Args: name: Workflow name (required). description: Workflow description. status: Initial status (default: "draft"). workflow_type: Execution mode: "batch" (default) or "streaming". tags: Tags for the workflow. nodes: Initial graph nodes (free-form node objects). connections: Initial graph connections (free-form connection objects). settings: Workflow settings (free-form, e.g. spanLevel, retryAttempts).

retrieve​

retrieve(workflow_id: str) -> Workflow

Get a single workflow by ID.

update​

update(workflow_id: str, *, name: Optional[str] = None, description: Optional[str] = None, status: Optional[str] = None, tags: Optional[Sequence[str]] = None, nodes: Optional[Sequence[Mapping[str, Any]]] = None, connections: Optional[Sequence[Mapping[str, Any]]] = None, config: Optional[Mapping[str, Any]] = None, settings: Optional[Mapping[str, Any]] = None) -> Workflow

Update a workflow.

Args: workflow_id: The workflow ID. name, description, status, tags: Flat metadata to update. nodes, connections: Full graph node/connection lists (free-form). config: Workflow config object (free-form). settings: Workflow settings object (free-form). Replaces the stored settings object, so read-modify-write to preserve other keys.

delete​

delete(workflow_id: str) -> None

Delete a workflow.

Lifecycle & actions​

start​

start(workflow_id: str) -> Dict[str, Any]

Start a deployed workflow (long-running types).

stop​

stop(workflow_id: str) -> Dict[str, Any]

Stop a running workflow.

deploy​

deploy(workflow_id: str, **kwargs) -> Dict[str, Any]

Deploy a workflow.

Args: workflow_id: The workflow ID. **kwargs: Optional deployment configuration.

undeploy​

undeploy(workflow_id: str) -> Dict[str, Any]

Undeploy a workflow.

execute​

execute(workflow_id: str, *, definition: Optional[Mapping[str, Any]] = None, trigger_inputs: Optional[Mapping[str, Any]] = None, capacity_type: Optional[str] = None) -> Dict[str, Any]

Execute a workflow.

Args: workflow_id: The workflow ID. definition: Optional graph definition override (nodes + connections); defaults to the stored workflow definition. trigger_inputs: Input data for the trigger node. capacity_type: Optional compute capacity type (e.g. spot instances).

Returns: Dict containing executionId.

trigger​

trigger(workflow_id: str, *, inputs: Optional[Dict[str, Any]] = None, trigger_type: Optional[str] = None, sync: bool = False) -> Dict[str, Any]

Trigger a deployed (production) workflow.

Unlike execute which starts a dev run, this sends a request to the deployed pod's /trigger endpoint inside K8s.

Args: workflow_id: The workflow ID. inputs: Input data for the workflow trigger node. trigger_type: Optional trigger type (default "api"). sync: If True, wait for the workflow to complete and return the output.

Sharing & scope​

share​

share(workflow_id: str, *, user_id: str, permission: Optional[str] = None) -> Dict[str, Any]

Share a workflow with a user.

Args: workflow_id: The workflow ID. user_id: The user ID to share with. permission: Optional permission level.

unshare​

unshare(workflow_id: str, *, user_id: str) -> Dict[str, Any]

Unshare a workflow from a user.

Args: workflow_id: The workflow ID. user_id: The user ID to remove sharing for.

Other​

add_connection​

add_connection(workflow_id: str, *, source_node_id: str, target_node_id: str, source_port: Optional[str] = None, target_port: Optional[str] = None, feedback: Optional[bool] = None, max_iterations: Optional[int] = None) -> Dict[str, Any]

Add a connection between two nodes.

Args: workflow_id: The workflow ID. source_node_id: Source node ID. target_node_id: Target node ID. source_port: Source port (default "output"). target_port: Target port (default "input"; use "ai" or "tools" for agent nodes). feedback: Mark this as a streaming feedback edge (closes a loop). Must be paired with max_iterations. max_iterations: Bound on feedback-edge iterations per session. Required when feedback is True.

add_node​

add_node(workflow_id: str, *, node_id: str, label: Optional[str] = None, version: Optional[str] = None, config: Optional[Mapping[str, Any]] = None, position: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]

Add a node to a workflow graph.

Args: workflow_id: The workflow ID. node_id: Catalog node identifier (e.g. "webhook", "ai-gateway"). label: Optional custom label for the node. version: Optional catalog version to pin (default: catalog latest). config: Initial node configuration (free-form). position: Canvas position {x, y} (free-form).

compute_scopes​

compute_scopes(workflow_id: str) -> Dict[str, Any]

Compute required scopes for a workflow run.

create_version​

create_version(workflow_id: str, *, version_tag: str, description: Optional[str] = None) -> Dict[str, Any]

Create a new workflow version.

Args: workflow_id: The workflow ID. version_tag: The version tag (e.g. "v2"). description: Optional description for this version.

delete_connection​

delete_connection(workflow_id: str, connection_id: str) -> None

Remove a connection.

delete_node​

delete_node(workflow_id: str, node_id: str) -> None

Remove a node from a workflow graph.

discover​

discover(**params) -> Dict[str, Any]

Discover workflows by capability / tag / search.

duplicate​

duplicate(workflow_id: str) -> Dict[str, Any]

Duplicate a workflow.

Returns: Dict containing workflowId of the new workflow.

email_trigger​

email_trigger(workflow_id: str, *, from_address: Optional[str] = None, to: Optional[str] = None, subject: Optional[str] = None, body_text: Optional[str] = None, body_html: Optional[str] = None, attachments: Optional[Sequence[Mapping[str, Any]]] = None, email_headers: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]

Trigger a workflow from an inbound email payload.

Args: workflow_id: The workflow ID (must have an email trigger node). from_address: Sender email address (sent as from). to: Recipient email address. subject: Email subject. body_text: Plain-text email body. body_html: HTML email body. attachments: Email attachments (free-form objects). email_headers: Raw email headers (free-form).

emit_event​

emit_event(*, event_type: str, event_data: Optional[Mapping[str, Any]] = None, source: Optional[str] = None) -> Dict[str, Any]

Emit a global workflow event that may trigger event-listening workflows.

Args: event_type: The event type to emit (required). event_data: Event payload (free-form). source: Optional event source label.

enqueue​

enqueue(workflow_id: str, *, message: Optional[Mapping[str, Any]] = None, priority: Optional[int] = None) -> Dict[str, Any]

Enqueue a workflow execution via its queue trigger node.

Args: workflow_id: The workflow ID. message: Queue message payload (free-form). priority: Message priority (default 0).

export​

export(workflow_id: str) -> Dict[str, Any]

Export a workflow as a portable JSON bundle.

import_bundle​

import_bundle(*, export_data: Mapping[str, Any], resolved_deps: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]

Import a workflow from a previously exported JSON bundle.

Args: export_data: The exported workflow JSON data (free-form bundle). resolved_deps: Resolved dependency mappings (free-form). Omit for a validation-only pass; supply to execute the import.

layout​

layout(workflow_id: str) -> Dict[str, Any]

Compute an auto-layout for the workflow graph.

lifecycle​

lifecycle(workflow_id: str) -> Dict[str, Any]

Get lifecycle metadata for a workflow.

list_nodes​

list_nodes(workflow_id: str) -> Dict[str, Any]

List nodes in a workflow graph.

shared_users​

shared_users(workflow_id: str) -> List[WorkflowSharedUser]

List users a workflow is shared with.

stats​

stats() -> WorkflowStats

Get workflow statistics (total, active, paused, draft, archived).

templates​

templates() -> List[Workflow]

List workflow templates.

update_lifecycle​

update_lifecycle(workflow_id: str, *, type: str, idle_timeout_minutes: Optional[int] = None, schedule: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]

Update the lifecycle policy for a workflow.

Args: workflow_id: The workflow ID. type: Policy type: "always-on", "idle-shutdown", "on-demand" or "scheduled-window". idle_timeout_minutes: Minutes before idle shutdown (5-1440, default 30); used with "idle-shutdown". schedule: Schedule config (free-form), required for "scheduled-window": {timezone, windows: [{days, startTime, endTime}]}.

update_node​

update_node(workflow_id: str, node_id: str, *, config: Optional[Mapping[str, Any]] = None, label: Optional[str] = None) -> Dict[str, Any]

Update a node in a workflow graph.

Args: workflow_id: The workflow ID. node_id: The node ID within the workflow. config: Configuration key-value pairs to merge (free-form). label: Optional new label for the node.

update_node_input_mappings​

update_node_input_mappings(workflow_id: str, node_id: str, *, mappings: Mapping[str, Any]) -> Dict[str, Any]

Replace a node's input mappings.

Args: workflow_id: The workflow ID. node_id: The node ID within the workflow. mappings: Input mapping object (free-form), e.g. {targetField: "data.sourceField"}.

update_node_passthrough_values​

update_node_passthrough_values(workflow_id: str, node_id: str, *, values: Mapping[str, Any]) -> Dict[str, Any]

Replace a node's passthrough values.

Args: workflow_id: The workflow ID. node_id: The node ID within the workflow. values: Passthrough mapping object (free-form), e.g. {outputKey: "inputKey"}.

update_status​

update_status(workflow_id: str, *, status: str) -> Dict[str, Any]

Update workflow status.

Args: workflow_id: The workflow ID. status: New status. Must be one of "draft", "active", "paused", "archived".

validate​

validate(workflow_id: str) -> Dict[str, Any]

Validate a workflow graph and return errors/warnings.

versions​

versions(workflow_id: str) -> WorkflowVersionInfo

List version history for a workflow.