feat(cli): add API v2 AgentInstance workflows

.github/workflows/ci.yaml

@@ -141,6 +141,8 @@ jobs:
           test "$READY" = True
           export KAGENT_GRPC_URL="$(kubectl get svc -n kagent kagent-controller -o jsonpath='{.status.loadBalancer.ingress[0].ip}'):8084"
           echo "KAGENT_GRPC_URL: $KAGENT_GRPC_URL"
+          make core/bin/kagent-local
+          export KAGENT_E2E_CLI="$PWD/core/bin/kagent-local"
           go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast -shuffle=on
 
       - name: fail print info

go/Makefile

@@ -91,6 +91,7 @@ govulncheck: ## Run govulncheck.
 
 ##@ Build (core module — CLI binaries)
 
+.PHONY: core/bin/kagent-local
 core/bin/kagent-local:
 	go test -race ./core/cli/...
 	CGO_ENABLED=0 go build -ldflags "$(LDFLAGS)" -o core/bin/kagent-local ./core/cli/cmd/kagent
@@ -148,13 +149,13 @@ test: ## Run all unit tests.
 KIND_CLUSTER_NAME ?= kagent
 
 .PHONY: e2e
-e2e: ## Run end-to-end tests.
+e2e: core/bin/kagent-local ## Run end-to-end tests.
 	@kind get clusters 2>/dev/null | grep -qx '$(KIND_CLUSTER_NAME)' || { \
 	  echo "Error: kind cluster '$(KIND_CLUSTER_NAME)' not found. Run 'make create-kind-cluster' first." >&2; \
 	  exit 1; \
 	}
 	@kind get kubeconfig --name $(KIND_CLUSTER_NAME) > /tmp/kind-config-e2e
-	KUBECONFIG=/tmp/kind-config-e2e go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast
+	KAGENT_E2E_CLI=$(CURDIR)/core/bin/kagent-local KUBECONFIG=/tmp/kind-config-e2e go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast
 
 ##@ Dependencies
 

go/api/a2a/routing.go

@@ -0,0 +1,9 @@
+// Package a2a defines the public A2A routing contract for AgentInstances.
+package a2a
+
+const (
+	// AgentInstanceNamespaceHeader selects the namespace scope for the AgentInstance.
+	AgentInstanceNamespaceHeader = "x-kagent-agent-instance-namespace"
+	// AgentInstanceIDHeader selects the AgentInstance within that namespace scope.
+	AgentInstanceIDHeader = "x-kagent-agent-instance-id"
+)

go/api/client/a2a.go

@@ -0,0 +1,85 @@
+package client
+
+import (
+	"context"
+	"time"
+
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	"github.com/a2aproject/a2a-go/v2/a2aclient"
+	a2agrpc "github.com/a2aproject/a2a-go/v2/a2agrpc/v1"
+	a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1"
+	kagenta2a "github.com/kagent-dev/kagent/go/api/a2a"
+)
+
+const userIDHeader = "x-user-id"
+
+// A2AClient creates upstream A2A clients routed to an AgentInstance.
+type A2AClient struct {
+	client *BaseClient
+}
+
+// NewA2AClient creates an A2A client factory over the shared gRPC connection.
+func NewA2AClient(client *BaseClient) *A2AClient {
+	return &A2AClient{client: client}
+}
+
+// ForAgentInstance creates an upstream A2A client routed to one AgentInstance.
+func (c *A2AClient) ForAgentInstance(ctx context.Context, namespace, id string) (*a2aclient.Client, error) {
+	connection, err := c.client.grpcConnection()
+	if err != nil {
+		return nil, err
+	}
+	transport := a2agrpc.NewGRPCTransportFromClient(a2apb.NewA2AServiceClient(connection))
+	return a2aclient.NewFromEndpoints(ctx, []*a2atype.AgentInterface{{
+		URL:             c.client.grpc.target,
+		ProtocolBinding: a2atype.TransportProtocolGRPC,
+		ProtocolVersion: a2atype.Version,
+	}},
+		a2aclient.WithDefaultsDisabled(),
+		a2aclient.WithTransport(a2atype.TransportProtocolGRPC, a2aclient.TransportFactoryFn(
+			func(context.Context, *a2atype.AgentCard, *a2atype.AgentInterface) (a2aclient.Transport, error) {
+				return transport, nil
+			},
+		)),
+		a2aclient.WithCallInterceptors(&agentInstanceRoutingInterceptor{
+			namespace: namespace,
+			id:        id,
+			userID:    c.client.UserID,
+			timeout:   c.client.grpc.timeout,
+		}),
+	)
+}
+
+type cancelCallContextKey struct{}
+
+type agentInstanceRoutingInterceptor struct {
+	a2aclient.PassthroughInterceptor
+	namespace string
+	id        string
+	userID    string
+	timeout   time.Duration
+}
+
+func (i *agentInstanceRoutingInterceptor) Before(ctx context.Context, request *a2aclient.Request) (context.Context, any, error) {
+	request.ServiceParams.Append(kagenta2a.AgentInstanceNamespaceHeader, i.namespace)
+	request.ServiceParams.Append(kagenta2a.AgentInstanceIDHeader, i.id)
+	if i.userID != "" {
+		request.ServiceParams.Append(userIDHeader, i.userID)
+	}
+	if i.timeout <= 0 || isStreamingA2AMethod(request.Method) {
+		return ctx, nil, nil
+	}
+	callContext, cancel := context.WithTimeout(ctx, i.timeout)
+	return context.WithValue(callContext, cancelCallContextKey{}, cancel), nil, nil
+}
+
+func (i *agentInstanceRoutingInterceptor) After(ctx context.Context, _ *a2aclient.Response) error {
+	if cancel, ok := ctx.Value(cancelCallContextKey{}).(context.CancelFunc); ok {
+		cancel()
+	}
+	return nil
+}
+
+func isStreamingA2AMethod(method string) bool {
+	return method == "SendStreamingMessage" || method == "SubscribeToTask"
+}

go/api/client/agentinstance.go

@@ -0,0 +1,62 @@
+package client
+
+import (
+	"context"
+
+	apiv1alpha1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1alpha1"
+)
+
+// AgentInstanceClient provides supported AgentInstance operations.
+type AgentInstanceClient struct {
+	client *BaseClient
+}
+
+// NewAgentInstanceClient creates an AgentInstance client over the shared gRPC connection.
+func NewAgentInstanceClient(client *BaseClient) *AgentInstanceClient {
+	return &AgentInstanceClient{client: client}
+}
+
+func (c *AgentInstanceClient) CreateAgentInstance(ctx context.Context, request *apiv1alpha1.CreateAgentInstanceRequest) (*apiv1alpha1.CreateAgentInstanceResponse, error) {
+	client, callContext, cancel, err := c.client.agentInstanceCall(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer cancel()
+	return client.CreateAgentInstance(callContext, request)
+}
+
+func (c *AgentInstanceClient) GetAgentInstance(ctx context.Context, request *apiv1alpha1.GetAgentInstanceRequest) (*apiv1alpha1.GetAgentInstanceResponse, error) {
+	client, callContext, cancel, err := c.client.agentInstanceCall(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer cancel()
+	return client.GetAgentInstance(callContext, request)
+}
+
+func (c *AgentInstanceClient) ListAgentInstances(ctx context.Context, request *apiv1alpha1.ListAgentInstancesRequest) (*apiv1alpha1.ListAgentInstancesResponse, error) {
+	client, callContext, cancel, err := c.client.agentInstanceCall(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer cancel()
+	return client.ListAgentInstances(callContext, request)
+}
+
+func (c *AgentInstanceClient) DeleteAgentInstance(ctx context.Context, request *apiv1alpha1.DeleteAgentInstanceRequest) (*apiv1alpha1.DeleteAgentInstanceResponse, error) {
+	client, callContext, cancel, err := c.client.agentInstanceCall(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer cancel()
+	return client.DeleteAgentInstance(callContext, request)
+}
+
+func (c *BaseClient) agentInstanceCall(ctx context.Context) (apiv1alpha1.AgentInstanceServiceClient, context.Context, context.CancelFunc, error) {
+	connection, err := c.grpcConnection()
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	callContext, cancel := c.grpcCallContext(ctx)
+	return apiv1alpha1.NewAgentInstanceServiceClient(connection), callContext, cancel, nil
+}

go/api/client/agentinstancegrpc_test.go

@@ -0,0 +1,151 @@
+package client
+
+import (
+	"context"
+	"net"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	"github.com/a2aproject/a2a-go/v2/a2aclient"
+	a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1"
+	"github.com/a2aproject/a2a-go/v2/a2apb/v1/pbconv"
+	kagenta2a "github.com/kagent-dev/kagent/go/api/a2a"
+	apiv1alpha1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/test/bufconn"
+)
+
+const agentInstanceClientTestID = "8bd650a8-9775-488f-8bc1-0d52bf7bdcab"
+
+type recordingAgentInstanceService struct {
+	apiv1alpha1.UnimplementedAgentInstanceServiceServer
+	observation callObservation
+}
+
+func (s *recordingAgentInstanceService) CreateAgentInstance(ctx context.Context, _ *apiv1alpha1.CreateAgentInstanceRequest) (*apiv1alpha1.CreateAgentInstanceResponse, error) {
+	values, _ := metadata.FromIncomingContext(ctx)
+	_, hasDeadline := ctx.Deadline()
+	s.observation = callObservation{userID: first(values.Get(userIDHeader)), hasDeadline: hasDeadline}
+	return &apiv1alpha1.CreateAgentInstanceResponse{}, nil
+}
+
+type a2aCallObservation struct {
+	namespace     string
+	id            string
+	userID        string
+	authorization string
+	hasDeadline   bool
+}
+
+type recordingA2AService struct {
+	a2apb.UnimplementedA2AServiceServer
+	mu           sync.Mutex
+	observations []a2aCallObservation
+}
+
+func (s *recordingA2AService) SendMessage(ctx context.Context, _ *a2apb.SendMessageRequest) (*a2apb.SendMessageResponse, error) {
+	s.observe(ctx)
+	return pbconv.ToProtoSendMessageResponse(a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart("hello")))
+}
+
+func (s *recordingA2AService) SendStreamingMessage(_ *a2apb.SendMessageRequest, stream grpc.ServerStreamingServer[a2apb.StreamResponse]) error {
+	s.observe(stream.Context())
+	response, err := pbconv.ToProtoStreamResponse(a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart("hello")))
+	if err != nil {
+		return err
+	}
+	return stream.Send(response)
+}
+
+func (s *recordingA2AService) SubscribeToTask(_ *a2apb.SubscribeToTaskRequest, stream grpc.ServerStreamingServer[a2apb.StreamResponse]) error {
+	s.observe(stream.Context())
+	response, err := pbconv.ToProtoStreamResponse(a2atype.NewMessage(a2atype.MessageRoleAgent, a2atype.NewTextPart("hello")))
+	if err != nil {
+		return err
+	}
+	return stream.Send(response)
+}
+
+func (s *recordingA2AService) observe(ctx context.Context) {
+	values, _ := metadata.FromIncomingContext(ctx)
+	_, hasDeadline := ctx.Deadline()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.observations = append(s.observations, a2aCallObservation{
+		namespace:     first(values.Get(kagenta2a.AgentInstanceNamespaceHeader)),
+		id:            first(values.Get(kagenta2a.AgentInstanceIDHeader)),
+		userID:        first(values.Get(userIDHeader)),
+		authorization: first(values.Get("authorization")),
+		hasDeadline:   hasDeadline,
+	})
+}
+
+func TestAgentInstanceAndA2AClientsShareGRPCConnection(t *testing.T) {
+	listener := bufconn.Listen(1024 * 1024)
+	agentInstanceService := &recordingAgentInstanceService{}
+	a2aService := &recordingA2AService{}
+	server := grpc.NewServer()
+	apiv1alpha1.RegisterAgentInstanceServiceServer(server, agentInstanceService)
+	a2apb.RegisterA2AServiceServer(server, a2aService)
+	go func() { _ = server.Serve(listener) }()
+	t.Cleanup(func() {
+		server.Stop()
+		_ = listener.Close()
+	})
+
+	var dialCount atomic.Int32
+	clientSet := New(
+		"http://rest-must-not-be-used.invalid",
+		WithUserID("caller"),
+		WithGRPCTarget("passthrough:///bufnet"),
+		WithGRPCTimeout(5*time.Second),
+		WithGRPCDialOptions(grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			dialCount.Add(1)
+			return listener.Dial()
+		})),
+	)
+	t.Cleanup(func() { require.NoError(t, clientSet.Close()) })
+
+	_, err := clientSet.AgentInstance.CreateAgentInstance(context.Background(), &apiv1alpha1.CreateAgentInstanceRequest{})
+	require.NoError(t, err)
+	assert.Equal(t, callObservation{userID: "caller", hasDeadline: true}, agentInstanceService.observation)
+
+	a2aClient, err := clientSet.A2A.ForAgentInstance(context.Background(), "kagent", agentInstanceClientTestID)
+	require.NoError(t, err)
+	a2aCtx := a2aclient.AttachServiceParams(context.Background(), a2aclient.ServiceParams{
+		"authorization": {"Bearer model-key"},
+	})
+	request := &a2atype.SendMessageRequest{Message: a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("hi"))}
+	_, err = a2aClient.SendMessage(a2aCtx, request)
+	require.NoError(t, err)
+	for _, streamErr := range a2aClient.SendStreamingMessage(a2aCtx, request) {
+		require.NoError(t, streamErr)
+	}
+	for _, streamErr := range a2aClient.SubscribeToTask(a2aCtx, &a2atype.SubscribeToTaskRequest{ID: "task-id"}) {
+		require.NoError(t, streamErr)
+	}
+
+	a2aService.mu.Lock()
+	require.Equal(t, []a2aCallObservation{
+		{namespace: "kagent", id: agentInstanceClientTestID, userID: "caller", authorization: "Bearer model-key", hasDeadline: true},
+		{namespace: "kagent", id: agentInstanceClientTestID, userID: "caller", authorization: "Bearer model-key", hasDeadline: false},
+		{namespace: "kagent", id: agentInstanceClientTestID, userID: "caller", authorization: "Bearer model-key", hasDeadline: false},
+	}, a2aService.observations)
+	a2aService.mu.Unlock()
+	assert.Equal(t, int32(1), dialCount.Load())
+}
+
+func TestStreamingA2AMethodsMatchUpstreamService(t *testing.T) {
+	methods := make([]string, 0, len(a2apb.A2AService_ServiceDesc.Streams))
+	for _, stream := range a2apb.A2AService_ServiceDesc.Streams {
+		methods = append(methods, stream.StreamName)
+		assert.True(t, isStreamingA2AMethod(stream.StreamName), "streaming method %q must not receive a unary timeout", stream.StreamName)
+	}
+	assert.Equal(t, []string{"SendStreamingMessage", "SubscribeToTask"}, methods)
+}

go/api/client/clientset.go

@@ -15,6 +15,8 @@ type ClientSet struct {
 	Model               Model
 	Namespace           Namespace
 	Feedback            Feedback
+	AgentInstance       *AgentInstanceClient
+	A2A                 *A2AClient
 }
 
 // New creates a new KAgent client set
@@ -34,6 +36,8 @@ func New(baseURL string, options ...ClientOption) *ClientSet {
 		Model:               NewModelClient(baseClient),
 		Namespace:           NewNamespaceClient(baseClient),
 		Feedback:            NewFeedbackClient(baseClient),
+		AgentInstance:       NewAgentInstanceClient(baseClient),
+		A2A:                 NewA2AClient(baseClient),
 	}
 }
 

go/api/client/grpc.go

@@ -16,7 +16,9 @@ import (
 )
 
 const (
-	defaultGRPCTarget         = "localhost:8084"
+	// DefaultGRPCTarget is the implicit local target used when WithGRPCTarget is omitted.
+	// Callers can use it to limit local connection setup to the default endpoint.
+	DefaultGRPCTarget         = "localhost:8084"
 	defaultGRPCTimeout        = 30 * time.Second
 	defaultGRPCMaxMessageSize = 16 << 20
 )
@@ -42,7 +44,7 @@ type grpcTransport struct {
 
 func newGRPCTransport() grpcTransport {
 	return grpcTransport{
-		target:          defaultGRPCTarget,
+		target:          DefaultGRPCTarget,
 		timeout:         defaultGRPCTimeout,
 		maxMessageBytes: defaultGRPCMaxMessageSize,
 	}