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)

Methods​

Core​

list​

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

List workflows with pagination and filtering.

Parameters

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

create​

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

Create a new workflow.

Parameters

  • name (str): Workflow name (required).
  • description (str | None, optional): Workflow description.
  • status (str | None, optional): Initial status (default: "draft").
  • workflow_type (str | None, optional): Execution mode: "batch" (default) or "streaming".
  • tags (Sequence[str] | None, optional): Tags for the workflow.
  • nodes (Sequence[Mapping[str, Any]] | None, optional): Initial graph nodes (free-form node objects).
  • connections (Sequence[Mapping[str, Any]] | None, optional): Initial graph connections (free-form connection objects).
  • settings (Mapping[str, Any] | None, optional): 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: str | None = None, description: str | None = None, status: str | None = None, tags: Sequence[str] | None = None, nodes: Sequence[Mapping[str, Any]] | None = None, connections: Sequence[Mapping[str, Any]] | None = None, config: Mapping[str, Any] | None = None, settings: Mapping[str, Any] | None = None, max_concurrent_runs: int | None = None) -> Workflow

Update a workflow.

Parameters

  • workflow_id (str): The workflow ID.
  • name, description, status, tags: Flat metadata to update.
  • nodes, connections (Sequence[Mapping[str, Any]] | None, optional): Full graph node/connection lists (free-form).
  • config (Mapping[str, Any] | None, optional): Workflow config object (free-form).
  • settings (Mapping[str, Any] | None, optional): Workflow settings object (free-form). Replaces the stored settings object, so read-modify-write to preserve other keys.
  • max_concurrent_runs (int | None, optional): How many of the workflow's runs may be in flight at once, 1-10 (default 1). A Queue trigger keeps this many messages' runs going. Like every update, refused while the workflow is deployed.

delete​

delete(workflow_id: str) -> None

Delete a workflow.

Lifecycle & actions​

start​

start(workflow_id: str) -> Workflow

Start a deployed workflow (long-running types).

stop​

stop(workflow_id: str) -> Workflow

Stop a running workflow.

deploy​

deploy(workflow_id: str, **kwargs) -> Workflow

Deploy a workflow.

Parameters

  • workflow_id (str): The workflow ID.
  • **kwargs (Any): Optional deployment configuration.

undeploy​

undeploy(workflow_id: str) -> Workflow

Undeploy a workflow.

execute​

execute(workflow_id: str, *, definition: Mapping[str, Any] | None = None, trigger_inputs: Mapping[str, Any] | None = None) -> dict[str, Any]

Execute a workflow.

Parameters

  • workflow_id (str): The workflow ID.
  • definition (Mapping[str, Any] | None, optional): Optional graph definition override (nodes + connections); defaults to the stored workflow definition.
  • trigger_inputs (Mapping[str, Any] | None, optional): Input data for the trigger node.

Returns

  • dict[str, Any]: Dict containing executionId.

Other​

add_connection​

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

Add a connection between two nodes.

Parameters

  • workflow_id (str): The workflow ID.
  • source_node_id (str): Source node ID.
  • target_node_id (str): Target node ID.
  • source_port (str | None, optional): Source port, one of the source node's declared outputs (default "output"): "continue" for a loop body, "output" for a map body, "completed" from a loop or map into its Loop Accumulator, "if" / "else" from a conditional, "case_<n>" / "default" from a switch-case.
  • target_port (str | None, optional): Target port (default "input"; use "ai" or "tools" for agent nodes).
  • feedback (bool | None, optional): Mark this as a streaming feedback edge (closes a loop). Must be paired with max_iterations.
  • max_iterations (int | None, optional): Bound on feedback-edge iterations per session. Required when feedback is True.

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).

add_node​

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

Add a node to a workflow graph.

Parameters

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

build​

build(*, name: str, nodes: Sequence[Mapping[str, Any]], connections: Sequence[Mapping[str, Any]] | None = None, workflow_type: str | None = None, **extra) -> Workflow

Create a whole workflow graph in one call and return the workflow.

Give every node (id, type, label, config, input_mappings) and every connection (source, target and their ports); nodes and connections are sent as given. extra takes the workflow's other fields (description, tags, ...).

create_alert_rule​

create_alert_rule(*, name: str, condition: str, channels: Sequence[Mapping[str, Any]], workflow_id: str | None = None, condition_config: Mapping[str, Any] | None = None, throttle: Mapping[str, Any] | None = None, is_enabled: bool | None = None, description: str | None = None) -> dict[str, Any]

Create a workflow alert rule.

You are alerted on each channel when a run of the workflow (or of any of your workflows, with no workflow_id) ends meeting condition.

Parameters

  • name (str): The rule's name.
  • condition (str): execution_failed, execution_timeout, execution_completed, duration_exceeded, node_failed, consecutive_failures or dlq_threshold.
  • channels (Sequence[Mapping[str, Any]]): {"type": "in_app", "config": {}} (your bell), email (emailAddresses; none = your address), slack (slackWebhookUrl) or webhook (webhookUrl).
  • workflow_id (str | None, optional): The workflow; omit for all your workflows.
  • condition_config (Mapping[str, Any] | None, optional): durationThresholdMs, nodeTypes, failureCount or dlqCountThreshold, per condition.
  • throttle (Mapping[str, Any] | None, optional): {"enabled": bool, "intervalMinutes": int} (default off).
  • is_enabled (bool | None, optional): Default True.
  • description (str | None, optional): What the rule is for.
  • Returns ``{"id": <rule id>}``.

create_from_template​

create_from_template(template_id: str, *, name: str | None = None, description: str | None = None, tags: Sequence[str] | None = None) -> Workflow

Create a draft workflow from a template (from templates).

create_version​

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

Create a new workflow version.

Parameters

  • workflow_id (str): The workflow ID.
  • version_tag (str): The version tag (e.g. "v2").
  • description (str | None, optional): Optional description for this version.

delete_alert_rule​

delete_alert_rule(rule_id: str) -> None

Delete one of your workflow alert rules; its alerts stay in the history.

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.

dependency_access​

dependency_access(workflow_id: str, *, for_: str | None = None) -> dict[str, Any]

List the dependencies you may not use, with their owners.

for_ is run or deploy.

deploy_version​

deploy_version(workflow_id: str, version_id: str, **resources) -> Workflow

Deploy a saved version as the live deployment (roll back or forward).

Returns the workflow, its deployment_status queued: the deployment runs on after this answers; poll status. resources may set environment_id, cpu, memory, disk, gpu and gpu_type.

discover​

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

Discover workflows by capability / tag / search.

duplicate​

duplicate(workflow_id: str) -> Workflow

Duplicate a workflow and return the copy.

emit_event​

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

Emit a platform event that starts the deployed workflows listening to it.

It starts every deployed workflow you may use whose Event trigger listens for event_type, each run as you. Drafts never start. workflow.completed / workflow.failed are also emitted by the platform when a run ends; an Event trigger with a Source Workflow hears those only from that workflow. Returns {event_type, workflows_triggered, workflows_failed, results}, each result naming the started run's executionId or the error that kept that workflow from starting.

Parameters

  • event_type (str): The event type to emit (required).
  • event_data (Mapping[str, Any] | None, optional): Event payload (free-form).
  • source (str | None, optional): Optional event source label.

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: Mapping[str, Any] | None = None) -> dict[str, Any]

Import a workflow from a previously exported JSON bundle.

Parameters

  • export_data (Mapping[str, Any]): The exported workflow JSON data (free-form bundle).
  • resolved_deps (Mapping[str, Any] | None, optional): 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_alert_rules​

list_alert_rules(*, workflow_id: str | None = None, limit: int | None = None) -> SyncPaginator[WorkflowAlertRule]

List your workflow alert rules.

With workflow_id, the rules that apply to that workflow: its own and your rules for all your workflows.

list_alerts​

list_alerts(*, workflow_id: str | None = None, rule_id: str | None = None, limit: int | None = None) -> SyncPaginator[WorkflowAlert]

List the alerts your rules sent, newest first, a page at a time as you iterate.

Each alert has its run, condition, message, details and every channel's outcome (sent to whom, or why not).

list_nodes​

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

List nodes in a workflow graph.

logs​

logs(workflow_id: str, *, lines: int | None = None, container: str | None = None) -> dict[str, Any]

Get recent log lines of a deployed workflow's pod.

metrics​

metrics(workflow_id: str, *, window: str | None = None) -> dict[str, Any]

Get run counts by status, average duration, error rate and the last run.

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.

retrieve_version​

retrieve_version(workflow_id: str, version_id: str) -> dict[str, Any]

Get one saved version of a workflow, with its graph.

save_as_template​

save_as_template(workflow_id: str) -> Workflow

Save a copy of a workflow as a reusable template and return it.

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".

stats​

stats() -> WorkflowStats

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

status​

status(workflow_id: str) -> dict[str, Any]

Return a workflow's deployment status and how to invoke it.

Its deploymentStatus, pod and replica health, and, once it has a trigger, the invocation block: the URL, auth mode and signing header callers use.

structure_check​

structure_check(workflow_id: str) -> dict[str, Any]

Check the graph: broken connections, disconnected nodes, a missing trigger.

templates​

templates() -> list[Workflow]

List workflow templates.

update_alert_rule​

update_alert_rule(rule_id: str, *, name: str | None = None, condition: str | None = None, channels: Sequence[Mapping[str, Any]] | None = None, workflow_id: str | None = None, condition_config: Mapping[str, Any] | None = None, throttle: Mapping[str, Any] | None = None, is_enabled: bool | None = None, description: str | None = None) -> dict[str, Any]

Update one of your workflow alert rules.

Only the fields you pass change; is_enabled=False pauses it.

update_lifecycle​

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

Update the lifecycle policy for a workflow.

Parameters

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

update_node​

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

Update a node in a workflow graph.

Parameters

  • workflow_id (str): The workflow ID.
  • node_id (str): The node ID within the workflow.
  • config (Mapping[str, Any] | None, optional): Configuration key-value pairs to merge (free-form).
  • label (str | None, optional): 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.

Parameters

  • workflow_id (str): The workflow ID.
  • node_id (str): The node ID within the workflow.
  • mappings (Mapping[str, Any]): 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 (saved to node.config.passThroughValues).

Each entry copies one of the node's inputs into its output, so the next node can map it as data.<outputKey>.

Parameters

  • workflow_id (str): The workflow ID.
  • node_id (str): The node ID within the workflow.
  • values (Mapping[str, Any]): {outputKey: "inputName"}, where each inputName is one of the node's inputs (an inputMappings key). An input name the node does not map is skipped at run time.

update_status​

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

Update workflow status.

Parameters

  • workflow_id (str): The workflow ID.
  • status (str): 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.