查看工作流节点定义

列出可用节点，从目录中取得节点 ID，再查询节点定义。

准备

1. 配置 SDK 和个人访问令牌。

2. 选择工作区并获取工作区 ID。

列出节点并读取定义

示例输出目录中的节点 ID、名称和推荐版本，并读取首个节点的定义。首项仅用于演示查询；实际编写工作流时按任务选择节点。

Python

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_id = os.environ["WORKSPACE_ID"]
workspace = client.workspace(workspace_id)
catalog = workspace.workflows().workitem_catalog()
for node in catalog.items:
    print(node.node_id, node.display_name, node.preferred_version)
if not catalog.items:
    raise SystemExit("当前没有可用节点")
node_id = catalog.items[0].node_id
detail = workspace.workflows().workitem_catalog_item(node_id)
print("节点定义:", detail.item)

Go

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)
	}
	workspaceID := os.Getenv("WORKSPACE_ID")
	workspace, err := client.Workspace(workspaceID)
	if err != nil {
		panic(err)
	}
	_ = workspace
	catalog, err := workspace.Workflows().WorkitemCatalog(ctx)
	if err != nil {
		panic(err)
	}
	for _, node := range catalog.GetItems() {
		fmt.Println(node.GetNodeId(), node.GetDisplayName(), node.GetPreferredVersion())
	}
	if len(catalog.GetItems()) == 0 {
		fmt.Println("当前没有可用节点")
		return
	}
	detail, err := workspace.Workflows().WorkitemCatalogItem(ctx, catalog.GetItems()[0].GetNodeId())
	if err != nil {
		panic(err)
	}
	fmt.Println("节点定义:", detail.GetItem())
}

从目录选择所需节点，使用返回的节点 ID 读取配置定义。

最后更新于 2026 年 11 月 11 日