管理连接器¶
以 Dify 知识库导出连接器为例，构造连接配置、校验外部服务，并在校验通过后创建连接器、回读确认资源 ID。默认运行只校验；创建连接器不会执行数据导出。
任务流程¶
从 Dify 的知识库服务 API 页面取得地址和密钥。
配置 Dify 来源类型和导出用途，校验连接。
确认校验通过后，显式运行创建命令，再用返回 ID 回读连接器。
准备¶
Dify 地址和知识库 API 密钥：在 Dify 的 Knowledge > Service API > API Key 创建密钥，并从同一面板取得 API 基础地址。具体操作见Dify 创建 API 密钥。需要由 Dify 工作区所有者或管理员操作，密钥须能列出全部知识库。
连接器名称：由调用方为本连接器命名，在目标工作区中使用尚未占用的名称。
本例的类型取值如下。配置在代码中完整构造，无需另行提供请求 JSON。
参数 |
本例取值 |
业务意义 |
|---|---|---|
|
|
Dify；表示连接的外部系统类型。 |
|
|
导出；作为数据导出的目标连接器。 |
|
Dify 面板显示的 API 基础地址 |
包含 API 前缀，例如 |
|
Dify 知识库 API 密钥 |
用于访问 Dify，不填 AI Studio 个人访问令牌。 |
设置以下环境变量。Dify 地址必须能从 AI Studio 服务所在网络访问；替换示例值，不输出密钥或完整配置。
export CONNECTOR_NAME="dify-export-demo"
export DIFY_API_URL="替换为 Dify 面板显示的 API 基础地址"
export DIFY_API_KEY="替换为 Dify 知识库 API 密钥"
校验连接配置¶
将 Python 示例保存为 connector_demo.py，运行 python connector_demo.py；或将 Go 示例保存为 main.go，在已配置 SDK 依赖的 Go module 中运行 go run .。两者默认只校验，不创建连接器。
import argparse
import os
import moi_product_sdk as sdk
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--create", action="store_true", help="校验通过后创建连接器")
args = parser.parse_args()
client = sdk.new_with_personal_access_token(
os.environ["PRODUCT_API_BASE_URL"], os.environ["PRODUCT_API_KEY"]
)
workspace = client.workspace(os.environ["WORKSPACE_ID"])
name = os.environ["CONNECTOR_NAME"]
config = {"dify": {
"api_url": os.environ["DIFY_API_URL"].rstrip("/"),
"api_key": os.environ["DIFY_API_KEY"],
}}
source_type, usage_type = 6, 2 # Dify，导出
result = workspace.connectors().validate(
name, source_type, usage_type, sdk.with_connector_config(config)
)
if not result.HasField("valid"):
raise RuntimeError("校验响应缺少 valid，停止创建")
if not result.valid:
raise RuntimeError("valid=false：检查 Dify 地址、密钥、权限和网络，停止创建")
print("valid=true：Dify 连接校验通过")
if not args.create:
print("仅校验，未创建连接器")
return
handle, created = workspace.create_connector(
name, source_type, usage_type, sdk.with_connector_config(config)
)
if not created.id or handle.id != created.id:
raise RuntimeError("创建响应缺少有效连接器 ID")
print(f"创建返回：id={created.id}，name={created.name}，status={created.status}")
info = handle.info()
if info.id != created.id or info.name != name or info.source_type != source_type:
raise RuntimeError("回读的连接器身份或类型不一致")
print(f"回读确认：id={info.id}，source_type={info.source_type}，"
f"usage_type={list(info.usage_type)}，status={info.status}")
if __name__ == "__main__":
main()
package main
import (
"context"
"flag"
"fmt"
"log"
"os"
"strings"
sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)
func required(name string) string {
value := os.Getenv(name)
if value == "" {
log.Fatalf("请设置环境变量 %s", name)
}
return value
}
func run(create bool) error {
client, err := sdk.NewWithPersonalAccessToken(required("PRODUCT_API_BASE_URL"), required("PRODUCT_API_KEY"))
if err != nil {
return err
}
workspace, err := client.Workspace(required("WORKSPACE_ID"))
if err != nil {
return err
}
ctx := context.Background()
name := required("CONNECTOR_NAME")
config := map[string]any{"dify": map[string]any{
"api_url": strings.TrimRight(required("DIFY_API_URL"), "/"),
"api_key": required("DIFY_API_KEY"),
}}
const sourceType, usageType = 6, 2 // Dify，导出
result, err := workspace.Connectors().Validate(ctx, name, sourceType, usageType, sdk.WithConnectorConfig(config))
if err != nil {
return err
}
if result.Valid == nil {
return fmt.Errorf("校验响应缺少 valid，停止创建")
}
if !result.GetValid() {
return fmt.Errorf("valid=false：检查 Dify 地址、密钥、权限和网络，停止创建")
}
fmt.Println("valid=true：Dify 连接校验通过")
if !create {
fmt.Println("仅校验，未创建连接器")
return nil
}
handle, created, err := workspace.CreateConnector(ctx, name, sourceType, usageType, sdk.WithConnectorConfig(config))
if err != nil {
return err
}
if created.GetId() == "" || handle.ID() != created.GetId() {
return fmt.Errorf("创建响应缺少有效连接器 ID")
}
fmt.Printf("创建返回：id=%s，name=%s，status=%s\n", created.GetId(), created.GetName(), created.GetStatus())
info, err := handle.Info(ctx)
if err != nil {
return err
}
if info.GetId() != created.GetId() || info.GetName() != name || info.GetSourceType() != sourceType {
return fmt.Errorf("回读的连接器身份或类型不一致")
}
fmt.Printf("回读确认：id=%s，source_type=%d，usage_type=%v，status=%s\n", info.GetId(), info.GetSourceType(), info.GetUsageType(), info.GetStatus())
return nil
}
func main() {
create := flag.Bool("create", false, "校验通过后创建连接器")
flag.Parse()
if err := run(*create); err != nil {
log.Fatal(err)
}
}
校验结果必须包含 valid 且值为 true。对本例，这表示服务端使用所填密钥请求 Dify 知识库列表时收到 HTTP 200；不表示数据已导出。valid=false 和字段缺失均会停止程序，不能继续创建。
创建连接器¶
确认配置后，Python 运行 python connector_demo.py --create，Go 运行 go run . --create。该命令会重新校验同一配置，校验通过才在目标工作区新增连接器。只选择一种语言执行创建，重复创建同名连接器会失败。
程序输出创建响应的 id、name 和 status，随后用返回 ID 读取详情，检查 ID、名称及来源类型，并显示使用类型和状态。保存真实返回的 ID，用于后续选择连接器；不要自行填一个预期 ID。
确认连接器已保存¶
校验通过后，显式执行创建命令。创建成功会输出连接器 ID，程序再回读 ID、名称和来源类型确认保存结果。
回读失败时，先按已输出的 ID 核对资源，避免重复创建。密钥失效、权限不足或网络不可达时，修正配置后重新校验。数据导出需另行创建任务。