diff --git a/ci/metrics_agent.py b/ci/metrics_agent.py
new file mode 100644
index 0000000..bc2479e
--- /dev/null
+++ b/ci/metrics_agent.py
@@ -0,0 +1,116 @@
+# Copyright 2024, The Android Open Source Project
+#
+# Licensed 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.
+
+"""MetricsAgent is a singleton class that collects metrics for optimized build."""
+
+from enum import Enum
+import time
+import metrics_pb2
+import os
+import logging
+
+
+class MetricsAgent:
+  _SOONG_METRICS_PATH = 'logs/soong_metrics'
+  _DIST_DIR = 'DIST_DIR'
+  _instance = None
+
+  def __init__(self):
+    raise RuntimeError(
+        'MetricsAgent cannot be instantialized, use instance() instead'
+    )
+
+  @classmethod
+  def instance(cls):
+    if not cls._instance:
+      cls._instance = cls.__new__(cls)
+      cls._instance._proto = metrics_pb2.OptimizedBuildMetrics()
+      cls._instance._init_proto()
+      cls._instance._target_results = dict()
+
+    return cls._instance
+
+  def _init_proto(self):
+    self._proto.analysis_perf.name = 'Optimized build analysis time.'
+    self._proto.packaging_perf.name = 'Optimized build total packaging time.'
+
+  def analysis_start(self):
+    self._proto.analysis_perf.start_time = time.time_ns()
+
+  def analysis_end(self):
+    self._proto.analysis_perf.real_time = (
+        time.time_ns() - self._proto.analysis_perf.start_time
+    )
+
+  def packaging_start(self):
+    self._proto.packaging_perf.start_time = time.time_ns()
+
+  def packaging_end(self):
+    self._proto.packaging_perf.real_time = (
+        time.time_ns() - self._proto.packaging_perf.start_time
+    )
+
+  def report_optimized_target(self, name: str):
+    target_result = metrics_pb2.OptimizedBuildMetrics.TargetOptimizationResult()
+    target_result.name = name
+    target_result.optimized = True
+    self._target_results[name] = target_result
+
+  def report_unoptimized_target(self, name: str, optimization_rationale: str):
+    target_result = metrics_pb2.OptimizedBuildMetrics.TargetOptimizationResult()
+    target_result.name = name
+    target_result.optimization_rationale = optimization_rationale
+    target_result.optimized = False
+    self._target_results[name] = target_result
+
+  def target_packaging_start(self, name: str):
+    target_result = self._target_results.get(name)
+    target_result.packaging_perf.start_time = time.time_ns()
+    self._target_results[name] = target_result
+
+  def target_packaging_end(self, name: str):
+    target_result = self._target_results.get(name)
+    target_result.packaging_perf.real_time = (
+        time.time_ns() - target_result.packaging_perf.start_time
+    )
+
+  def add_target_artifact(
+      self,
+      target_name: str,
+      artifact_name: str,
+      size: int,
+      included_modules: set[str],
+  ):
+    target_result = self.target_results.get(target_name)
+    artifact = (
+        metrics_pb2.OptimizedBuildMetrics.TargetOptimizationResult.OutputArtifact()
+    )
+    artifact.name = artifact_name
+    artifact.size = size
+    for module in included_modules:
+      artifact.included_modules.add(module)
+    target_result.output_artifacts.add(artifact)
+
+  def end_reporting(self):
+    for target_result in self._target_results.values():
+      self._proto.target_result.append(target_result)
+    soong_metrics_proto = metrics_pb2.MetricsBase()
+    # Read in existing metrics that should have been written out by the soong
+    # build command so that we don't overwrite them.
+    with open(os.path.join(os.environ[self._DIST_DIR], self._SOONG_METRICS_PATH), 'rb') as f:
+      soong_metrics_proto.ParseFromString(f.read())
+    soong_metrics_proto.optimized_build_metrics.CopyFrom(self._proto)
+    logging.info(soong_metrics_proto)
+    with open(os.path.join(os.environ[self._DIST_DIR], self._SOONG_METRICS_PATH), 'wb') as f:
+      f.write(soong_metrics_proto.SerializeToString())