Run and Track Workflows

Use the AI Studio SDK to execute deployed workflows and inspect results or control execution based on current run status. Save the run ID upon launch; subsequent operations identify the same execution using this ID.

Task workflow

An execution sequentially passes through input preparation, launch submission, obtaining the run ID, querying current status, and reading results. The run ID serves as the correlation token between launching, querying, controlling, and retrieving outputs.

Workspace + Deployed workflow + Run inputs
              │
              ▼
          Start run
              │
              ▼
       Run ID + Current status
         ├── Inspect run status and available actions
         ├── Read run results
         └── Suspend, resume, retry, or cancel

Prerequisites

Required item

How to obtain

Role on this page

Authenticated client

Created during initial setup.

Access AI Studio resources.

Workspace ID

Obtained from workspace creation, listing, or operator selection.

Define the resource scope of the workflow.

Deployed workflow ID

Obtained from workflow deployment results or operator selection.

Identify the workflow to launch.

Workflow inputs

Derived from the workflow’s input definitions and runtime data.

Provide parameter values for this execution.

This page does not deploy workflows or validate input schemas. If these are missing, obtain them via corresponding workflow management workflows first.

Start an execution

Upon submitting a run, the SDK returns the execution ID and current status. Preserve the run ID; request success does not imply that the workflow has reached a terminal state.

import os

import moi_product_sdk as sdk

client = sdk.new_with_personal_access_token(
    os.environ["PRODUCT_API_BASE_URL"],
    os.environ["PRODUCT_API_KEY"],
)
workspace = client.workspace("<workspace-id>")
workflow = workspace.workflow("<workflow-id>")
started = workflow.run(
    sdk.with_workflow_run_input(
        {"<input-field-id>": "<input-value>"}
    )
)

execution_id = started.workflow_run.execution_id
print(execution_id)
print(started.workflow_run.status)

package main

import (
	"context"
	"fmt"
	"os"

	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)

func main() {
	ctx := context.Background()
	client, err := sdk.NewWithPersonalAccessToken(
		os.Getenv("PRODUCT_API_BASE_URL"),
		os.Getenv("PRODUCT_API_KEY"),
	)
	if err != nil {
		panic(err)
	}
	workspace, err := client.Workspace("<workspace-id>")
	if err != nil {
		panic(err)
	}
	workflow, err := workspace.Workflow("<workflow-id>")
	if err != nil {
		panic(err)
	}
	started, err := workflow.Run(
		ctx,
		sdk.WithWorkflowRunInput(map[string]any{
			"<input-field-id>": "<input-value>",
		}),
	)
	if err != nil {
		panic(err)
	}

	executionID := started.GetWorkflowRun().GetExecutionId()
	fmt.Println(executionID)
	fmt.Println(started.GetWorkflowRun().GetStatus())
}

Inspect the execution

Use the execution ID to query the current status, available actions, and error messages. The SDK does not provide generic polling or retry loops; application logic should determine polling intervals and branching based on returned states.

run = workflow.run_handle(execution_id)
current = run.refresh()

print(current.execution.status)
print(current.execution.available_actions)
print(current.execution.error)

run, err := workflow.RunHandle(executionID)
if err != nil {
	panic(err)
}
current, err := run.Refresh(ctx)
if err != nil {
	panic(err)
}

fmt.Println(current.GetExecution().GetStatus())
fmt.Println(current.GetExecution().GetAvailableActions())
fmt.Println(current.GetExecution().GetError())

Read execution results

When output records are needed, use the same run handle to read results. The payload contains overall run status, case results, case errors, node statuses, and available actions. The overall workflow status cannot replace individual node statuses.

result = run.result()

print(result.result.status)
print(result.result.case_result)
print(result.result.case_error)

result, err := run.Result(ctx)
if err != nil {
	panic(err)
}

fmt.Println(result.GetResult().GetStatus())
fmt.Println(result.GetResult().GetCaseResult())
fmt.Println(result.GetResult().GetCaseError())

Control execution

When status and available actions permit, you can suspend, resume, retry, or cancel the run. Retrying returns new run details; retain the new execution ID. Other control actions return updated state for the current run, which still requires verification.

Goal

Operation result

Suspend or resume

Returns current run metadata.

Retry

Returns new execution metadata.

Cancel

Returns current run metadata.

Next steps

Last updated on