Run Workflow Execution

async AsyncCogniteClient.workflows.executions.run(
workflow_external_id: str,
version: str,
input: dict | None = None,
metadata: dict | None = None,
client_credentials: ClientCredentials | None = None,
nonce: str | None = None,
) → WorkflowExecution

Run a workflow execution.

Parameters:

  • workflow_external_id (str) – External id of the workflow.

  • version (str) – Version of the workflow.

  • input (dict | None) – The input to the workflow execution. This will be available for tasks that have specified it as an input with the string “${workflow.input}” See tip below for more information.

  • metadata (dict | None) – Application specific metadata. Keys have a maximum length of 32 characters, values a maximum of 255, and there can be a maximum of 10 key-value pairs.

  • client_credentials (ClientCredentials | None) – Specific credentials that should be used to trigger the workflow execution. When passed will take precedence over the current credentials.

  • nonce (str | None) – The nonce to use to bind the session. If not provided, a new session will be created using the given ‘client_credentials’. If this is not given, the current credentials will be used.

Tip

The workflow input can be available in the workflow tasks. For example, if you have a Task with function parameters then you can specify it as follows

>>> from cognite.client.data_classes import WorkflowTask, FunctionTaskParameters
>>> task = WorkflowTask(
...     external_id="my_workflow-task1",
...     parameters=FunctionTaskParameters(
...         external_id="cdf_deployed_function:my_function",
...         data={"workflow_data": "${workflow.input}"},
...     ),
... )

Tip

You can create a session via the Sessions API, using the client.iam.session.create() method.

Returns:

The created workflow execution.

Return type:

WorkflowExecution

Examples

Trigger a workflow execution for the workflow “foo”, version 1:

>>> from cognite.client import CogniteClient, AsyncCogniteClient
>>> client = CogniteClient()
>>> # async_client = AsyncCogniteClient()  # another option
>>> res = client.workflows.executions.run("foo", "1")

Trigger a workflow execution with input data:

>>> res = client.workflows.executions.run("foo", "1", input={"a": 1, "b": 2})

Trigger a workflow execution using a specific set of client credentials (i.e. not your current credentials):

>>> import os
>>> from cognite.client.data_classes import ClientCredentials
>>> credentials = ClientCredentials("my-client-id", os.environ["MY_CLIENT_SECRET"])
>>> res = client.workflows.executions.run("foo", "1", client_credentials=credentials)