diff --git a/samcli/commands/exceptions.py b/samcli/commands/exceptions.py
index 8fdaaaf438..84200f5e25 100644
--- a/samcli/commands/exceptions.py
+++ b/samcli/commands/exceptions.py
@@ -76,6 +76,12 @@ class AppPipelineTemplateManifestException(UserException):
     """
 
 
+class AppPipelineTemplateMetadataException(UserException):
+    """
+    Exception class when SAM is not able to parse the "metadata.json" file located in the SAM pipeline templates
+    """
+
+
 class PipelineFileAlreadyExistsError(UserException):
     """
     Exception class when the files to-be-generated by the pipeline template already exists on the SAM project. Instead
diff --git a/samcli/commands/pipeline/init/interactive_init_flow.py b/samcli/commands/pipeline/init/interactive_init_flow.py
index 8287531218..a126d693b3 100644
--- a/samcli/commands/pipeline/init/interactive_init_flow.py
+++ b/samcli/commands/pipeline/init/interactive_init_flow.py
@@ -2,8 +2,10 @@
 Interactive flow that prompts that users for pipeline template (cookiecutter template) and used it to generate
 pipeline configuration file
 """
+import json
 import logging
 import os
+from json import JSONDecodeError
 from pathlib import Path
 from textwrap import dedent
 from typing import Dict, List, Tuple
@@ -11,7 +13,11 @@
 import click
 
 from samcli.cli.main import global_cfg
-from samcli.commands.exceptions import PipelineTemplateCloneException, PipelineFileAlreadyExistsError
+from samcli.commands.exceptions import (
+    PipelineTemplateCloneException,
+    PipelineFileAlreadyExistsError,
+    AppPipelineTemplateMetadataException,
+)
 from samcli.lib.config.samconfig import SamConfig
 from samcli.lib.cookiecutter.interactive_flow import InteractiveFlow
 from samcli.lib.cookiecutter.interactive_flow_creator import InteractiveFlowCreator
@@ -118,7 +124,7 @@ def _generate_from_custom_location(
             )
             return self._generate_from_pipeline_template(pipeline_template_local_dir)
 
-    def _prompt_run_bootstrap_within_pipeline_init(self, stage_names: List[str], required_env_number: int) -> bool:
+    def _prompt_run_bootstrap_within_pipeline_init(self, stage_names: List[str], number_of_stages: int) -> bool:
         """
         Prompt bootstrap if `--bootstrap` flag is provided. Return True if bootstrap process is executed.
         """
@@ -128,7 +134,7 @@ def _prompt_run_bootstrap_within_pipeline_init(self, stage_names: List[str], req
             click.echo(
                 Colored().yellow(
                     f"Only {len(stage_names)} stage(s) were detected, "
-                    f"fewer than what the template requires: {required_env_number}."
+                    f"fewer than what the template requires: {number_of_stages}."
                 )
             )
         click.echo()
@@ -192,14 +198,17 @@ def _generate_from_pipeline_template(self, pipeline_template_dir: Path) -> List[
         and return the list of generated files.
         """
         pipeline_template: Template = _initialize_pipeline_template(pipeline_template_dir)
-        required_env_number = 2  # TODO: read from template
-        click.echo(f"You are using the {required_env_number}-stage pipeline template.")
-        _draw_stage_diagram(required_env_number)
+        number_of_stages = (pipeline_template.metadata or {}).get("number_of_stages")
+        if not number_of_stages:
+            LOG.debug("Cannot find number_of_stages from template's metadata, set to default 2.")
+            number_of_stages = 2
+        click.echo(f"You are using the {number_of_stages}-stage pipeline template.")
+        _draw_stage_diagram(number_of_stages)
         while True:
             click.echo("Checking for existing stages...\n")
             stage_names, bootstrap_context = _load_pipeline_bootstrap_resources()
-            if len(stage_names) < required_env_number and self._prompt_run_bootstrap_within_pipeline_init(
-                stage_names, required_env_number
+            if len(stage_names) < number_of_stages and self._prompt_run_bootstrap_within_pipeline_init(
+                stage_names, number_of_stages
             ):
                 # the customers just went through the bootstrap process,
                 # refresh the pipeline bootstrap resources and see whether bootstrap is still needed
@@ -399,7 +408,26 @@ def _initialize_pipeline_template(pipeline_template_dir: Path) -> Template:
         The initialized pipeline's cookiecutter template
     """
     interactive_flow = _get_pipeline_template_interactive_flow(pipeline_template_dir)
-    return Template(location=str(pipeline_template_dir), interactive_flows=[interactive_flow])
+    metadata = _get_pipeline_template_metadata(pipeline_template_dir)
+    return Template(location=str(pipeline_template_dir), interactive_flows=[interactive_flow], metadata=metadata)
+
+
+def _get_pipeline_template_metadata(pipeline_template_dir: Path) -> Dict:
+    """
+    Load the metadata from the file metadata.json located in the template directory,
+    raise an exception if anything wrong.
+    """
+    metadata_path = Path(pipeline_template_dir, "metadata.json")
+    if not metadata_path.exists():
+        raise AppPipelineTemplateMetadataException(f"Cannot find metadata file {metadata_path}")
+    try:
+        with open(metadata_path, "r", encoding="utf-8") as file:
+            metadata = json.load(file)
+            if isinstance(metadata, dict):
+                return metadata
+            raise AppPipelineTemplateMetadataException(f"Invalid content found in {metadata_path}")
+    except JSONDecodeError as ex:
+        raise AppPipelineTemplateMetadataException(f"Invalid JSON found in {metadata_path}") from ex
 
 
 def _get_pipeline_template_interactive_flow(pipeline_template_dir: Path) -> InteractiveFlow:
diff --git a/samcli/lib/cookiecutter/template.py b/samcli/lib/cookiecutter/template.py
index 678303dd08..46b851985e 100644
--- a/samcli/lib/cookiecutter/template.py
+++ b/samcli/lib/cookiecutter/template.py
@@ -43,6 +43,8 @@ class Template:
         An optional series of plugins to be plugged in. A plugin defines its own interactive_flow, preprocessor
         and postprocessor. A plugin is a sub-set of the template, if there is a common behavior among multiple
         templates, it is better to be extracted to a plugin that can then be plugged in to each of these templates.
+    metadata: Optional[Dict]
+        An optional dictionary with extra information about the template
 
     Methods
     -------
@@ -63,6 +65,7 @@ def __init__(
         preprocessors: Optional[List[Processor]] = None,
         postprocessors: Optional[List[Processor]] = None,
         plugins: Optional[List[Plugin]] = None,
+        metadata: Optional[Dict] = None,
     ):
         """
         Initialize the class
@@ -86,6 +89,8 @@ def __init__(
             An optional series of plugins to be plugged in. A plugin defines its own interactive_flow, preprocessor
             and postprocessor. A plugin is a sub-set of the template, if there is a common behavior among multiple
             templates, it is better to be extracted to a plugin that can then be plugged in to each of these templates.
+        metadata: Optional[Dict]
+            An optional dictionary with extra information about the template
         """
         self._location = location
         self._interactive_flows = interactive_flows or []
@@ -99,6 +104,7 @@ def __init__(
                 self._preprocessors.append(plugin.preprocessor)
             if plugin.postprocessor:
                 self._postprocessors.append(plugin.postprocessor)
+        self.metadata = metadata
 
     def run_interactive_flows(self, context: Optional[Dict] = None) -> Dict:
         """
diff --git a/tests/unit/commands/pipeline/init/test_initeractive_init_flow.py b/tests/unit/commands/pipeline/init/test_initeractive_init_flow.py
index 843b39008f..c9670332ed 100644
--- a/tests/unit/commands/pipeline/init/test_initeractive_init_flow.py
+++ b/tests/unit/commands/pipeline/init/test_initeractive_init_flow.py
@@ -1,3 +1,5 @@
+import json
+import tempfile
 from unittest import TestCase
 from unittest.mock import patch, Mock, ANY, call
 import os
@@ -5,6 +7,7 @@
 
 from parameterized import parameterized
 
+from samcli.commands.exceptions import AppPipelineTemplateMetadataException
 from samcli.commands.pipeline.init.interactive_init_flow import (
     InteractiveInitFlow,
     PipelineTemplateCloneException,
@@ -13,6 +16,7 @@
     CUSTOM_PIPELINE_TEMPLATE_REPO_LOCAL_NAME,
     _prompt_cicd_provider,
     _prompt_provider_pipeline_template,
+    _get_pipeline_template_metadata,
 )
 from samcli.commands.pipeline.init.pipeline_templates_manifest import AppPipelineTemplateManifestException
 from samcli.lib.utils.git_repo import CloneRepoException
@@ -111,10 +115,12 @@ def test_app_pipeline_templates_with_invalid_manifest(
     @patch("samcli.commands.pipeline.init.interactive_init_flow.PipelineTemplatesManifest")
     @patch("samcli.commands.pipeline.init.interactive_init_flow.GitRepo.clone")
     @patch("samcli.commands.pipeline.init.interactive_init_flow._copy_dir_contents_to_cwd_fail_on_exist")
+    @patch("samcli.commands.pipeline.init.interactive_init_flow._get_pipeline_template_metadata")
     @patch("samcli.lib.cookiecutter.question.click")
     def test_generate_pipeline_configuration_file_from_app_pipeline_template_happy_case(
         self,
         click_mock,
+        _get_pipeline_template_metadata_mock,
         _copy_dir_contents_to_cwd_fail_on_exist_mock,
         clone_mock,
         PipelineTemplatesManifest_mock,
@@ -158,6 +164,7 @@ def test_generate_pipeline_configuration_file_from_app_pipeline_template_happy_c
             "2",  # choose "Jenkins" when prompt for CI/CD system. (See pipeline_templates_manifest_mock, Jenkins is the 2nd provider)
             "1",  # choose "Jenkins pipeline template" when prompt for pipeline template
         ]
+        _get_pipeline_template_metadata_mock.return_value = {"number_of_stages": 2}
 
         # trigger
         InteractiveInitFlow(allow_bootstrap=False).do_interactive()
@@ -254,10 +261,12 @@ def test_generate_pipeline_configuration_file_from_custom_local_existing_path_wi
     @patch("samcli.commands.pipeline.init.interactive_init_flow.GitRepo.clone")
     @patch("samcli.commands.pipeline.init.interactive_init_flow.click")
     @patch("samcli.commands.pipeline.init.interactive_init_flow._copy_dir_contents_to_cwd_fail_on_exist")
+    @patch("samcli.commands.pipeline.init.interactive_init_flow._get_pipeline_template_metadata")
     @patch("samcli.lib.cookiecutter.question.click")
     def test_generate_pipeline_configuration_file_from_custom_remote_pipeline_template_happy_case(
         self,
         questions_click_mock,
+        _get_pipeline_template_metadata_mock,
         _copy_dir_contents_to_cwd_fail_on_exist_mock,
         init_click_mock,
         clone_mock,
@@ -280,6 +289,7 @@ def test_generate_pipeline_configuration_file_from_custom_remote_pipeline_templa
 
         questions_click_mock.prompt.return_value = "2"  # Custom pipeline templates
         init_click_mock.prompt.return_value = "https://github.com/any-custom-pipeline-template-repo.git"
+        _get_pipeline_template_metadata_mock.return_value = {"number_of_stages": 2}
 
         # trigger
         InteractiveInitFlow(allow_bootstrap=False).do_interactive()
@@ -336,6 +346,31 @@ def test_prompt_provider_pipeline_template_will_not_prompt_if_the_list_of_templa
         click_mock.prompt.assert_called_once()
         self.assertEqual(chosen_template, template2)
 
+    def test_get_pipeline_template_metadata_can_load(self):
+        with tempfile.TemporaryDirectory() as dir:
+            metadata = {"number_of_stages": 2}
+            with open(Path(dir, "metadata.json"), "w") as f:
+                json.dump(metadata, f)
+            self.assertEquals(metadata, _get_pipeline_template_metadata(dir))
+
+    def test_get_pipeline_template_metadata_not_exist(self):
+        with tempfile.TemporaryDirectory() as dir:
+            with self.assertRaises(AppPipelineTemplateMetadataException):
+                _get_pipeline_template_metadata(dir)
+
+    @parameterized.expand(
+        [
+            ('["not_a_dict"]',),
+            ("not a json"),
+        ]
+    )
+    def test_get_pipeline_template_metadata_not_valid(self, metadata_str):
+        with tempfile.TemporaryDirectory() as dir:
+            with open(Path(dir, "metadata.json"), "w") as f:
+                f.write(metadata_str)
+            with self.assertRaises(AppPipelineTemplateMetadataException):
+                _get_pipeline_template_metadata(dir)
+
 
 class TestInteractiveInitFlowWithBootstrap(TestCase):
     @patch("samcli.commands.pipeline.init.interactive_init_flow.SamConfig")
@@ -348,10 +383,12 @@ class TestInteractiveInitFlowWithBootstrap(TestCase):
     @patch("samcli.commands.pipeline.init.interactive_init_flow.PipelineTemplatesManifest")
     @patch("samcli.commands.pipeline.init.interactive_init_flow.GitRepo.clone")
     @patch("samcli.commands.pipeline.init.interactive_init_flow._copy_dir_contents_to_cwd_fail_on_exist")
+    @patch("samcli.commands.pipeline.init.interactive_init_flow._get_pipeline_template_metadata")
     @patch("samcli.lib.cookiecutter.question.click")
     def test_with_bootstrap_but_answer_no(
         self,
         click_mock,
+        _get_pipeline_template_metadata_mock,
         _copy_dir_contents_to_cwd_fail_on_exist_mock,
         clone_mock,
         PipelineTemplatesManifest_mock,
@@ -389,6 +426,7 @@ def test_with_bootstrap_but_answer_no(
         config_file.exists.return_value = True
         config_file.get_stage_names.return_value = ["testing"]
         config_file.get_all.return_value = {"pipeline_execution_role": "arn:aws:iam::123456789012:role/execution-role"}
+        _get_pipeline_template_metadata_mock.return_value = {"number_of_stages": 2}
 
         click_mock.prompt.side_effect = [
             "1",  # App pipeline templates
@@ -421,12 +459,14 @@ def test_with_bootstrap_but_answer_no(
     @patch("samcli.commands.pipeline.init.interactive_init_flow.PipelineTemplatesManifest")
     @patch("samcli.commands.pipeline.init.interactive_init_flow.GitRepo.clone")
     @patch("samcli.commands.pipeline.init.interactive_init_flow._copy_dir_contents_to_cwd_fail_on_exist")
+    @patch("samcli.commands.pipeline.init.interactive_init_flow._get_pipeline_template_metadata")
     @patch("samcli.lib.cookiecutter.question.click")
     def test_with_bootstrap_answer_yes(
         self,
         get_stage_name_side_effects,
         _prompt_run_bootstrap_expected_calls,
         click_mock,
+        _get_pipeline_template_metadata_mock,
         _copy_dir_contents_to_cwd_fail_on_exist_mock,
         clone_mock,
         PipelineTemplatesManifest_mock,
@@ -464,6 +504,7 @@ def test_with_bootstrap_answer_yes(
         config_file.exists.return_value = True
         config_file.get_stage_names.side_effect = get_stage_name_side_effects
         config_file.get_all.return_value = {"pipeline_execution_role": "arn:aws:iam::123456789012:role/execution-role"}
+        _get_pipeline_template_metadata_mock.return_value = {"number_of_stages": 2}
 
         click_mock.prompt.side_effect = [
             "1",  # App pipeline templates