Refactor Pulsar Admin Brokers Tools with Builder Pattern

- Introduced a new builder pattern for Pulsar admin brokers tools, enhancing modularity and maintainability.
- Migrated tool definitions and handler logic to the new `PulsarAdminBrokersToolBuilder`, improving code organization.
- Updated tool registration logic to utilize the builder, ensuring a unified approach for managing Pulsar broker 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/pulsar/brokers.go

@@ -0,0 +1,388 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pulsar
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/mark3labs/mcp-go/mcp"
+	"github.com/mark3labs/mcp-go/server"
+	"github.com/streamnative/pulsarctl/pkg/cmdutils"
+	"github.com/streamnative/streamnative-mcp-server/pkg/mcp/builders"
+	"github.com/streamnative/streamnative-mcp-server/pkg/pulsar"
+)
+
+// PulsarAdminBrokersToolBuilder implements the ToolBuilder interface for Pulsar admin brokers
+type PulsarAdminBrokersToolBuilder struct {
+	*builders.BaseToolBuilder
+}
+
+// NewPulsarAdminBrokersToolBuilder creates a new Pulsar admin brokers tool builder instance
+func NewPulsarAdminBrokersToolBuilder() *PulsarAdminBrokersToolBuilder {
+	metadata := builders.ToolMetadata{
+		Name:        "pulsar_admin_brokers",
+		Version:     "1.0.0",
+		Description: "Pulsar admin brokers management tools",
+		Category:    "pulsar_admin",
+		Tags:        []string{"pulsar", "admin", "brokers"},
+	}
+
+	features := []string{
+		"pulsar-admin-brokers",
+		"all",
+		"all-pulsar",
+		"pulsar-admin",
+	}
+
+	return &PulsarAdminBrokersToolBuilder{
+		BaseToolBuilder: builders.NewBaseToolBuilder(metadata, features),
+	}
+}
+
+// BuildTools builds the Pulsar admin brokers tool list
+func (b *PulsarAdminBrokersToolBuilder) BuildTools(_ context.Context, config builders.ToolBuildConfig) ([]server.ServerTool, error) {
+	// Check features - return empty list if no required features are present
+	if !b.HasAnyRequiredFeature(config.Features) {
+		return nil, nil
+	}
+
+	// Validate configuration (only validate when matching features are present)
+	if err := b.Validate(config); err != nil {
+		return nil, err
+	}
+
+	// Build tools
+	tool := b.buildPulsarAdminBrokersTool()
+	handler := b.buildPulsarAdminBrokersHandler(config.ReadOnly)
+
+	return []server.ServerTool{
+		{
+			Tool:    tool,
+			Handler: handler,
+		},
+	}, nil
+}
+
+// buildPulsarAdminBrokersTool builds the Pulsar admin brokers MCP tool definition
+func (b *PulsarAdminBrokersToolBuilder) buildPulsarAdminBrokersTool() mcp.Tool {
+	return mcp.NewTool("pulsar_admin_brokers",
+		mcp.WithDescription("Unified tool for managing Apache Pulsar broker resources. This tool integrates multiple broker management functions, including:\n"+
+			"1. List active brokers in a cluster (resource=brokers, operation=list)\n"+
+			"2. Check broker health status (resource=health, operation=get)\n"+
+			"3. Manage broker configurations (resource=config, operation=get/update/delete)\n"+
+			"4. View namespaces owned by a broker (resource=namespaces, operation=get)\n\n"+
+			"Different functions are accessed by combining resource and operation parameters, with other parameters used selectively based on operation type.\n"+
+			"Example: {\"resource\": \"config\", \"operation\": \"get\", \"configType\": \"dynamic\"} retrieves all dynamic configuration names.\n"+
+			"This tool requires Pulsar super-user permissions."),
+		mcp.WithString("resource", mcp.Required(),
+			mcp.Description("Type of resource to access, available options:\n"+
+				"- brokers: Manage broker listings\n"+
+				"- health: Check broker health status\n"+
+				"- config: Manage broker configurations\n"+
+				"- namespaces: Manage namespaces owned by a broker"),
+		),
+		mcp.WithString("operation", mcp.Required(),
+			mcp.Description("Operation to perform, available options:\n"+
+				"- list: List resources (used with brokers)\n"+
+				"- get: Retrieve resource information (used with health, config, namespaces)\n"+
+				"- update: Update a resource (used with config)\n"+
+				"- delete: Delete a resource (used with config)"),
+		),
+		mcp.WithString("clusterName",
+			mcp.Description("Pulsar cluster name, required for these operations:\n"+
+				"- When resource=brokers, operation=list\n"+
+				"- When resource=namespaces, operation=get"),
+		),
+		mcp.WithString("brokerUrl",
+			mcp.Description("Broker URL, such as '127.0.0.1:8080', required for these operations:\n"+
+				"- When resource=namespaces, operation=get"),
+		),
+		mcp.WithString("configType",
+			mcp.Description("Configuration type, required when resource=config, operation=get, available options:\n"+
+				"- dynamic: Get list of dynamically modifiable configuration names\n"+
+				"- runtime: Get all runtime configurations (including static and dynamic configs)\n"+
+				"- internal: Get internal configuration information\n"+
+				"- all_dynamic: Get all dynamic configurations and their current values"),
+		),
+		mcp.WithString("configName",
+			mcp.Description("Configuration parameter name, required for these operations:\n"+
+				"- When resource=config, operation=update\n"+
+				"- When resource=config, operation=delete"),
+		),
+		mcp.WithString("configValue",
+			mcp.Description("Configuration parameter value, required for these operations:\n"+
+				"- When resource=config, operation=update"),
+		),
+	)
+}
+
+// buildPulsarAdminBrokersHandler builds the Pulsar admin brokers handler function
+func (b *PulsarAdminBrokersToolBuilder) buildPulsarAdminBrokersHandler(readOnly bool) func(context.Context, mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+		// Get Pulsar session from context
+		session := b.getPulsarSession(ctx)
+		if session == nil {
+			return mcp.NewToolResultError("Pulsar session not found in context"), nil
+		}
+
+		// Get admin client
+		client, err := session.GetAdminClient()
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to get admin client: %v", err)), nil
+		}
+
+		// Get required parameters
+		resource, err := request.RequireString("resource")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required resource parameter. "+
+				"Please specify one of: brokers, health, config, namespaces."), nil
+		}
+
+		operation, err := request.RequireString("operation")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required operation parameter. "+
+				"Please specify one of: list, get, update, delete based on the resource type."), nil
+		}
+
+		// Validate if the parameter combination is valid
+		validCombination, errMsg := b.validateResourceOperation(resource, operation)
+		if !validCombination {
+			return mcp.NewToolResultError(errMsg), nil
+		}
+
+		// Process request based on resource type
+		switch resource {
+		case "brokers":
+			return b.handleBrokersResource(client, operation, request)
+		case "health":
+			return b.handleHealthResource(client, operation, request)
+		case "config":
+			// Check write operation permissions
+			if (operation == "update" || operation == "delete") && readOnly {
+				return mcp.NewToolResultError("Configuration update/delete operations not allowed in read-only mode. " +
+					"Please contact your administrator if you need to modify broker configurations."), nil
+			}
+			return b.handleConfigResource(client, operation, request)
+		case "namespaces":
+			return b.handleNamespacesResource(client, operation, request)
+		default:
+			return mcp.NewToolResultError(fmt.Sprintf("Unsupported resource: %s. "+
+				"Please use one of: brokers, health, config, namespaces.", resource)), nil
+		}
+	}
+}
+
+// Helper functions
+
+// getPulsarSession gets the Pulsar session from context
+// Uses the same context key as defined in pkg/mcp/ctx.go to ensure consistency
+func (b *PulsarAdminBrokersToolBuilder) getPulsarSession(ctx context.Context) *pulsar.Session {
+	type contextKey string
+	const pulsarSessionContextKey contextKey = "pulsar_session"
+	
+	session, ok := ctx.Value(pulsarSessionContextKey).(*pulsar.Session)
+	if !ok {
+		return nil
+	}
+	return session
+}
+
+// validateResourceOperation validates if the resource and operation combination is valid
+func (b *PulsarAdminBrokersToolBuilder) validateResourceOperation(resource, operation string) (bool, string) {
+	validCombinations := map[string][]string{
+		"brokers":    {"list"},
+		"health":     {"get"},
+		"config":     {"get", "update", "delete"},
+		"namespaces": {"get"},
+	}
+
+	if operations, ok := validCombinations[resource]; ok {
+		for _, op := range operations {
+			if op == operation {
+				return true, ""
+			}
+		}
+		return false, fmt.Sprintf("Invalid operation '%s' for resource '%s'. Valid operations are: %v",
+			operation, resource, validCombinations[resource])
+	}
+
+	return false, fmt.Sprintf("Invalid resource '%s'. Valid resources are: brokers, health, config, namespaces", resource)
+}
+
+// handleBrokersResource handles brokers resource
+func (b *PulsarAdminBrokersToolBuilder) handleBrokersResource(client cmdutils.Client, operation string, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	switch operation {
+	case "list":
+		clusterName, err := request.RequireString("clusterName")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'clusterName'. " +
+				"Please provide the name of the Pulsar cluster to list brokers for."), nil
+		}
+
+		brokers, err := client.Brokers().GetActiveBrokers(clusterName)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to get active brokers: %v. "+
+				"Please verify the cluster name and ensure the Pulsar service is running.", err)), nil
+		}
+
+		brokersJSON, err := json.Marshal(brokers)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to serialize brokers list: %v", err)), nil
+		}
+
+		return mcp.NewToolResultText(string(brokersJSON)), nil
+	default:
+		return mcp.NewToolResultError(fmt.Sprintf("Unsupported operation '%s' for brokers resource. "+
+			"The only supported operation is 'list'.", operation)), nil
+	}
+}
+
+// handleHealthResource handles health resource
+func (b *PulsarAdminBrokersToolBuilder) handleHealthResource(client cmdutils.Client, operation string, _ mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	switch operation {
+	case "get":
+		//nolint:staticcheck
+		err := client.Brokers().HealthCheck()
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Broker health check failed: %v. "+
+				"The broker might be down or experiencing issues.", err)), nil
+		}
+		return mcp.NewToolResultText("ok"), nil
+	default:
+		return mcp.NewToolResultError(fmt.Sprintf("Unsupported operation '%s' for health resource. "+
+			"The only supported operation is 'get'.", operation)), nil
+	}
+}
+
+// handleConfigResource handles config resource
+func (b *PulsarAdminBrokersToolBuilder) handleConfigResource(client cmdutils.Client, operation string, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	switch operation {
+	case "get":
+		configType, err := request.RequireString("configType")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'configType'. " +
+				"Please specify one of: dynamic, runtime, internal, all_dynamic."), nil
+		}
+
+		var result interface{}
+		var fetchErr error
+
+		switch configType {
+		case "dynamic":
+			result, fetchErr = client.Brokers().GetDynamicConfigurationNames()
+		case "runtime":
+			result, fetchErr = client.Brokers().GetRuntimeConfigurations()
+		case "internal":
+			result, fetchErr = client.Brokers().GetInternalConfigurationData()
+		case "all_dynamic":
+			result, fetchErr = client.Brokers().GetAllDynamicConfigurations()
+		default:
+			return mcp.NewToolResultError(fmt.Sprintf("Invalid config type: '%s'. "+
+				"Valid types are: dynamic, runtime, internal, all_dynamic.", configType)), nil
+		}
+
+		if fetchErr != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to get %s configuration: %v", configType, fetchErr)), nil
+		}
+
+		resultJSON, err := json.Marshal(result)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to serialize configuration: %v", err)), nil
+		}
+
+		return mcp.NewToolResultText(string(resultJSON)), nil
+
+	case "update":
+		configName, err := request.RequireString("configName")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'configName'. " +
+				"Please provide the name of the configuration parameter to update."), nil
+		}
+
+		configValue, err := request.RequireString("configValue")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'configValue'. " +
+				"Please provide the new value for the configuration parameter."), nil
+		}
+
+		err = client.Brokers().UpdateDynamicConfiguration(configName, configValue)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to update configuration: %v. "+
+				"Please verify the configuration name is valid and the value is of the correct type.", err)), nil
+		}
+
+		return mcp.NewToolResultText(fmt.Sprintf("Dynamic configuration '%s' updated successfully to '%s'",
+			configName, configValue)), nil
+
+	case "delete":
+		configName, err := request.RequireString("configName")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'configName'. " +
+				"Please provide the name of the configuration parameter to delete."), nil
+		}
+
+		err = client.Brokers().DeleteDynamicConfiguration(configName)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to delete configuration: %v. "+
+				"Please verify the configuration name is valid and exists.", err)), nil
+		}
+
+		return mcp.NewToolResultText(fmt.Sprintf("Dynamic configuration '%s' deleted successfully", configName)), nil
+
+	default:
+		return mcp.NewToolResultError(fmt.Sprintf("Unsupported operation '%s' for config resource. "+
+			"Supported operations are: get, update, delete.", operation)), nil
+	}
+}
+
+// handleNamespacesResource handles namespaces resource
+func (b *PulsarAdminBrokersToolBuilder) handleNamespacesResource(client cmdutils.Client, operation string, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+	switch operation {
+	case "get":
+		clusterName, err := request.RequireString("clusterName")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'clusterName'. " +
+				"Please provide the name of the Pulsar cluster."), nil
+		}
+
+		brokerURL, err := request.RequireString("brokerUrl")
+		if err != nil {
+			return mcp.NewToolResultError("Missing required parameter 'brokerUrl'. " +
+				"Please provide the URL of the broker (e.g., '127.0.0.1:8080')."), nil
+		}
+
+		namespaces, err := client.Brokers().GetOwnedNamespaces(clusterName, brokerURL)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to get owned namespaces: %v. "+
+				"Please verify the cluster name and broker URL are correct.", err)), nil
+		}
+
+		namespacesJSON, err := json.Marshal(namespaces)
+		if err != nil {
+			return mcp.NewToolResultError(fmt.Sprintf("Failed to serialize namespaces: %v", err)), nil
+		}
+
+		return mcp.NewToolResultText(string(namespacesJSON)), nil
+
+	default:
+		return mcp.NewToolResultError(fmt.Sprintf("Unsupported operation '%s' for namespaces resource. "+
+			"The only supported operation is 'get'.", operation)), nil
+	}
+}