Create Custom Operators and Workflow Templates

Use the AI Studio SDK to create custom operators or workflow templates. Creation results return the resource handles needed for reading, updating, toggling, testing, or deleting resources.

Task workflow

Creating an operator requires a complete operator definition; after successful creation, read its details or submit a test execution before updating, enabling, or deleting it. Creating a template requires a name and DSL YAML; creation results can be used to query, update, or remove the template.

Prerequisites

Required item

How to obtain

Role on this page

Authenticated client and workspace ID

Created or selected in initial setup.

Define resource access and creation scope.

Operator definition

Name, identifier, description, I/O schemas, runtime language, and code provided by operator developer.

Provision testable custom operators.

Test inputs

Provided by operator input schema and test cases.

Submit test runs for operators.

Template name and DSL YAML

Provided by workflow designs.

Create workflow templates.

Creating and deleting modifies workspace resources. Perform mutation operations only on operators and templates safe to manage.

Create and test custom operators

The following example creates an operator, reads its details, and submits a test run. The test run returns execution status and output logs; request return does not imply business assertions have passed.

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>")
operator, created = workspace.create_custom_operator(
    "<operator-name>",
    "<operator-identifier>",
    "<operator-description>",
    {"type": "object"},
    {"type": "object"},
    sdk.with_custom_operator_language("python"),
    sdk.with_custom_operator_code("<operator-code>"),
)
operator_info = operator.info()
test_run = operator.test({"<input-name>": "<input-value>"})

print(created)
print(operator_info)
print(test_run)

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)
	}
	operator, created, err := workspace.CreateCustomOperator(
		ctx,
		"<operator-name>",
		"<operator-identifier>",
		"<operator-description>",
		map[string]any{"type": "object"},
		map[string]any{"type": "object"},
		sdk.WithCustomOperatorLanguage("python"),
		sdk.WithCustomOperatorCode("<operator-code>"),
	)
	if err != nil {
		panic(err)
	}
	operatorInfo, err := operator.Info(ctx)
	if err != nil {
		panic(err)
	}
	testRun, err := operator.Test(ctx, map[string]any{"<input-name>": "<input-value>"})
	if err != nil {
		panic(err)
	}

	fmt.Println(created)
	fmt.Println(operatorInfo)
	fmt.Println(testRun)
}

Maintain custom operators

After reading an operator using creation outputs or verified operator IDs, you can update, enable, disable, or delete it. Before deletion, verify that no active workflows reference it; do not reuse original resource handles after deletion.

Create workflow templates

Continue using the workspace scope established on this page. After creating a template, use the creation output to make updates; template IDs come from creation outputs or verified resource lists, not display names.

template, created_template = workspace.create_workflow_template(
    "<template-name>",
    "<workflow-dsl-yaml>",
    sdk.with_workflow_template_key("<template-key>"),
)
updated = template.update("<updated-template-name>", "<updated-workflow-dsl-yaml>")

print(created_template)
print(updated)

template, createdTemplate, err := workspace.CreateWorkflowTemplate(
	ctx,
	"<template-name>",
	"<workflow-dsl-yaml>",
	sdk.WithWorkflowTemplateKey("<template-key>"),
)
if err != nil {
	panic(err)
}
updated, err := template.Update(ctx, "<updated-template-name>", "<updated-workflow-dsl-yaml>")
if err != nil {
	panic(err)
}

fmt.Println(createdTemplate)
fmt.Println(updated)

Next steps

Last updated on