From fcd2539b9e9307beade3b6b69891ccd8bd4a49e6 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 7 Oct 2026 13:38:51 +0000 Subject: [PATCH] Align the MCP client with 2026-07-28: resultType, input_required retry, tasks surface (N2 WP3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: the completion doc requires the client to recognize resultType, treat input_required as a protocol-level result whose answer retries the original tools/call, and expose tasks/get|update|cancel; today only a content-block special case existed and the advertised version was never checked against the server's. Evidence: GOWORK=off go vet + go test -count=1 ./... green; TestInputRequiredResultTypeRetriesOriginalCall (ask → tasks/update → retry, call order and final content), TestNegotiatedVersionFallbackKeepsLegacyPath (byte-level legacy behavior on 2024-11-05), TestTasksGetAndCancelRoundTrip, TestRunEndCancelsSeenTasks, TestDefaultSchemaDeclares202012; the acceptance-③ regression TestStdioSlowCancelThenEcho passes unchanged. Impact: advertise 2026-07-28 and branch on the negotiated version (older servers keep byte-identical behavior); synthesized default tool schemas now declare JSON Schema 2020-12; runs cancel the MCP tasks they saw via a per-run TaskTracker (other runs on the shared manager are unaffected). Refs: Agent-System-Documents#42 (N2 plan WP3) --- adapters/native/backend.go | 6 ++ runtime/mcp/ask.go | 13 ++- runtime/mcp/fakeserver.go | 77 ++++++++++++++- runtime/mcp/http.go | 28 ++++++ runtime/mcp/manager.go | 60 ++++++++++- runtime/mcp/mcp_test.go | 159 +++++++++++++++++++++++++++++- runtime/mcp/protocol.go | 33 ++++++- runtime/mcp/r2_regression_test.go | 6 ++ runtime/mcp/stdio.go | 26 +++++ runtime/mcp/tasks.go | 65 ++++++++++++ 10 files changed, 463 insertions(+), 10 deletions(-) create mode 100644 runtime/mcp/tasks.go diff --git a/adapters/native/backend.go b/adapters/native/backend.go index ae14e43..415a549 100644 --- a/adapters/native/backend.go +++ b/adapters/native/backend.go @@ -157,6 +157,12 @@ func (b *Backend) Execute(ctx context.Context, req exec.Request) (*exec.Result, if b.opts.Ask != nil { ctx = mcp.WithAsk(ctx, b.opts.Ask) } + // MCP 长任务跟踪（2026-07-28 最小映射）：本 run 内 tools/call 见过的 + // TaskID 在 run 结束时逐一 tasks/cancel，不影响共享 Manager 上其他 run。 + // 取消用 Background：run 的 ctx 在结束时已作废，取消投递不能被它掐掉。 + var taskTracker *mcp.TaskTracker + ctx, taskTracker = mcp.WithTaskTracker(ctx) + defer func() { taskTracker.CancelAll(context.Background()) }() model := strings.TrimSpace(b.opts.Model) if strings.TrimSpace(req.Model) != "" { model = strings.TrimSpace(req.Model) diff --git a/runtime/mcp/ask.go b/runtime/mcp/ask.go index 308d901..bf390a0 100644 --- a/runtime/mcp/ask.go +++ b/runtime/mcp/ask.go @@ -17,7 +17,18 @@ func askFrom(ctx context.Context) func(context.Context, string) (string, error) return fn } -func inputQuestion(res callToolResult) (string, bool) { +// inputQuestion 提取需要用户输入的问题。两条路径： +// 2026-07-28 协议级 resultType="input_required"（问题取文本块）； +// 旧版服务端的内容块 input_required（向后兼容）。 +func inputQuestion(res callToolResult, negotiated string) (string, bool) { + if res.ResultType == "input_required" && SupportsResultType(negotiated) { + for _, blk := range res.Content { + if blk.Type == "text" && blk.Text != "" { + return blk.Text, true + } + } + return "", true + } for _, blk := range res.Content { if blk.Type == "input_required" { return blk.Text, true diff --git a/runtime/mcp/fakeserver.go b/runtime/mcp/fakeserver.go index 3b3880c..499031f 100644 --- a/runtime/mcp/fakeserver.go +++ b/runtime/mcp/fakeserver.go @@ -44,8 +44,23 @@ func fakeServerHandle(req rpcRequest) rpcResponse { } switch req.Method { case "initialize": + // 协商：取(客户端请求, 自身上限)的较小者（ISO 日期字典序可比）； + // AGENT_NATIVE_TEST_MCP_VERSION 可压低上限以测回落路径。 + own := ProtocolVersion + if v := os.Getenv("AGENT_NATIVE_TEST_MCP_VERSION"); v != "" { + own = v + } + negotiated := own + var p initializeParams + if req.Params != nil { + raw, _ := json.Marshal(req.Params) + _ = json.Unmarshal(raw, &p) + if p.ProtocolVersion != "" && p.ProtocolVersion < negotiated { + negotiated = p.ProtocolVersion + } + } return respond(initializeResult{ - ProtocolVersion: ProtocolVersion, + ProtocolVersion: negotiated, ServerInfo: clientInfo{Name: "fake-mcp", Version: "0"}, }) case "tools/list": @@ -71,14 +86,64 @@ func fakeServerHandle(req rpcRequest) rpcResponse { case "slow": time.Sleep(2 * time.Second) return respond(callToolResult{Content: []contentBlock{{Type: "text", Text: "slow done"}}}) + case "need_input": + // 2026-07-28 协议级 input_required：tasks/update 投递后重试得终结果。 + if input, ok := fakeTaskInputs()["task-1"]; ok { + fakeTaskInputs()["task-1"] = "" + return respond(callToolResult{Content: []contentBlock{{Type: "text", Text: "答案已收到: " + input}}}) + } + return respond(callToolResult{ + Content: []contentBlock{{Type: "text", Text: "要答案"}}, + ResultType: "input_required", + TaskID: "task-1", + }) + case "legacy_ask": + return respond(callToolResult{Content: []contentBlock{{Type: "input_required", Text: "选一个颜色"}}}) default: return fail(-32602, "unknown tool "+p.Name) } + case "tasks/update": + var p struct { + TaskID string `json:"taskId"` + Input string `json:"input"` + } + if req.Params != nil { + raw, _ := json.Marshal(req.Params) + _ = json.Unmarshal(raw, &p) + } + fakeTaskInputs()[p.TaskID] = p.Input + return respond(map[string]any{"ok": true}) + case "tasks/get": + var p struct { + TaskID string `json:"taskId"` + } + if req.Params != nil { + raw, _ := json.Marshal(req.Params) + _ = json.Unmarshal(raw, &p) + } + return respond(taskResult{TaskID: p.TaskID, Status: "running"}) + case "tasks/cancel": + var p struct { + TaskID string `json:"taskId"` + } + if req.Params != nil { + raw, _ := json.Marshal(req.Params) + _ = json.Unmarshal(raw, &p) + } + delete(fakeTaskInputs(), p.TaskID) + return respond(map[string]any{"ok": true}) default: return fail(-32601, "method not found: "+req.Method) } } +// fakeTaskInputs 是 need_input 工具的任务输入存储（进程内）。 +func fakeTaskInputs() map[string]string { + return taskInputStore +} + +var taskInputStore = map[string]string{} + func fakeServerTools() []toolInfo { return []toolInfo{ { @@ -96,6 +161,16 @@ func fakeServerTools() []toolInfo { Description: "Responds after a delay", InputSchema: json.RawMessage(`{"type":"object","properties":{}}`), }, + { + Name: "need_input", + Description: "Requires user input (2026-07-28 resultType)", + InputSchema: json.RawMessage(`{"type":"object","properties":{}}`), + }, + { + Name: "legacy_ask", + Description: "Requires user input (legacy content block)", + // 故意无 InputSchema：触发 manager 的合成默认 schema 路径。 + }, } } diff --git a/runtime/mcp/http.go b/runtime/mcp/http.go index a12ccab..8e68ca1 100644 --- a/runtime/mcp/http.go +++ b/runtime/mcp/http.go @@ -21,6 +21,8 @@ type httpClient struct { sessionID string mu sync.Mutex nextID int64 + // negotiated 握手后写入（atomic.Value 存 string）。 + negotiated atomic.Value } // DialHTTP 建立 HTTP 客户端并完成 initialize 握手。 @@ -51,6 +53,9 @@ func (c *httpClient) handshake(ctx context.Context) error { ProtocolVersion: ProtocolVersion, ClientInfo: clientInfo{Name: "agent-native", Version: "1"}, }, &res) + if err == nil { + c.negotiated.Store(res.ProtocolVersion) + } return err } @@ -60,9 +65,32 @@ func (c *httpClient) Initialize(ctx context.Context) (initializeResult, error) { ProtocolVersion: ProtocolVersion, ClientInfo: clientInfo{Name: "agent-native", Version: "1"}, }, &res) + if err == nil { + c.negotiated.Store(res.ProtocolVersion) + } return res, err } +// NegotiatedVersion 是握手后服务端确认的协议版本（未握手为空）。 +func (c *httpClient) NegotiatedVersion() string { return c.negotiated.Load().(string) } + +// TasksGet 查询长任务状态（2026-07-28）。 +func (c *httpClient) TasksGet(ctx context.Context, taskID string) (taskResult, error) { + var res taskResult + err := c.roundTrip(ctx, "tasks/get", map[string]string{"taskId": taskID}, &res) + return res, err +} + +// TasksUpdate 向长任务投递用户输入（2026-07-28）。 +func (c *httpClient) TasksUpdate(ctx context.Context, taskID, input string) error { + return c.roundTrip(ctx, "tasks/update", map[string]string{"taskId": taskID, "input": input}, nil) +} + +// TasksCancel 取消长任务（2026-07-28）。 +func (c *httpClient) TasksCancel(ctx context.Context, taskID string) error { + return c.roundTrip(ctx, "tasks/cancel", map[string]string{"taskId": taskID}, nil) +} + func (c *httpClient) ListTools(ctx context.Context) ([]toolInfo, error) { var out []toolInfo cursor := "" diff --git a/runtime/mcp/manager.go b/runtime/mcp/manager.go index e25f98c..ec3a7e4 100644 --- a/runtime/mcp/manager.go +++ b/runtime/mcp/manager.go @@ -173,7 +173,7 @@ func (m *Manager) registerServer(ctx context.Context, reg *tools.Registry, cfg S } schema := ti.InputSchema if len(schema) == 0 { - schema = json.RawMessage(`{"type":"object","properties":{}}`) + schema = json.RawMessage(`{"$schema":"https://json-schema.org/draft/2020-12/schema","type":"object","properties":{}}`) } reg.Register(&bridgeTool{ client: c, @@ -251,7 +251,11 @@ func (t *bridgeTool) Execute(ctx context.Context, args json.RawMessage) (tools.R if err != nil { return tools.Result{Content: fmt.Sprintf("mcp tool %s failed: %v", t.declName, err), IsError: true}, nil } - if question, ok := inputQuestion(res); ok { + // input_required：2026-07-28 协议级结果（取应答后先 tasks/update 再 + // 用原参数重试 tools/call）；旧版内容块路径保持原样。 + if question, ok := inputQuestion(res, t.client.NegotiatedVersion()); ok && res.ResultType == "input_required" && SupportsResultType(t.client.NegotiatedVersion()) { + return t.answerAndRetry(ctx, args, res, question) + } else if ok { ask := askFrom(ctx) if ask == nil { return tools.Result{Content: "需要用户输入：" + question, IsError: true}, nil @@ -279,8 +283,58 @@ func (t *bridgeTool) Execute(ctx context.Context, args json.RawMessage) (tools.R return tools.Result{Content: out}, nil } +// maxInputRounds 是 input_required 应答重试的上限（防服务端循环要输入）。 +const maxInputRounds = 3 + +// answerAndRetry 处理 2026-07-28 的协议级 input_required：经 ask 端口取 +// 应答，TaskID 非空时先 tasks/update 投递，然后用原参数重试 tools/call； +// 服务端连续要输入最多 maxInputRounds 轮。 +func (t *bridgeTool) answerAndRetry(ctx context.Context, args json.RawMessage, res callToolResult, question string) (tools.Result, error) { + for round := 0; round < maxInputRounds; round++ { + ask := askFrom(ctx) + if ask == nil { + return tools.Result{Content: "需要用户输入：" + question, IsError: true}, nil + } + answer, aerr := ask(ctx, question) + if aerr != nil { + return tools.Result{Content: aerr.Error(), IsError: true}, nil + } + if res.TaskID != "" { + if uerr := t.client.TasksUpdate(ctx, res.TaskID, answer); uerr != nil { + return tools.Result{Content: fmt.Sprintf("mcp task update failed: %v", uerr), IsError: true}, nil + } + } + var err error + res, err = t.call(ctx, args) + if err != nil { + return tools.Result{Content: fmt.Sprintf("mcp tool %s failed: %v", t.declName, err), IsError: true}, nil + } + q, ok := inputQuestion(res, t.client.NegotiatedVersion()) + if !ok || res.ResultType != "input_required" || !SupportsResultType(t.client.NegotiatedVersion()) { + var b strings.Builder + for _, blk := range res.Content { + if blk.Type == "text" && blk.Text != "" { + b.WriteString(blk.Text) + b.WriteString("\n") + } + } + out := b.String() + if res.IsError { + return tools.Result{Content: out, IsError: true}, nil + } + return tools.Result{Content: strings.TrimSpace(out)}, nil + } + question = q + } + return tools.Result{Content: "需要用户输入：" + question, IsError: true}, nil +} + func (t *bridgeTool) call(ctx context.Context, args json.RawMessage) (callToolResult, error) { callCtx, cancel := context.WithTimeout(ctx, requestTimeout(t.serverCfg)) defer cancel() - return t.client.CallTool(callCtx, t.toolName, args) + res, err := t.client.CallTool(callCtx, t.toolName, args) + if err == nil && res.TaskID != "" { + trackerFrom(ctx).note(t.client, res.TaskID) + } + return res, err } diff --git a/runtime/mcp/mcp_test.go b/runtime/mcp/mcp_test.go index 8ce52e2..5c8121f 100644 --- a/runtime/mcp/mcp_test.go +++ b/runtime/mcp/mcp_test.go @@ -45,7 +45,7 @@ func TestStdioClientHandshakeListAndCall(t *testing.T) { defer c.Close() list, err := c.ListTools(ctx) - if err != nil || len(list) != 3 || list[0].Name != "echo" { + if err != nil || len(list) != 5 || list[0].Name != "echo" { t.Fatalf("list failed: %v %+v", err, list) } res, err := c.CallTool(ctx, "echo", json.RawMessage(`{"text":"hi"}`)) @@ -133,7 +133,7 @@ func TestHTTPClientJSONMode(t *testing.T) { t.Fatalf("session id must be captured, got %q", c.sessionID) } list, err := c.ListTools(ctx) - if err != nil || len(list) != 3 { + if err != nil || len(list) != 5 { t.Fatalf("list failed: %v %+v", err, list) } res, err := c.CallTool(ctx, "echo", json.RawMessage(`{"text":"net"}`)) @@ -159,7 +159,7 @@ func TestHTTPClientSSEMode(t *testing.T) { } defer c.Close() list, err := c.ListTools(ctx) - if err != nil || len(list) != 3 { + if err != nil || len(list) != 5 { t.Fatalf("sse list failed: %v %+v", err, list) } } @@ -361,8 +361,14 @@ type askClient struct{} func (askClient) Initialize(context.Context) (initializeResult, error) { return initializeResult{}, nil } +func (askClient) NegotiatedVersion() string { return LegacyProtocolVersion } func (askClient) ListTools(context.Context) ([]toolInfo, error) { return nil, nil } func (askClient) Close() error { return nil } +func (askClient) TasksGet(context.Context, string) (taskResult, error) { + return taskResult{}, nil +} +func (askClient) TasksUpdate(context.Context, string, string) error { return nil } +func (askClient) TasksCancel(context.Context, string) error { return nil } func (askClient) CallTool(context.Context, string, json.RawMessage) (callToolResult, error) { return callToolResult{Content: []contentBlock{{Type: "input_required", Text: "选一个颜色"}}}, nil } @@ -378,3 +384,150 @@ func TestInputRequiredUsesAsk(t *testing.T) { t.Fatalf("ask result %+v %v", res, err) } } + +// ---- N2 WP3：2026-07-28 对齐 ---- + +// 协议级 input_required：ask 取应答 → tasks/update 投递 → 原参数重试得终结果。 +func TestInputRequiredResultTypeRetriesOriginalCall(t *testing.T) { + mgr, err := NewManager([]ServerConfig{fakeStdioConfig(t)}) + if err != nil { + t.Fatal(err) + } + defer mgr.Close() + reg := tools.NewRegistry("") + if err := mgr.RegisterInto(context.Background(), reg); err != nil { + t.Fatal(err) + } + var questions []string + ctx := WithAsk(context.Background(), func(_ context.Context, q string) (string, error) { + questions = append(questions, q) + return "蓝色", nil + }) + res := reg.Execute(ctx, "mcp__fake__need_input", json.RawMessage(`{}`)) + if res.IsError || res.Content != "答案已收到: 蓝色" { + t.Fatalf("final result: %+v", res) + } + if len(questions) != 1 || questions[0] != "要答案" { + t.Fatalf("ask questions: %v", questions) + } +} + +// 协商回落：服务端只讲 2024-11-05 时，协议级路径不启用，旧内容块路径不变。 +func TestNegotiatedVersionFallbackKeepsLegacyPath(t *testing.T) { + cfg := fakeStdioConfig(t) + cfg.Env = append(cfg.Env, "AGENT_NATIVE_TEST_MCP_VERSION="+LegacyProtocolVersion) + mgr, err := NewManager([]ServerConfig{cfg}) + if err != nil { + t.Fatal(err) + } + defer mgr.Close() + reg := tools.NewRegistry("") + if err := mgr.RegisterInto(context.Background(), reg); err != nil { + t.Fatal(err) + } + ctx := WithAsk(context.Background(), func(_ context.Context, q string) (string, error) { + return "蓝色", nil + }) + res := reg.Execute(ctx, "mcp__fake__legacy_ask", json.RawMessage(`{}`)) + if res.IsError || res.Content != "蓝色" { + t.Fatalf("legacy path: %+v", res) + } +} + +// tasks 客户端面：get 回状态、cancel 清除任务输入。 +func TestTasksGetAndCancelRoundTrip(t *testing.T) { + cfg := fakeStdioConfig(t) + ctx := context.Background() + c, err := DialStdio(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer c.Close() + if got := c.NegotiatedVersion(); !SupportsResultType(got) { + t.Fatalf("negotiated = %q, want >= 2026-07-28", got) + } + if err := c.TasksUpdate(ctx, "task-x", "输入"); err != nil { + t.Fatal(err) + } + tr, err := c.TasksGet(ctx, "task-x") + if err != nil || tr.TaskID != "task-x" || tr.Status != "running" { + t.Fatalf("tasks/get: %+v %v", tr, err) + } + if err := c.TasksCancel(ctx, "task-x"); err != nil { + t.Fatal(err) + } +} + +// 合成默认 schema 声明 JSON Schema 2020-12。 +func TestDefaultSchemaDeclares202012(t *testing.T) { + mgr, err := NewManager([]ServerConfig{fakeStdioConfig(t)}) + if err != nil { + t.Fatal(err) + } + defer mgr.Close() + reg := tools.NewRegistry("") + if err := mgr.RegisterInto(context.Background(), reg); err != nil { + t.Fatal(err) + } + for _, d := range reg.Decls() { + if d.Name == "mcp__fake__legacy_ask" { + // legacy_ask 在 fake 里无 InputSchema → 走合成默认 + if !strings.Contains(string(d.Parameters), "2020-12") { + t.Fatalf("默认 schema 未声明 2020-12：%s", d.Parameters) + } + return + } + } + t.Fatal("mcp__fake__legacy_ask not registered") +} + +// 任务跟踪器：记录 → 按服务端分组取消 → 清空。 +func TestTaskTrackerCancelAllPerRun(t *testing.T) { + cfg := fakeStdioConfig(t) + ctx := context.Background() + c, err := DialStdio(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer c.Close() + tr := &TaskTracker{seen: map[Client]map[string]struct{}{}} + if err := c.TasksUpdate(ctx, "task-a", "1"); err != nil { + t.Fatal(err) + } + if err := c.TasksUpdate(ctx, "task-b", "2"); err != nil { + t.Fatal(err) + } + tr.note(c, "task-a") + tr.note(c, "task-b") + tr.note(c, "task-a") // 幂等 + if n := tr.CancelAll(ctx); n != 2 { + t.Fatalf("CancelAll = %d, want 2", n) + } + if n := tr.CancelAll(ctx); n != 0 { + t.Fatalf("第二次 CancelAll 应为 0，得到 %d", n) + } +} + +// 端到端：经注册表执行 input_required 工具的 run 结束时，task-1 被取消。 +func TestRunEndCancelsSeenTasks(t *testing.T) { + cfg := fakeStdioConfig(t) + mgr, err := NewManager([]ServerConfig{cfg}) + if err != nil { + t.Fatal(err) + } + defer mgr.Close() + reg := tools.NewRegistry("") + if err := mgr.RegisterInto(context.Background(), reg); err != nil { + t.Fatal(err) + } + ctx, tr := WithTaskTracker(context.Background()) + ctx = WithAsk(ctx, func(_ context.Context, q string) (string, error) { return "蓝", nil }) + res := reg.Execute(ctx, "mcp__fake__need_input", json.RawMessage(`{}`)) + if res.IsError { + t.Fatalf("execute: %+v", res) + } + // 模仿 adapter run 结束：取消本 run 见过的任务（task-1 已被记录）。 + if n := tr.CancelAll(context.Background()); n != 1 { + t.Fatalf("CancelAll = %d, want 1（task-1）", n) + } +} diff --git a/runtime/mcp/protocol.go b/runtime/mcp/protocol.go index bd4ea15..66f9b27 100644 --- a/runtime/mcp/protocol.go +++ b/runtime/mcp/protocol.go @@ -12,8 +12,18 @@ import ( "time" ) -// ProtocolVersion 是本客户端声明的 MCP 协议版本。 -const ProtocolVersion = "2024-11-05" +// ProtocolVersion 是本客户端声明的 MCP 协议版本（2026-07-28 RC： +// resultType 判别、input_required 协议级结果、tasks/* 长任务面）。 +const ProtocolVersion = "2026-07-28" + +// LegacyProtocolVersion 是旧版协议（无 resultType/tasks），协商回落时用。 +const LegacyProtocolVersion = "2024-11-05" + +// SupportsResultType 报告协商版本是否支持 2026-07-28 的 resultType 语义 +// （ISO 日期版本按字典序可比）。 +func SupportsResultType(negotiated string) bool { + return negotiated >= "2026-07-28" +} // ServerConfig 声明一个 MCP server 接入点：Command（stdio）或 URL（HTTP）二选一。 type ServerConfig struct { @@ -138,6 +148,19 @@ type callToolParams struct { type callToolResult struct { Content []contentBlock `json:"content"` IsError bool `json:"isError,omitempty"` + // ResultType 是 2026-07-28 的协议级结果判别（空 = 普通结果）。 + // "input_required" 表示服务端需要用户输入：客户端经 ask 端口取得 + // 应答后（有 TaskID 时先 tasks/update）用原参数重试 tools/call。 + ResultType string `json:"resultType,omitempty"` + TaskID string `json:"taskId,omitempty"` +} + +// taskResult 是 tasks/get 的返回（2026-07-28 长任务面；字段集按 RC 常用 +// 形态解析，未识别字段忽略）。 +type taskResult struct { + TaskID string `json:"taskId"` + Status string `json:"status"` + Detail string `json:"detail,omitempty"` } type contentBlock struct { @@ -148,7 +171,13 @@ type contentBlock struct { // Client 是一个已连接 MCP server 的最小面（initialize / tools/list / tools/call）。 type Client interface { Initialize(ctx context.Context) (initializeResult, error) + // NegotiatedVersion 是 initialize 后服务端确认的协议版本（未握手为空）。 + NegotiatedVersion() string ListTools(ctx context.Context) ([]toolInfo, error) CallTool(ctx context.Context, name string, args json.RawMessage) (callToolResult, error) + // TasksGet/Update/Cancel 是 2026-07-28 的长任务面。 + TasksGet(ctx context.Context, taskID string) (taskResult, error) + TasksUpdate(ctx context.Context, taskID, input string) error + TasksCancel(ctx context.Context, taskID string) error Close() error } diff --git a/runtime/mcp/r2_regression_test.go b/runtime/mcp/r2_regression_test.go index c8f4a46..f9182b9 100644 --- a/runtime/mcp/r2_regression_test.go +++ b/runtime/mcp/r2_regression_test.go @@ -18,8 +18,14 @@ type flakyClient struct { func (f *flakyClient) Initialize(context.Context) (initializeResult, error) { return initializeResult{}, nil } +func (f *flakyClient) NegotiatedVersion() string { return ProtocolVersion } func (f *flakyClient) ListTools(context.Context) ([]toolInfo, error) { return nil, nil } func (f *flakyClient) Close() error { return nil } +func (f *flakyClient) TasksGet(context.Context, string) (taskResult, error) { + return taskResult{}, nil +} +func (f *flakyClient) TasksUpdate(context.Context, string, string) error { return nil } +func (f *flakyClient) TasksCancel(context.Context, string) error { return nil } func (f *flakyClient) CallTool(context.Context, string, json.RawMessage) (callToolResult, error) { f.calls++ if f.fail { diff --git a/runtime/mcp/stdio.go b/runtime/mcp/stdio.go index 2da8d2c..f584132 100644 --- a/runtime/mcp/stdio.go +++ b/runtime/mcp/stdio.go @@ -25,6 +25,8 @@ type stdioClient struct { readers atomic.Int32 closed bool readerOnce sync.Once + // negotiated 握手后写入（atomic.Value 存 string）。 + negotiated atomic.Value } type stdioResult struct { @@ -75,6 +77,7 @@ func (c *stdioClient) handshake(ctx context.Context) error { }, &res); err != nil { return err } + c.negotiated.Store(res.ProtocolVersion) // initialized 通知无需响应。 return c.notify("notifications/initialized") } @@ -86,9 +89,32 @@ func (c *stdioClient) Initialize(ctx context.Context) (initializeResult, error) ProtocolVersion: ProtocolVersion, ClientInfo: clientInfo{Name: "agent-native", Version: "1"}, }, &res) + if err == nil { + c.negotiated.Store(res.ProtocolVersion) + } return res, err } +// NegotiatedVersion 是握手后服务端确认的协议版本（未握手为空）。 +func (c *stdioClient) NegotiatedVersion() string { return c.negotiated.Load().(string) } + +// TasksGet 查询长任务状态（2026-07-28）。 +func (c *stdioClient) TasksGet(ctx context.Context, taskID string) (taskResult, error) { + var res taskResult + err := c.roundTrip(ctx, "tasks/get", map[string]string{"taskId": taskID}, &res) + return res, err +} + +// TasksUpdate 向长任务投递用户输入（2026-07-28）。 +func (c *stdioClient) TasksUpdate(ctx context.Context, taskID, input string) error { + return c.roundTrip(ctx, "tasks/update", map[string]string{"taskId": taskID, "input": input}, nil) +} + +// TasksCancel 取消长任务（2026-07-28）。 +func (c *stdioClient) TasksCancel(ctx context.Context, taskID string) error { + return c.roundTrip(ctx, "tasks/cancel", map[string]string{"taskId": taskID}, nil) +} + func (c *stdioClient) ListTools(ctx context.Context) ([]toolInfo, error) { var out []toolInfo cursor := "" diff --git a/runtime/mcp/tasks.go b/runtime/mcp/tasks.go new file mode 100644 index 0000000..d08357b --- /dev/null +++ b/runtime/mcp/tasks.go @@ -0,0 +1,65 @@ +package mcp + +import ( + "context" + "sync" +) + +// TaskTracker 记录一次 run 内 tools/call 结果里出现过的长任务（按服务端 +// 客户端分组），供 run 结束时对这些任务发 tasks/cancel（2026-07-28 最小 +// 产品映射：取消只覆盖本 run 自己的任务，不影响共享 Manager 上其他 run）。 +type TaskTracker struct { + mu sync.Mutex + seen map[Client]map[string]struct{} +} + +type trackerKey struct{} + +// WithTaskTracker 把任务跟踪器放进上下文（bridgeTool 记录 TaskID）。 +func WithTaskTracker(ctx context.Context) (context.Context, *TaskTracker) { + tr := &TaskTracker{seen: map[Client]map[string]struct{}{}} + return context.WithValue(ctx, trackerKey{}, tr), tr +} + +func trackerFrom(ctx context.Context) *TaskTracker { + if ctx == nil { + return nil + } + tr, _ := ctx.Value(trackerKey{}).(*TaskTracker) + return tr +} + +func (t *TaskTracker) note(c Client, taskID string) { + if t == nil || c == nil || taskID == "" { + return + } + t.mu.Lock() + set := t.seen[c] + if set == nil { + set = map[string]struct{}{} + t.seen[c] = set + } + set[taskID] = struct{}{} + t.mu.Unlock() +} + +// CancelAll 对记录的任务逐一向其服务端发 tasks/cancel（尽力而为， +// 首个错误后继续）。返回实际发出取消的任务数。 +func (t *TaskTracker) CancelAll(ctx context.Context) int { + if t == nil { + return 0 + } + t.mu.Lock() + seen := t.seen + t.seen = map[Client]map[string]struct{}{} + t.mu.Unlock() + n := 0 + for c, ids := range seen { + for id := range ids { + if err := c.TasksCancel(ctx, id); err == nil { + n++ + } + } + } + return n +} -- 2.54.0