发布和调用 WorkItem API 服务¶

使用 AI Studio SDK 管理一个指定 WorkItem 节点和版本的 API 服务。服务的发布、启用、调用和停用都使用同一节点 ID 与版本。

任务流程¶

先用节点 ID 和版本绑定服务，再提交发布配置并读取服务信息。启用后可发起调用；调用结果包含状态、结果或错误信息。停止接受调用时，停用服务并重新读取服务信息确认状态。

准备¶

需要的内容	如何取得	在本页中的作用
已认证的客户端和工作区 ID	在首次接入流程中创建或选择。	确定资源访问和发布范围。
WorkItem 节点 ID 和版本	从已确认的 WorkItem 资源结果取得。	选择要管理的服务版本。
发布配置	由服务负责人确定版本，必要时提供计算资源、超时、并发和限流设置。	发布或更新服务。
调用信息	由服务定义提供服务名、类型、版本和负载。	发起一次服务调用。

发布并启用服务¶

下面的示例发布服务并启用它。发布和启用都返回服务信息；在把服务用于业务调用前，读取返回信息或执行应用自己的可用性检查。

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 = client.workspace("<workspace-id>")
service = workspace.workitem_api_service("<node-id>", "<version>")
published = service.publish(
    sdk.WorkitemAPIServicePublishSpec(
        version="<version>",
        compute_resource_id="<compute-resource-id>",
        timeout_seconds=30,
        max_concurrency=1,
        rate_limit_per_min=60,
    )
)
enabled = service.enable()

print(published.service.status)
print(enabled.service.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)
	}
	service, err := workspace.WorkitemAPIService("<node-id>", "<version>")
	if err != nil {
		panic(err)
	}
	published, err := service.Publish(ctx, sdk.WorkitemAPIServicePublishSpec{
		Version:           "<version>",
		ComputeResourceID: "<compute-resource-id>",
		TimeoutSeconds:    30,
		MaxConcurrency:    1,
		RateLimitPerMin:   60,
	})
	if err != nil {
		panic(err)
	}
	enabled, err := service.Enable(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println(published.GetService().GetStatus())
	fmt.Println(enabled.GetService().GetStatus())
}

调用服务¶

继续使用同一服务引用，提供服务名、类型、版本和非空负载。调用响应中的状态、结果和错误信息用于判断本次业务调用的结果；请求返回不代表业务已经成功。

Python

response = service.invoke(
    sdk.WorkitemAPIServiceInvokeSpec(
        "<service-name>",
        "<service-type>",
        "<version>",
        {"<input-name>": "<input-value>"},
    )
)

print(response.result.status)
print(response.result.result)
print(response.result.error)

response, err := service.Invoke(ctx, sdk.WorkitemAPIServiceInvokeSpec{
	ServiceName: "<service-name>",
	Type:        "<service-type>",
	Version:     "<version>",
	Payload:     map[string]any{"<input-name>": "<input-value>"},
})
if err != nil {
	panic(err)
}

fmt.Println(response.GetResult().GetStatus())
fmt.Println(response.GetResult().GetResult())
fmt.Println(response.GetResult().GetError())

停用服务¶

不再接受调用前，停用服务并读取服务信息确认当前状态。服务引用没有独立删除操作；WorkItem 本身的生命周期应按对应资源流程处理。

Python

disabled = service.disable()
current = service.info()

print(disabled.service.status)
print(current.service.status)

disabled, err := service.Disable(ctx)
if err != nil {
	panic(err)
}
current, err := service.Info(ctx)
if err != nil {
	panic(err)
}

fmt.Println(disabled.GetService().GetStatus())
fmt.Println(current.GetService().GetStatus())

下一步¶

最后更新于 2026 年 9 月 1 日