Add warnings against using Kafka tools for Pulsar operations and vice versa in respective tool descriptions.

Author: freeznet

pkg/mcp/kafka_admin_topics_tools.go

@@ -49,6 +49,7 @@ func KafkaAdminAddTopicTools(s *server.MCPServer, readOnly bool, features []stri
 	toolDesc := "Unified tool for managing Apache Kafka topics.\n" +
 		"This tool provides access to various Kafka topic operations, including creation, deletion, listing, and configuration retrieval.\n" +
 		"Kafka topics are distributed append-only logs that store ordered, immutable records. Topics can be partitioned for parallel processing.\n\n" +
+		"Do not use this tool for Pulsar protocol operations. Use 'pulsar_admin_topics' instead." +
 		"Usage Examples:\n\n" +
 		"1. List all topics:\n" +
 		"   resource: \"topics\"\n" +

pkg/mcp/kafka_client_consume_tools.go

@@ -47,7 +47,7 @@ func KafkaClientAddConsumeTools(s *server.MCPServer, _ bool, logrusLogger *logru
 		"- Offsets track the position of consumers in each partition, allowing resumption after failures\n" +
 		"- Partitions are independent ordered sequences of messages that enable parallel processing\n\n" +
 		"This tool provides a temporary consumer instance for diagnostic and testing purposes. " +
-		"It does not commit offsets back to Kafka unless the 'group' parameter is explicitly specified.\n\n" +
+		"It does not commit offsets back to Kafka unless the 'group' parameter is explicitly specified. Do not use this tool for Pulsar protocol operations. Use 'pulsar_client_consume' instead.\n\n" +
 		"Usage Examples:\n\n" +
 		"1. Basic consumption - Get 10 earliest messages from a topic:\n" +
 		"   topic: \"my-topic\"\n" +

pkg/mcp/kafka_client_produce_tools.go

@@ -49,7 +49,7 @@ func KafkaClientAddProduceTools(s *server.MCPServer, readOnly bool, features []s
 		"- Messages can include a key, which determines the partition assignment (consistent hashing)\n" +
 		"- Headers can be added to messages to include metadata without affecting the message payload\n" +
 		"- Partitions enable parallel processing and ordered delivery within a single partition\n\n" +
-		"This tool provides a simple producer instance for diagnostic and testing purposes.\n\n" +
+		"This tool provides a simple producer instance for diagnostic and testing purposes. Do not use this tool for Pulsar protocol operations. Use 'pulsar_client_produce' instead.\n\n" +
 		"Usage Examples:\n\n" +
 		"1. Basic message production - Send a simple message to a topic:\n" +
 		"   topic: \"my-topic\"\n" +

pkg/mcp/pulsar_admin_topic_tools.go

@@ -47,6 +47,7 @@ func PulsarAdminAddTopicTools(s *server.MCPServer, readOnly bool, features []str
 		"Topics follow a hierarchical naming structure: persistent://tenant/namespace/topic. " +
 		"This tool supports various operations on topics including creation, deletion, lookup, compaction, " +
 		"offloading, and retrieving statistics. " +
+		"Do not use this tool for Kafka protocol operations. Use 'kafka_admin_topics' instead." +
 		"Most operations require namespace admin permissions."
 
 	resourceDesc := "Resource to operate on. Available resources:\n" +

pkg/mcp/pulsar_client_consume_tools.go

@@ -41,7 +41,7 @@ func PulsarClientAddConsumerTools(s *server.MCPServer, _ bool, features []string
 	consumeTool := mcp.NewTool("pulsar_client_consume",
 		mcp.WithDescription("Consume messages from a Pulsar topic. This tool allows you to consume messages "+
 			"from a specified Pulsar topic with various options to control the subscription behavior, "+
-			"message processing, and display format."),
+			"message processing, and display format. Do not use this tool for Kafka protocol operations. Use 'kafka_client_consume' instead."),
 		mcp.WithString("topic", mcp.Required(),
 			mcp.Description("Topic to consume from"),
 		),

pkg/mcp/pulsar_client_produce_tools.go

@@ -42,12 +42,12 @@ func PulsarClientAddProducerTools(s *server.MCPServer, _ bool, features []string
 	produceTool := mcp.NewTool("pulsar_client_produce",
 		mcp.WithDescription("Produce messages to a Pulsar topic. This tool allows you to send messages "+
 			"to a specified Pulsar topic with various options to control message format, "+
-			"batching, and properties."),
+			"batching, and properties. Do not use this tool for Kafka protocol operations. Use 'kafka_client_produce' instead."),
 		mcp.WithString("topic", mcp.Required(),
 			mcp.Description("Topic to produce to"),
 		),
 		mcp.WithArray("messages",
-			mcp.Description("Messages to send. Specify multiple times for multiple messages."),
+			mcp.Description("Messages to send. Specify multiple times for multiple messages. IMPORTANT: Use this parameter to provide message content."),
 		),
 		mcp.WithArray("files",
 			mcp.Description("Files to send as message content. Specify multiple times for multiple files."),
@@ -73,30 +73,6 @@ func PulsarClientAddProducerTools(s *server.MCPServer, _ bool, features []string
 		mcp.WithString("key",
 			mcp.Description("Partitioning key to add to each message"),
 		),
-		mcp.WithString("key-value-key",
-			mcp.Description("Value to add as message key in KeyValue schema"),
-		),
-		mcp.WithString("key-value-key-file",
-			mcp.Description("Path to file containing the value to add as message key in KeyValue schema"),
-		),
-		mcp.WithString("value-schema",
-			mcp.Description("Schema type for the value (can be string, bytes, json, avro) (default: bytes)"),
-		),
-		mcp.WithString("key-schema",
-			mcp.Description("Schema type for the key (can be string, bytes, json, avro) (default: string)"),
-		),
-		mcp.WithString("key-value-encoding-type",
-			mcp.Description("Key Value Encoding Type (can be separated or inline)"),
-		),
-		mcp.WithString("encryption-key-name",
-			mcp.Description("The public key name to encrypt payload"),
-		),
-		mcp.WithString("encryption-key-value",
-			mcp.Description("The URI of public key to encrypt payload"),
-		),
-		mcp.WithBoolean("disable-replication",
-			mcp.Description("Disable geo-replication for messages (default: false)"),
-		),
 	)
 	s.AddTool(produceTool, handleClientProduce)
 }
@@ -128,7 +104,7 @@ func handleClientProduce(ctx context.Context, request mcp.CallToolRequest) (*mcp
 	}
 
 	if len(messages) == 0 && len(files) == 0 {
-		return mcp.NewToolResultError("Please supply message content with either --messages or --files"), nil
+		return mcp.NewToolResultError("Please supply message content with 'messages' parameter."), nil
 	}
 
 	numProduce := 1
@@ -170,50 +146,6 @@ func handleClientProduce(ctx context.Context, request mcp.CallToolRequest) (*mcp
 		key = val
 	}
 
-	keyValueKey := ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "key-value-key"); exists {
-		keyValueKey = val
-	}
-
-	keyValueKeyFile := ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "key-value-key-file"); exists {
-		keyValueKeyFile = val
-	}
-
-	// Read but not used due to API compatibility issues
-	_ = ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "value-schema"); exists && val != "" {
-		_ = val
-	}
-
-	_ = ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "key-schema"); exists && val != "" {
-		_ = val
-	}
-
-	_ = ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "key-value-encoding-type"); exists {
-		_ = val
-		if val != "separated" && val != "inline" {
-			return mcp.NewToolResultError("Invalid key-value-encoding-type: must be 'separated' or 'inline'"), nil
-		}
-	}
-
-	_ = ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "encryption-key-name"); exists {
-		_ = val
-	}
-
-	_ = ""
-	if val, exists := optionalParam[string](request.Params.Arguments, "encryption-key-value"); exists {
-		_ = val
-	}
-
-	disableReplication := false
-	if val, exists := optionalParam[bool](request.Params.Arguments, "disable-replication"); exists {
-		disableReplication = val
-	}
-
 	// Split messages by separator if needed
 	if separator != "" && len(messages) > 0 {
 		var splitMessages []string
@@ -269,20 +201,6 @@ func handleClientProduce(ctx context.Context, request mcp.CallToolRequest) (*mcp
 		}
 	}
 
-	// Get key value key bytes
-	_ = []byte(nil)
-	if keyValueKey != "" {
-		// We'd validate but not using for now
-		_ = []byte(keyValueKey)
-	} else if keyValueKeyFile != "" {
-		// We'd validate but not using for now
-		data, err := os.ReadFile(keyValueKeyFile)
-		if err != nil {
-			return mcp.NewToolResultError(fmt.Sprintf("Failed to read key-value-key-file: %v", err)), nil
-		}
-		_ = data
-	}
-
 	// Setup rate limiter
 	var limiter *time.Ticker
 	if rate > 0 {
@@ -294,7 +212,7 @@ func handleClientProduce(ctx context.Context, request mcp.CallToolRequest) (*mcp
 	// Send messages
 	numMessagesSent := 0
 	var lastMessageID pulsar.MessageID
-	for i := 0; i < numProduce; i++ {
+	for range numProduce {
 		for _, payload := range messagePayloads {
 			// Apply rate limiting if enabled
 			if limiter != nil {
@@ -316,16 +234,6 @@ func handleClientProduce(ctx context.Context, request mcp.CallToolRequest) (*mcp
 				msg.Key = key
 			}
 
-			// Handle geo-replication
-			if disableReplication {
-				msg.DisableReplication = true
-			}
-
-			// Note: KeyValue handling is commented out until correct API is confirmed
-			// if keyValueEncodingType != "" && keyValueKeyBytes != nil {
-			//     // TODO: Implement KeyValue schema handling when API is confirmed
-			// }
-
 			// Send the message
 			msgID, err := producer.Send(ctx, msg)
 			if err != nil {