feat: add token passthrough for remote MCP servers

docs/architecture/credential-injection.md

@@ -80,10 +80,97 @@ ModelConfig, RemoteMCPServer, or [EgressPolicy](egress-policies.md) for gateway
 AWS IAM signing keys, Google service-account keys, and OAuth client credentials
 require mechanisms beyond static header injection and are rejected rather than
 serialized into runtimes.
-Caller-token passthrough retains its existing behavior. A passthrough model
+`ModelConfig.spec.tokenPassthrough: true` enables caller-token passthrough
+instead of a static model API key. For example:
+
+```yaml
+spec:
+  provider: OpenAI
+  model: gpt-4.1
+  tokenPassthrough: true
+```
+
+A passthrough model
 cannot share an origin with static gateway credentials, which would override
 the caller's authentication.
 
+## Remote MCP caller bearer tokens
+
+`RemoteMCPServer.spec.tokenPassthrough` is optional and defaults to
+`false`. When enabled, the server receives `Authorization: Bearer <token>`
+from the **current authenticated A2A invocation**. Go ADK reads only
+`models.PassthroughToken` (the restricted `BearerTokenKey`); Python reads the
+dedicated `bearer_token` ContextVar. Neither path falls back to general request
+metadata, an STS token, a runtime service credential, or a previous invocation.
+A missing token fails before an MCP connection starts.
+A resumed/approved call requires the new invocation's token.
+
+```yaml
+apiVersion: api.kagent.dev/v1alpha3
+kind: RemoteMCPServer
+metadata:
+  name: caller-tools
+  namespace: kagent
+  labels:
+    kagent.dev/discovery: disabled
+spec:
+  description: Tools authorized by the invoking caller
+  url: https://tools.example.com/mcp
+  tokenPassthrough: true
+```
+
+Controller discovery has no caller identity:
+
+- The existing `kagent.dev/discovery: disabled` label takes precedence. It
+  retains `Accepted=True`, with no discovered tools or catalog connectivity.
+- Without that label, passthrough requires an `Authorization` entry in
+  `headersFrom` (case-insensitive) for discovery. Without one, the controller
+  does not dial, clears stale discovered tools and catalog connectivity, and
+  reports `Accepted=False`, reason `DiscoveryCredentialRequired`. It uses the
+  ordinary discovery refresh interval.
+- When supplied, that Authorization is **discovery-only**. All case variants
+  are removed before runtime header resolution, including ConfigMap lookup,
+  Secret environment placeholders, and gateway credential compilation. Other
+  configured headers remain unchanged. Discovery failures do not publish
+  upstream errors that could echo the credential.
+
+Model and MCP passthrough destinations cannot share an origin with **any**
+static gateway credential, including an MCP server's own Secret-backed
+non-Authorization header. Matching lowercases DNS names and removes a trailing
+dot and normalizes default ports, just as egress credential canonicalization
+does. Different schemes or ports are separate origins; paths are not.
+Multiple passthrough users of one origin are
+allowed. Discovery-only Authorization creates no gateway binding.
+
+The ADK protobuf carries only the flag, not a token. Go ADK creates an ordinary
+ADK MCP toolset for each listing and tool call, closing every transport connection
+at the operation boundary, including initialization failures and reconnects.
+Only ADK-generated declarations survive discovery; the native toolset and its
+caller-bound sessions are never shared between operations. Execution adds a
+fresh discovery to recheck selected tools and model visibility, including after
+a call reconnects. ADK owns successful result conversion; the local wrapper
+sanitizes upstream failures. Calls inherit ordinary ADK's single retry on
+connection loss, so they do not provide at-most-once execution if a server
+completes a tool call before the connection fails.
+Python creates agents per request and pools MCP sessions
+by effective headers within that lifetime. Request headers use ADK's
+`temp:headers` state, so credentials are not written to its native SQLite
+conversation or event history. Go ADK and Python passthrough HTTP clients reject
+redirects and foreign SSE endpoint destinations.
+
+This setting is supported only by Go and Python ADK. Claude and Codex
+compilers reject it; native runtime support is not implemented.
+
+This per-server setting does not enable the older global ADK token-propagation
+switch. That independently configured behavior remains unchanged for servers
+without passthrough; it cannot override a passthrough server's credential.
+
+MCP Apps `CallTool` and `ReadResource` control-plane calls do not carry the
+restricted A2A invocation context. They are unsupported for passthrough
+servers and fail before resolving discovery credentials or opening a
+connection. The controller must never use discovery Authorization to perform
+an app's tool call.
+
 ## Deferred API fields
 
 The source retains commented declarations for Harness and SandboxTemplate

go/adk/pkg/mcp/baggage_test.go

@@ -196,8 +196,11 @@ type toolContext struct {
 	callID string
 }
 
-func (c toolContext) Value(key any) any      { return c.ctx.Value(key) }
-func (c toolContext) FunctionCallID() string { return c.callID }
+func (c toolContext) Value(key any) any           { return c.ctx.Value(key) }
+func (c toolContext) FunctionCallID() string      { return c.callID }
+func (c toolContext) Deadline() (time.Time, bool) { return c.ctx.Deadline() }
+func (c toolContext) Done() <-chan struct{}       { return c.ctx.Done() }
+func (c toolContext) Err() error                  { return c.ctx.Err() }
 
 type readonlyContext struct {
 	adkagent.ReadonlyContext

go/adk/pkg/mcp/passthrough.go

@@ -0,0 +1,175 @@
+package mcp
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"slices"
+
+	mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
+	adkagent "google.golang.org/adk/v2/agent"
+	"google.golang.org/adk/v2/model"
+	"google.golang.org/adk/v2/tool"
+	"google.golang.org/adk/v2/tool/mcptoolset"
+	"google.golang.org/adk/v2/tool/toolutils"
+	"google.golang.org/genai"
+
+	"github.com/kagent-dev/kagent/go/adk/pkg/models"
+	"github.com/kagent-dev/kagent/go/pkg/logging"
+)
+
+// passthroughToolset keeps only configuration, never a client session or caller
+// credential. Each operation owns an ordinary ADK toolset and its connections.
+type passthroughToolset struct {
+	params mcpServerParams
+	filter map[string]bool
+}
+
+func (*passthroughToolset) Name() string { return "mcp_tool_set" }
+
+// passthroughTransport observes connections without wrapping them: the SDK uses
+// private connection hooks to update session identity and protocol negotiation.
+// ADK connects synchronously, including retries, within the owning operation.
+type passthroughTransport struct {
+	params      mcpServerParams
+	connections []mcpsdk.Connection
+}
+
+func (p *passthroughTransport) Connect(ctx context.Context) (mcpsdk.Connection, error) {
+	// SDK transports are single-use; ADK may reconnect through this wrapper.
+	transport, err := createTransport(ctx, p.params)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := transport.Connect(ctx)
+	if conn != nil {
+		p.connections = append(p.connections, conn)
+	}
+	return conn, err
+}
+
+func (p *passthroughTransport) close(ctx context.Context) {
+	for _, conn := range p.connections {
+		// Cancellation alone does not stop the SDK's detached background work.
+		// Close is idempotent, including after failed initialization or retries.
+		if err := conn.Close(); err != nil {
+			logging.FromContext(ctx).WarnContext(ctx, "failed to close passthrough MCP connection")
+		}
+	}
+	p.connections = nil
+}
+
+func (p *passthroughToolset) tools(ctx adkagent.ReadonlyContext, transport *passthroughTransport) ([]tool.Tool, error) {
+	if _, ok := models.PassthroughToken(ctx, true); !ok {
+		return nil, fmt.Errorf("MCP bearer token passthrough requires an authenticated invocation bearer token")
+	}
+	client := mcpsdk.NewClient(&mcpsdk.Implementation{Name: "kagent", Version: "1"},
+		&mcpsdk.ClientOptions{Capabilities: mcpUIClientCapabilities()})
+	// ADK drops MCP metadata during conversion. Filter each discovery page at
+	// the SDK boundary, before delegating declarations and results to ADK.
+	var listedSession mcpsdk.Session
+	client.AddSendingMiddleware(func(next mcpsdk.MethodHandler) mcpsdk.MethodHandler {
+		return func(ctx context.Context, method string, req mcpsdk.Request) (mcpsdk.Result, error) {
+			if method == "tools/call" && req.GetSession() != listedSession {
+				// A native ADK call can reconnect after discovery. Recheck access
+				// on the replacement session before allowing the retried call.
+				allowed := false
+				for item, err := range req.GetSession().(*mcpsdk.ClientSession).Tools(ctx, nil) {
+					if err != nil {
+						return nil, err
+					}
+					if item.Name == req.GetParams().(*mcpsdk.CallToolParams).Name {
+						allowed = true
+					}
+				}
+				if !allowed {
+					return nil, fmt.Errorf("passthrough MCP tool is no longer available")
+				}
+			}
+			result, err := next(ctx, method, req)
+			if err == nil && method == "tools/list" {
+				list := result.(*mcpsdk.ListToolsResult)
+				list.Tools = slices.DeleteFunc(list.Tools, func(item *mcpsdk.Tool) bool {
+					return item == nil || len(p.filter) > 0 && !p.filter[item.Name] ||
+						mcpToolKindOf(item.Meta) == mcpToolKindAppInternal
+				})
+				listedSession = req.GetSession()
+			}
+			return result, err
+		}
+	})
+	native, err := mcptoolset.New(mcptoolset.Config{Client: client, Transport: transport})
+	if err != nil {
+		return nil, passthroughError(err)
+	}
+	tools, err := native.Tools(ctx)
+	if err != nil {
+		return nil, passthroughError(err)
+	}
+	return tools, nil
+}
+
+func (p *passthroughToolset) Tools(ctx adkagent.ReadonlyContext) ([]tool.Tool, error) {
+	transport := &passthroughTransport{params: p.params}
+	defer transport.close(ctx)
+	tools, err := p.tools(ctx, transport)
+	if err != nil {
+		return nil, err
+	}
+	for i, t := range tools {
+		// Keep only ADK's declaration, not its tool/client and closed session
+		// (which can still retain the discovery caller's context).
+		tools[i] = &passthroughTool{set: p, declaration: t.(functionTool).Declaration()}
+	}
+	return tools, nil
+}
+
+type passthroughTool struct {
+	set         *passthroughToolset
+	declaration *genai.FunctionDeclaration
+}
+
+func (p *passthroughTool) Name() string                            { return p.declaration.Name }
+func (p *passthroughTool) Description() string                     { return p.declaration.Description }
+func (*passthroughTool) IsLongRunning() bool                       { return false }
+func (p *passthroughTool) Declaration() *genai.FunctionDeclaration { return p.declaration }
+func (p *passthroughTool) ProcessRequest(_ adkagent.Context, req *model.LLMRequest) error {
+	return toolutils.PackTool(req, p)
+}
+
+func (p *passthroughTool) Run(ctx adkagent.Context, args any) (map[string]any, error) {
+	transport := &passthroughTransport{params: p.set.params}
+	defer transport.close(ctx)
+	tools, err := p.set.tools(ctx, transport)
+	if err != nil {
+		return nil, err
+	}
+	for _, t := range tools {
+		if t.Name() == p.Name() {
+			result, err := t.(functionTool).Run(ctx, args)
+			if err != nil {
+				return nil, passthroughError(err)
+			}
+			return result, nil
+		}
+	}
+	return nil, fmt.Errorf("passthrough MCP tool is no longer available")
+}
+
+// Upstream errors may contain server content or credentials. Preserve only
+// framework classifications, never the original error or its wrapping text.
+func passthroughError(err error) error {
+	for _, safe := range []error{
+		context.Canceled, context.DeadlineExceeded,
+		tool.ErrConfirmationRequired, tool.ErrConfirmationRejected,
+	} {
+		if errors.Is(err, safe) {
+			return safe
+		}
+	}
+	return fmt.Errorf("passthrough MCP operation failed")
+}
+
+var _ mcpsdk.Transport = (*passthroughTransport)(nil)
+var _ tool.Toolset = (*passthroughToolset)(nil)
+var _ functionTool = (*passthroughTool)(nil)