deploy/agent-assistant/skills/fwh-assistant/references/write-operations.md

@@ -12,15 +12,13 @@
 
 ## 站点信息同步
 
-同步在主服务后台执行，耗时可能超过 1.5 小时；`site_info_refresh` 立即返回，不等同步完成。
-
-1. **先查状态**：调用只读工具 `mcp__gateway__fwh__site_info_sync_status`（无需审批），按 `state` 处理：
-   - `syncing`：同步进行中，**不要触发**，告诉用户已进行多久，之后再用本工具查看；
-   - `stale`：超过 6 小时未完成，疑似中断（如主服务重启）。向用户说明情况，用户确认后才用 `force=true` 重新触发；
-   - `idle` 且最近一次成功同步在 24 小时内（`advice` 会说明）：数据较新，告诉用户数据截至 `lastSuccess.completedAt`，一般无需同步；用户坚持要最新数据时用 `force=true`；
-   - `idle` 且超过 24 小时、上次失败，或 `never`：可以触发。
-2. **再触发**：调用 `site_info_refresh`。工具会再次检查状态，按返回的 `action` 回复：`triggered`（已触发）、`tracking`（同步中，未重复触发）、`skipped_fresh` / `skipped_stale`（未触发，原因见 `message`）。
-3. **跟踪**：触发后告诉用户同步需要较长时间、完成后 `site_search` 才会更新；用户之后询问进度时用 `site_info_sync_status` 查看，不要在一轮对话中反复轮询。
+同步在主服务后台执行，约 1 小时，可能超过 1.5 小时。`site_info_refresh` 以长任务方式执行：工具会先检查同步状态再决定是否触发，不会重复触发。
+
+1. **用户要求同步**：直接调用 `site_info_refresh`（不带 `force`）。
+   - 触发了新同步，或已有同步在进行（改为跟踪它）：工具立即返回“已转为后台执行”。告诉用户已提交、需要较长时间、完成后 `site_search` 才会更新，然后结束这一轮。**同步结束后平台会把结果作为新消息发给你**（完成时间、站点数、耗时，或失败原因），收到后向用户汇报。不要轮询。
+   - 未触发：最近 24 小时内已成功同步（`skipped_fresh`），或上次同步超过 6 小时未完成、疑似中断（`skipped_stale`）。把原因和数据截至的时间告诉用户；用户确认仍要同步时，再调用 `site_info_refresh` 并设置 `force=true`。
+2. **用户只问数据是否最新、同步进度如何**：用只读工具 `mcp__gateway__fwh__site_info_sync_status`（无需审批），按 `advice` 回答，不触发同步。
+3. **结果为失败**：说明失败原因，询问用户是否稍后重新同步；不要自动重试。
 
 通用规则：
 

docs/ai-assistant-agentserver/agent-mcp.md

@@ -67,7 +67,7 @@ fwh-agent-mcp -set=ops      -listen=127.0.0.1:8092   # 写服务（二期）
 
 ### 3.2 工具定义
 
-使用 mcp-go v0.57（`go.mod` 当前版本）的结构体生成参数 schema（`mcp.WithInputSchema[T]`）和类型化处理函数（`mcp.NewTypedToolHandler`），参数校验交给 `server.WithInputSchemaValidation()`：
+使用 mcp-go v1.4（`go.mod` 当前版本）的结构体生成参数 schema（`mcp.WithInputSchema[T]`）和类型化处理函数（`mcp.NewTypedToolHandler`），参数校验交给 `server.WithInputSchemaValidation()`：
 
 ```go
 // kit/tool.go
@@ -86,7 +86,7 @@ func (d Def[T]) Register(s *server.MCPServer) {
             mcp.WithInputSchema[T](),
             mcp.WithTitleAnnotation(d.Title),
             mcp.WithReadOnlyHintAnnotation(d.ReadOnly),
-            mcp.WithDestructiveHintAnnotation(false),
+            mcp.WithDestructiveHintAnnotation(d.Destructive),   // 覆盖、删除类写操作为 true：自动模式下也每次审批
             mcp.WithOpenWorldHintAnnotation(false),
         ),
         mcp.NewTypedToolHandler(func(ctx context.Context, _ mcp.CallToolRequest, args T) (*mcp.CallToolResult, error) {
@@ -142,7 +142,7 @@ schema 由 mcp-go 依赖的 `google/jsonschema-go` 生成：`jsonschema` 标签
 | `List(rows, limit)` | 列表 | `{"total":N,"returned":n,"truncated":bool,"items":[...]}` |
 | `Table(cols, rows, limit)` | 行多、字段固定的数据 | 首行列名的 TSV 文本，比 JSON 省一半以上字符 |
 | `Export(name, csv)` | 导出工具（第 5 节） | `EmbeddedResource`（`text/csv`） |
-| `Fail(err)` / `Upstream(name, err)` | 错误 | `isError` + 一句中文说明 + 处理建议 |
+| `Fail(err)` / `Upstream(name, err)` | 错误 | `isError` + 一句中文说明 + 处理建议。`Upstream` 只返回数据源、错误类别与错误编号，原始错误（可能含内部地址、SQL）只写日志 |
 
 统一限制：单次结果（非导出）不超过 30,000 字符；超过时按行截断并标明 `truncated`，附一句“请缩小范围或使用 xxx_export”。这样不会触发 AgentServer 的 10 万字截断（那条路径会把前 10 万字仍放进上下文）。
 
@@ -226,6 +226,34 @@ ADR-0021 的结果处理（`mcp-gateway/server.ts` 的 `postProcess`）目前：
 
 故障录波原始文件、日志包等二进制内容在一期不直接交给 Agent。需要时提供“解析后的结果”工具（如录波通道数据导出为 CSV），由 Agent MCP 在服务端解析；是否需要见 Q-03。上述 5.2 的改动落地后，也可以用 `blob` 资源交付原始文件。
 
+### 5.4 长任务（MCP Tasks）
+
+超过 3 分钟的操作以 MCP 长任务执行（AgentServer 接入指南 §5、`docs/design/16-mcp-support.md` §3）：工具立即返回任务，AgentServer 在后台按 `pollInterval` 查询，结束后把结果作为新消息发到会话、唤醒模型。不做“提交 + 查询”让模型轮询；查询进度的只读工具保留，供用户主动询问。
+
+**任务状态不放在 MCP 进程里。** mcp-go 的任务只存在进程内存，fwh-agent-mcp 重启后平台查询得到“任务不存在”而按失败通知用户，而底层作业其实仍在执行。所以长任务工具用 `kit.TaskDef` 定义，由 `kit.TaskRouter` 在 mcp-go 之前处理任务协议：
+
+| 请求 | 处理 |
+| --- | --- |
+| `tools/call` 带 `task` | `Start`：启动或挂到已有作业上，立即返回任务（含 `pollInterval` 与给模型的即时说明 `model-immediate-response`） |
+| `tasks/get` | `Status`：读取业务数据（如 Mongo 中的作业记录）翻译为任务状态 |
+| `tasks/result` | 任务结束后返回工具结果（失败为 `isError`），带 `related-task` |
+| `tasks/cancel` | `Cancel`：由工具按任务 id 中的标记决定是否真的停止作业 |
+| 其他请求、不是本服务签发的任务 id | 原样交给 mcp-go |
+
+- 任务 id：`fwhtask.<工具>.<业务键>.<标记>.<随机数>.<签名>`，签名为以访问令牌为密钥的 HMAC，伪造或篡改的任务 id 一律返回“找不到任务”。
+- MCP 进程不保存任何任务状态，重启后照常回答查询；业务数据长期保存，满足平台至少 24 小时的要求。
+- 不带 `task` 的普通调用照常由 `Def.Handler` 同步返回。
+
+`site_info_refresh` 是第一个长任务工具：业务键为同步记录（`SiteInfoManageRecord`）的 id，标记为本次调用的动作。
+
+| 动作（标记） | 任务 |
+| --- | --- |
+| `triggered`（本次触发） | 跟随同步记录：pending / running → working（进度说明“同步中，已进行 N 分钟”）；complete → completed（站点数、耗时、完成时间）；error → failed；cancelled → cancelled；超过 6 小时未完成 → failed（疑似中断）。取消时停止同步 |
+| `tracking`（已有同步在进行） | 同上，但取消只停止跟踪，不停止同步（其他会话可能也在等它） |
+| `skipped_fresh` / `skipped_stale`（未触发） | 立即完成，结果为不触发的原因 |
+
+取消同步由主服务配合：同步循环每页检查 ctx；`fwhQueue.CancelSiteInfoUpdate` 对执行中的任务发出取消（worker 停止后记为 cancelled），对排队中的直接删除。同步任务超时为 4 小时（`SiteInfoUpdateTimeout`），超时同样会停止同步并记为失败。
+
 ## 6. 部署与接入
 
 ### 6.1 Agent MCP
@@ -259,7 +287,7 @@ Profile `fwh-assistant` 勾选 `fwh`。工具使用的业务说明（字段含
 | --- | --- | --- |
 | `feedback_create` | 在 Plane 创建 Intake Issue | 提交人固定为服务账号“AI 助手”；描述中由 Agent 写明背景。需要真实提交人时由 ToolBox 页面提交，不经过 MCP |
 | `tou_template_save` | 保存测试服 TOU 模版 | 只允许测试服；写入前按调度配置需求中的规则校验，返回差异 |
-| `site_info_refresh` | 触发站点信息表同步，立即返回 | 触发前检查同步状态（与 `site_info_sync_status` 相同的判断）：同步中不重复触发；最近 24 小时内已成功、或超过 6 小时未完成疑似中断时不触发，`force=true` 可强制。任务投递到主服务的队列执行（MCP 只创建入队客户端，Redis 库号与主服务一致） |
+| `site_info_refresh` | 触发站点信息表同步，以长任务执行（5.4） | 触发前检查同步状态（与 `site_info_sync_status` 相同的判断）：同步中改为跟踪，不重复触发；最近 24 小时内已成功、或超过 6 小时未完成疑似中断时不触发，`force=true` 可强制。同步结束后平台把结果发给模型。任务投递到主服务的队列执行（MCP 只创建入队客户端与 Inspector，Redis 库号与主服务一致） |
 
 无用户模型下，写工具的“谁做的”只记录在 ToolBox 会话归属和 AgentServer 审计中，业务系统里看到的是服务账号。如果某个写操作在业务上必须归属到具体的人，就不放进 MCP。
 

service/agentmcp/http.go

@@ -10,16 +10,19 @@ import (
 
 	"github.com/mark3labs/mcp-go/server"
 	"go.uber.org/zap"
+
+	"fwh-toolbox-server/agentmcp/kit"
 )
 
 // Handler 返回 HTTP 处理器：/mcp（Streamable HTTP，无状态，需 Bearer 令牌）与 /healthz。
-func Handler(s *server.MCPServer, token string) http.Handler {
+// tasks 为持久化长任务工具：任务协议请求由 kit.TaskRouter 在 mcp-go 之前处理（任务 id 以访问令牌签名）。
+func Handler(s *server.MCPServer, token string, tasks []kit.TaskTool, log *zap.Logger) http.Handler {
 	mcpHandler := server.NewStreamableHTTPServer(s,
 		server.WithStateLess(true),
 		server.WithEndpointPath("/mcp"),
 	)
 	mux := http.NewServeMux()
-	mux.Handle("/mcp", bearerAuth(token, mcpHandler))
+	mux.Handle("/mcp", bearerAuth(token, kit.NewTaskRouter(mcpHandler, tasks, token, log)))
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusOK)
 		_, _ = w.Write([]byte("ok"))

service/agentmcp/kit/errors_test.go

@@ -0,0 +1,31 @@
+package kit
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+func TestUpstreamHidesRawError(t *testing.T) {
+	cases := []struct {
+		err  error
+		kind string
+	}{
+		{errors.New(`pq: relation "site_info" does not exist (host=10.0.0.5:5432)`), "数据源返回错误"},
+		{fmt.Errorf("query: %w", context.DeadlineExceeded), "请求超时"},
+		{fmt.Errorf("find: %w", mongo.ErrNoDocuments), "没有找到对应记录"},
+	}
+	for _, c := range cases {
+		text := UpstreamText("测试数据源", c.err)
+		if strings.Contains(text, "10.0.0.5") || strings.Contains(text, "relation") || strings.Contains(text, "mongo") {
+			t.Errorf("错误说明泄露了原始错误：%s", text)
+		}
+		if !strings.Contains(text, c.kind) || !strings.Contains(text, "错误编号") {
+			t.Errorf("错误说明应包含类别 %s 与错误编号：%s", c.kind, text)
+		}
+	}
+}

service/agentmcp/kit/result.go

@@ -2,15 +2,21 @@ package kit
 
 import (
 	"bytes"
+	"context"
+	"crypto/rand"
 	"encoding/csv"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"reflect"
 	"strings"
 	"unicode/utf8"
 
 	"github.com/mark3labs/mcp-go/mcp"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.uber.org/zap"
 
 	"fwh-toolbox-server/utils/pii"
 )
@@ -40,7 +46,38 @@ func Failf(format string, a ...any) *mcp.CallToolResult {
 
 // Upstream 上游（FWH 平台、数据库、Plane 等）调用失败。
 func Upstream(source string, err error) *mcp.CallToolResult {
-	return mcp.NewToolResultError(fmt.Sprintf("调用%s失败：%v。可以稍后重试，或缩小查询范围。", source, err))
+	return mcp.NewToolResultError(UpstreamText(source, err) + "。可以稍后重试，或缩小查询范围。")
+}
+
+// UpstreamText 写给模型的上游错误说明：只给数据源、错误类别与错误编号。
+// 原始错误可能含内部地址、SQL、库表名，只写入日志（按错误编号检索），不返回给模型。
+func UpstreamText(source string, err error) string {
+	ref := errorRef()
+	zap.L().Warn("[agent-mcp] 上游调用失败", zap.String("source", source), zap.String("ref", ref), zap.Error(err))
+	return fmt.Sprintf("调用%s失败（%s，错误编号 %s）", source, errorKind(err), ref)
+}
+
+func errorKind(err error) string {
+	var ne net.Error
+	switch {
+	case errors.Is(err, context.DeadlineExceeded):
+		return "请求超时"
+	case errors.Is(err, context.Canceled):
+		return "请求已取消"
+	case errors.Is(err, mongo.ErrNoDocuments):
+		return "没有找到对应记录"
+	case errors.As(err, &ne) && ne.Timeout():
+		return "请求超时"
+	case errors.As(err, &ne):
+		return "网络连接失败"
+	}
+	return "数据源返回错误"
+}
+
+func errorRef() string {
+	b := make([]byte, 4)
+	_, _ = rand.Read(b)
+	return hex.EncodeToString(b)
 }
 
 // Empty 无数据时的统一返回（不是错误）。

service/agentmcp/kit/task.go

@@ -0,0 +1,372 @@
+package kit
+
+// 持久化长任务（MCP Tasks，规范 2025-11-25；AgentServer 接入指南 §5）。
+//
+// mcp-go 的任务只保存在进程内存里，fwh-agent-mcp 重启后平台查询会得到“任务不存在”而按失败通知用户，
+// 而底层作业（如主服务的站点信息同步）其实还在执行。所以长任务工具不使用 mcp-go 的任务表：
+//   - 任务状态由业务数据给出（如 Mongo 中的同步记录），MCP 进程不保存任何任务状态，重启后照常回答查询；
+//   - 任务 id 携带工具名、业务键与标记，并带 HMAC 签名，不可伪造、不可猜测；
+//   - TaskRouter 在 mcp-go 之前处理“以任务方式调用工具”与 tasks/get、tasks/result、tasks/cancel，其余请求原样交给 mcp-go。
+// 普通调用（不带 task 参数）照常由 mcp-go 按 Def.Handler 同步返回。
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/mark3labs/mcp-go/mcp"
+	"github.com/mark3labs/mcp-go/server"
+	"go.uber.org/zap"
+)
+
+const (
+	taskIDPrefix  = "fwhtask"
+	immediateMeta = "io.modelcontextprotocol/model-immediate-response"
+	relatedMeta   = "io.modelcontextprotocol/related-task"
+	// defaultTaskTTL 调用方未指定 ttl 时声明的保留时间（毫秒）。业务数据本身长期保存，这里只是告知调用方
+	defaultTaskTTL = int64(24 * time.Hour / time.Millisecond)
+	// resultWaitStep tasks/result 在任务未结束时的查询间隔（规范要求阻塞到任务结束）
+	resultWaitStep = 5 * time.Second
+	maxRPCBody     = 1 << 20
+)
+
+// ErrTaskNotFound 业务键对应的数据不存在（返回给调用方“找不到任务”）
+var ErrTaskNotFound = errors.New("task not found")
+
+// TaskRef 任务 id 中携带的信息
+type TaskRef struct {
+	Tool string // 工具名
+	Key  string // 业务键，如同步记录 id（只含字母数字）
+	Flag string // 工具自定义的标记，如本次调用是否触发了作业（只含小写字母、数字、下划线）
+}
+
+// TaskStart 以任务方式调用时 Start 的返回
+type TaskStart struct {
+	Key  string
+	Flag string
+	Note string // 立即交给模型的说明（model-immediate-response），如“已开始同步，预计约 1 小时”
+}
+
+// TaskState 由业务数据计算出的任务状态
+type TaskState struct {
+	Status    mcp.TaskStatus
+	Message   string              // statusMessage，显示在平台的后台调用栏
+	Result    *mcp.CallToolResult // 终态时的结果；failed 时应为 isError
+	CreatedAt time.Time
+}
+
+// TaskTool 持久化长任务工具
+type TaskTool interface {
+	Registrar
+	PollInterval() time.Duration
+	// StartTask 立即返回（几秒内）：启动或挂到已有作业上。参数或业务错误时返回 reject（作为错误返回给调用方）
+	StartTask(ctx context.Context, args json.RawMessage) (start TaskStart, reject *mcp.CallToolResult)
+	// TaskStatus 按业务数据计算任务状态；数据不存在时返回 ErrTaskNotFound
+	TaskStatus(ctx context.Context, ref TaskRef) (TaskState, error)
+	// CancelTask 取消任务；由工具按 ref.Flag 决定是否真的停止作业（只是跟踪他人触发的作业时不停止）
+	CancelTask(ctx context.Context, ref TaskRef) error
+}
+
+// TaskDef 类型化的持久化长任务工具：Def 用于普通调用，其余回调用于任务方式调用
+type TaskDef[T any] struct {
+	Def[T]
+	Poll   time.Duration
+	Start  func(ctx context.Context, args T) (TaskStart, *mcp.CallToolResult)
+	Status func(ctx context.Context, ref TaskRef) (TaskState, error)
+	Cancel func(ctx context.Context, ref TaskRef) error
+}
+
+func (d TaskDef[T]) Tool() mcp.Tool {
+	t := d.Def.Tool()
+	mcp.WithTaskSupport(mcp.TaskSupportOptional)(&t)
+	return t
+}
+
+func (d TaskDef[T]) Register(s *server.MCPServer) {
+	s.AddTool(d.Tool(), mcp.NewTypedToolHandler(func(ctx context.Context, _ mcp.CallToolRequest, args T) (*mcp.CallToolResult, error) {
+		return d.Handler(ctx, args)
+	}))
+}
+
+func (d TaskDef[T]) PollInterval() time.Duration { return d.Poll }
+
+func (d TaskDef[T]) StartTask(ctx context.Context, raw json.RawMessage) (TaskStart, *mcp.CallToolResult) {
+	var args T
+	if len(raw) > 0 && string(raw) != "null" {
+		if err := json.Unmarshal(raw, &args); err != nil {
+			return TaskStart{}, Failf("参数格式不正确：%v", err)
+		}
+	}
+	return d.Start(ctx, args)
+}
+
+func (d TaskDef[T]) TaskStatus(ctx context.Context, ref TaskRef) (TaskState, error) {
+	return d.Status(ctx, ref)
+}
+
+func (d TaskDef[T]) CancelTask(ctx context.Context, ref TaskRef) error { return d.Cancel(ctx, ref) }
+
+// ---- 任务 id ----
+
+type taskIDCodec struct{ secret []byte }
+
+func (c taskIDCodec) encode(ref TaskRef) string {
+	body := strings.Join([]string{taskIDPrefix, ref.Tool, ref.Key, ref.Flag, errorRef()}, ".")
+	return body + "." + c.sign(body)
+}
+
+// decode 返回 ok=false 表示不是本服务签发的任务 id（交给 mcp-go）；err 表示前缀匹配但校验失败
+func (c taskIDCodec) decode(id string) (ref TaskRef, ok bool, err error) {
+	parts := strings.Split(id, ".")
+	if len(parts) == 0 || parts[0] != taskIDPrefix {
+		return TaskRef{}, false, nil
+	}
+	if len(parts) != 6 || !hmac.Equal([]byte(c.sign(strings.Join(parts[:5], "."))), []byte(parts[5])) {
+		return TaskRef{}, true, ErrTaskNotFound
+	}
+	return TaskRef{Tool: parts[1], Key: parts[2], Flag: parts[3]}, true, nil
+}
+
+func (c taskIDCodec) sign(body string) string {
+	m := hmac.New(sha256.New, c.secret)
+	m.Write([]byte(body))
+	return hex.EncodeToString(m.Sum(nil))[:32]
+}
+
+// ---- 协议处理 ----
+
+// TaskRouter 在 mcp-go 之前处理持久化长任务的协议请求
+type TaskRouter struct {
+	next  http.Handler
+	tools map[string]TaskTool
+	ids   taskIDCodec
+	log   *zap.Logger
+}
+
+// NewTaskRouter secret 用于任务 id 签名（使用 MCP 访问令牌即可，只有持有令牌的调用方能访问本服务）
+func NewTaskRouter(next http.Handler, tools []TaskTool, secret string, log *zap.Logger) *TaskRouter {
+	m := make(map[string]TaskTool, len(tools))
+	for _, t := range tools {
+		m[t.Meta().Name] = t
+	}
+	return &TaskRouter{next: next, tools: m, ids: taskIDCodec{secret: []byte(secret)}, log: log}
+}
+
+type rpcRequest struct {
+	JSONRPC string          `json:"jsonrpc"`
+	ID      json.RawMessage `json:"id"`
+	Method  string          `json:"method"`
+	Params  json.RawMessage `json:"params"`
+}
+
+func (r *TaskRouter) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	if req.Method != http.MethodPost || len(r.tools) == 0 {
+		r.next.ServeHTTP(w, req)
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(req.Body, maxRPCBody))
+	if err != nil {
+		http.Error(w, "read body failed", http.StatusBadRequest)
+		return
+	}
+	req.Body = io.NopCloser(bytes.NewReader(body))
+
+	var rpc rpcRequest
+	if json.Unmarshal(body, &rpc) != nil || len(rpc.ID) == 0 || !r.handle(w, req.Context(), rpc) {
+		r.next.ServeHTTP(w, req)
+	}
+}
+
+// handle 处理本服务负责的请求，返回 false 表示交给 mcp-go
+func (r *TaskRouter) handle(w http.ResponseWriter, ctx context.Context, rpc rpcRequest) bool {
+	switch rpc.Method {
+	case "tools/call":
+		var p struct {
+			Name      string          `json:"name"`
+			Arguments json.RawMessage `json:"arguments"`
+			Task      *mcp.TaskParams `json:"task"`
+		}
+		if json.Unmarshal(rpc.Params, &p) != nil || p.Task == nil {
+			return false
+		}
+		tool, ok := r.tools[p.Name]
+		if !ok {
+			return false
+		}
+		r.create(w, ctx, rpc.ID, tool, p.Arguments, p.Task.TTL)
+		return true
+	case "tasks/get", "tasks/result", "tasks/cancel":
+		var p struct {
+			TaskID string `json:"taskId"`
+		}
+		if json.Unmarshal(rpc.Params, &p) != nil {
+			return false
+		}
+		ref, ours, err := r.ids.decode(p.TaskID)
+		if !ours {
+			return false
+		}
+		tool, known := r.tools[ref.Tool]
+		if err != nil || !known {
+			writeRPCError(w, rpc.ID, mcp.INVALID_PARAMS, "task not found")
+			return true
+		}
+		switch rpc.Method {
+		case "tasks/get":
+			r.get(w, ctx, rpc.ID, tool, ref, p.TaskID)
+		case "tasks/result":
+			r.result(w, ctx, rpc.ID, tool, ref, p.TaskID)
+		default:
+			r.cancel(w, ctx, rpc.ID, tool, ref, p.TaskID)
+		}
+		return true
+	}
+	return false
+}
+
+func (r *TaskRouter) create(w http.ResponseWriter, ctx context.Context, id json.RawMessage, tool TaskTool, args json.RawMessage, ttl *int64) {
+	start, reject := tool.StartTask(ctx, args)
+	if reject != nil {
+		writeRPCError(w, id, mcp.INVALID_PARAMS, resultText(reject))
+		return
+	}
+	ref := TaskRef{Tool: tool.Meta().Name, Key: start.Key, Flag: start.Flag}
+	taskID := r.ids.encode(ref)
+	state, err := tool.TaskStatus(ctx, ref)
+	if err != nil {
+		r.log.Warn("[agent-mcp] 任务创建后查询状态失败", zap.String("taskId", taskID), zap.Error(err))
+		writeRPCError(w, id, mcp.INTERNAL_ERROR, "任务已创建，但查询状态失败，请稍后重试")
+		return
+	}
+	if ttl == nil {
+		t := defaultTaskTTL
+		ttl = &t
+	}
+	res := map[string]any{"task": r.task(taskID, tool, state, ttl)}
+	if start.Note != "" {
+		res["_meta"] = map[string]any{immediateMeta: start.Note}
+	}
+	r.log.Info("[agent-mcp] 任务已创建", zap.String("tool", ref.Tool), zap.String("taskId", taskID), zap.String("flag", ref.Flag), zap.String("status", string(state.Status)))
+	writeRPCResult(w, id, res)
+}
+
+func (r *TaskRouter) get(w http.ResponseWriter, ctx context.Context, id json.RawMessage, tool TaskTool, ref TaskRef, taskID string) {
+	state, ok := r.status(w, ctx, id, tool, ref)
+	if ok {
+		writeRPCResult(w, id, r.task(taskID, tool, state, nil))
+	}
+}
+
+// result 阻塞到任务结束（规范要求），然后返回工具结果
+func (r *TaskRouter) result(w http.ResponseWriter, ctx context.Context, id json.RawMessage, tool TaskTool, ref TaskRef, taskID string) {
+	for {
+		state, ok := r.status(w, ctx, id, tool, ref)
+		if !ok {
+			return
+		}
+		if state.Status.IsTerminal() {
+			res := state.Result
+			if res == nil {
+				res = mcp.NewToolResultText(state.Message)
+				res.IsError = state.Status != mcp.TaskStatusCompleted
+			}
+			out := *res
+			out.Meta = &mcp.Meta{AdditionalFields: map[string]any{relatedMeta: map[string]any{"taskId": taskID}}}
+			writeRPCResult(w, id, out)
+			return
+		}
+		select {
+		case <-ctx.Done():
+			writeRPCError(w, id, mcp.INTERNAL_ERROR, "等待任务结束时连接已断开")
+			return
+		case <-time.After(resultWaitStep):
+		}
+	}
+}
+
+func (r *TaskRouter) cancel(w http.ResponseWriter, ctx context.Context, id json.RawMessage, tool TaskTool, ref TaskRef, taskID string) {
+	state, ok := r.status(w, ctx, id, tool, ref)
+	if !ok {
+		return
+	}
+	if state.Status.IsTerminal() {
+		writeRPCError(w, id, mcp.INVALID_PARAMS, fmt.Sprintf("cannot cancel task in terminal status: %s", state.Status))
+		return
+	}
+	if err := tool.CancelTask(ctx, ref); err != nil {
+		r.log.Warn("[agent-mcp] 取消任务失败", zap.String("taskId", taskID), zap.Error(err))
+		writeRPCError(w, id, mcp.INTERNAL_ERROR, "取消任务失败，请稍后重试")
+		return
+	}
+	r.log.Info("[agent-mcp] 任务已取消", zap.String("taskId", taskID), zap.String("flag", ref.Flag))
+	state.Status, state.Message = mcp.TaskStatusCancelled, "已取消"
+	writeRPCResult(w, id, r.task(taskID, tool, state, nil))
+}
+
+func (r *TaskRouter) status(w http.ResponseWriter, ctx context.Context, id json.RawMessage, tool TaskTool, ref TaskRef) (TaskState, bool) {
+	state, err := tool.TaskStatus(ctx, ref)
+	switch {
+	case errors.Is(err, ErrTaskNotFound):
+		writeRPCError(w, id, mcp.INVALID_PARAMS, "task not found")
+		return state, false
+	case err != nil:
+		// 暂时性错误：平台下次再查（连续失败 30 分钟才按失败处理）
+		r.log.Warn("[agent-mcp] 查询任务状态失败", zap.String("tool", ref.Tool), zap.String("key", ref.Key), zap.Error(err))
+		writeRPCError(w, id, mcp.INTERNAL_ERROR, "查询任务状态失败，请稍后重试")
+		return state, false
+	}
+	return state, true
+}
+
+func (r *TaskRouter) task(taskID string, tool TaskTool, s TaskState, ttl *int64) mcp.Task {
+	if ttl == nil {
+		t := defaultTaskTTL
+		ttl = &t
+	}
+	poll := int64(tool.PollInterval() / time.Millisecond)
+	created := s.CreatedAt
+	if created.IsZero() {
+		created = time.Now()
+	}
+	return mcp.Task{
+		TaskId:        taskID,
+		Status:        s.Status,
+		StatusMessage: s.Message,
+		CreatedAt:     created.UTC().Format(time.RFC3339),
+		LastUpdatedAt: time.Now().UTC().Format(time.RFC3339),
+		TTL:           ttl,
+		PollInterval:  &poll,
+	}
+}
+
+func resultText(r *mcp.CallToolResult) string {
+	var parts []string
+	for _, c := range r.Content {
+		if t, ok := c.(mcp.TextContent); ok {
+			parts = append(parts, t.Text)
+		}
+	}
+	return strings.Join(parts, "\n")
+}
+
+func writeRPCResult(w http.ResponseWriter, id json.RawMessage, result any) {
+	writeRPC(w, map[string]any{"jsonrpc": "2.0", "id": id, "result": result})
+}
+
+func writeRPCError(w http.ResponseWriter, id json.RawMessage, code int, msg string) {
+	writeRPC(w, map[string]any{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": code, "message": msg}})
+}
+
+func writeRPC(w http.ResponseWriter, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(v)
+}

service/agentmcp/kit/task_test.go

@@ -0,0 +1,172 @@
+package kit
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/mark3labs/mcp-go/mcp"
+	"go.uber.org/zap"
+)
+
+type fakeTaskArgs struct {
+	Name string `json:"name"`
+}
+
+// 业务数据：key -> 状态；cancelled 记录被真正取消的 key
+type fakeJobs struct {
+	status    map[string]mcp.TaskStatus
+	cancelled []string
+}
+
+func newFakeTaskTool(jobs *fakeJobs) TaskDef[fakeTaskArgs] {
+	return TaskDef[fakeTaskArgs]{
+		Def: Def[fakeTaskArgs]{Name: "job_run", Title: "运行作业", Description: "测试", Handler: func(context.Context, fakeTaskArgs) (*mcp.CallToolResult, error) {
+			return mcp.NewToolResultText("sync"), nil
+		}},
+		Poll: time.Minute,
+		Start: func(_ context.Context, a fakeTaskArgs) (TaskStart, *mcp.CallToolResult) {
+			if a.Name == "" {
+				return TaskStart{}, Failf("参数 name 必填")
+			}
+			jobs.status[a.Name] = mcp.TaskStatusWorking
+			return TaskStart{Key: a.Name, Flag: "owner", Note: "已开始"}, nil
+		},
+		Status: func(_ context.Context, ref TaskRef) (TaskState, error) {
+			st, ok := jobs.status[ref.Key]
+			if !ok {
+				return TaskState{}, ErrTaskNotFound
+			}
+			s := TaskState{Status: st, Message: string(st)}
+			if st == mcp.TaskStatusCompleted {
+				s.Result = mcp.NewToolResultText("done " + ref.Key)
+			}
+			if st == mcp.TaskStatusFailed {
+				s.Result = Failf("failed %s", ref.Key)
+			}
+			return s, nil
+		},
+		Cancel: func(_ context.Context, ref TaskRef) error {
+			if ref.Flag == "owner" {
+				jobs.cancelled = append(jobs.cancelled, ref.Key)
+			}
+			return nil
+		},
+	}
+}
+
+type rpcReply struct {
+	Result json.RawMessage `json:"result"`
+	Error  *struct {
+		Code    int    `json:"code"`
+		Message string `json:"message"`
+	} `json:"error"`
+}
+
+func TestTaskRouter(t *testing.T) {
+	jobs := &fakeJobs{status: map[string]mcp.TaskStatus{}}
+	passed := 0
+	next := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		passed++
+		_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":{}}`))
+	})
+	router := NewTaskRouter(next, []TaskTool{newFakeTaskTool(jobs)}, "secret", zap.NewNop())
+
+	call := func(method string, params any) rpcReply {
+		t.Helper()
+		body, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": 7, "method": method, "params": params})
+		rec := httptest.NewRecorder()
+		router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/mcp", bytes.NewReader(body)))
+		var r rpcReply
+		if err := json.Unmarshal(rec.Body.Bytes(), &r); err != nil {
+			t.Fatalf("%s: 响应不是 JSON-RPC：%s", method, rec.Body.String())
+		}
+		return r
+	}
+
+	// 以任务方式调用：立即返回任务与即时说明
+	r := call("tools/call", map[string]any{"name": "job_run", "arguments": map[string]any{"name": "a"}, "task": map[string]any{"ttl": 1000}})
+	var created struct {
+		Task mcp.Task       `json:"task"`
+		Meta map[string]any `json:"_meta"`
+	}
+	if r.Error != nil || json.Unmarshal(r.Result, &created) != nil {
+		t.Fatalf("创建任务失败：%+v", r)
+	}
+	id := created.Task.TaskId
+	if !strings.HasPrefix(id, taskIDPrefix+".job_run.a.owner.") || created.Task.Status != mcp.TaskStatusWorking ||
+		*created.Task.TTL != 1000 || *created.Task.PollInterval != 60000 || created.Meta[immediateMeta] != "已开始" {
+		t.Errorf("创建任务的返回不符合预期：%+v", created)
+	}
+
+	// 参数错误：返回 JSON-RPC 错误
+	if r := call("tools/call", map[string]any{"name": "job_run", "arguments": map[string]any{}, "task": map[string]any{}}); r.Error == nil || !strings.Contains(r.Error.Message, "name 必填") {
+		t.Errorf("参数错误应返回错误：%+v", r)
+	}
+
+	// 查询状态
+	var got mcp.Task
+	if r := call("tasks/get", map[string]any{"taskId": id}); r.Error != nil || json.Unmarshal(r.Result, &got) != nil || got.Status != mcp.TaskStatusWorking {
+		t.Errorf("tasks/get 应为 working：%+v", r)
+	}
+
+	// 篡改任务 id：找不到任务
+	tampered := strings.Replace(id, ".a.", ".b.", 1)
+	if r := call("tasks/get", map[string]any{"taskId": tampered}); r.Error == nil || r.Error.Code != mcp.INVALID_PARAMS {
+		t.Errorf("篡改的任务 id 应返回 INVALID_PARAMS：%+v", r)
+	}
+
+	// 完成后取结果：带 related-task
+	jobs.status["a"] = mcp.TaskStatusCompleted
+	var res struct {
+		Content []mcp.TextContent `json:"content"`
+		IsError bool              `json:"isError"`
+		Meta    map[string]any    `json:"_meta"`
+	}
+	if r := call("tasks/result", map[string]any{"taskId": id}); r.Error != nil || json.Unmarshal(r.Result, &res) != nil ||
+		res.Content[0].Text != "done a" || res.IsError || res.Meta[relatedMeta] == nil {
+		t.Errorf("tasks/result 不符合预期：%+v %+v", r, res)
+	}
+
+	// 已结束的任务不能取消
+	if r := call("tasks/cancel", map[string]any{"taskId": id}); r.Error == nil {
+		t.Errorf("已完成的任务取消应报错：%+v", r)
+	}
+
+	// 取消进行中的任务：调用业务取消
+	r = call("tools/call", map[string]any{"name": "job_run", "arguments": map[string]any{"name": "b"}, "task": map[string]any{}})
+	_ = json.Unmarshal(r.Result, &created)
+	if r := call("tasks/cancel", map[string]any{"taskId": created.Task.TaskId}); r.Error != nil || json.Unmarshal(r.Result, &got) != nil || got.Status != mcp.TaskStatusCancelled {
+		t.Errorf("取消应返回 cancelled：%+v", r)
+	}
+	if len(jobs.cancelled) != 1 || jobs.cancelled[0] != "b" {
+		t.Errorf("应取消作业 b，实际 %v", jobs.cancelled)
+	}
+
+	// 失败的任务：结果为 isError
+	jobs.status["b"] = mcp.TaskStatusFailed
+	if r := call("tasks/result", map[string]any{"taskId": created.Task.TaskId}); json.Unmarshal(r.Result, &res) != nil || !res.IsError {
+		t.Errorf("失败任务的结果应为 isError：%+v", r)
+	}
+
+	// 不归本服务处理的请求交给 mcp-go：普通调用、别的任务 id、其他方法
+	passed = 0
+	call("tools/call", map[string]any{"name": "job_run", "arguments": map[string]any{"name": "c"}})
+	call("tasks/get", map[string]any{"taskId": "other-task"})
+	call("tools/list", map[string]any{})
+	if passed != 3 {
+		t.Errorf("应有 3 个请求交给 mcp-go，实际 %d", passed)
+	}
+}
+
+func TestTaskDefDeclaresTaskSupport(t *testing.T) {
+	tool := newFakeTaskTool(&fakeJobs{}).Tool()
+	if tool.Execution == nil || tool.Execution.TaskSupport != mcp.TaskSupportOptional {
+		t.Errorf("TaskDef 应声明 taskSupport=optional：%+v", tool.Execution)
+	}
+}

service/agentmcp/kit/tool.go

@@ -41,7 +41,12 @@ type Def[T any] struct {
 	Title       string // 中文短标题（审批卡片与后台显示）
 	Description string // 用途、适用场景、返回字段与单位
 	ReadOnly    bool
-	Handler     func(ctx context.Context, args T) (*mcp.CallToolResult, error)
+	// Destructive 覆盖、删除等不可恢复的写操作。AgentServer 据此“总是审批”：自动模式下也每次询问用户
+	Destructive bool
+	// Idempotent 同样参数重复调用结果相同（只读工具默认幂等）。AgentServer 网络出错时会自动重试一次，
+	// 所以有副作用且不去重的工具不能声明
+	Idempotent bool
+	Handler    func(ctx context.Context, args T) (*mcp.CallToolResult, error)
 	// Properties 可选：补充生成的 schema 中各字段的约束（枚举、取值范围、数组长度等），
 	// key 为字段名，value 合并进该字段的 schema（同名键覆盖）。服务端按补充后的 schema 校验参数。
 	Properties map[string]map[string]any
@@ -57,8 +62,8 @@ func (d Def[T]) Tool() mcp.Tool {
 		mcp.WithInputSchema[T](),
 		mcp.WithTitleAnnotation(d.Title),
 		mcp.WithReadOnlyHintAnnotation(d.ReadOnly),
-		mcp.WithDestructiveHintAnnotation(false),
-		mcp.WithIdempotentHintAnnotation(d.ReadOnly),
+		mcp.WithDestructiveHintAnnotation(d.Destructive),
+		mcp.WithIdempotentHintAnnotation(d.ReadOnly || d.Idempotent),
 		mcp.WithOpenWorldHintAnnotation(false),
 	)
 	if len(d.Properties) > 0 {

service/agentmcp/kit/tool_test.go

@@ -55,3 +55,22 @@ func TestDefPropertiesUnknownField(t *testing.T) {
 	}()
 	Def[patchArgs]{Name: "patch_bad", Properties: map[string]map[string]any{"nope": {"type": "string"}}}.Tool()
 }
+
+func TestDefHints(t *testing.T) {
+	h := func(context.Context, NoArgs) (*mcp.CallToolResult, error) { return nil, nil }
+	cases := []struct {
+		def               Def[NoArgs]
+		destructive, idem bool
+	}{
+		{Def[NoArgs]{Name: "ro", ReadOnly: true, Handler: h}, false, true},
+		{Def[NoArgs]{Name: "write", Handler: h}, false, false},
+		{Def[NoArgs]{Name: "overwrite", Destructive: true, Handler: h}, true, false},
+		{Def[NoArgs]{Name: "dedup", Idempotent: true, Handler: h}, false, true},
+	}
+	for _, c := range cases {
+		a := c.def.Tool().Annotations
+		if *a.DestructiveHint != c.destructive || *a.IdempotentHint != c.idem {
+			t.Errorf("%s: destructive=%v idempotent=%v，期望 %v %v", c.def.Name, *a.DestructiveHint, *a.IdempotentHint, c.destructive, c.idem)
+		}
+	}
+}

service/agentmcp/server.go

@@ -33,21 +33,30 @@ func Tools(set string) ([]kit.Registrar, error) {
 	return nil, fmt.Errorf("未知的工具集：%s（可选 readonly / ops）", set)
 }
 
-// NewServer 创建 MCPServer 并注册工具集。
-func NewServer(set string, log *zap.Logger, maxConcurrency int) (*server.MCPServer, error) {
+// NewServer 创建 MCPServer 并注册工具集；同时返回其中的持久化长任务工具（交给 Handler 的 TaskRouter 处理任务协议）。
+func NewServer(set string, log *zap.Logger, maxConcurrency int) (*server.MCPServer, []kit.TaskTool, error) {
 	list, err := Tools(set)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	s := server.NewMCPServer(
-		ServerName+"-"+set,
-		ServerVersion,
+	var tasks []kit.TaskTool
+	for _, t := range list {
+		if tt, ok := t.(kit.TaskTool); ok {
+			tasks = append(tasks, tt)
+		}
+	}
+	opts := []server.ServerOption{
 		server.WithToolCapabilities(false),
 		server.WithInputSchemaValidation(),
 		server.WithToolHandlerMiddleware(kit.Middleware(log, maxConcurrency)),
-	)
+	}
+	if len(tasks) > 0 {
+		// 声明支持以任务方式调用工具与取消（tasks.requests.tools.call、tasks.cancel）；任务协议由 kit.TaskRouter 处理
+		opts = append(opts, server.WithTaskCapabilities(false, true, true))
+	}
+	s := server.NewMCPServer(ServerName+"-"+set, ServerVersion, opts...)
 	for _, t := range list {
 		t.Register(s)
 	}
-	return s, nil
+	return s, tasks, nil
 }

service/agentmcp/tools/logs.go

@@ -96,8 +96,8 @@ var EmsLogRecords = kit.Def[emsLogRecordsArgs]{
 		}
 		rec, err := repositories.QueryEmsLogCallRecord(sn, date, a.LogType)
 		if err != nil {
-			// 上游在"没有召唤日志"时也返回 error，这里按无数据处理并带上原因
-			return kit.Empty("未找到 EMS 日志召唤记录：" + err.Error()), nil
+			// 上游在"没有召唤日志"时也返回 error，这里按无数据处理；原始错误只写日志
+			return kit.Empty("未找到 EMS 日志召唤记录（" + kit.UpstreamText("FWH 平台 EMS 日志接口", err) + "）。"), nil
 		}
 		return kit.Object(rec), nil
 	},

service/agentmcp/tools/misc.go

@@ -139,7 +139,7 @@ var PlaneIssueGet = kit.Def[planeIssueGetArgs]{
 						out["comments"] = comments
 					}
 				} else {
-					out["commentsError"] = err.Error()
+					out["commentsError"] = kit.UpstreamText("Plane 评论接口", err)
 				}
 			}
 		}

service/agentmcp/tools/ops.go

@@ -73,6 +73,8 @@ var TouTemplateSave = kit.Def[touTemplateSaveArgs]{
 	Description: "在 FWH 测试服新增或更新 TOU 调度模版，只支持测试服。保存前校验每个策略的时段首尾衔接覆盖全天、sourceType 为 0；" +
 		"返回 templateId 与按调度模式识别后的策略摘要。这是写操作，执行前会请求确认。",
 	ReadOnly: false,
+	// 更新会覆盖测试服上已有的模版，不可恢复：声明为破坏性，自动模式下也每次审批
+	Destructive: true,
 	Handler: func(ctx context.Context, a touTemplateSaveArgs) (*mcp.CallToolResult, error) {
 		raw, err := kit.Required("payload", a.Payload)
 		if err != nil {

service/agentmcp/tools/site.go

@@ -70,17 +70,17 @@ var SiteGet = kit.Def[siteGetArgs]{
 		var errs []string
 
 		if info, err := services.GetSiteInfoData(sn); err != nil {
-			errs = append(errs, "info: "+err.Error())
+			errs = append(errs, "info: "+kit.UpstreamText("站点信息表", err))
 		} else {
 			out["info"] = info
 		}
 		if mode, err := repositories.QuerySiteModeInfo(sn); err != nil {
-			errs = append(errs, "mode: "+err.Error())
+			errs = append(errs, "mode: "+kit.UpstreamText("FWH 平台工作模式接口", err))
 		} else if mode != nil {
 			out["mode"] = mode.Result
 		}
 		if alarms, err := repositories.QuerySiteAlarmList(sn); err != nil {
-			errs = append(errs, "currentAlarms: "+err.Error())
+			errs = append(errs, "currentAlarms: "+kit.UpstreamText("FWH 平台实时告警接口", err))
 		} else if alarms != nil && alarms.Result != nil {
 			list := *alarms.Result
 			out["currentAlarmTotal"] = len(list)

service/agentmcp/tools/siteinfosync.go

@@ -6,11 +6,13 @@ import (
 	"time"
 
 	"github.com/mark3labs/mcp-go/mcp"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 
 	"fwh-toolbox-server/agentmcp/kit"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/repositories"
 	"fwh-toolbox-server/services"
+	"fwh-toolbox-server/task/queue/fwhQueue"
 )
 
 // 站点信息表同步（SiteInfoManageRecord）的状态判断。
@@ -65,6 +67,10 @@ func loadSiteInfoSync() (*siteInfoSync, error) {
 
 // evalSiteInfoSync 根据最近一次记录与最近一次成功记录判断同步状态
 func evalSiteInfoSync(latest, lastComplete *domain.SiteInfoUpdateRecord, now time.Time) *siteInfoSync {
+	// 最近一条已完成时它就是最近一次成功（调用方可能只传了最近一条）
+	if lastComplete == nil && latest != nil && latest.Progress == domain.ProgressComplete {
+		lastComplete = latest
+	}
 	s := &siteInfoSync{Latest: latest, LastSuccess: lastComplete, now: now}
 	switch {
 	case latest == nil:
@@ -181,29 +187,126 @@ type siteInfoRefreshArgs struct {
 	Force bool `json:"force,omitempty" jsonschema:"为 true 时：最近 24 小时内已成功同步、或上次同步疑似中断（stale）时仍重新触发。同步进行中（syncing）时任何情况下都不会重复触发"`
 }
 
-var SiteInfoRefresh = kit.Def[siteInfoRefreshArgs]{
-	Name:  "site_info_refresh",
-	Title: "触发站点信息表同步",
-	Description: "触发 ToolBox 站点信息表从 FWH 平台重新同步，立即返回，同步在主服务后台执行，耗时可能超过 1.5 小时。" +
-		"触发前先检查同步状态：同步进行中时不重复触发，返回当前进度（action=tracking）；最近 24 小时内已成功同步时不触发（action=skipped_fresh）；" +
-		"上次同步超过 6 小时未完成、疑似中断时不触发（action=skipped_stale）；以上两种不触发的情况可用 force=true 强制触发。" +
-		"已触发时 action=triggered。返回中的 status 与 site_info_sync_status 相同，之后用 site_info_sync_status 跟踪进度。" +
-		"建议先调用 site_info_sync_status 判断是否需要同步，再调用本工具。这是写操作，执行前会请求确认。",
-	ReadOnly: false,
-	Handler: func(ctx context.Context, a siteInfoRefreshArgs) (*mcp.CallToolResult, error) {
-		s, err := loadSiteInfoSync()
+// siteInfoTaskPoll 平台查询任务状态的间隔：同步约 1 小时，每分钟查一次足够
+const siteInfoTaskPoll = time.Minute
+
+var SiteInfoRefresh = kit.TaskDef[siteInfoRefreshArgs]{
+	Def: kit.Def[siteInfoRefreshArgs]{
+		Name:  "site_info_refresh",
+		Title: "触发站点信息表同步",
+		Description: "触发 ToolBox 站点信息表从 FWH 平台重新同步，同步在主服务后台执行，耗时约 1 小时，可能超过 1.5 小时。" +
+			"触发前会检查同步状态：同步进行中时不重复触发，改为跟踪这次同步（action=tracking）；最近 24 小时内已成功同步时不触发（action=skipped_fresh）；" +
+			"上次同步超过 6 小时未完成、疑似中断时不触发（action=skipped_stale）；这两种不触发的情况可用 force=true 强制触发。已触发时 action=triggered。" +
+			"以长任务方式执行：触发或跟踪时立即返回“已转为后台”，同步结束后平台会把结果（完成时间、站点数、耗时，或失败原因）发给你，无需轮询；" +
+			"用户询问进度时用 site_info_sync_status。这是写操作，执行前会请求确认。",
+		ReadOnly: false,
+		// 同步中或数据较新时不重复触发，网络出错时平台重试一次不会产生第二次同步
+		Idempotent: true,
+		Handler: func(ctx context.Context, a siteInfoRefreshArgs) (*mcp.CallToolResult, error) {
+			action, s, fail := refreshSiteInfo(a.Force)
+			if fail != nil {
+				return fail, nil
+			}
+			return kit.Object(map[string]any{"action": action, "message": refreshMessages[action], "status": s.view()}), nil
+		},
+	},
+	Poll: siteInfoTaskPoll,
+	Start: func(ctx context.Context, a siteInfoRefreshArgs) (kit.TaskStart, *mcp.CallToolResult) {
+		action, s, fail := refreshSiteInfo(a.Force)
+		if fail != nil {
+			return kit.TaskStart{}, fail
+		}
+		// 每种结果都对应一条同步记录：触发的是新记录，跟踪、不触发的是最近一条记录
+		note := refreshMessages[action]
+		if action == "triggered" || action == "tracking" {
+			note += "同步结束后平台会把结果发给你，现在可以先告诉用户已提交。"
+		}
+		return kit.TaskStart{Key: s.Latest.Id.Hex(), Flag: action, Note: note}, nil
+	},
+	Status: func(ctx context.Context, ref kit.TaskRef) (kit.TaskState, error) {
+		id, err := primitive.ObjectIDFromHex(ref.Key)
 		if err != nil {
-			return kit.Upstream("ToolBox 站点信息同步记录", err), nil
+			return kit.TaskState{}, kit.ErrTaskNotFound
 		}
-		action, trigger := s.decideRefresh(a.Force)
-		if trigger {
-			if err := services.UpdateSiteInfoDate(); err != nil {
-				return kit.Upstream("ToolBox 站点信息同步", err), nil
-			}
-			if s, err = loadSiteInfoSync(); err != nil {
-				return kit.Upstream("ToolBox 站点信息同步记录", err), nil
-			}
+		rec, err := repositories.GetSiteInfoUpdateRecord(id)
+		if err != nil {
+			return kit.TaskState{}, err
 		}
-		return kit.Object(map[string]any{"action": action, "message": refreshMessages[action], "status": s.view()}), nil
+		if rec == nil {
+			return kit.TaskState{}, kit.ErrTaskNotFound
+		}
+		return siteInfoTaskState(ref.Flag, rec, time.Now()), nil
 	},
+	Cancel: func(ctx context.Context, ref kit.TaskRef) error {
+		// 只有本次调用触发的同步才真正停止；跟踪他人触发的同步时只是停止跟踪（其他会话可能还在等它）
+		if ref.Flag != "triggered" {
+			return nil
+		}
+		id, err := primitive.ObjectIDFromHex(ref.Key)
+		if err != nil {
+			return kit.ErrTaskNotFound
+		}
+		return fwhQueue.CancelSiteInfoUpdate(id)
+	},
+}
+
+// refreshSiteInfo 判断并在需要时触发同步，返回动作与触发后的状态；普通调用与任务方式调用共用
+func refreshSiteInfo(force bool) (string, *siteInfoSync, *mcp.CallToolResult) {
+	s, err := loadSiteInfoSync()
+	if err != nil {
+		return "", nil, kit.Upstream("ToolBox 站点信息同步记录", err)
+	}
+	action, trigger := s.decideRefresh(force)
+	if trigger {
+		if err := services.UpdateSiteInfoDate(); err != nil {
+			return "", nil, kit.Upstream("ToolBox 站点信息同步", err)
+		}
+		if s, err = loadSiteInfoSync(); err != nil {
+			return "", nil, kit.Upstream("ToolBox 站点信息同步记录", err)
+		}
+	}
+	return action, s, nil
+}
+
+// siteInfoTaskState 把同步记录翻译为任务状态。flag 为发起任务时的动作：
+// skipped_* 时任务立即完成（结果是不触发的原因）；triggered / tracking 时跟随这条同步记录
+func siteInfoTaskState(flag string, rec *domain.SiteInfoUpdateRecord, now time.Time) kit.TaskState {
+	st := kit.TaskState{CreatedAt: rec.CreateTime}
+	if flag == "skipped_fresh" || flag == "skipped_stale" {
+		st.Status, st.Message = mcp.TaskStatusCompleted, "未触发同步"
+		st.Result = kit.Object(map[string]any{"action": flag, "message": refreshMessages[flag], "status": evalSiteInfoSync(rec, nil, now).view()})
+		return st
+	}
+	elapsed := now.Sub(rec.CreateTime)
+	switch rec.Progress {
+	case domain.ProgressPending, domain.ProgressRunning:
+		if elapsed > syncStaleAfter {
+			st.Status = mcp.TaskStatusFailed
+			st.Message = fmt.Sprintf("同步已 %s 仍未完成，疑似中断", fmtHours(elapsed))
+			st.Result = kit.Failf("站点信息同步已请求 %s 仍未完成，超过 %s，疑似中断（如主服务重启）。可以告诉用户，确认后用 site_info_refresh 并设置 force=true 重新同步。",
+				fmtHours(elapsed), fmtHours(syncStaleAfter))
+			return st
+		}
+		st.Status = mcp.TaskStatusWorking
+		if rec.Progress == domain.ProgressPending {
+			st.Message = fmt.Sprintf("排队中，已等待 %s", fmtHours(elapsed))
+		} else {
+			st.Message = fmt.Sprintf("同步中，已进行 %s", fmtHours(elapsed))
+		}
+	case domain.ProgressComplete:
+		st.Status, st.Message = mcp.TaskStatusCompleted, "同步完成"
+		st.Result = kit.Object(map[string]any{
+			"message":     fmt.Sprintf("站点信息表同步完成：%d 个站点，耗时 %s，完成于 %s（北京时间）。", rec.Total, fmtHours(rec.UpdateTime.Sub(rec.CreateTime)), fmtBeijing(rec.UpdateTime)),
+			"requestedAt": fmtBeijing(rec.CreateTime),
+			"completedAt": fmtBeijing(rec.UpdateTime),
+			"total":       rec.Total,
+		})
+	case domain.ProgressCancelled:
+		st.Status, st.Message = mcp.TaskStatusCancelled, "同步已取消"
+		st.Result = kit.Failf("站点信息同步已取消。")
+	default: // error / failed
+		st.Status, st.Message = mcp.TaskStatusFailed, "同步失败"
+		st.Result = kit.Failf("站点信息同步失败（主服务同步任务出错，详情见主服务日志）。可以稍后用 site_info_refresh 重新同步。")
+	}
+	return st
 }

service/agentmcp/tools/siteinfosync_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/mark3labs/mcp-go/mcp"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 
 	"fwh-toolbox-server/domain"
@@ -55,3 +56,37 @@ func TestSiteInfoSyncDecide(t *testing.T) {
 		}
 	}
 }
+
+func TestSiteInfoTaskState(t *testing.T) {
+	now := time.Date(2026, 10, 29, 12, 0, 0, 0, time.UTC)
+	ago := func(h float64) time.Time { return now.Add(-time.Duration(h * float64(time.Hour))) }
+
+	cases := []struct {
+		name    string
+		flag    string
+		rec     *domain.SiteInfoUpdateRecord
+		status  mcp.TaskStatus
+		isError bool
+	}{
+		{"排队中", "triggered", syncRecord(domain.ProgressPending, ago(0.1), ago(0.1)), mcp.TaskStatusWorking, false},
+		{"同步中", "tracking", syncRecord(domain.ProgressRunning, ago(1), ago(1)), mcp.TaskStatusWorking, false},
+		{"超过 6 小时疑似中断", "triggered", syncRecord(domain.ProgressRunning, ago(7), ago(7)), mcp.TaskStatusFailed, true},
+		{"完成", "triggered", syncRecord(domain.ProgressComplete, ago(2), ago(1)), mcp.TaskStatusCompleted, false},
+		{"失败", "triggered", syncRecord(domain.ProgressError, ago(2), ago(2)), mcp.TaskStatusFailed, true},
+		{"取消", "triggered", syncRecord(domain.ProgressCancelled, ago(2), ago(2)), mcp.TaskStatusCancelled, true},
+		{"未触发：数据较新，立即完成", "skipped_fresh", syncRecord(domain.ProgressComplete, ago(3), ago(2)), mcp.TaskStatusCompleted, false},
+		{"未触发：疑似中断，立即完成", "skipped_stale", syncRecord(domain.ProgressRunning, ago(7), ago(7)), mcp.TaskStatusCompleted, false},
+	}
+	for _, c := range cases {
+		st := siteInfoTaskState(c.flag, c.rec, now)
+		if st.Status != c.status {
+			t.Errorf("%s: status = %s，期望 %s", c.name, st.Status, c.status)
+		}
+		if st.Status.IsTerminal() && (st.Result == nil || st.Result.IsError != c.isError) {
+			t.Errorf("%s: 终态结果不符合预期（isError 期望 %v）：%+v", c.name, c.isError, st.Result)
+		}
+		if !st.Status.IsTerminal() && st.Message == "" {
+			t.Errorf("%s: 进行中应有进度说明", c.name)
+		}
+	}
+}

service/cmd/agent-mcp/main.go

@@ -47,7 +47,7 @@ func main() {
 		log.Fatal("[agent-mcp] 初始化失败", zap.Error(err))
 	}
 
-	s, err := agentmcp.NewServer(*set, log, settings.MaxConcurrency)
+	s, tasks, err := agentmcp.NewServer(*set, log, settings.MaxConcurrency)
 	if err != nil {
 		log.Fatal("[agent-mcp] 创建 MCP Server 失败", zap.Error(err))
 	}
@@ -56,7 +56,7 @@ func main() {
 	defer stop()
 
 	log.Info("[agent-mcp] 启动", zap.String("set", *set), zap.String("version", agentmcp.ServerVersion), zap.Int("maxConcurrency", settings.MaxConcurrency))
-	if err := agentmcp.Serve(ctx, addr, agentmcp.Handler(s, settings.Token), log); err != nil {
+	if err := agentmcp.Serve(ctx, addr, agentmcp.Handler(s, settings.Token, tasks, log), log); err != nil {
 		log.Fatal("[agent-mcp] HTTP 服务异常退出", zap.Error(err))
 	}
 	log.Info("[agent-mcp] 已退出")

service/domain/taskQueueDomain.go

@@ -5,11 +5,12 @@ type Trigger string  // 'auto' | 'manual'
 
 // Progress枚举
 const (
-	ProgressPending  Progress = "pending"
-	ProgressRunning  Progress = "running"
-	ProgressFailed   Progress = "failed"
-	ProgressError    Progress = "error"
-	ProgressComplete Progress = "complete"
+	ProgressPending   Progress = "pending"
+	ProgressRunning   Progress = "running"
+	ProgressFailed    Progress = "failed"
+	ProgressError     Progress = "error"
+	ProgressComplete  Progress = "complete"
+	ProgressCancelled Progress = "cancelled" // 用户取消（目前用于站点信息同步）
 )
 
 // Trigger枚举

service/go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hibiken/asynq v0.26.0
 	github.com/hibiken/asynqmon v0.7.2
-	github.com/mark3labs/mcp-go v0.57.0
+	github.com/mark3labs/mcp-go v1.4.0
 	github.com/redis/go-redis/v9 v9.21.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/sigurn/crc16 v0.0.0-20240131213347-83fcde1e29d1

service/go.sum

@@ -194,6 +194,8 @@ github.com/mailru/easyjson v0.7.6 h1:8yTIVnZgCoiM1TgqoeTl+LfU5Jg6/xL3QhGQnimLYnA
 github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mark3labs/mcp-go v0.57.0 h1:jzWKyCzdWnwnZt05cvcQQ+ngiUl2RnixXJa7Kj4qP1E=
 github.com/mark3labs/mcp-go v0.57.0/go.mod h1:+8WclSK1ZUweCP3hvktSji8n8ABG/95QaEkeVE/Uwas=
+github.com/mark3labs/mcp-go v1.4.0 h1:gEWEv7YOTJjW4K9cxuwfr0FLRFgSWOE1Bju1SxPlctI=
+github.com/mark3labs/mcp-go v1.4.0/go.mod h1:6L8eBboqlAMnYrvl8rpbUeV2qE/SJh52aVGrXIomVb0=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
 github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=

service/repositories/mongo/siteInfoManageRepo.go

@@ -96,6 +96,24 @@ func GetSiteInfoUpdateNewRecord() (*domain.SiteInfoUpdateRecord, error) {
 	return &result, nil
 }
 
+// 按 id 获取站点信息更新记录；不存在时返回 nil
+func GetSiteInfoUpdateRecord(id primitive.ObjectID) (*domain.SiteInfoUpdateRecord, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	collection := database.GetCollection(CollectionName_SiteInfoManageRecord)
+
+	var result domain.SiteInfoUpdateRecord
+	err := collection.FindOne(ctx, bson.M{"_id": id}).Decode(&result)
+	if err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &result, nil
+}
+
 // 获取最近一次成功完成的站点信息更新记录；没有时返回 nil
 func GetSiteInfoUpdateLastComplete() (*domain.SiteInfoUpdateRecord, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

service/repositories/repositories.go

@@ -423,6 +423,7 @@ var CreateSiteInfoUpdateInfo = mongo.CreateSiteInfoUpdateInfo
 var UpdateSiteInfoUpdateResult = mongo.UpdateSiteInfoUpdateResult
 var GetSiteInfoUpdateNewRecord = mongo.GetSiteInfoUpdateNewRecord
 var GetSiteInfoUpdateLastComplete = mongo.GetSiteInfoUpdateLastComplete
+var GetSiteInfoUpdateRecord = mongo.GetSiteInfoUpdateRecord
 var GetSiteInfoManageList = mongo.GetSiteInfoManageList
 var InsertSiteInfoData = mongo.InsertSiteInfoData
 var GetSiteInfoManageListFromGiccid = mongo.GetSiteInfoManageListFromGiccid

service/task/queue/fwhQueue/fwhEnqueue.go

@@ -120,7 +120,8 @@ func EnqueueSiteInfoUpdateTask(payload domain.TaskSiteInfoUpdate) error {
 	taskId := newTaskID(TaskTypeSiteInfoUpdate, payload.Id.Hex())
 	task := asynq.NewTask(TaskTypeSiteInfoUpdate, data,
 		asynq.MaxRetry(0),
-		asynq.Timeout(60*time.Minute),
+		// 全量同步 6 万多个站点约 1 小时，可能超过 1.5 小时；同步循环会响应 ctx，超时即停止，所以留足余量
+		asynq.Timeout(SiteInfoUpdateTimeout),
 		asynq.Queue(QueueMain),
 		asynq.TaskID(taskId),
 		asynq.Unique(30*time.Minute), // 30 分钟内去重（保证唯一性）

service/task/queue/fwhQueue/fwhTaskQueue.go

@@ -33,10 +33,11 @@ var (
 	inspector  *asynq.Inspector
 )
 
-// InitEnqueueClient 只创建入队客户端，不启动 worker、不注册处理函数。
+// InitEnqueueClient 只创建入队客户端与 Inspector（查询、取消），不启动 worker、不注册处理函数。
 // 供 fwh-agent-mcp 等独立进程把任务投递给主服务执行；主服务使用 InitFwhTaskQueue。
 func InitEnqueueClient(redisOpt asynq.RedisClientOpt) {
 	TaskClient = asynq.NewClient(redisOpt)
+	inspector = asynq.NewInspector(redisOpt)
 }
 
 // InitFwhTaskQueue 初始化新的 FWH 任务队列

service/task/queue/fwhQueue/fwhTaskSiteInfoUpdate.go

@@ -3,12 +3,15 @@ package fwhQueue
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/dto"
 	"fwh-toolbox-server/repositories"
+	"time"
 
 	"github.com/hibiken/asynq"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.uber.org/zap"
 )
 
@@ -29,6 +32,12 @@ func handleSiteInfoUpdate(ctx context.Context, task *asynq.Task) error {
 
 	// 执行任务
 	total, err := executeSiteInfoUpdate(ctx, payload)
+	if errors.Is(err, context.Canceled) {
+		// CancelSiteInfoUpdate 发出的取消（asynq 超时为 DeadlineExceeded，按失败处理）
+		zap.L().Info("[SiteInfoUpdate] 任务已取消", zap.String("id", taskId))
+		repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressCancelled, 0)
+		return err
+	}
 	if err != nil {
 		zap.L().Error("[SiteInfoUpdate] 任务执行失败", zap.String("id", taskId), zap.String("err", err.Error()))
 		repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressError, 0)
@@ -48,6 +57,10 @@ func executeSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdat
 	var realTotal uint64 = 0
 
 	for {
+		// 每页检查一次：取消（CancelSiteInfoUpdate）或超时（SiteInfoUpdateTimeout）时停止
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
 		resp, err := repositories.QuerySiteInfoList(dto.SiteInfoPara{
 			PageSize: pageSize,
 			PageNum:  pageNum,
@@ -72,3 +85,33 @@ func executeSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdat
 
 	return realTotal, nil
 }
+
+// SiteInfoUpdateTimeout 站点信息同步任务的执行超时
+const SiteInfoUpdateTimeout = 4 * time.Hour
+
+// CancelSiteInfoUpdate 取消一次站点信息同步：
+// 正在执行的通知 worker 停止（worker 停下后把记录标记为 cancelled）；还在排队或已不在队列中的直接删除并标记。
+// 可在主服务或只入队的独立进程（fwh-agent-mcp）中调用，需要 inspector 已初始化。
+func CancelSiteInfoUpdate(id primitive.ObjectID) error {
+	if inspector == nil {
+		return fmt.Errorf("任务队列未初始化")
+	}
+	taskId := newTaskID(TaskTypeSiteInfoUpdate, id.Hex())
+	info, err := inspector.GetTaskInfo(QueueMain, taskId)
+	switch {
+	case errors.Is(err, asynq.ErrTaskNotFound) || errors.Is(err, asynq.ErrQueueNotFound):
+		// 队列中已没有这个任务（如主服务重启时丢失）：只更新记录
+	case err != nil:
+		return err
+	case info.State == asynq.TaskStateActive:
+		return inspector.CancelProcessing(taskId)
+	case info.State == asynq.TaskStateCompleted || info.State == asynq.TaskStateArchived:
+		// 任务已经结束，worker 已写入最终状态，不覆盖
+		return nil
+	case info.State == asynq.TaskStatePending || info.State == asynq.TaskStateScheduled || info.State == asynq.TaskStateRetry:
+		if err := inspector.DeleteTask(QueueMain, taskId); err != nil {
+			return err
+		}
+	}
+	return repositories.UpdateSiteInfoUpdateResult(id, domain.ProgressCancelled, 0)
+}