weichen/Agent-Native
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align the MCP client with 2026-07-28: resultType, input_required retry, tasks surface (N2 WP3) #27

Merged
weichen merged 1 commits from feature/n2-mcp-wp3 into main 2026-10-07 13:40:14 +00:00
Conversation 0 Commits 1 Files Changed 10
+463 -10
10 changed files with 463 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adapters/native/backend.go
+6
View File
@@ -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)
runtime/mcp/ask.go
+12 -1
View File
@@ -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
runtime/mcp/fakeserver.go
+76 -1
View File
@@ -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 路径。
},
}
}
runtime/mcp/http.go
+28
View File
@@ -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 := ""
runtime/mcp/manager.go
+57 -3
View File
@@ -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
}
runtime/mcp/mcp_test.go
+156 -3
View File
@@ -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)
}
}
runtime/mcp/protocol.go
+31 -2
View File
@@ -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
}
runtime/mcp/r2_regression_test.go
+6
View File
@@ -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 {
runtime/mcp/stdio.go
+26
View File
@@ -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 := ""
runtime/mcp/tasks.go
+65
View File
@@ -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
}