diff --git a/azure_functions_worker/dispatcher.py b/azure_functions_worker/dispatcher.py
index 24c25ece1..1e60917aa 100644
--- a/azure_functions_worker/dispatcher.py
+++ b/azure_functions_worker/dispatcher.py
@@ -9,6 +9,7 @@
 import concurrent.futures
 import logging
 import os
+import platform
 import queue
 import sys
 import threading
@@ -24,7 +25,8 @@
                         PYTHON_THREADPOOL_THREAD_COUNT_DEFAULT,
                         PYTHON_THREADPOOL_THREAD_COUNT_MAX_37,
                         PYTHON_THREADPOOL_THREAD_COUNT_MIN,
-                        PYTHON_ENABLE_DEBUG_LOGGING, SCRIPT_FILE_NAME)
+                        PYTHON_ENABLE_DEBUG_LOGGING, SCRIPT_FILE_NAME,
+                        PYTHON_LANGUAGE_RUNTIME)
 from .extension import ExtensionManager
 from .logging import disable_console_logging, enable_console_logging
 from .logging import enable_debug_logging_recommendation
@@ -93,6 +95,16 @@ def __init__(self, loop: BaseEventLoop, host: str, port: int,
         self._grpc_thread: threading.Thread = threading.Thread(
             name='grpc-thread', target=self.__poll_grpc)
 
+    @staticmethod
+    def get_worker_metadata():
+        return protos.WorkerMetadata(
+            runtime_name=PYTHON_LANGUAGE_RUNTIME,
+            runtime_version=f"{sys.version_info.major}."
+                            f"{sys.version_info.minor}",
+            worker_version=VERSION,
+            worker_bitness=platform.machine(),
+            custom_properties={})
+
     def get_sync_tp_workers_set(self):
         """We don't know the exact value of the threadcount set for the Python
          3.9 scenarios (as we'll start passing only None by default), and we
@@ -280,6 +292,7 @@ async def _handle__worker_init_request(self, request):
             request_id=self.request_id,
             worker_init_response=protos.WorkerInitResponse(
                 capabilities=capabilities,
+                worker_metadata=self.get_worker_metadata(),
                 result=protos.StatusResult(
                     status=protos.StatusResult.Success)))
 
@@ -552,6 +565,8 @@ async def _handle__function_environment_reload_request(self, request):
                     func_env_reload_request.function_app_directory)
 
             success_response = protos.FunctionEnvironmentReloadResponse(
+                capabilities={},
+                worker_metadata=self.get_worker_metadata(),
                 result=protos.StatusResult(
                     status=protos.StatusResult.Success))
 
diff --git a/azure_functions_worker/protos/__init__.py b/azure_functions_worker/protos/__init__.py
index 45c31d375..8e74b5c62 100644
--- a/azure_functions_worker/protos/__init__.py
+++ b/azure_functions_worker/protos/__init__.py
@@ -31,7 +31,8 @@
     CloseSharedMemoryResourcesRequest,
     CloseSharedMemoryResourcesResponse,
     FunctionsMetadataRequest,
-    FunctionMetadataResponse)
+    FunctionMetadataResponse,
+    WorkerMetadata)
 
 from .shared.NullableTypes_pb2 import (
     NullableString,
diff --git a/azure_functions_worker/protos/_src/src/proto/FunctionRpc.proto b/azure_functions_worker/protos/_src/src/proto/FunctionRpc.proto
index 155af606e..f5cc48b30 100644
--- a/azure_functions_worker/protos/_src/src/proto/FunctionRpc.proto
+++ b/azure_functions_worker/protos/_src/src/proto/FunctionRpc.proto
@@ -238,6 +238,13 @@ message FunctionEnvironmentReloadRequest {
 }
 
 message FunctionEnvironmentReloadResponse {
+  // After specialization, worker sends capabilities & metadata.
+  // Worker metadata captured for telemetry purposes
+  WorkerMetadata worker_metadata = 1;
+
+  // A map of worker supported features/capabilities
+  map<string, string> capabilities = 2;
+
   // Status of the response
   StatusResult result = 3;
 }
@@ -431,6 +438,7 @@ message TypedData {
     CollectionDouble collection_double = 10;
     CollectionSInt64 collection_sint64 = 11;
     ModelBindingData model_binding_data = 12;
+    CollectionModelBindingData collection_model_binding_data = 13;
   }
 }
 
@@ -668,4 +676,9 @@ message ModelBindingData
 
     // The binding data content
     bytes content = 4;
+}
+
+// Used to encapsulate collection model_binding_data
+message CollectionModelBindingData {
+  repeated ModelBindingData model_binding_data = 1;
 }
\ No newline at end of file
diff --git a/tests/unittests/test_dispatcher.py b/tests/unittests/test_dispatcher.py
index 5aab8d38a..4c8b058fc 100644
--- a/tests/unittests/test_dispatcher.py
+++ b/tests/unittests/test_dispatcher.py
@@ -8,6 +8,7 @@
 from unittest.mock import patch
 
 from azure_functions_worker import protos
+from azure_functions_worker.version import VERSION
 from tests.utils import testutils
 from azure_functions_worker.constants import PYTHON_THREADPOOL_THREAD_COUNT, \
     PYTHON_THREADPOOL_THREAD_COUNT_DEFAULT, \
@@ -62,6 +63,27 @@ async def test_dispatcher_initialize_worker(self):
         async with self._ctrl as host:
             r = await host.init_worker('3.0.12345')
             self.assertIsInstance(r.response, protos.WorkerInitResponse)
+            self.assertIsInstance(r.response.worker_metadata,
+                                  protos.WorkerMetadata)
+            self.assertEquals(r.response.worker_metadata.runtime_name,
+                              "python")
+            self.assertEquals(r.response.worker_metadata.worker_version,
+                              VERSION)
+
+    async def test_dispatcher_environment_reload(self):
+        """Test function environment reload response
+        """
+        async with self._ctrl as host:
+            # Reload environment variable on specialization
+            r = await host.reload_environment(environment={})
+            self.assertIsInstance(r.response,
+                                  protos.FunctionEnvironmentReloadResponse)
+            self.assertIsInstance(r.response.worker_metadata,
+                                  protos.WorkerMetadata)
+            self.assertEquals(r.response.worker_metadata.runtime_name,
+                              "python")
+            self.assertEquals(r.response.worker_metadata.worker_version,
+                              VERSION)
 
     async def test_dispatcher_initialize_worker_logging(self):
         """Test if the dispatcher's log can be flushed out during worker