diff --git a/go/adk/pkg/mcp/mcp_ui.go b/go/adk/pkg/mcp/mcp_ui.go index 404a9831e..f28122012 100644 --- a/go/adk/pkg/mcp/mcp_ui.go +++ b/go/adk/pkg/mcp/mcp_ui.go @@ -19,7 +19,9 @@ package mcp import ( "context" "fmt" + "sync" + "github.com/kagent-dev/kagent/go/pkg/logging" mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp" adkagent "google.golang.org/adk/v2/agent" "google.golang.org/adk/v2/tool" @@ -53,14 +55,58 @@ type mcpAppToolset struct { // results render as interactive MCP App (UI) widgets (the tool declares a // `_meta.ui.resourceUri`). Used as a set, so only key presence is meaningful. appToolNames map[string]bool + // serverURL identifies the backend in degradation logs. + serverURL string + + mu sync.Mutex + // lastTools is the tool list from the most recent successful ListTools. + // The MCP toolset lists tools lazily on every invocation, so a backend + // that goes away mid-conversation would otherwise fail every turn of + // every agent that references it. Serving the last known list instead + // makes a dead backend behave like a set of failing tools: the model + // still sees them and receives a tool error when it calls one. + lastTools []tool.Tool + // listed records whether lastTools holds a real result, so an empty + // list from the server is not confused with "never reached". + listed bool } func (m *mcpAppToolset) Name() string { return m.inner.Name() } +// Tools returns the server's model-visible tools. When the server cannot be +// listed, the failure is logged at error level with the server URL and the +// last successfully listed tools (or none, if the server was never reachable) +// are returned so the agent keeps running. Only a cancelled or expired +// invocation context is still surfaced as an error. func (m *mcpAppToolset) Tools(ctx adkagent.ReadonlyContext) ([]tool.Tool, error) { - return m.inner.Tools(ctx) + tools, err := m.inner.Tools(ctx) + if err == nil { + m.mu.Lock() + m.lastTools = append([]tool.Tool(nil), tools...) + m.listed = true + m.mu.Unlock() + return tools, nil + } + if ctx.Err() != nil { + return nil, err + } + + m.mu.Lock() + cached := append([]tool.Tool(nil), m.lastTools...) + listed := m.listed + m.mu.Unlock() + + log := logging.FromContext(ctx) + if listed { + log.ErrorContext(ctx, "MCP server unreachable; continuing with the last known tool list", + "url", m.serverURL, "toolset", m.inner.Name(), "tools", len(cached), "error", err) + } else { + log.ErrorContext(ctx, "MCP server unreachable and its tools were never listed; continuing without them", + "url", m.serverURL, "toolset", m.inner.Name(), "error", err) + } + return cached, nil } // MCPAppToolNamesFromToolsets returns the union of MCP App-capable tool names diff --git a/go/adk/pkg/mcp/mcp_ui_degrade_test.go b/go/adk/pkg/mcp/mcp_ui_degrade_test.go new file mode 100644 index 000000000..a8e14a01b --- /dev/null +++ b/go/adk/pkg/mcp/mcp_ui_degrade_test.go @@ -0,0 +1,132 @@ +package mcp + +import ( + "bytes" + "context" + "log/slog" + "net" + "net/http" + "strings" + "testing" + + mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp" + "google.golang.org/adk/v2/tool" + + "github.com/kagent-dev/kagent/go/pkg/logging" +) + +// startWeatherMCPServer serves a single getWeather tool on a fresh port and +// returns its URL plus a function that shuts it down. +func startWeatherMCPServer(t *testing.T) (string, func()) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen() error = %v", err) + } + mcpServer := mcpsdk.NewServer(&mcpsdk.Implementation{Name: "weather-test", Version: "1.0.0"}, nil) + mcpsdk.AddTool(mcpServer, &mcpsdk.Tool{Name: "getWeather"}, func(context.Context, *mcpsdk.CallToolRequest, map[string]any) (*mcpsdk.CallToolResult, map[string]any, error) { + return nil, map[string]any{"weather": "sunny"}, nil + }) + httpServer := &http.Server{Handler: mcpsdk.NewStreamableHTTPHandler(func(*http.Request) *mcpsdk.Server { + return mcpServer + }, nil)} + go func() { _ = httpServer.Serve(listener) }() + stop := func() { _ = httpServer.Close() } + t.Cleanup(stop) + return "http://" + listener.Addr().String(), stop +} + +// loggingContext returns a ReadonlyContext whose logger writes JSON to buf. +func loggingContext(t *testing.T, buf *bytes.Buffer) testReadonlyContext { + t.Helper() + logger := slog.New(slog.NewJSONHandler(buf, nil)) + return testReadonlyContext{Context: logging.IntoContext(t.Context(), logger)} +} + +func toolNames(tools []tool.Tool) []string { + names := make([]string, 0, len(tools)) + for _, tl := range tools { + names = append(names, tl.Name()) + } + return names +} + +func TestMCPAppToolsetServesLastKnownToolsWhenServerDies(t *testing.T) { + serverURL, stop := startWeatherMCPServer(t) + + toolset, err := initializeToolSet(t.Context(), mcpServerParams{URL: serverURL, ServerType: "http"}, nil) + if err != nil { + t.Fatalf("initializeToolSet() error = %v", err) + } + + var logs bytes.Buffer + ctx := loggingContext(t, &logs) + + tools, err := toolset.Tools(ctx) + if err != nil { + t.Fatalf("Tools() with live server error = %v", err) + } + if got := toolNames(tools); len(got) != 1 || got[0] != "getWeather" { + t.Fatalf("Tools() with live server = %v, want [getWeather]", got) + } + if logs.Len() != 0 { + t.Fatalf("unexpected log output with live server: %s", logs.String()) + } + + stop() + + tools, err = toolset.Tools(ctx) + if err != nil { + t.Fatalf("Tools() after server died error = %v, want degraded success", err) + } + if got := toolNames(tools); len(got) != 1 || got[0] != "getWeather" { + t.Fatalf("Tools() after server died = %v, want last known [getWeather]", got) + } + out := logs.String() + if !strings.Contains(out, `"level":"ERROR"`) || !strings.Contains(out, serverURL) || !strings.Contains(out, "last known tool list") { + t.Fatalf("expected an error-level log naming %s and the degraded mode, got: %s", serverURL, out) + } +} + +func TestMCPAppToolsetReturnsNoToolsWhenServerNeverReachable(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("net.Listen() error = %v", err) + } + serverURL := "http://" + listener.Addr().String() + if err := listener.Close(); err != nil { + t.Fatalf("listener.Close() error = %v", err) + } + + toolset, err := initializeToolSet(t.Context(), mcpServerParams{URL: serverURL, ServerType: "http"}, nil) + if err != nil { + t.Fatalf("initializeToolSet() error = %v, want lazy fallback", err) + } + + var logs bytes.Buffer + tools, err := toolset.Tools(loggingContext(t, &logs)) + if err != nil { + t.Fatalf("Tools() against dead server error = %v, want degraded success", err) + } + if len(tools) != 0 { + t.Fatalf("Tools() against dead server = %v, want none", toolNames(tools)) + } + out := logs.String() + if !strings.Contains(out, `"level":"ERROR"`) || !strings.Contains(out, serverURL) || !strings.Contains(out, "never listed") { + t.Fatalf("expected an error-level log naming %s, got: %s", serverURL, out) + } +} + +func TestMCPAppToolsetPropagatesCancelledContext(t *testing.T) { + serverURL, _ := startWeatherMCPServer(t) + toolset, err := initializeToolSet(t.Context(), mcpServerParams{URL: serverURL, ServerType: "http"}, nil) + if err != nil { + t.Fatalf("initializeToolSet() error = %v", err) + } + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + if _, err := toolset.Tools(testReadonlyContext{Context: cancelled}); err == nil { + t.Fatal("Tools() with cancelled context error = nil, want the cancellation surfaced") + } +} diff --git a/go/adk/pkg/mcp/registry.go b/go/adk/pkg/mcp/registry.go index 3872d22e8..8d9a17ac7 100644 --- a/go/adk/pkg/mcp/registry.go +++ b/go/adk/pkg/mcp/registry.go @@ -385,5 +385,5 @@ func initializeToolSet(ctx context.Context, params mcpServerParams, toolFilter m if toolPredicate != nil { visibleTools = tool.FilterToolset(toolset, toolPredicate) } - return &mcpAppToolset{inner: visibleTools, appToolNames: appToolNames}, nil + return &mcpAppToolset{inner: visibleTools, appToolNames: appToolNames, serverURL: params.URL}, nil }