Migrate from A2A v0 to v1 (part 1) (#1921)

The 1st of 3 PRs for A2A v0 to v1 zero downtime migration. This will go
into release 0.10.0

Makes the controller compatible with A2A v0 (trpc-a2a-go) and v1
(a2a-go) data and both v0 and v1 JSON-RPC routes / agent card routes
depending on `A2A-Version` header (designed explicitly by the protocol
for version negotiations). Replaces **all** internal usages of
`trpc-a2a-go` with `a2a-go` except in data conversion code and CLI
client code.

This ensure that the controller is compatible with old / existing agents
and UI in A2A v0 and existing v0 data in db, while adding the capability
to communicate with v1 agents and UI during upgrade as they are added in
the next part.

Follows
https://a2a-protocol.org/latest/whats-new-v1/#migration-strategy-recommendations
and db conversion reuses as much of existing a2a backward compatible
code as possible.

---------

Signed-off-by: Jet Chiang <pokyuen.jetchiang-ext@solo.io>
Signed-off-by: JM Huibonhoa <jm.huibonhoa@solo.io>
Co-authored-by: JM Huibonhoa <jm.huibonhoa@solo.io>

Author: supreme-gg-gg

go/adk/pkg/tools/skills_test.go

@@ -21,8 +21,12 @@ func TestResolveReadPath_AllowsSymlinkedSkillsDirectory(t *testing.T) {
 	if err != nil {
 		t.Fatalf("resolveReadPath() error = %v", err)
 	}
-	if resolved != skillFile {
-		t.Fatalf("resolveReadPath() = %q, want %q", resolved, skillFile)
+	want, err := filepath.EvalSymlinks(skillFile)
+	if err != nil {
+		t.Fatalf("EvalSymlinks() error = %v", err)
+	}
+	if resolved != want {
+		t.Fatalf("resolveReadPath() = %q, want %q", resolved, want)
 	}
 }
 

go/api/config/crd/bases/kagent.dev_agents.yaml

@@ -7739,35 +7739,36 @@ spec:
                               description: Examples are optional usage examples.
                               items:
                                 type: string
+                              maxItems: 20
                               type: array
                             id:
                               description: ID is the unique identifier for the skill.
                               type: string
                             inputModes:
-                              description: InputModes are the supported input data
-                                modes/types.
+                              description: InputModes are the supported input MIME
+                                types for this skill, overriding the agent's defaults.
                               items:
                                 type: string
                               type: array
                             name:
                               description: Name is the human-readable name of the
                                 skill.
+                              minLength: 1
                               type: string
                             outputModes:
-                              description: OutputModes are the supported output data
-                                modes/types.
+                              description: OutputModes are the supported output MIME
+                                types for this skill, overriding the agent's defaults.
                               items:
                                 type: string
                               type: array
                             tags:
                               description: Tags are optional tags for categorization.
                               items:
                                 type: string
+                              maxItems: 20
                               type: array
                           required:
-                          - id
                           - name
-                          - tags
                           type: object
                         minItems: 1
                         type: array

go/api/config/crd/bases/kagent.dev_sandboxagents.yaml

@@ -5397,35 +5397,36 @@ spec:
                               description: Examples are optional usage examples.
                               items:
                                 type: string
+                              maxItems: 20
                               type: array
                             id:
                               description: ID is the unique identifier for the skill.
                               type: string
                             inputModes:
-                              description: InputModes are the supported input data
-                                modes/types.
+                              description: InputModes are the supported input MIME
+                                types for this skill, overriding the agent's defaults.
                               items:
                                 type: string
                               type: array
                             name:
                               description: Name is the human-readable name of the
                                 skill.
+                              minLength: 1
                               type: string
                             outputModes:
-                              description: OutputModes are the supported output data
-                                modes/types.
+                              description: OutputModes are the supported output MIME
+                                types for this skill, overriding the agent's defaults.
                               items:
                                 type: string
                               type: array
                             tags:
                               description: Tags are optional tags for categorization.
                               items:
                                 type: string
+                              maxItems: 20
                               type: array
                           required:
-                          - id
                           - name
-                          - tags
                           type: object
                         minItems: 1
                         type: array

go/api/database/client.go

@@ -4,9 +4,9 @@ import (
 	"context"
 	"time"
 
+	a2a "github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
 	"github.com/pgvector/pgvector-go"
-	"trpc.group/trpc-go/trpc-a2a-go/protocol"
 )
 
 type QueryOptions struct {
@@ -24,8 +24,8 @@ type Client interface {
 	StoreFeedback(ctx context.Context, feedback *Feedback) error
 	StoreSession(ctx context.Context, session *Session) error
 	StoreAgent(ctx context.Context, agent *Agent) error
-	StoreTask(ctx context.Context, task *protocol.Task) error
-	StorePushNotification(ctx context.Context, config *protocol.TaskPushNotificationConfig) error
+	StoreTask(ctx context.Context, task *a2a.Task) error
+	StorePushNotification(ctx context.Context, config *a2a.PushConfig) error
 	StoreToolServer(ctx context.Context, toolServer *ToolServer) (*ToolServer, error)
 	StoreEvents(ctx context.Context, messages ...*Event) error
 
@@ -40,23 +40,23 @@ type Client interface {
 	// Get methods
 	GetSession(ctx context.Context, sessionID string, userID string) (*Session, error)
 	GetAgent(ctx context.Context, name string) (*Agent, error)
-	GetTask(ctx context.Context, id string) (*protocol.Task, error)
+	GetTask(ctx context.Context, id string) (*a2a.Task, error)
 	GetTool(ctx context.Context, name string) (*Tool, error)
 	GetToolServer(ctx context.Context, name string) (*ToolServer, error)
-	GetPushNotification(ctx context.Context, taskID string, configID string) (*protocol.TaskPushNotificationConfig, error)
+	GetPushNotification(ctx context.Context, taskID string, configID string) (*a2a.PushConfig, error)
 
 	// List methods
 	ListTools(ctx context.Context) ([]Tool, error)
 	ListFeedback(ctx context.Context, userID string) ([]Feedback, error)
-	ListTasksForSession(ctx context.Context, sessionID string) ([]*protocol.Task, error)
+	ListTasksForSession(ctx context.Context, sessionID string) ([]*a2a.Task, error)
 	ListSessions(ctx context.Context, userID string) ([]Session, error)
 	ListSessionsForAgent(ctx context.Context, agentID string, userID string) ([]Session, error)
 	ListSessionsForAgentAllUsers(ctx context.Context, agentID string) ([]Session, error)
 	ListAgents(ctx context.Context) ([]Agent, error)
 	ListToolServers(ctx context.Context) ([]ToolServer, error)
 	ListToolsForServer(ctx context.Context, serverName string, groupKind string) ([]Tool, error)
 	ListEventsForSession(ctx context.Context, sessionID, userID string, options QueryOptions) ([]*Event, error)
-	ListPushNotifications(ctx context.Context, taskID string) ([]*protocol.TaskPushNotificationConfig, error)
+	ListPushNotifications(ctx context.Context, taskID string) ([]*a2a.PushConfig, error)
 
 	// Helper methods
 	RefreshToolsForServer(ctx context.Context, serverName string, groupKind string, tools ...*v1alpha2.MCPTool) error

go/api/database/models.go

@@ -4,10 +4,10 @@ import (
 	"encoding/json"
 	"time"
 
+	a2a "github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/kagent-dev/kagent/go/api/adk"
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
 	"github.com/pgvector/pgvector-go"
-	"trpc.group/trpc-go/trpc-a2a-go/protocol"
 )
 
 type Agent struct {
@@ -32,16 +32,16 @@ type Event struct {
 	Data string `json:"data"` // JSON-serialized protocol.Message
 }
 
-func (m *Event) Parse() (protocol.Message, error) {
-	var data protocol.Message
+func (m *Event) Parse() (a2a.Message, error) {
+	var data a2a.Message
 	if err := json.Unmarshal([]byte(m.Data), &data); err != nil {
-		return protocol.Message{}, err
+		return a2a.Message{}, err
 	}
 	return data, nil
 }
 
-func ParseMessages(messages []Event) ([]*protocol.Message, error) {
-	result := make([]*protocol.Message, 0, len(messages))
+func ParseMessages(messages []Event) ([]*a2a.Message, error) {
+	result := make([]*a2a.Message, 0, len(messages))
 	for _, message := range messages {
 		parsed, err := message.Parse()
 		if err != nil {
@@ -77,24 +77,25 @@ type Session struct {
 }
 
 type Task struct {
-	ID        string     `json:"id"`
-	CreatedAt time.Time  `json:"created_at"`
-	UpdatedAt time.Time  `json:"updated_at"`
-	DeletedAt *time.Time `json:"deleted_at,omitempty"`
-	Data      string     `json:"data"` // JSON-serialized task data
-	SessionID string     `json:"session_id"`
+	ID              string     `json:"id"`
+	CreatedAt       time.Time  `json:"created_at"`
+	UpdatedAt       time.Time  `json:"updated_at"`
+	DeletedAt       *time.Time `json:"deleted_at,omitempty"`
+	Data            string     `json:"data"` // JSON-serialized task data
+	ProtocolVersion *string    `json:"protocol_version,omitempty"`
+	SessionID       string     `json:"session_id"`
 }
 
-func (t *Task) Parse() (protocol.Task, error) {
-	var data protocol.Task
+func (t *Task) Parse() (a2a.Task, error) {
+	var data a2a.Task
 	if err := json.Unmarshal([]byte(t.Data), &data); err != nil {
-		return protocol.Task{}, err
+		return a2a.Task{}, err
 	}
 	return data, nil
 }
 
-func ParseTasks(tasks []Task) ([]*protocol.Task, error) {
-	result := make([]*protocol.Task, 0, len(tasks))
+func ParseTasks(tasks []Task) ([]*a2a.Task, error) {
+	result := make([]*a2a.Task, 0, len(tasks))
 	for _, task := range tasks {
 		parsed, err := task.Parse()
 		if err != nil {
@@ -106,12 +107,13 @@ func ParseTasks(tasks []Task) ([]*protocol.Task, error) {
 }
 
 type PushNotification struct {
-	ID        string     `json:"id"`
-	TaskID    string     `json:"task_id"`
-	CreatedAt time.Time  `json:"created_at"`
-	UpdatedAt time.Time  `json:"updated_at"`
-	DeletedAt *time.Time `json:"deleted_at,omitempty"`
-	Data      string     `json:"data"` // JSON-serialized push notification config
+	ID              string     `json:"id"`
+	TaskID          string     `json:"task_id"`
+	CreatedAt       time.Time  `json:"created_at"`
+	UpdatedAt       time.Time  `json:"updated_at"`
+	DeletedAt       *time.Time `json:"deleted_at,omitempty"`
+	Data            string     `json:"data"` // JSON-serialized push notification config
+	ProtocolVersion *string    `json:"protocol_version,omitempty"`
 }
 
 // FeedbackIssueType represents the category of feedback issue

go/api/v1alpha2/agent_types.go

@@ -26,8 +26,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"trpc.group/trpc-go/trpc-a2a-go/server"
 )
 
 // AgentType represents the agent type
@@ -557,7 +555,33 @@ type A2AConfig struct {
 	Skills []AgentSkill `json:"skills,omitempty"`
 }
 
-type AgentSkill server.AgentSkill
+// AgentSkill describes a specific capability or function of the agent.
+type AgentSkill struct {
+	// ID is the unique identifier for the skill.
+	// +optional
+	ID string `json:"id,omitempty"`
+	// Name is the human-readable name of the skill.
+	// +kubebuilder:validation:MinLength=1
+	// +required
+	Name string `json:"name"`
+	// Description is an optional detailed description of the skill.
+	// +optional
+	Description string `json:"description,omitempty"`
+	// Tags are optional tags for categorization.
+	// +optional
+	// +kubebuilder:validation:MaxItems=20
+	Tags []string `json:"tags,omitempty"`
+	// Examples are optional usage examples.
+	// +optional
+	// +kubebuilder:validation:MaxItems=20
+	Examples []string `json:"examples,omitempty"`
+	// InputModes are the supported input MIME types for this skill, overriding the agent's defaults.
+	// +optional
+	InputModes []string `json:"inputModes,omitempty"`
+	// OutputModes are the supported output MIME types for this skill, overriding the agent's defaults.
+	// +optional
+	OutputModes []string `json:"outputModes,omitempty"`
+}
 
 const (
 	AgentConditionTypeAccepted            = "Accepted"

go/api/v1alpha2/zz_generated.deepcopy.go

@@ -3,24 +3,27 @@ package a2a
 import (
 	"fmt"
 	"net/http"
+	"slices"
 	"strings"
 	"sync"
 
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	a2aclient "github.com/a2aproject/a2a-go/v2/a2aclient"
+	"github.com/a2aproject/a2a-go/v2/a2acompat/a2av0"
+	"github.com/a2aproject/a2a-go/v2/a2asrv"
 	"github.com/gorilla/mux"
 	authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth"
 	common "github.com/kagent-dev/kagent/go/core/internal/utils"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
-	"trpc.group/trpc-go/trpc-a2a-go/client"
-	"trpc.group/trpc-go/trpc-a2a-go/server"
 )
 
 // A2AHandlerMux is an interface that defines methods for adding, getting, and removing agentic task handlers.
 type A2AHandlerMux interface {
 	SetAgentHandler(
 		agentRef string,
-		client *client.A2AClient,
-		card server.AgentCard,
-		tracing server.Middleware,
+		client *a2aclient.Client,
+		card a2atype.AgentCard,
+		tracing middleware,
 	) error
 	RemoveAgentHandler(
 		agentRef string,
@@ -38,6 +41,10 @@ type handlerMux struct {
 
 var _ A2AHandlerMux = &handlerMux{}
 
+type middleware interface {
+	Wrap(next http.Handler) http.Handler
+}
+
 func NewA2AHttpMux(agentPathPrefix, sandboxPathPrefix string, authenticator auth.AuthProvider) *handlerMux {
 	return &handlerMux{
 		handlers:          make(map[string]http.Handler),
@@ -49,23 +56,47 @@ func NewA2AHttpMux(agentPathPrefix, sandboxPathPrefix string, authenticator auth
 
 func (a *handlerMux) SetAgentHandler(
 	agentRef string,
-	client *client.A2AClient,
-	card server.AgentCard,
-	tracing server.Middleware,
+	client *a2aclient.Client,
+	card a2atype.AgentCard,
+	tracing middleware,
 ) error {
-	middlewares := []server.Middleware{authimpl.NewA2AAuthenticator(a.authenticator)}
+	requestHandler := NewPassthroughRequestHandler(client, &card)
+	legacyJSONRPCHandler := a2av0.NewJSONRPCHandler(requestHandler)
+	v1JSONRPCHandler := a2asrv.NewJSONRPCHandler(requestHandler)
+	cardHandler := a2asrv.NewAgentCardHandler(a2av0.NewStaticAgentCardProducer(&card))
+	wellKnownPath := "/" + strings.TrimPrefix(a2asrv.WellKnownAgentCardPath, "/")
+
+	var handler http.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, wellKnownPath) {
+			cardHandler.ServeHTTP(w, r)
+			return
+		}
+		wireVersion, err := common.NegotiateA2AWireVersion(r)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		switch wireVersion {
+		case common.A2AWireVersionLegacy:
+			legacyJSONRPCHandler.ServeHTTP(w, r)
+		case common.A2AWireVersionV1:
+			v1JSONRPCHandler.ServeHTTP(w, r)
+		default:
+			http.Error(w, fmt.Sprintf("unknown negotiated A2A wire version %q", wireVersion), http.StatusBadRequest)
+		}
+	})
+	middlewares := []middleware{authimpl.NewA2AAuthenticator(a.authenticator)}
 	if tracing != nil {
 		middlewares = append(middlewares, tracing)
 	}
-	srv, err := server.NewA2AServer(card, NewPassthroughManager(client), server.WithMiddleWare(middlewares...))
-	if err != nil {
-		return fmt.Errorf("failed to create A2A server: %w", err)
+	for _, middleware := range slices.Backward(middlewares) {
+		handler = middleware.Wrap(handler)
 	}
 
 	a.lock.Lock()
 	defer a.lock.Unlock()
 
-	a.handlers[agentRef] = srv.Handler()
+	a.handlers[agentRef] = handler
 
 	return nil
 }