FranklinWH-Toolbox/service/agentmcp/tools/autoupdate.go

package tools

// 自动升级配置（只读，与 ToolBox“自动升级”页面同源）：按 aGate 硬件类型配置可自动升级的 EMS 版本、aPower 硬件与升级步骤（升级包）。

import (
	"context"
	"errors"
	"strings"

	"github.com/mark3labs/mcp-go/mcp"
	"go.mongodb.org/mongo-driver/mongo"

	"fwh-toolbox-server/agentmcp/kit"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/services"
)

// ---- auto_update_catalog ----

var AutoUpdateCatalog = kit.Def[kit.NoArgs]{
	Name:  "auto_update_catalog",
	Title: "查询自动升级目录",
	Description: "列出自动升级配置的目录：aGateHardware（已配置的 aGate 硬件类型，作为 auto_update_config_get 的参数）、" +
		"packages（升级包文件名，按 system 系统包、aGate、subDevice 子设备包、others 分类）、aPowerVersions（aPower 硬件版本号与名称）。",
	ReadOnly: true,
	Handler: func(ctx context.Context, _ kit.NoArgs) (*mcp.CallToolResult, error) {
		opt, err := services.GetAutoUpdateConfigOption()
		if err != nil {
			return kit.Upstream("ToolBox 自动升级配置", err), nil
		}
		ap, err := services.GetAutoUpdateAPowerVersion()
		if err != nil {
			return kit.Upstream("ToolBox 自动升级配置", err), nil
		}
		return kit.Object(map[string]any{"aGateHardware": opt.AGateVersions, "packages": opt.Packages, "aPowerVersions": ap}), nil
	},
}

// ---- auto_update_config_get ----

type autoUpdateConfigArgs struct {
	AGate string `json:"aGate" jsonschema:"aGate 硬件类型（auto_update_catalog 的 aGateHardware 中的值）"`
}

var AutoUpdateConfigGet = kit.Def[autoUpdateConfigArgs]{
	Name:  "auto_update_config_get",
	Title: "查询 aGate 自动升级配置",
	Description: "查询某个 aGate 硬件类型的自动升级配置：default（标准升级）与 special（备件换件、休眠等特殊场景）。" +
		"每项含 emsVersions（满足条件的当前 EMS 版本）、aPowerVersions（满足条件的 aPower 硬件版本）、steps（按顺序执行的升级包：file、title、invVersion）。" +
		"special 中 useDefaultUpdate=true 表示先执行特殊步骤再追加标准步骤，否则只执行特殊步骤。用于回答“某类站点会升级到什么版本”。",
	ReadOnly: true,
	Handler: func(ctx context.Context, a autoUpdateConfigArgs) (*mcp.CallToolResult, error) {
		hw, err := kit.Required("aGate", a.AGate)
		if err != nil {
			return kit.Fail(err), nil
		}
		cfg, err := services.GetAutoUpdateConfig(hw)
		if errors.Is(err, mongo.ErrNoDocuments) {
			return kit.Failf("没有 aGate 硬件类型 %s 的自动升级配置，可用 auto_update_catalog 查看已配置的类型", hw), nil
		}
		if err != nil {
			return kit.Upstream("ToolBox 自动升级配置", err), nil
		}
		packages := map[string]map[string]any{}
		steps := func(files []string) []map[string]any {
			out := make([]map[string]any, 0, len(files))
			for _, f := range files {
				p, ok := packages[f]
				if !ok {
					p = map[string]any{"file": f}
					if info, err := services.GetAutoUpdatePackageInfo(f); err == nil && info != nil {
						p["title"], p["invVersion"] = info.Title, info.InvVersion
					} else {
						p["note"] = "升级包信息缺失"
					}
					packages[f] = p
				}
				out = append(out, p)
			}
			return out
		}
		def := cfg.Config.DefaultType
		special := make([]map[string]any, 0, len(cfg.Config.SpecialType))
		for _, s := range cfg.Config.SpecialType {
			special = append(special, map[string]any{
				"title": s.Title, "emsVersions": versionList(s.EmsVersions), "aPowerVersions": s.APowerVersions,
				"steps": steps(s.SpecialUpdateOrder), "useDefaultUpdate": s.UseDefaultUpdate,
			})
		}
		return kit.Object(map[string]any{
			"aGate": cfg.Id,
			"default": map[string]any{
				"emsVersions": versionList(def.EmsVersions), "aPowerVersions": def.APowerVersions, "steps": steps(def.UpdateOrder),
			},
			"special": special,
		}), nil
	},
}

// versionList EMS 版本条件：版本号（带备注时附在后面）
func versionList(vs []domain.AutoUpdateSoftwareVersionInfo[string]) []string {
	out := make([]string, 0, len(vs))
	for _, v := range vs {
		s := v.Version
		if r := strings.TrimSpace(v.Remark); r != "" {
			s += "（" + r + "）"
		}
		out = append(out, s)
	}
	return out
}