Publish and Invoke WorkItem API Services

Use the AI Studio SDK to manage API services for a designated WorkItem node and version. Publishing, enabling, invoking, and disabling the service all use the same node ID and version.

Task workflow

First bind the service using the node ID and version, submit the publish configuration, and read service details. Once enabled, calls can be dispatched; call responses provide status, results, or error messages. When terminating public access, disable the service and re-read service details to confirm the status transition.

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 publication scope.

WorkItem node ID and version

Obtained from verified WorkItem outputs.

Select service version to manage.

Publication configuration

Defined by service owner, including compute instance, timeout, concurrency, and rate limits.

Publish or update the service.

Invocation details

Service name, type, version, and payload specified by service schema.

Dispatch a service call.

Publish and enable service

The following example publishes and enables a service. Both actions return service metadata; before routing production traffic, verify returned details or run application health checks.

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())
}

Invoke service

Continue using the same service handle, providing service name, type, version, and non-empty payload. Use status, result, and error in the response to evaluate the invocation; HTTP response return does not imply business success.

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())

Disable service

Before discontinuing invocations, disable the service and re-read service details to verify current status. Service handles do not offer separate deletion operations; manage WorkItem lifecycles through their respective resource workflows.

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())

Next steps

Last updated on