@@ -17,12 +17,14 @@
 import itertools
 import re
 import os
+import pathlib
 import typing
-from typing import Any, DefaultDict, Dict, Mapping
+from typing import Any, DefaultDict, Dict, Mapping, Tuple
 from hashlib import sha256
 from collections import OrderedDict, defaultdict
 from gapic.samplegen_utils.utils import coerce_response_name, is_valid_sample_cfg, render_format_string
 from gapic.samplegen_utils.types import DuplicateSample
+from gapic.samplegen_utils import snippet_index, snippet_metadata_pb2
 from gapic.samplegen import manifest, samplegen
 from gapic.generator import formatter
 from gapic.schema import api
@@ -93,6 +95,17 @@ def get_response(
             self._env.loader.list_templates(),  # type: ignore
         )
 
+        # We generate code snippets *before* the library code so snippets
+        # can be inserted into method docstrings.
+        snippet_idx = snippet_index.SnippetIndex(api_schema)
+        if sample_templates:
+            sample_output, snippet_idx = self._generate_samples_and_manifest(
+                api_schema, snippet_idx, self._env.get_template(
+                    sample_templates[0]),
+                opts=opts,
+            )
+            output_files.update(sample_output)
+
         # Iterate over each template and add the appropriate output files
         # based on that template.
         # Sample templates work differently: there's (usually) only one,
@@ -107,15 +120,8 @@ def get_response(
             # Append to the output files dictionary.
             output_files.update(
                 self._render_template(
-                    template_name, api_schema=api_schema, opts=opts)
-            )
-
-        if sample_templates:
-            sample_output = self._generate_samples_and_manifest(
-                api_schema, self._env.get_template(sample_templates[0]),
-                opts=opts,
+                    template_name, api_schema=api_schema, opts=opts, snippet_index=snippet_idx)
             )
-            output_files.update(sample_output)
 
         # Return the CodeGeneratorResponse output.
         res = CodeGeneratorResponse(
@@ -124,7 +130,7 @@ def get_response(
         return res
 
     def _generate_samples_and_manifest(
-            self, api_schema: api.API, sample_template: jinja2.Template, *, opts: Options) -> Dict:
+            self, api_schema: api.API, index: snippet_index.SnippetIndex, sample_template: jinja2.Template, *, opts: Options) -> Tuple[Dict, snippet_index.SnippetIndex]:
         """Generate samples and samplegen manifest for the API.
 
         Arguments:
@@ -133,7 +139,7 @@ def _generate_samples_and_manifest(
             opts (Options): Additional generator options.
 
         Returns:
-            Dict[str, CodeGeneratorResponse.File]: A dict mapping filepath to rendered file.
+            Tuple[Dict[str, CodeGeneratorResponse.File], snippet_index.SnippetIndex] : A dict mapping filepath to rendered file.
         """
         # The two-layer data structure lets us do two things:
         # * detect duplicate samples, which is an error
@@ -181,7 +187,7 @@ def _generate_samples_and_manifest(
                 if not id_is_unique:
                     spec["id"] += f"_{spec_hash}"
 
-                sample = samplegen.generate_sample(
+                sample, snippet_metadata = samplegen.generate_sample(
                     spec, api_schema, sample_template,)
 
                 fpath = utils.to_snake_case(spec["id"]) + ".py"
@@ -190,36 +196,30 @@ def _generate_samples_and_manifest(
                     sample,
                 )
 
+                snippet_metadata.file = fpath
+
+                index.add_snippet(
+                    snippet_index.Snippet(sample, snippet_metadata))
+
         output_files = {
             fname: CodeGeneratorResponse.File(
                 content=formatter.fix_whitespace(sample), name=fname
             )
             for fname, (_, sample) in fpath_to_spec_and_rendered.items()
         }
 
-        # TODO(busunkim): Re-enable manifest generation once metadata
-        # format has been formalized.
-        # https://docs.google.com/document/d/1ghBam8vMj3xdoe4xfXhzVcOAIwrkbTpkMLgKc9RPD9k/edit#heading=h.sakzausv6hue
-        #
-        # if output_files:
-
-        # manifest_fname, manifest_doc = manifest.generate(
-        #     (
-        #         (fname, spec)
-        #         for fname, (spec, _) in fpath_to_spec_and_rendered.items()
-        #     ),
-        #     api_schema,
-        # )
-
-        # manifest_fname = os.path.join(out_dir, manifest_fname)
-        # output_files[manifest_fname] = CodeGeneratorResponse.File(
-        #     content=manifest_doc.render(), name=manifest_fname
-        # )
+        if index.metadata_index.snippets:
+            # NOTE(busunkim): Not all fields are yet populated in the snippet metadata.
+            # Expected filename: snippet_metadata_{apishortname}_{apiversion}.json
+            snippet_metadata_path = str(pathlib.Path(
+                out_dir) / f"snippet_metadata_{api_schema.naming.name}_{api_schema.naming.version}.json").lower()
+            output_files[snippet_metadata_path] = CodeGeneratorResponse.File(
+                content=formatter.fix_whitespace(index.get_metadata_json()), name=snippet_metadata_path)
 
-        return output_files
+        return output_files, index
 
     def _render_template(
-            self, template_name: str, *, api_schema: api.API, opts: Options,
+            self, template_name: str, *, api_schema: api.API, opts: Options, snippet_index: snippet_index.SnippetIndex,
     ) -> Dict[str, CodeGeneratorResponse.File]:
         """Render the requested templates.
 
@@ -258,7 +258,7 @@ def _render_template(
             for subpackage in api_schema.subpackages.values():
                 answer.update(
                     self._render_template(
-                        template_name, api_schema=subpackage, opts=opts
+                        template_name, api_schema=subpackage, opts=opts, snippet_index=snippet_index
                     )
                 )
                 skip_subpackages = True
@@ -275,7 +275,7 @@ def _render_template(
 
                 answer.update(
                     self._get_file(
-                        template_name, api_schema=api_schema, proto=proto, opts=opts
+                        template_name, api_schema=api_schema, proto=proto, opts=opts, snippet_index=snippet_index
                     )
                 )
 
@@ -304,14 +304,15 @@ def _render_template(
                         api_schema=api_schema,
                         service=service,
                         opts=opts,
+                        snippet_index=snippet_index,
                     )
                 )
             return answer
 
         # This file is not iterating over anything else; return back
         # the one applicable file.
         answer.update(self._get_file(
-            template_name, api_schema=api_schema, opts=opts))
+            template_name, api_schema=api_schema, opts=opts, snippet_index=snippet_index))
         return answer
 
     def _is_desired_transport(self, template_name: str, opts: Options) -> bool:
@@ -324,8 +325,8 @@ def _get_file(
         template_name: str,
         *,
         opts: Options,
-        api_schema=api.API,
-        **context: Mapping,
+        api_schema: api.API,
+        **context,
     ):
         """Render a template to a protobuf plugin File object."""
         # Determine the target filename.
 
@@ -24,13 +24,13 @@
 
 from gapic import utils
 
-from gapic.samplegen_utils import types
+from gapic.samplegen_utils import types, snippet_metadata_pb2  # type: ignore
 from gapic.samplegen_utils.utils import is_valid_sample_cfg
 from gapic.schema import api
 from gapic.schema import wrappers
 
 from collections import defaultdict, namedtuple, ChainMap as chainmap
-from typing import Any, ChainMap, Dict, FrozenSet, Generator, List, Mapping, Optional, Sequence
+from typing import Any, ChainMap, Dict, FrozenSet, Generator, List, Mapping, Optional, Sequence, Tuple
 
 # There is no library stub file for this module, so ignore it.
 from google.api import resource_pb2  # type: ignore
@@ -915,8 +915,6 @@ def _validate_loop(self, loop):
 def parse_handwritten_specs(sample_configs: Sequence[str]) -> Generator[Dict[str, Any], None, None]:
     """Parse a handwritten sample spec"""
 
-    STANDALONE_TYPE = "standalone"
-
     for config_fpath in sample_configs:
         with open(config_fpath) as f:
             configs = yaml.safe_load_all(f.read())
@@ -925,13 +923,9 @@ def parse_handwritten_specs(sample_configs: Sequence[str]) -> Generator[Dict[str
                 valid = is_valid_sample_cfg(cfg)
                 if not valid:
                     raise types.InvalidConfig(
-                        "Sample config is invalid", valid)
+                        "Sample config in '{}' is invalid\n\n{}".format(config_fpath, cfg), valid)
                 for spec in cfg.get("samples", []):
-                    # If unspecified, assume a sample config describes a standalone.
-                    # If sample_types are specified, standalone samples must be
-                    # explicitly enabled.
-                    if STANDALONE_TYPE in spec.get("sample_type", [STANDALONE_TYPE]):
-                        yield spec
+                    yield spec
 
 
 def _generate_resource_path_request_object(field_name: str, message: wrappers.MessageType) -> List[Dict[str, str]]:
@@ -1050,7 +1044,6 @@ def generate_sample_specs(api_schema: api.API, *, opts) -> Generator[Dict[str, A
                 # [{START|END} ${apishortname}_generated_${api}_${apiVersion}_${serviceName}_${rpcName}_{sync|async}_${overloadDisambiguation}]
                 region_tag = f"{api_short_name}_generated_{api_schema.naming.versioned_module_name}_{service_name}_{rpc_name}_{transport_type}"
                 spec = {
-                    "sample_type": "standalone",
                     "rpc": rpc_name,
                     "transport": transport,
                     # `request` and `response` is populated in `preprocess_sample`
@@ -1062,7 +1055,7 @@ def generate_sample_specs(api_schema: api.API, *, opts) -> Generator[Dict[str, A
                 yield spec
 
 
-def generate_sample(sample, api_schema, sample_template: jinja2.Template) -> str:
+def generate_sample(sample, api_schema, sample_template: jinja2.Template) -> Tuple[str, Any]:
     """Generate a standalone, runnable sample.
 
     Writing the rendered output is left for the caller.
@@ -1073,7 +1066,7 @@ def generate_sample(sample, api_schema, sample_template: jinja2.Template) -> str
         sample_template (jinja2.Template): The template representing a generic sample.
 
     Returns:
-        str: The rendered sample.
+        Tuple(str, snippet_metadata_pb2.Snippet): The rendered sample.
     """
     service_name = sample["service"]
     service = api_schema.services.get(service_name)
@@ -1100,11 +1093,22 @@ def generate_sample(sample, api_schema, sample_template: jinja2.Template) -> str
 
     v.validate_response(sample["response"])
 
+    # Snippet Metadata can't be fully filled out in any one function
+    # In this function we add information from
+    # the API schema and sample dictionary.
+    snippet_metadata = snippet_metadata_pb2.Snippet()  # type: ignore
+    snippet_metadata.region_tag = sample["region_tag"]
+    setattr(snippet_metadata.client_method, "async",
+            sample["transport"] == api.TRANSPORT_GRPC_ASYNC)
+    snippet_metadata.client_method.method.short_name = sample["rpc"]
+    snippet_metadata.client_method.method.service.short_name = sample["service"].split(
+        ".")[-1]
+
     return sample_template.render(
         sample=sample,
         imports=[],
         calling_form=calling_form,
         calling_form_enum=types.CallingForm,
         trim_blocks=True,
         lstrip_blocks=True,
-    )
+    ), snippet_metadata
@@ -12,8 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Optional, Dict
 import re
+from typing import Optional, Dict
 
 from google.protobuf import json_format
 
@@ -88,6 +88,7 @@ def full_snippet(self) -> str:
         """The portion between the START and END region tags."""
         start_idx = self._full_snippet.start - 1
         end_idx = self._full_snippet.end
+        self.sample_lines[start_idx] = self.sample_lines[start_idx].strip()
         return "".join(self.sample_lines[start_idx:end_idx])
 
 
@@ -124,7 +125,7 @@ def add_snippet(self, snippet: Snippet) -> None:
             RpcMethodNotFound: If the method indicated by the snippet metadata is not found.
         """
         service_name = snippet.metadata.client_method.method.service.short_name
-        rpc_name = snippet.metadata.client_method.method.full_name
+        rpc_name = snippet.metadata.client_method.method.short_name
 
         service = self._index.get(service_name)
         if service is None:
@@ -172,4 +173,8 @@ def get_snippet(self, service_name: str, rpc_name: str, sync: bool = True) -> Op
 
     def get_metadata_json(self) -> str:
         """JSON representation of Snippet Index."""
+
+        # Downstream tools assume the generator will produce the exact
+        # same output when run over the same API multiple times
+        self.metadata_index.snippets.sort(key=lambda s: s.region_tag)
         return json_format.MessageToJson(self.metadata_index, sort_keys=True)
@@ -41,7 +41,7 @@
 from google.api import resource_pb2
 from google.api_core import exceptions
 from google.api_core import path_template
-from google.cloud import extended_operations_pb2 as ex_ops_pb2
+from google.cloud import extended_operations_pb2 as ex_ops_pb2  # type: ignore
 from google.protobuf import descriptor_pb2  # type: ignore
 from google.protobuf.json_format import MessageToDict  # type: ignore
 
 
@@ -207,6 +207,14 @@ class {{ service.async_client_name }}:
             {% endif %}
         r"""{{ method.meta.doc|rst(width=72, indent=8) }}
 
+        {% with snippet = snippet_index.get_snippet(service.name, method.name, sync=True) %}
+        {% if snippet is not none %}
+        .. code-block::
+
+{{ snippet.full_snippet|indent(width=12, first=True) }}
+        {% endif %}
+        {% endwith %}
+
         Args:
             {% if not method.client_streaming %}
             request (Union[{{ method.input.ident.sphinx }}, dict]):
 
@@ -363,6 +363,15 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
             {% endif %}
         r"""{{ method.meta.doc|rst(width=72, indent=8) }}
 
+        
+        {% with snippet = snippet_index.get_snippet(service.name, method.name, sync=True) %}
+        {% if snippet is not none %}
+        .. code-block::
+
+{{ snippet.full_snippet|indent(width=12, first=True) }}
+        {% endif %}
+        {% endwith %}
+
         Args:
             {% if not method.client_streaming %}
             request (Union[{{ method.input.ident.sphinx }}, dict]):
 
@@ -32,8 +32,6 @@ from {{ sample.module_namespace|join(".") }} import {{ sample.module_name }}
 
 {# also need calling form #}
 {% if sample.transport == "grpc-async" %}async {% endif %}def sample_{{ frags.render_method_name(sample.rpc)|trim }}({{ frags.print_input_params(sample.request)|trim }}):
-    """{{ sample.description }}"""
-
     {{ frags.render_client_setup(sample.module_name, sample.client_name)|indent }}
     {{ frags.render_request_setup(sample.request, sample.module_name, sample.request_type, calling_form, calling_form_enum)|indent }}
     {% with method_call = frags.render_method_call(sample, calling_form, calling_form_enum, sample.transport) %}