Refactor Pulsar Admin Tools with Builder Pattern

- Introduced builder pattern for various Pulsar admin tools, enhancing modularity and maintainability.
- Migrated tool definitions and handler logic to respective builders, improving code organization and readability.
- Updated tool registration logic to utilize the new builders, ensuring a unified approach for managing Pulsar operations.
- Enhanced error handling and added comments for better clarity and understanding of the code structure.
- Maintained backward compatibility while integrating the new builder pattern.

Author: freeznet

pkg/mcp/builders/kafka/schema_registry.go

@@ -348,7 +348,8 @@ func (b *KafkaSchemaRegistryToolBuilder) handleSchemaSubjectDelete(ctx context.C
 		return b.handleError("get subject name", err), nil
 	}
 
-	versions, err := client.DeleteSubject(ctx, subject)
+	// Delete subject using correct API signature (soft delete by default)
+	versions, err := client.DeleteSubject(ctx, subject, sr.SoftDelete)
 	if err != nil {
 		return b.handleError("delete schema subject", err), nil
 	}
@@ -416,30 +417,41 @@ func (b *KafkaSchemaRegistryToolBuilder) handleSchemaVersionDelete(ctx context.C
 		return b.handleError("parse version number", err), nil
 	}
 
-	deletedVersion, err := client.DeleteSchemaVersion(ctx, subject, version)
+	// Delete schema version using correct API signature (soft delete by default)
+	err = client.DeleteSchema(ctx, subject, version, sr.SoftDelete)
 	if err != nil {
 		return b.handleError("delete schema version", err), nil
 	}
-	return b.marshalResponse(deletedVersion)
+	
+	return mcp.NewToolResultText(fmt.Sprintf("Schema version %d for subject %s deleted successfully", version, subject)), nil
 }
 
 // handleSchemaCompatibilityGet handles getting compatibility setting
 func (b *KafkaSchemaRegistryToolBuilder) handleSchemaCompatibilityGet(ctx context.Context, client *sr.Client, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-	subject := request.GetString("subject") // Optional for global compatibility
-
-	var compatibility sr.CompatibilityLevel
-	var err error
+	subject := request.GetString("subject", "") // Optional for global compatibility
 
+	var results []sr.CompatibilityResult
 	if subject != "" {
-		compatibility, err = client.Compatibility(ctx, subject)
+		// Get compatibility for specific subject
+		results = client.Compatibility(ctx, subject)
 	} else {
-		compatibility, err = client.Config(ctx)
+		// Get global compatibility
+		results = client.Compatibility(ctx)
 	}
 
-	if err != nil {
-		return b.handleError("get compatibility setting", err), nil
+	// Check for errors in results
+	for _, result := range results {
+		if result.Err != nil {
+			return b.handleError("get compatibility setting", result.Err), nil
+		}
+	}
+
+	// Return the first result (there should only be one)
+	if len(results) > 0 {
+		return b.marshalResponse(map[string]string{"compatibility": results[0].Level.String()})
 	}
-	return b.marshalResponse(map[string]string{"compatibility": string(compatibility)})
+
+	return mcp.NewToolResultError("No compatibility result returned"), nil
 }
 
 // handleSchemaCompatibilitySet handles setting compatibility level
@@ -449,31 +461,42 @@ func (b *KafkaSchemaRegistryToolBuilder) handleSchemaCompatibilitySet(ctx contex
 		return b.handleError("get compatibility level", err), nil
 	}
 
-	subject := request.GetString("subject") // Optional for global compatibility
+	subject := request.GetString("subject", "") // Optional for global compatibility
 
 	// Parse compatibility level
 	var compatibility sr.CompatibilityLevel
 	switch strings.ToUpper(compatibilityStr) {
 	case "BACKWARD":
-		compatibility = sr.CompatibilityBackward
+		compatibility = sr.CompatBackward
 	case "FORWARD":
-		compatibility = sr.CompatibilityForward
+		compatibility = sr.CompatForward
 	case "FULL":
-		compatibility = sr.CompatibilityFull
+		compatibility = sr.CompatFull
 	case "NONE":
-		compatibility = sr.CompatibilityNone
+		compatibility = sr.CompatNone
 	default:
 		return mcp.NewToolResultError(fmt.Sprintf("Invalid compatibility level: %s. Valid levels: BACKWARD, FORWARD, FULL, NONE", compatibilityStr)), nil
 	}
 
+	// Create SetCompatibility request
+	setCompat := sr.SetCompatibility{
+		Level: compatibility,
+	}
+
+	var results []sr.CompatibilityResult
 	if subject != "" {
-		_, err = client.SetCompatibility(ctx, subject, compatibility)
+		// Set compatibility for specific subject
+		results = client.SetCompatibility(ctx, setCompat, subject)
 	} else {
-		_, err = client.SetConfig(ctx, compatibility)
+		// Set global compatibility
+		results = client.SetCompatibility(ctx, setCompat)
 	}
 
-	if err != nil {
-		return b.handleError("set compatibility level", err), nil
+	// Check for errors in results
+	for _, result := range results {
+		if result.Err != nil {
+			return b.handleError("set compatibility level", result.Err), nil
+		}
 	}
 
 	return mcp.NewToolResultText(fmt.Sprintf("Compatibility level set to %s", compatibilityStr)), nil

pkg/mcp/builders/kafka/topics.go

@@ -21,12 +21,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"strconv"
 	"strings"
 
 	"github.com/mark3labs/mcp-go/mcp"
 	"github.com/mark3labs/mcp-go/server"
-	"github.com/streamnative/streamnative-mcp-server/pkg/common"
 	"github.com/streamnative/streamnative-mcp-server/pkg/kafka"
 	"github.com/streamnative/streamnative-mcp-server/pkg/mcp/builders"
 	"github.com/twmb/franz-go/pkg/kadm"
@@ -266,12 +264,7 @@ func (b *KafkaTopicsToolBuilder) getKafkaAdminClient(ctx context.Context) (*kadm
 
 // handleKafkaTopicsList handles listing all topics
 func (b *KafkaTopicsToolBuilder) handleKafkaTopicsList(ctx context.Context, admin *kadm.Client, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
-	includeInternal := false
-	if val, exists := request.Arguments["includeInternal"]; exists {
-		if boolVal, ok := val.(bool); ok {
-			includeInternal = boolVal
-		}
-	}
+	includeInternal := request.GetBool("includeInternal", false)
 
 	topics, err := admin.ListTopics(ctx)
 	if err != nil {
@@ -314,12 +307,12 @@ func (b *KafkaTopicsToolBuilder) handleKafkaTopicCreate(ctx context.Context, adm
 		return b.handleError("get topic name", err), nil
 	}
 
-	partitionsNum, err := request.RequireNumber("partitions")
+	partitionsNum, err := request.RequireInt("partitions")
 	if err != nil {
 		return b.handleError("get partitions", err), nil
 	}
 
-	replicationFactorNum, err := request.RequireNumber("replicationFactor")
+	replicationFactorNum, err := request.RequireInt("replicationFactor")
 	if err != nil {
 		return b.handleError("get replication factor", err), nil
 	}
@@ -329,7 +322,8 @@ func (b *KafkaTopicsToolBuilder) handleKafkaTopicCreate(ctx context.Context, adm
 
 	// Parse optional configs
 	var configs map[string]*string
-	if configsParam, exists := request.Arguments["configs"]; exists {
+	arguments := request.GetArguments()
+	if configsParam, exists := arguments["configs"]; exists {
 		if configsMap, ok := configsParam.(map[string]interface{}); ok {
 			configs = make(map[string]*string)
 			for key, value := range configsMap {
@@ -344,15 +338,8 @@ func (b *KafkaTopicsToolBuilder) handleKafkaTopicCreate(ctx context.Context, adm
 		}
 	}
 
-	// Create topic request
-	req := kadm.TopicRequest{
-		Topic:             topicName,
-		NumPartitions:     partitions,
-		ReplicationFactor: replicationFactor,
-		Configs:           configs,
-	}
-
-	results, err := admin.CreateTopics(ctx, req)
+	// Create topic using the correct CreateTopics API
+	results, err := admin.CreateTopics(ctx, partitions, replicationFactor, configs, topicName)
 	if err != nil {
 		return b.handleError("create Kafka topic", err), nil
 	}