feat: add API v2 MCP tasks gateway (#2576)

## Summary

- replace the legacy MCP implementation with an authenticated stateless
API v2 endpoint
- expose AgentInstance discovery and invocation through the in-process
public A2A gateway
- support the MCP Tasks extension for durable polling, cancellation, and
input-required continuation, with synchronous fallback
- remove the legacy MCP handler and in-memory agent client registry

## Testing

- go test ./core/...
- make -C go lint
- go test ./core/v2/mcp ./core/v2/a2agateway -timeout=2m
- local Kind E2E: TestMCPAgentInstanceInteraction,
TestMCPAskUserContinuation, TestMCPCancelTask

---------

Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io>

Author: EItanya

docs/plans/api-v2-execution-plan.md

@@ -326,6 +326,10 @@ MCP:
 
 - Discover ready AgentInstances.
 - Invoke them only through the public gateway A2A path.
+- Expose durable A2A turns through the MCP Tasks extension, including polling,
+  cancellation, and input-required continuation.
+- Keep synchronous `tools/call` fallback for clients without Tasks support.
+- Store no MCP-owned task or session state.
 - Never expose Actor or private MCP endpoints.
 
 Content:

go/core/cmd/controller-v2/main.go

@@ -35,12 +35,14 @@ import (
 	"github.com/kagent-dev/kagent/go/core/internal/service/kubecrud"
 	sessionservice "github.com/kagent-dev/kagent/go/core/internal/service/session"
 	taskservice "github.com/kagent-dev/kagent/go/core/internal/service/task"
+	"github.com/kagent-dev/kagent/go/core/pkg/auth"
 	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
 	legacysubstrate "github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
 	"github.com/kagent-dev/kagent/go/core/v2/a2agateway"
 	"github.com/kagent-dev/kagent/go/core/v2/agentinstance"
 	"github.com/kagent-dev/kagent/go/core/v2/checkpoint"
 	v2controller "github.com/kagent-dev/kagent/go/core/v2/controller"
+	v2mcp "github.com/kagent-dev/kagent/go/core/v2/mcp"
 	"golang.org/x/sync/errgroup"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -126,6 +128,12 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+	gateway := a2agateway.New(store, authorizer, gatewayDialer, instanceWorkflow,
+		env("A2A_GATEWAY_URL", "http://127.0.0.1:8084"))
+	mcpHandler, err := v2mcp.New(instances, gateway)
+	if err != nil {
+		log.Fatal(err)
+	}
 	server, err := grpcserver.New(grpcserver.Config{
 		BindAddress:          env("GRPC_BIND_ADDRESS", ":8084"),
 		Reflection:           envBool("GRPC_REFLECTION"),
@@ -139,10 +147,7 @@ func main() {
 		AgentTemplateService: kubecrud.NewService(manager.GetClient(), authorizer, &kagentv1alpha3.AgentTemplate{}, &kagentv1alpha3.AgentTemplateList{}, "AgentTemplate"),
 		HarnessService:       kubecrud.NewService(manager.GetClient(), authorizer, &kagentv1alpha3.Harness{}, &kagentv1alpha3.HarnessList{}, "Harness"),
 		CheckpointService:    checkpoints,
-		// `instanceWorkflow` is what upstream added: the gateway needs it to suspend an
-		// instance once a turn reaches a quiescent boundary.
-		A2AHandler: a2agateway.New(store, authorizer, gatewayDialer, instanceWorkflow,
-			env("A2A_GATEWAY_URL", "http://127.0.0.1:8084")),
+		A2AHandler:           gateway,
 	})
 	if err != nil {
 		log.Fatal(err)
@@ -156,12 +161,14 @@ func main() {
 	// It answered every path with an empty 200 and ignored the request entirely, so
 	// a browser calling an RPC got a success with no body — which reads as a
 	// serialisation fault in the client rather than as a server that never had the
-	// endpoint. The router below hands anything that is not gRPC-Web to the same
-	// health response as before.
-	httpHandler := server.WebHandlerOr(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+	// endpoint. The router below serves MCP and hands other non-gRPC-Web requests
+	// to the same health response as before.
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusOK)
-	}))
-	health := &http.Server{Addr: env("HTTP_BIND_ADDRESS", ":8083"), Handler: httpHandler}
+	})
+	mux.Handle("/mcp", auth.AuthnMiddleware(authenticator)(mcpHandler))
+	health := &http.Server{Addr: env("HTTP_BIND_ADDRESS", ":8083"), Handler: server.WebHandlerOr(mux)}
 	group, ctx := errgroup.WithContext(ctx)
 	group.Go(func() error { return runtime.Start(ctx) })
 	group.Go(func() error { return manager.Start(ctx) })