diff --git a/internal/app/bootstrap_test.go b/internal/app/bootstrap_test.go index a81903c0..981cba06 100644 --- a/internal/app/bootstrap_test.go +++ b/internal/app/bootstrap_test.go @@ -271,6 +271,9 @@ func TestDefaultRegisterMCPStdioServerRefreshFailure(t *testing.T) { if !strings.Contains(strings.ToLower(err.Error()), "list tools failed") { t.Fatalf("unexpected error: %v", err) } + if snapshots := registry.Snapshot(); len(snapshots) != 0 { + t.Fatalf("expected failed registration to rollback server, got %+v", snapshots) + } } func TestBuildToolRegistryIncludesMCPFromConfig(t *testing.T) { @@ -434,6 +437,53 @@ func TestBuildMCPRegistryRegisterError(t *testing.T) { } } +func TestBuildMCPRegistryRollbackRegisteredServersOnFailure(t *testing.T) { + t.Parallel() + + cfg := config.Default().Clone() + cfg.Workdir = t.TempDir() + cfg.Tools.MCP.Servers = []config.MCPServerConfig{ + {ID: "docs", Enabled: true, Source: "stdio"}, + {ID: "search", Enabled: true, Source: "stdio"}, + } + + closedByServer := map[string]*bool{ + "docs": new(bool), + "search": new(bool), + } + + originalRegister := registerMCPStdioServer + t.Cleanup(func() { registerMCPStdioServer = originalRegister }) + registerMCPStdioServer = func(registry *mcp.Registry, cfg config.Config, server config.MCPServerConfig) error { + client := &closeableStubMCPServerClient{closed: closedByServer[strings.TrimSpace(server.ID)]} + if err := registry.RegisterServer(server.ID, "stdio", server.Version, client); err != nil { + return err + } + if strings.EqualFold(strings.TrimSpace(server.ID), "search") { + return errors.New("search register failed") + } + return nil + } + + registry, err := buildMCPRegistry(cfg) + if err == nil || !strings.Contains(err.Error(), "search register failed") { + t.Fatalf("expected wrapped register error, got %v", err) + } + if registry != nil { + t.Fatalf("expected nil registry on build failure") + } + if !*closedByServer["docs"] || !*closedByServer["search"] { + t.Fatalf("expected rollback to close all registered servers, got %+v", closedByServer) + } +} + +func TestRollbackMCPServersBoundaries(t *testing.T) { + t.Parallel() + + rollbackMCPServers(nil, []string{"docs"}) + rollbackMCPServers(mcp.NewRegistry(), nil) +} + func TestInitialMCPRefreshTimeoutAndDurationConversion(t *testing.T) { t.Parallel() @@ -614,6 +664,29 @@ func (s *stubMCPServerClient) HealthCheck(ctx context.Context) error { return nil } +type closeableStubMCPServerClient struct { + closed *bool +} + +func (s *closeableStubMCPServerClient) ListTools(ctx context.Context) ([]mcp.ToolDescriptor, error) { + return nil, nil +} + +func (s *closeableStubMCPServerClient) CallTool(ctx context.Context, toolName string, arguments []byte) (mcp.CallResult, error) { + return mcp.CallResult{}, nil +} + +func (s *closeableStubMCPServerClient) HealthCheck(ctx context.Context) error { + return nil +} + +func (s *closeableStubMCPServerClient) Close() error { + if s.closed != nil { + *s.closed = true + } + return nil +} + func TestHelperProcessAppMCPStdioServer(t *testing.T) { if os.Getenv("GO_WANT_APP_MCP_STDIO_HELPER") != "1" { return diff --git a/internal/app/mcp_bootstrap.go b/internal/app/mcp_bootstrap.go index 27982e14..d0253b0c 100644 --- a/internal/app/mcp_bootstrap.go +++ b/internal/app/mcp_bootstrap.go @@ -23,6 +23,7 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) { registry := mcp.NewRegistry() enabledCount := 0 + registeredServerIDs := make([]string, 0, len(cfg.Tools.MCP.Servers)) for index := range cfg.Tools.MCP.Servers { server := cfg.Tools.MCP.Servers[index] if !server.Enabled { @@ -33,9 +34,12 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) { switch strings.ToLower(strings.TrimSpace(server.Source)) { case "", "stdio": if err := registerMCPStdioServer(registry, cfg, server); err != nil { + rollbackMCPServers(registry, append(registeredServerIDs, strings.TrimSpace(server.ID))) return nil, fmt.Errorf("app: register mcp server %q: %w", strings.TrimSpace(server.ID), err) } + registeredServerIDs = append(registeredServerIDs, strings.TrimSpace(server.ID)) default: + rollbackMCPServers(registry, registeredServerIDs) return nil, fmt.Errorf("app: unsupported mcp source %q", server.Source) } } @@ -46,6 +50,16 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) { return registry, nil } +// rollbackMCPServers 在批量注册失败时回滚已注册 server，避免残留子进程或脏状态。 +func rollbackMCPServers(registry *mcp.Registry, serverIDs []string) { + if registry == nil || len(serverIDs) == 0 { + return + } + for index := len(serverIDs) - 1; index >= 0; index-- { + _ = registry.UnregisterServer(serverIDs[index]) + } +} + // defaultRegisterMCPStdioServer 创建 stdio client 并完成 server 注册与 tools 快照初始化。 func defaultRegisterMCPStdioServer(registry *mcp.Registry, cfg config.Config, server config.MCPServerConfig) error { env, err := resolveMCPServerEnv(server) @@ -79,6 +93,7 @@ func defaultRegisterMCPStdioServer(registry *mcp.Registry, cfg config.Config, se refreshCtx, cancel := context.WithTimeout(context.Background(), initialMCPRefreshTimeout(cfg)) defer cancel() if err := registry.RefreshServerTools(refreshCtx, serverID); err != nil { + _ = registry.UnregisterServer(serverID) return err } return nil diff --git a/internal/tools/mcp/adapter.go b/internal/tools/mcp/adapter.go index d9ecb470..a0082f3f 100644 --- a/internal/tools/mcp/adapter.go +++ b/internal/tools/mcp/adapter.go @@ -132,10 +132,22 @@ func ensureObjectSchema(schema map[string]any) map[string]any { } } - if strings.TrimSpace(fmt.Sprintf("%v", cloned["type"])) == "" { + rawType, hasType := cloned["type"] + trimmedType := strings.TrimSpace(fmt.Sprintf("%v", rawType)) + if !hasType || rawType == nil || trimmedType == "" || strings.EqualFold(trimmedType, "") { cloned["type"] = "object" + if _, ok := cloned["properties"].(map[string]any); !ok { + cloned["properties"] = map[string]any{} + } + return cloned + } + + if !strings.EqualFold(trimmedType, "object") { + cloned["type"] = "object" + cloned["properties"] = map[string]any{} + return cloned } - if _, ok := cloned["properties"]; !ok { + if _, ok := cloned["properties"].(map[string]any); !ok { cloned["properties"] = map[string]any{} } return cloned diff --git a/internal/tools/mcp/adapter_test.go b/internal/tools/mcp/adapter_test.go index 2b38691f..20d8a381 100644 --- a/internal/tools/mcp/adapter_test.go +++ b/internal/tools/mcp/adapter_test.go @@ -227,3 +227,61 @@ func TestAdapterEnsureObjectSchemaDefaults(t *testing.T) { t.Fatalf("expected properties object, got %+v", schema["properties"]) } } + +func TestAdapterEnsureObjectSchemaNormalizesInvalidType(t *testing.T) { + t.Parallel() + + registry := NewRegistry() + adapter, err := NewAdapter(registry, "docs", ToolDescriptor{ + Name: "search", + InputSchema: map[string]any{ + "type": "array", + "items": map[string]any{ + "type": "string", + }, + }, + }) + if err != nil { + t.Fatalf("NewAdapter() error = %v", err) + } + schema := adapter.Schema() + if schema["type"] != "object" { + t.Fatalf("expected normalized object type, got %v", schema["type"]) + } + if _, ok := schema["properties"].(map[string]any); !ok { + t.Fatalf("expected normalized properties object, got %+v", schema["properties"]) + } +} + +func TestAdapterEnsureObjectSchemaKeepsPropertiesWhenTypeMissing(t *testing.T) { + t.Parallel() + + registry := NewRegistry() + adapter, err := NewAdapter(registry, "docs", ToolDescriptor{ + Name: "search", + InputSchema: map[string]any{ + "properties": map[string]any{ + "q": map[string]any{"type": "string"}, + }, + }, + }) + if err != nil { + t.Fatalf("NewAdapter() error = %v", err) + } + + schema := adapter.Schema() + if schema["type"] != "object" { + t.Fatalf("expected type object, got %v", schema["type"]) + } + properties, ok := schema["properties"].(map[string]any) + if !ok { + t.Fatalf("expected properties object, got %+v", schema["properties"]) + } + querySchema, ok := properties["q"].(map[string]any) + if !ok { + t.Fatalf("expected q schema map, got %+v", properties["q"]) + } + if querySchema["type"] != "string" { + t.Fatalf("expected q type string, got %v", querySchema["type"]) + } +} diff --git a/internal/tools/mcp/registry.go b/internal/tools/mcp/registry.go index 490b5142..ea85e0cc 100644 --- a/internal/tools/mcp/registry.go +++ b/internal/tools/mcp/registry.go @@ -55,6 +55,11 @@ type ServerClient interface { HealthCheck(ctx context.Context) error } +// closeableServerClient 描述支持主动关闭资源的 MCP client 扩展能力。 +type closeableServerClient interface { + Close() error +} + type serverEntry struct { snapshot ServerSnapshot client ServerClient @@ -116,11 +121,14 @@ func (r *Registry) UnregisterServer(serverID string) bool { } r.mu.Lock() - defer r.mu.Unlock() - if _, exists := r.servers[normalizedID]; !exists { + entry, exists := r.servers[normalizedID] + if !exists { + r.mu.Unlock() return false } delete(r.servers, normalizedID) + r.mu.Unlock() + closeServerClient(entry.client) return true } @@ -337,3 +345,15 @@ func cloneAny(value any) any { return value } } + +// closeServerClient 在 server 注销时尽力释放 client 持有的底层资源。 +func closeServerClient(client ServerClient) { + if client == nil { + return + } + closeableClient, ok := client.(closeableServerClient) + if !ok { + return + } + _ = closeableClient.Close() +} diff --git a/internal/tools/mcp/registry_test.go b/internal/tools/mcp/registry_test.go index 30c5cd3f..249e2505 100644 --- a/internal/tools/mcp/registry_test.go +++ b/internal/tools/mcp/registry_test.go @@ -19,6 +19,16 @@ type stubServerClient struct { lastArguments []byte } +type closableStubServerClient struct { + stubServerClient + closed bool +} + +func (s *closableStubServerClient) Close() error { + s.closed = true + return nil +} + func (s *stubServerClient) ListTools(ctx context.Context) ([]ToolDescriptor, error) { if s.listErr != nil { return nil, s.listErr @@ -229,6 +239,35 @@ func TestRegistryRegisterAndUnregisterBoundaries(t *testing.T) { } } +func TestRegistryUnregisterServerClosesClient(t *testing.T) { + t.Parallel() + + registry := NewRegistry() + client := &closableStubServerClient{} + if err := registry.RegisterServer("docs", "stdio", "v1", client); err != nil { + t.Fatalf("register server: %v", err) + } + if !registry.UnregisterServer("docs") { + t.Fatalf("expected unregister success") + } + if !client.closed { + t.Fatalf("expected client to be closed on unregister") + } +} + +func TestCloseServerClientBoundaries(t *testing.T) { + t.Parallel() + + closeServerClient(nil) + closeServerClient(&stubServerClient{}) + + client := &closableStubServerClient{} + closeServerClient(client) + if !client.closed { + t.Fatalf("expected closeable client to be closed") + } +} + func TestRegistrySetServerStatusValidation(t *testing.T) { t.Parallel() diff --git a/internal/tools/mcp/stdio_client.go b/internal/tools/mcp/stdio_client.go index b6808d51..210b9698 100644 --- a/internal/tools/mcp/stdio_client.go +++ b/internal/tools/mcp/stdio_client.go @@ -799,35 +799,6 @@ func decodeCallResult(raw json.RawMessage) CallResult { } } - content := "" - switch typed := payload["content"].(type) { - case string: - content = strings.TrimSpace(typed) - case []any: - lines := make([]string, 0, len(typed)) - for _, item := range typed { - switch value := item.(type) { - case map[string]any: - text, _ := value["text"].(string) - if strings.TrimSpace(text) != "" { - lines = append(lines, strings.TrimSpace(text)) - } - case string: - if strings.TrimSpace(value) != "" { - lines = append(lines, strings.TrimSpace(value)) - } - } - } - content = strings.Join(lines, "\n") - default: - if typed != nil { - content = strings.TrimSpace(fmt.Sprintf("%v", typed)) - } - } - if content == "" { - content = "ok" - } - isError := false if value, ok := payload["isError"].(bool); ok { isError = value @@ -836,6 +807,15 @@ func decodeCallResult(raw json.RawMessage) CallResult { isError = isError || value } + content := decodeCallContent(payload["content"]) + if content == "" { + if isError { + content = "mcp tool returned empty error content" + } else { + content = "ok" + } + } + metadata := map[string]any{} for key, value := range payload { if key == "content" || key == "isError" || key == "is_error" { @@ -851,3 +831,47 @@ func decodeCallResult(raw json.RawMessage) CallResult { Metadata: metadata, } } + +// decodeCallContent 将 MCP tools/call 的 content 字段归一为可回灌文本，避免结构化内容丢失。 +func decodeCallContent(content any) string { + switch typed := content.(type) { + case string: + return strings.TrimSpace(typed) + case []any: + parts := make([]string, 0, len(typed)) + for _, item := range typed { + formatted := decodeCallContentItem(item) + if formatted != "" { + parts = append(parts, formatted) + } + } + return strings.Join(parts, "\n") + default: + return decodeCallContentItem(typed) + } +} + +// decodeCallContentItem 对单个 MCP content item 做兜底格式化，保留非文本对象信息。 +func decodeCallContentItem(item any) string { + switch typed := item.(type) { + case nil: + return "" + case string: + return strings.TrimSpace(typed) + case map[string]any: + if text, ok := typed["text"].(string); ok && strings.TrimSpace(text) != "" { + return strings.TrimSpace(text) + } + raw, err := json.Marshal(typed) + if err != nil { + return strings.TrimSpace(fmt.Sprintf("%v", typed)) + } + return strings.TrimSpace(string(raw)) + default: + raw, err := json.Marshal(typed) + if err != nil { + return strings.TrimSpace(fmt.Sprintf("%v", typed)) + } + return strings.TrimSpace(string(raw)) + } +} diff --git a/internal/tools/mcp/stdio_client_test.go b/internal/tools/mcp/stdio_client_test.go index 01d8d20b..07163406 100644 --- a/internal/tools/mcp/stdio_client_test.go +++ b/internal/tools/mcp/stdio_client_test.go @@ -355,9 +355,19 @@ func TestDecodeCallResultVariants(t *testing.T) { t.Fatalf("unexpected list content decode: %+v", result) } + result = decodeCallResult(json.RawMessage(`{"content":[{"type":"resource_link","uri":"https://example.com"},{"type":"image","mimeType":"image/png"}]}`)) + if !strings.Contains(result.Content, `"type":"resource_link"`) || !strings.Contains(result.Content, `"type":"image"`) { + t.Fatalf("expected non-text items to be preserved, got %q", result.Content) + } + result = decodeCallResult(json.RawMessage(`{"content":{"nested":"x"}}`)) - if result.Content == "" { - t.Fatalf("expected fallback string content") + if !strings.Contains(result.Content, `"nested":"x"`) { + t.Fatalf("expected structured map content, got %q", result.Content) + } + + result = decodeCallResult(json.RawMessage(`{"content":[],"isError":true}`)) + if !result.IsError || strings.Contains(strings.ToLower(result.Content), "ok") { + t.Fatalf("expected non-ok error fallback content, got %+v", result) } result = decodeCallResult(json.RawMessage(`not-json`)) @@ -369,6 +379,44 @@ func TestDecodeCallResultVariants(t *testing.T) { } } +func TestDecodeCallContentItemVariants(t *testing.T) { + t.Parallel() + + if got := decodeCallContentItem(nil); got != "" { + t.Fatalf("expected empty for nil, got %q", got) + } + if got := decodeCallContentItem(" text "); got != "text" { + t.Fatalf("expected trimmed text, got %q", got) + } + if got := decodeCallContentItem(map[string]any{"text": " hello "}); got != "hello" { + t.Fatalf("expected text extraction, got %q", got) + } + if got := decodeCallContentItem(map[string]any{"type": "resource_link", "uri": "https://example.com"}); !strings.Contains(got, `"type":"resource_link"`) { + t.Fatalf("expected json fallback for object item, got %q", got) + } + if got := decodeCallContentItem(123); got != "123" { + t.Fatalf("expected scalar json fallback, got %q", got) + } + if got := decodeCallContentItem(map[string]any{"bad": func() {}}); !strings.Contains(got, "bad") { + t.Fatalf("expected fmt fallback for non-marshalable map, got %q", got) + } + if got := decodeCallContentItem(func() {}); !strings.Contains(got, "0x") { + t.Fatalf("expected fmt fallback for non-marshalable scalar, got %q", got) + } +} + +func TestDecodeCallResultContentFallbackOK(t *testing.T) { + t.Parallel() + + result := decodeCallResult(json.RawMessage(`{"content":[]}`)) + if result.IsError { + t.Fatalf("expected non-error result") + } + if result.Content != "ok" { + t.Fatalf("expected ok fallback content, got %q", result.Content) + } +} + func TestFailAllPendingLocked(t *testing.T) { t.Parallel() diff --git a/internal/tools/registry.go b/internal/tools/registry.go index a392f629..3f7af5a3 100644 --- a/internal/tools/registry.go +++ b/internal/tools/registry.go @@ -173,6 +173,13 @@ func (r *Registry) Execute(ctx context.Context, input ToolCallInput) (ToolResult result = ApplyOutputLimit(result, DefaultOutputLimitBytes) return result, callErr } + if result.IsError { + if strings.TrimSpace(result.Content) == "" { + result.Content = FormatError(result.Name, "mcp tool returned isError=true", "") + } + result = ApplyOutputLimit(result, DefaultOutputLimitBytes) + return result, errors.New("mcp: tool returned error result") + } if result.Content == "" { result.Content = "ok" } diff --git a/internal/tools/registry_test.go b/internal/tools/registry_test.go index a26843e4..09529b75 100644 --- a/internal/tools/registry_test.go +++ b/internal/tools/registry_test.go @@ -413,6 +413,78 @@ func TestRegistryExecuteMCPCallErrorDoesNotReturnOK(t *testing.T) { } } +func TestRegistryExecuteMCPIsErrorResultDoesNotFallbackToOK(t *testing.T) { + t.Parallel() + + registry := NewRegistry() + mcpRegistry := mcp.NewRegistry() + if err := mcpRegistry.RegisterServer("docs", "stdio", "v1", &stubMCPClient{ + tools: []mcp.ToolDescriptor{ + {Name: "search", Description: "search docs", InputSchema: map[string]any{"type": "object"}}, + }, + callResult: mcp.CallResult{ + Content: "", + IsError: true, + }, + }); err != nil { + t.Fatalf("register mcp server: %v", err) + } + if err := mcpRegistry.RefreshServerTools(context.Background(), "docs"); err != nil { + t.Fatalf("refresh mcp tools: %v", err) + } + registry.SetMCPRegistry(mcpRegistry) + + result, err := registry.Execute(context.Background(), ToolCallInput{ + ID: "mcp-call-iserror", + Name: "mcp.docs.search", + }) + if err == nil { + t.Fatalf("expected mcp isError to return error") + } + if !result.IsError { + t.Fatalf("expected IsError true") + } + if strings.EqualFold(strings.TrimSpace(result.Content), "ok") || strings.TrimSpace(result.Content) == "" { + t.Fatalf("expected non-ok error content, got %q", result.Content) + } +} + +func TestRegistryExecuteMCPIsErrorWithContentKeepsContent(t *testing.T) { + t.Parallel() + + registry := NewRegistry() + mcpRegistry := mcp.NewRegistry() + if err := mcpRegistry.RegisterServer("docs", "stdio", "v1", &stubMCPClient{ + tools: []mcp.ToolDescriptor{ + {Name: "search", Description: "search docs", InputSchema: map[string]any{"type": "object"}}, + }, + callResult: mcp.CallResult{ + Content: "explicit mcp error", + IsError: true, + }, + }); err != nil { + t.Fatalf("register mcp server: %v", err) + } + if err := mcpRegistry.RefreshServerTools(context.Background(), "docs"); err != nil { + t.Fatalf("refresh mcp tools: %v", err) + } + registry.SetMCPRegistry(mcpRegistry) + + result, err := registry.Execute(context.Background(), ToolCallInput{ + ID: "mcp-call-iserror-content", + Name: "mcp.docs.search", + }) + if err == nil { + t.Fatalf("expected mcp isError to return error") + } + if !result.IsError { + t.Fatalf("expected IsError true") + } + if result.Content != "explicit mcp error" { + t.Fatalf("expected explicit error content preserved, got %q", result.Content) + } +} + func TestRegistrySupportsMCPToolAndHelpers(t *testing.T) { t.Parallel()