[feat]: [DBOPS-1395]: Add support for dbops changeset agent (#103)

* [feat]: [DBOPS-1395]: Add support for dbops changeset agent

* Merge branch 'master' into DBOPS-1395

# Conflicts:
#	cmd/harness-mcp-server/main.go
#	pkg/modules/unlicensed.go

* [feat]: [DBOPS-1395]: Add support for dbops changeset agent

Author: archit-harness

client/dto/genai.go

@@ -52,21 +52,74 @@ const (
 	CallerUnifiedAgent Caller = "unified_agent"
 )
 
-type ServiceChatParameters struct {
+// GenAIRequest defines the interface for all GenAI request types
+type GenAIRequest interface {
+	GetBaseParameters() *BaseRequestParameters
+	IsStreaming() bool
+}
+
+// BaseRequestParameters defines common fields for all GenAI request types
+type BaseRequestParameters struct {
 	Prompt          string          `json:"prompt"`
 	Provider        string          `json:"provider,omitempty"`
 	ModelName       string          `json:"model_name,omitempty"`
 	ConversationID  string          `json:"conversation_id"`
 	InteractionID   string          `json:"interaction_id,omitempty"`
-	Capabilities    []Capability    `json:"capabilities"`
+	Capabilities    []Capability    `json:"capabilities,omitempty"`
 	ConversationRaw []any           `json:"conversation_raw,omitempty"`
 	Context         []ContextItem   `json:"context,omitempty"`
-	Action          RequestAction   `json:"action,omitempty"`
 	HarnessContext  *HarnessContext `json:"harness_context,omitempty"`
 	Stream          bool            `json:"stream,omitempty"`
 	Caller          Caller          `json:"caller,omitempty"`
 }
 
+// GetBaseParameters returns the base parameters
+func (b *BaseRequestParameters) GetBaseParameters() *BaseRequestParameters {
+	return b
+}
+
+// IsStreaming returns whether the request is streaming
+func (b *BaseRequestParameters) IsStreaming() bool {
+	return b.Stream
+}
+
+// ServiceChatParameters extends BaseRequestParameters for AI DevOps agent requests
+type ServiceChatParameters struct {
+	BaseRequestParameters
+	Action RequestAction `json:"action,omitempty"`
+}
+
+// GetBaseParameters returns the base parameters
+func (s *ServiceChatParameters) GetBaseParameters() *BaseRequestParameters {
+	return &s.BaseRequestParameters
+}
+
+// IsStreaming returns whether the request is streaming
+func (s *ServiceChatParameters) IsStreaming() bool {
+	return s.BaseRequestParameters.Stream
+}
+
+// DatabaseType defines the supported database types
+type DatabaseType string
+
+// DBChangesetParameters extends BaseRequestParameters for database changeset generation
+type DBChangesetParameters struct {
+	BaseRequestParameters
+	DatabaseType DatabaseType `json:"database_type"`
+	OldChangeset string       `json:"oldchangeset,omitempty"`
+	ErrorContext string       `json:"error_context,omitempty"`
+}
+
+// GetBaseParameters returns the base parameters
+func (d *DBChangesetParameters) GetBaseParameters() *BaseRequestParameters {
+	return &d.BaseRequestParameters
+}
+
+// IsStreaming returns whether the request is streaming
+func (d *DBChangesetParameters) IsStreaming() bool {
+	return d.BaseRequestParameters.Stream
+}
+
 type CapabilityToRun struct {
 	CallID string         `json:"call_id"`
 	Type   string         `json:"type"`

client/genai.go

@@ -15,16 +15,16 @@ import (
 
 const (
 	aiDevopsChatPath = "chat/platform"
+	dbChangesetPath  = "chat/db-changeset"
 )
 
 type GenaiService struct {
 	Client *Client
 }
 
-// SendAIDevOpsChat sends a request to the AI DevOps service and returns the response.
-// If request.Stream is true and onProgress is provided, it will handle streaming responses.
-// For non-streaming requests or when onProgress is nil, it will use the standard request flow.
-func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, request *dto.ServiceChatParameters, onProgress ...func(progressUpdate dto.ProgressUpdate) error) (*dto.ServiceChatResponse, error) {
+// sendGenAIRequest is a generic method to send requests to GenAI services
+func (g *GenaiService) sendGenAIRequest(ctx context.Context, path string, scope dto.Scope, request dto.GenAIRequest,
+	onProgress ...func(progressUpdate dto.ProgressUpdate) error) (*dto.ServiceChatResponse, error) {
 	if g == nil || g.Client == nil {
 		return nil, fmt.Errorf("genai service or client is nil")
 	}
@@ -56,11 +56,13 @@ func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, re
 		params["projectIdentifier"] = scope.ProjectID
 	}
 
+	// Determine if streaming is required
+	shouldStream := request.IsStreaming() && progressCB != nil
+
 	// Handle non-streaming case with early return
-	isStreaming := request.Stream && progressCB != nil
-	if !isStreaming {
+	if !shouldStream {
 		var response dto.ServiceChatResponse
-		err := g.Client.Post(ctx, aiDevopsChatPath, params, request, &response)
+		err := g.Client.Post(ctx, path, params, request, &response)
 		if err != nil {
 			return nil, fmt.Errorf("failed to send request to genai service: %w", err)
 		}
@@ -69,7 +71,7 @@ func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, re
 	}
 
 	// Execute the streaming request
-	resp, err := g.Client.PostStream(ctx, aiDevopsChatPath, params, request)
+	resp, err := g.Client.PostStream(ctx, path, params, request)
 	if err != nil {
 		slog.Warn("Failed to execute streaming request", "error", err.Error())
 		return nil, fmt.Errorf("failed to execute streaming request: %w", err)
@@ -84,7 +86,7 @@ func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, re
 
 	// Initialize the response with conversation ID from request
 	finalResponse := &dto.ServiceChatResponse{
-		ConversationID: request.ConversationID,
+		ConversationID: request.GetBaseParameters().ConversationID,
 	}
 
 	// Process the streaming response
@@ -97,6 +99,19 @@ func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, re
 	return finalResponse, nil
 }
 
+// SendAIDevOpsChat sends a request to the AI DevOps service and returns the response.
+// If request.Stream is true and onProgress is provided, it will handle streaming responses.
+// For non-streaming requests or when onProgress is nil, it will use the standard request flow.
+func (g *GenaiService) SendAIDevOpsChat(ctx context.Context, scope dto.Scope, request *dto.ServiceChatParameters, onProgress ...func(progressUpdate dto.ProgressUpdate) error) (*dto.ServiceChatResponse, error) {
+	return g.sendGenAIRequest(ctx, aiDevopsChatPath, scope, request, onProgress...)
+}
+
+// SendDBChangeset sends a request to generate database changesets and returns the response.
+// If request.Stream is true and onProgress is provided, it will handle streaming responses.
+func (g *GenaiService) SendDBChangeset(ctx context.Context, scope dto.Scope, request *dto.DBChangesetParameters, onProgress ...func(progressUpdate dto.ProgressUpdate) error) (*dto.ServiceChatResponse, error) {
+	return g.sendGenAIRequest(ctx, dbChangesetPath, scope, request, onProgress...)
+}
+
 // processStreamingResponse handles Server-Sent Events (SSE) streaming responses
 // and accumulates complete events before forwarding them with appropriate event types
 func (g *GenaiService) processStreamingResponse(body io.ReadCloser, finalResponse *dto.ServiceChatResponse, onProgress func(dto.ProgressUpdate) error) error {

pkg/harness/tools/dbops.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+
 	"github.com/harness/harness-mcp/client/dbops"
 	"github.com/harness/harness-mcp/cmd/harness-mcp-server/config"
 	"github.com/mark3labs/mcp-go/mcp"