fix(adk): defer MCP tool discovery until invocation (v1.0.x backport)

docs/architecture/mcp.md

@@ -75,3 +75,30 @@ not protected: registering a name or URI core already uses replaces it.
 Authorization inside an added tool stays with the program that adds it.
 
 The implementation is in [`go/core/internal/mcp`](../../go/core/internal/mcp).
+
+## Go ADK runtime discovery
+
+The Go ADK connects to MCP servers and discovers tools during invocation
+preparation. Startup and Actor resume do not wait for HTTP or SSE connections
+or start stdio MCP processes. This also applies to subagents.
+
+Discovery filters app-only tools and applies configured allowlists before tools
+reach the model. The MCP SDK caches filtered responses and owns TTL expiry and
+catalog-notification invalidation. Cached tools remain filtered. Credentials,
+approvals, and tool-call authorization continue to apply.
+
+If discovery fails, that server's toolset stays hidden for the current A2A
+execution, including subsequent model steps and subagent calls. Healthy toolsets
+remain available. The next execution retries, including an approval continuation.
+There is no fallback to unclassified tools. Discovery failures are logged as
+warnings. Cancellation and tool-call errors retain their existing handling.
+
+MCP App results are compacted for the model while stored results remain intact
+for UI rendering. Removed app names remain recognized for historical compaction
+during the process lifetime. Tools reclassified as ordinary keep their full
+output, including historical results. A new process rediscovers app metadata
+during invocation preparation.
+
+Standalone runners can use `mcp.WithDiscoveryScope(ctx)` for each `runner.Run`
+to get the same omission behavior. Without it, discovery errors propagate.
+Controller-side catalog discovery used by the UI is independent.

go/adk/examples/oneshot/main.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/kagent-dev/kagent/go/adk/pkg/agent"
 	"github.com/kagent-dev/kagent/go/adk/pkg/config"
+	"github.com/kagent-dev/kagent/go/adk/pkg/mcp"
 	"github.com/kagent-dev/kagent/go/pkg/logging"
 )
 
@@ -106,7 +107,7 @@ func main() {
 		streamMode = adkagent.StreamingModeSSE
 	}
 
-	for ev, err := range r.Run(ctx, "user", sess.Session.ID(), content, adkagent.RunConfig{
+	for ev, err := range r.Run(mcp.WithDiscoveryScope(ctx), "user", sess.Session.ID(), content, adkagent.RunConfig{
 		StreamingMode: streamMode,
 	}) {
 		if err != nil {

go/adk/pkg/a2a/executor.go

@@ -22,6 +22,7 @@ import (
 	"google.golang.org/genai"
 
 	"github.com/kagent-dev/kagent/go/adk/pkg/auth"
+	"github.com/kagent-dev/kagent/go/adk/pkg/mcp"
 	"github.com/kagent-dev/kagent/go/adk/pkg/models"
 	"github.com/kagent-dev/kagent/go/adk/pkg/telemetry"
 	apia2a "github.com/kagent-dev/kagent/go/api/a2a"
@@ -264,6 +265,7 @@ func (e *KAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.ExecutorCon
 
 		ctx = context.WithValue(ctx, publicContextIDKey{}, reqCtx.ContextID)
 		ctx = withBearerToken(ctx)
+		ctx = mcp.WithDiscoveryScope(ctx)
 		// The synthetic ADK user ID is only a native session lookup key. Memory
 		// and outgoing credentials must receive only the passed-through caller.
 		ctx = auth.WithUserID(ctx, trustedUserID)

go/adk/pkg/a2a/executor_mcp_test.go

@@ -0,0 +1,118 @@
+package a2a
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"sync/atomic"
+	"testing"
+
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	"github.com/a2aproject/a2a-go/v2/a2asrv"
+	mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
+	"github.com/stretchr/testify/require"
+	adkagent "google.golang.org/adk/v2/agent"
+
+	"github.com/kagent-dev/kagent/go/adk/pkg/mcp"
+	configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1"
+)
+
+func TestKAgentExecutorScopesMCPDiscoveryFailures(t *testing.T) {
+	// Given a temporarily unavailable MCP catalog, when execution continues and
+	// later resumes with reused IDs, assert that omission lasts only one execution.
+	var unavailable atomic.Bool
+	unavailable.Store(true)
+	var listings atomic.Int32
+	server := mcpsdk.NewServer(&mcpsdk.Implementation{Name: "execution-discovery", Version: "1"}, &mcpsdk.ServerOptions{
+		SupportedProtocolVersions: []string{"2025-11-25"},
+	})
+	mcpsdk.AddTool(server, &mcpsdk.Tool{Name: "available_tool"}, func(context.Context, *mcpsdk.CallToolRequest, map[string]any) (*mcpsdk.CallToolResult, map[string]any, error) {
+		return nil, map[string]any{"result": "ok"}, nil
+	})
+	server.AddReceivingMiddleware(func(next mcpsdk.MethodHandler) mcpsdk.MethodHandler {
+		return func(ctx context.Context, method string, req mcpsdk.Request) (mcpsdk.Result, error) {
+			if method == "tools/list" {
+				listings.Add(1)
+				if unavailable.Load() {
+					return nil, errors.New("catalog temporarily unavailable")
+				}
+			}
+			return next(ctx, method, req)
+		}
+	})
+	httpServer := httptest.NewServer(mcpsdk.NewStreamableHTTPHandler(func(*http.Request) *mcpsdk.Server { return server }, &mcpsdk.StreamableHTTPOptions{Stateless: true}))
+	t.Cleanup(httpServer.Close)
+	healthy := mcpsdk.NewServer(&mcpsdk.Implementation{Name: "healthy-discovery"}, &mcpsdk.ServerOptions{SupportedProtocolVersions: []string{"2025-11-25"}})
+	mcpsdk.AddTool(healthy, &mcpsdk.Tool{Name: "healthy_tool"}, func(context.Context, *mcpsdk.CallToolRequest, map[string]any) (*mcpsdk.CallToolResult, map[string]any, error) {
+		return nil, map[string]any{"result": "healthy"}, nil
+	})
+	healthyHTTP := httptest.NewServer(mcpsdk.NewStreamableHTTPHandler(func(*http.Request) *mcpsdk.Server { return healthy }, &mcpsdk.StreamableHTTPOptions{Stateless: true}))
+	t.Cleanup(healthyHTTP.Close)
+	toolsets := mcp.CreateToolsets(t.Context(), []*configv1.ADKHTTPMCPServerConfig{{
+		Params: &configv1.ADKHTTPConnectionParams{Url: httpServer.URL},
+	}, {
+		Params: &configv1.ADKHTTPConnectionParams{Url: healthyHTTP.URL},
+	}}, nil, nil, false, nil)
+	require.Len(t, toolsets, 2)
+
+	var visible [][]string
+	builtin := &recordingExecutor{onExecute: func(ctx context.Context) {
+		readonly := &executionDiscoveryContext{StrictContextMock: adkagent.NewStrictContextMock(ctx)}
+		for range 2 {
+			var names []string
+			for _, toolset := range toolsets {
+				tools, err := toolset.Tools(readonly)
+				require.NoError(t, err, "discovery failures must leave the execution usable")
+				for _, tool := range tools {
+					names = append(names, tool.Name())
+				}
+			}
+			slices.Sort(names)
+			visible = append(visible, names)
+			unavailable.Store(false)
+		}
+	}}
+	executor := &KAgentExecutor{builtin: builtin, logger: slog.New(slog.DiscardHandler)}
+	request := &a2asrv.ExecutorContext{
+		TaskID: "reused-task", ContextID: "reused-session",
+		Message: a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("continue")),
+	}
+	for _, err := range executor.Execute(t.Context(), request) {
+		require.NoError(t, err)
+	}
+	require.Equal(t, [][]string{{"healthy_tool"}, {"healthy_tool"}}, visible, "recovery during an execution must not retry its omitted server")
+	require.EqualValues(t, 1, listings.Load())
+
+	visible = nil
+	for _, err := range executor.Execute(t.Context(), request) {
+		require.NoError(t, err)
+	}
+	require.Equal(t, [][]string{{"available_tool", "healthy_tool"}, {"available_tool", "healthy_tool"}}, visible, "a later execution with the same IDs must retry discovery")
+	require.Greater(t, listings.Load(), int32(1))
+
+	// Given prior successful discovery, when the server goes down again, assert
+	// that healthy tools remain available and a fresh execution retries recovery.
+	unavailable.Store(true)
+	before := listings.Load()
+	visible = nil
+	for _, err := range executor.Execute(t.Context(), request) {
+		require.NoError(t, err)
+	}
+	require.Equal(t, [][]string{{"healthy_tool"}, {"healthy_tool"}}, visible)
+	require.Equal(t, before+1, listings.Load(), "the failed catalog must stay omitted across model steps")
+	visible = nil
+	for _, err := range executor.Execute(t.Context(), request) {
+		require.NoError(t, err)
+	}
+	require.Equal(t, [][]string{{"available_tool", "healthy_tool"}, {"available_tool", "healthy_tool"}}, visible)
+}
+
+type executionDiscoveryContext struct {
+	adkagent.StrictContextMock
+}
+
+func (*executionDiscoveryContext) InvocationID() string { return "reused-invocation" }
+func (*executionDiscoveryContext) SessionID() string    { return "reused-session" }

go/adk/pkg/a2a/executor_test.go

@@ -40,6 +40,7 @@ type recordingExecutor struct {
 	userID        string
 	cleanupCalled bool
 	events        []a2atype.Event
+	onExecute     func(context.Context)
 	// span, when set, is a span the turn emits, the way ADK does.
 	span string
 }
@@ -48,6 +49,9 @@ func (e *recordingExecutor) Execute(ctx context.Context, reqCtx *a2asrv.Executor
 	return func(yield func(a2atype.Event, error) bool) {
 		e.message = reqCtx.Message
 		e.userID = auth.UserIDFromContext(ctx)
+		if e.onExecute != nil {
+			e.onExecute(ctx)
+		}
 		if e.span != "" {
 			_, span := otel.Tracer("adk-test").Start(ctx, e.span)
 			span.End()

go/adk/pkg/a2a/server/lazy_mcp_startup_test.go

@@ -0,0 +1,76 @@
+package server
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/durationpb"
+
+	kagentagent "github.com/kagent-dev/kagent/go/adk/pkg/agent"
+	configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1"
+)
+
+func TestReadinessDoesNotWaitForMCPDiscovery(t *testing.T) {
+	t.Setenv("OPENAI_API_KEY", "test-key")
+	t.Run("Given slow MCP servers and MCP-bound subagents, when the runtime starts, assert that readiness precedes discovery", func(t *testing.T) {
+		var requests atomic.Int32
+		mcpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			requests.Add(1)
+			select {
+			case <-time.After(500 * time.Millisecond):
+			case <-r.Context().Done():
+			}
+			http.Error(w, "slow server", http.StatusServiceUnavailable)
+		}))
+		t.Cleanup(mcpServer.Close)
+		marker := filepath.Join(t.TempDir(), "stdio-started")
+		config := func(name string) *configv1.ADKAgentConfig {
+			return &configv1.ADKAgentConfig{
+				Name: name, Instruction: "Use the configured tools.", SkillsDirectory: t.TempDir(),
+				Model:     &configv1.ADKModelConfig{Model: "test-model", Provider: &configv1.ADKModelConfig_Openai{Openai: &configv1.ADKOpenAI{}}},
+				HttpTools: []*configv1.ADKHTTPMCPServerConfig{{Params: &configv1.ADKHTTPConnectionParams{Url: mcpServer.URL, Timeout: durationpb.New(time.Second)}}},
+				SseTools:  []*configv1.ADKSSEMCPServerConfig{{Params: &configv1.ADKSSEConnectionParams{Url: mcpServer.URL, Timeout: durationpb.New(time.Second)}}},
+			}
+		}
+		rootConfig := config("root")
+		rootConfig.StdioTools = []*configv1.ADKStdioMCPServerConfig{{Command: "touch", Args: []string{marker}}}
+		rootConfig.SubAgents = []*configv1.ADKAgentConfig{config("child")}
+		_, err := kagentagent.CreateGoogleADKAgent(t.Context(), &configv1.ADKConfig{Agent: rootConfig}, "root", nil)
+		require.NoError(t, err)
+		srv, err := NewA2AServer(a2atype.AgentCard{}, substrateExecutor{}, slog.New(slog.DiscardHandler), ServerConfig{Host: "127.0.0.1", Port: "0"})
+		require.NoError(t, err)
+		listener, err := net.Listen("tcp", "127.0.0.1:0")
+		require.NoError(t, err)
+		srv.readyServer.Addr = listener.Addr().String()
+		require.NoError(t, listener.Close())
+		require.NoError(t, srv.Start())
+		t.Cleanup(func() {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			require.NoError(t, srv.httpServer.Shutdown(ctx))
+			require.NoError(t, srv.readyServer.Shutdown(ctx))
+			srv.grpcServer.Stop()
+		})
+		client := &http.Client{Timeout: time.Second}
+		resp, err := client.Get("http://" + srv.readyServer.Addr + "/readyz")
+		require.NoError(t, err)
+		_, err = io.Copy(io.Discard, resp.Body)
+		require.NoError(t, err)
+		require.NoError(t, resp.Body.Close())
+		require.Equal(t, http.StatusOK, resp.StatusCode)
+		require.Zero(t, requests.Load(), "root and subagent MCP connections must be deferred")
+		_, err = os.Stat(marker)
+		require.ErrorIs(t, err, os.ErrNotExist, "stdio process must not start before readiness")
+	})
+}

go/adk/pkg/agent/agent.go

@@ -98,7 +98,6 @@ func createGoogleADKAgent(ctx context.Context, agentConfig *configv1.ADKAgentCon
 			log.InfoContext(ctx, "wired local skills", "skills_directory", skillsDirectory, "skill_count", len(skills))
 		}
 	}
-	mcpAppToolNames := mcp.MCPAppToolNamesFromToolsets(toolsets)
 
 	var remoteAgentTools []tool.Tool
 	for _, remoteAgent := range agentConfig.RemoteAgents {
@@ -150,12 +149,11 @@ func createGoogleADKAgent(ctx context.Context, agentConfig *configv1.ADKAgentCon
 	beforeToolCallbacks := []llmagent.BeforeToolCallback{}
 	beforeModelCallbacks := []llmagent.BeforeModelCallback{}
 
-	if len(mcpAppToolNames) > 0 {
-		// For MCP App-capable tools, keep rich tool payloads in chat history for UI rendering,
-		// but compact what is sent back to the model to avoid redundant polling/tool churn.
-		log.InfoContext(ctx, "wiring MCP App model result callback", "tool_count", len(mcpAppToolNames))
-		beforeModelCallbacks = append(beforeModelCallbacks, MakeMCPAppModelResultCallback(mcpAppToolNames))
-	}
+	// Tools are classified before each model request is prepared. Resolve app
+	// names here rather than capturing an empty snapshot during construction.
+	beforeModelCallbacks = append(beforeModelCallbacks, makeMCPAppModelResultCallback(func(ctx agent.Context) map[string]bool {
+		return mcp.MCPAppToolNamesFromToolsets(ctx, toolsets)
+	}))
 	beforeToolCallbacks = append(beforeToolCallbacks, makeBeforeToolCallback(log))
 
 	llmAgentConfig := llmagent.Config{

go/adk/pkg/agent/mcp_apps.go

@@ -19,20 +19,22 @@ import (
 // tool carrying a UI resourceUri, independent of the tool's name or payload keys.
 const mcpAppRenderedNotice = "The interactive UI for this tool has been rendered to the user and now updates live inside the app. Treat this as complete and do not call this tool again unless the user explicitly asks for it."
 
-// MakeMCPAppModelResultCallback replaces what the model sees for MCP App
+// makeMCPAppModelResultCallback replaces what the model sees for MCP App
 // (UI-rendering) tool results: instead of the heavy render payload it receives a
 // terminal directive (see mcpAppRenderedNotice). The full result is still
 // streamed to the UI separately, so this only changes the model's view and
 // prevents the model from looping on the rendering tool. Errors are passed
 // through so the model can still react to and recover from failures.
-func MakeMCPAppModelResultCallback(appToolNames map[string]bool) llmagent.BeforeModelCallback {
-	return func(_ agent.Context, req *adkmodel.LLMRequest) (*adkmodel.LLMResponse, error) {
+// Tool names are resolved during model preparation so discovery stays lazy.
+func makeMCPAppModelResultCallback(appToolNames func(agent.Context) map[string]bool) llmagent.BeforeModelCallback {
+	return func(ctx agent.Context, req *adkmodel.LLMRequest) (*adkmodel.LLMResponse, error) {
+		names := appToolNames(ctx)
 		for _, content := range req.Contents {
 			if content == nil {
 				continue
 			}
 			for _, part := range content.Parts {
-				if part == nil || part.FunctionResponse == nil || !appToolNames[part.FunctionResponse.Name] {
+				if part == nil || part.FunctionResponse == nil || !names[part.FunctionResponse.Name] {
 					continue
 				}
 				part.FunctionResponse.Response = compactMCPAppModelResponse(part.FunctionResponse.Response)