diff --git a/CHANGELOG.md b/CHANGELOG.md
index 64da93791c..3a5d2b73cb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -49,6 +49,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-sqlalchemy`: implement new semantic convention opt-in migration
   ([#4110](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4110))
 
+### Added
+
+- `opentelemetry-instrumentation`: Add experimental metrics attributes Labeler utility
+  ([#4288](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4288))
+- ``opentelemetry-instrumentation-wsgi`, `opentelemetry-instrumentation-asgi`: `enrich_metric_attributes` with any Labeler-stored attributes in Context for HTTP server duration metrics
+  ([#4300](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4300))
+- `opentelemetry-instrumentation-flask`: `enrich_metric_attributes` with any Labeler-stored attributes in Context
+  ([#4307](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4307))
+
 ### Fixed
 
 - `opentelemetry-instrumentation-celery`: Coerce non-string values to strings in `CeleryGetter.get()` to prevent `TypeError` in `TraceState.from_header()` when Celery request attributes contain ints
diff --git a/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py b/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py
index 8e7106010f..9694e872ca 100644
--- a/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py
@@ -204,6 +204,54 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 Note:
     The environment variable names used to capture HTTP headers are still experimental, and thus are subject to change.
 
+Custom Metrics Attributes using Labeler
+***************************************
+The ASGI instrumentation reads custom attributes from the labeler (when present)
+and applies them to all HTTP server metric points emitted by the middleware:
+
+- Active requests counter (`http.server.active_requests`)
+- Duration histograms (`http.server.duration` and/or
+    `http.server.request.duration` depending on semantic convention mode)
+- Response size histograms (`http.server.response.size` and/or
+    `http.server.response.body.size`)
+- Request size histograms (`http.server.request.size` and/or
+    `http.server.request.body.size`)
+
+Labeler attributes are request-scoped and merged without overriding base metric
+attributes at the same keys.
+
+
+.. code-block:: python
+
+    .. code-block:: python
+
+    from quart import Quart
+    from opentelemetry.instrumentation._labeler import get_labeler
+    from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
+
+    app = Quart(__name__)
+    app.asgi_app = OpenTelemetryMiddleware(app.asgi_app)
+
+    @app.route("/users/<user_id>/")
+    async def user_profile(user_id):
+        # Get the labeler for the current request
+        labeler = get_labeler()
+
+        # Add custom attributes to ASGI instrumentation metrics
+        labeler.add("user_id", user_id)
+        labeler.add("user_type", "registered")
+
+        # Or, add multiple attributes at once
+        labeler.add_attributes({
+            "feature_flag": "new_ui",
+            "experiment_group": "control"
+        })
+
+        return f"User profile for {user_id}"
+
+    if __name__ == "__main__":
+        app.run(debug=True)
+
 API
 ---
 """
@@ -220,6 +268,11 @@ def client_response_hook(span: Span, scope: Scope, message: dict[str, Any]):
 from asgiref.compatibility import guarantee_single_callable
 
 from opentelemetry import context, trace
+from opentelemetry.instrumentation._labeler import (
+    enrich_metric_attributes,
+    get_labeler,
+    get_labeler_attributes,
+)
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _filter_semconv_active_request_count_attr,
@@ -758,6 +811,9 @@ async def __call__(
         if self.excluded_urls and self.excluded_urls.url_disabled(url):
             return await self.app(scope, receive, send)
 
+        # Required to create new instance for custom attributes in async context
+        _ = get_labeler()
+
         span_name, additional_attributes = self.default_span_details(scope)
 
         attributes = collect_request_attributes(
@@ -776,6 +832,9 @@ async def __call__(
             attributes,
             self._sem_conv_opt_in_mode,
         )
+        active_requests_count_attrs = enrich_metric_attributes(
+            active_requests_count_attrs
+        )
 
         if scope["type"] == "http":
             self.active_requests_counter.add(1, active_requests_count_attrs)
@@ -806,12 +865,14 @@ async def __call__(
                     span_name, scope, receive
                 )
 
+                labeler_metric_attributes = {}
                 otel_send = self._get_otel_send(
                     current_span,
                     span_name,
                     scope,
                     send,
                     attributes,
+                    labeler_metric_attributes,
                 )
 
                 await self.app(scope, otel_receive, otel_send)
@@ -833,9 +894,21 @@ async def __call__(
                 )
                 if target:
                     duration_attrs_old[HTTP_TARGET] = target
+                duration_attrs_old = enrich_metric_attributes(
+                    duration_attrs_old
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_old:
+                        duration_attrs_old[key] = value
                 duration_attrs_new = _parse_duration_attrs(
                     attributes, _StabilityMode.HTTP
                 )
+                duration_attrs_new = enrich_metric_attributes(
+                    duration_attrs_new
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_new:
+                        duration_attrs_new[key] = value
                 span_ctx = set_span_in_context(span)
                 if self.duration_histogram_old:
                     self.duration_histogram_old.record(
@@ -985,6 +1058,7 @@ def _get_otel_send(
         scope,
         send,
         duration_attrs,
+        labeler_metric_attributes,
     ):
         expecting_trailers = False
 
@@ -992,6 +1066,9 @@ def _get_otel_send(
         async def otel_send(message: dict[str, Any]):
             nonlocal expecting_trailers
 
+            if not labeler_metric_attributes:
+                labeler_metric_attributes.update(get_labeler_attributes())
+
             status_code = None
             if message["type"] == "http.response.start":
                 status_code = message["status"]
diff --git a/instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py b/instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py
index b14f207d8a..fbb6b57f65 100644
--- a/instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py
+++ b/instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py
@@ -22,6 +22,7 @@
 
 import opentelemetry.instrumentation.asgi as otel_asgi
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._labeler import clear_labeler, get_labeler
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     OTEL_SEMCONV_STABILITY_OPT_IN,
@@ -273,6 +274,50 @@ async def background_execution_trailers_asgi(scope, receive, send):
         time.sleep(_SIMULATED_BACKGROUND_TASK_EXECUTION_TIME_S)
 
 
+async def custom_attrs_asgi(scope, receive, send):
+    assert isinstance(scope, dict)
+    assert scope["type"] == "http"
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.method", "POST")
+    message = await receive()
+    scope["headers"] = [(b"content-length", b"128")]
+    if message.get("type") == "http.request":
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"Content-Type", b"text/plain"],
+                    [b"content-length", b"1024"],
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"*"})
+
+
+async def custom_attrs_asgi_new_semconv(scope, receive, send):
+    assert isinstance(scope, dict)
+    assert scope["type"] == "http"
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.request.method", "POST")
+    message = await receive()
+    scope["headers"] = [(b"content-length", b"128")]
+    if message.get("type") == "http.request":
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"Content-Type", b"text/plain"],
+                    [b"content-length", b"1024"],
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": b"*"})
+
+
 async def error_asgi(scope, receive, send):
     assert isinstance(scope, dict)
     assert scope["type"] == "http"
@@ -314,6 +359,7 @@ def hook(*_):
 class TestAsgiApplication(AsyncAsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -1554,6 +1600,135 @@ async def test_asgi_metrics_both_semconv(self):
                     self.assertIn(attr, _recommended_attrs_both[metric.name])
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    async def test_asgi_metrics_custom_attributes_skip_override(self):
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        enriched_histogram_metric_names = {
+            "http.server.duration",
+            "http.server.response.size",
+            "http.server.request.size",
+        }
+        active_requests_point_seen = False
+        enriched_histograms_seen = set()
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+                self.assertNotIn("custom_attr", point.attributes)
+                active_requests_point_seen = True
+                continue
+
+            if metric.name not in enriched_histogram_metric_names:
+                continue
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            enriched_histograms_seen.add(metric.name)
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertSetEqual(
+            enriched_histogram_metric_names,
+            enriched_histograms_seen,
+        )
+
+    async def test_asgi_metrics_custom_attributes_skip_override_new_semconv(
+        self,
+    ):
+        app = otel_asgi.OpenTelemetryMiddleware(custom_attrs_asgi_new_semconv)
+        self.seed_app(app)
+        await self.send_default_request()
+        await self.get_all_output()
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        enriched_histogram_metric_names = {
+            "http.server.request.duration",
+            "http.server.response.body.size",
+            "http.server.request.body.size",
+        }
+        active_requests_point_seen = False
+        enriched_histograms_seen = set()
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+                self.assertNotIn("custom_attr", point.attributes)
+                active_requests_point_seen = True
+                continue
+
+            if metric.name not in enriched_histogram_metric_names:
+                continue
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            enriched_histograms_seen.add(metric.name)
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertSetEqual(
+            enriched_histogram_metric_names,
+            enriched_histograms_seen,
+        )
+
+    async def test_asgi_active_requests_attrs_use_enrich_old_semconv(self):
+        with mock.patch(
+            "opentelemetry.instrumentation.asgi.enrich_metric_attributes",
+            wraps=otel_asgi.enrich_metric_attributes,
+        ) as mock_enrich:
+            app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if HTTP_METHOD in attrs and HTTP_STATUS_CODE not in attrs:
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_seen)
+
+    async def test_asgi_active_requests_attrs_use_enrich_new_semconv(self):
+        with mock.patch(
+            "opentelemetry.instrumentation.asgi.enrich_metric_attributes",
+            wraps=otel_asgi.enrich_metric_attributes,
+        ) as mock_enrich:
+            app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+            self.seed_app(app)
+            await self.send_default_request()
+            await self.get_all_output()
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if (
+                HTTP_REQUEST_METHOD in attrs
+                and HTTP_RESPONSE_STATUS_CODE not in attrs
+            ):
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_seen)
+
     async def test_asgi_metrics_exemplars_expected_old_semconv(self):
         """Failing test placeholder asserting exemplars should be present for duration histogram (old semconv)."""
         app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
diff --git a/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py b/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
index 9e7d2e18d6..9a6a9ab812 100644
--- a/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
@@ -250,6 +250,51 @@ def response_hook(span: Span, status: str, response_headers: List):
 | ``controller``    | Flask controller/endpoint name.                    | ``controller='home_view'``             |
 +-------------------+----------------------------------------------------+----------------------------------------+
 
+Custom Metrics Attributes using Labeler
+***************************************
+The Flask instrumentation reads from a labeler utility that supports adding custom
+attributes to HTTP server metrics at emit time, including:
+
+- Active requests counter (``http.server.active_requests``)
+- Duration histogram (``http.server.duration``)
+- Request duration histogram (``http.server.request.duration``)
+
+The custom attributes are stored in the current OpenTelemetry context and are
+typically request-scoped for instrumented Flask handlers. In normal application
+flow, context detach at request teardown prevents these attributes from leaking
+to later requests. Application code typically should not call ``clear_labeler``;
+use it primarily for test isolation or manual context-lifecycle management. The
+instrumentor does not overwrite base attributes that exist at the same keys as
+any custom attributes.
+
+
+.. code-block:: python
+
+    from flask import Flask
+
+    from opentelemetry.instrumentation._labeler import get_labeler
+    from opentelemetry.instrumentation.flask import FlaskInstrumentor
+
+    app = Flask(__name__)
+    FlaskInstrumentor().instrument_app(app)
+
+    @app.route("/users/<user_id>/")
+    def user_profile(user_id):
+        # Get the labeler for the current request
+        labeler = get_labeler()
+
+        # Add custom attributes to Flask instrumentation metrics
+        labeler.add("user_id", user_id)
+        labeler.add("user_type", "registered")
+
+        # Or, add multiple attributes at once
+        labeler.add_attributes({
+            "feature_flag": "new_ui",
+            "experiment_group": "control"
+        })
+
+        return f"User profile for {user_id}"
+
 API
 ---
 """
@@ -265,6 +310,10 @@ def response_hook(span: Span, status: str, response_headers: List):
 
 import opentelemetry.instrumentation.wsgi as otel_wsgi
 from opentelemetry import context, trace
+from opentelemetry.instrumentation._labeler import (
+    enrich_metric_attributes,
+    get_labeler_attributes,
+)
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _get_schema_url,
@@ -311,6 +360,7 @@ def response_hook(span: Span, status: str, response_headers: List):
 _ENVIRON_ACTIVATION_KEY = "opentelemetry-flask.activation_key"
 _ENVIRON_REQCTX_REF_KEY = "opentelemetry-flask.reqctx_ref_key"
 _ENVIRON_TOKEN = "opentelemetry-flask.token"
+_ENVIRON_LABELER_ATTRIBUTES_KEY = "opentelemetry-flask.labeler_attributes"
 
 _excluded_urls_from_env = get_excluded_urls("FLASK")
 
@@ -350,6 +400,7 @@ def _rewrapped_app(
     duration_histogram_new=None,
 ):
     # pylint: disable=too-many-statements
+    # pylint: disable=too-many-locals
     def _wrapped_app(wrapped_app_environ, start_response):
         # We want to measure the time for route matching, etc.
         # In theory, we could start the span here and use
@@ -366,6 +417,9 @@ def _wrapped_app(wrapped_app_environ, start_response):
                 sem_conv_opt_in_mode,
             )
         )
+        active_requests_count_attrs = enrich_metric_attributes(
+            active_requests_count_attrs
+        )
 
         active_requests_counter.add(1, active_requests_count_attrs)
         request_route = None
@@ -436,6 +490,15 @@ def _start_response(status, response_headers, *args, **kwargs):
                 if request_route:
                     # http.target to be included in old semantic conventions
                     duration_attrs_old[HTTP_TARGET] = str(request_route)
+                duration_attrs_old = enrich_metric_attributes(
+                    duration_attrs_old
+                )
+                labeler_metric_attributes = wrapped_app_environ.get(
+                    _ENVIRON_LABELER_ATTRIBUTES_KEY, {}
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_old:
+                        duration_attrs_old[key] = value
                 duration_histogram_old.record(
                     max(round(duration_s * 1000), 0),
                     duration_attrs_old,
@@ -449,12 +512,32 @@ def _start_response(status, response_headers, *args, **kwargs):
                 if request_route:
                     duration_attrs_new[HTTP_ROUTE] = str(request_route)
 
+                duration_attrs_new = enrich_metric_attributes(
+                    duration_attrs_new
+                )
+                labeler_metric_attributes = wrapped_app_environ.get(
+                    _ENVIRON_LABELER_ATTRIBUTES_KEY, {}
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_new:
+                        duration_attrs_new[key] = value
+
                 duration_histogram_new.record(
                     max(duration_s, 0),
                     duration_attrs_new,
                     context=metrics_context,
                 )
 
+        active_requests_count_attrs = enrich_metric_attributes(
+            active_requests_count_attrs
+        )
+        labeler_metric_attributes = wrapped_app_environ.get(
+            _ENVIRON_LABELER_ATTRIBUTES_KEY, {}
+        )
+        for key, value in labeler_metric_attributes.items():
+            if key not in active_requests_count_attrs:
+                active_requests_count_attrs[key] = value
+
         active_requests_counter.add(-1, active_requests_count_attrs)
         return result
 
@@ -560,6 +643,9 @@ def _teardown_request(exc):
 
         activation = flask.request.environ.get(_ENVIRON_ACTIVATION_KEY)
         token = flask.request.environ.get(_ENVIRON_TOKEN)
+        flask.request.environ[_ENVIRON_LABELER_ATTRIBUTES_KEY] = dict(
+            get_labeler_attributes()
+        )
 
         original_reqctx_ref = flask.request.environ.get(
             _ENVIRON_REQCTX_REF_KEY
diff --git a/instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py b/instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py
index 039ae9db83..1680b327c7 100644
--- a/instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py
+++ b/instrumentation/opentelemetry-instrumentation-flask/tests/test_programmatic.py
@@ -18,7 +18,9 @@
 
 from flask import Flask, request
 
+import opentelemetry.instrumentation.flask as otel_flask
 from opentelemetry import trace
+from opentelemetry.instrumentation._labeler import clear_labeler, get_labeler
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     OTEL_SEMCONV_STABILITY_OPT_IN,
@@ -153,6 +155,7 @@ def expected_attributes_new(override_attributes):
 class TestProgrammatic(InstrumentationTest, WsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -180,6 +183,18 @@ def setUp(self):
         self.exclude_patch.start()
 
         self.app = Flask(__name__)
+
+        @self.app.route("/test_labeler")
+        def test_labeler_route():
+            labeler = get_labeler()
+            labeler.add("custom_attr", "test_value")
+            labeler.add("http.method", "POST")
+            return "OK"
+
+        @self.app.route("/no_labeler")
+        def test_no_labeler_route():
+            return "No labeler"
+
         FlaskInstrumentor().instrument_app(self.app)
 
         self._common_initialization()
@@ -521,6 +536,122 @@ def test_flask_metrics(self):
                     )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    def test_flask_metrics_custom_attributes_skip_override(self):
+        self.client.get("/test_labeler")
+        metrics = self.get_sorted_metrics(SCOPE)
+        active_requests_point_seen = False
+        histogram_point_seen = False
+
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                for point in data_points:
+                    self.assertIsInstance(point, NumberDataPoint)
+                    if point.attributes.get("custom_attr") != "test_value":
+                        continue
+                    self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+                    active_requests_point_seen = True
+                continue
+
+            if metric.name != "http.server.duration":
+                continue
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            histogram_point_seen = True
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertTrue(histogram_point_seen)
+
+    def test_flask_metrics_active_requests_custom_attributes_new_semconv(self):
+        self.client.get("/test_labeler")
+        metrics = self.get_sorted_metrics(SCOPE)
+        active_requests_point_seen = False
+
+        for metric in metrics:
+            if metric.name != "http.server.active_requests":
+                continue
+
+            data_points = list(metric.data.data_points)
+            for point in data_points:
+                self.assertIsInstance(point, NumberDataPoint)
+                if point.attributes.get("custom_attr") != "test_value":
+                    continue
+                self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+                active_requests_point_seen = True
+
+        self.assertTrue(active_requests_point_seen)
+
+    def test_flask_active_requests_attrs_use_enrich_old_semconv(self):
+        with patch(
+            "opentelemetry.instrumentation.flask.enrich_metric_attributes",
+            wraps=otel_flask.enrich_metric_attributes,
+        ) as mock_enrich:
+            self.client.get("/hello/123")
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if HTTP_METHOD in attrs and HTTP_STATUS_CODE not in attrs:
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_seen)
+
+    def test_flask_active_requests_attrs_use_enrich_new_semconv(self):
+        with patch(
+            "opentelemetry.instrumentation.flask.enrich_metric_attributes",
+            wraps=otel_flask.enrich_metric_attributes,
+        ) as mock_enrich:
+            self.client.get("/hello/123")
+
+        enriched_active_attrs_seen = False
+        for call in mock_enrich.call_args_list:
+            if not call.args:
+                continue
+            attrs = call.args[0]
+            if (
+                HTTP_REQUEST_METHOD in attrs
+                and HTTP_RESPONSE_STATUS_CODE not in attrs
+            ):
+                enriched_active_attrs_seen = True
+                break
+
+        self.assertTrue(enriched_active_attrs_seen)
+
+    def test_flask_metrics_no_labeler(self):
+        self.client.get("/test_labeler")
+        self.client.get("/no_labeler")
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        labeler_attrs = None
+        no_labeler_attrs = None
+
+        for metric in metrics:
+            if metric.name != "http.server.duration":
+                continue
+
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 2)
+
+            for point in data_points:
+                self.assertIsInstance(point, HistogramDataPoint)
+                attrs = point.attributes
+                if attrs.get(HTTP_TARGET) == "/test_labeler":
+                    labeler_attrs = attrs
+                elif attrs.get(HTTP_TARGET) == "/no_labeler":
+                    no_labeler_attrs = attrs
+
+        self.assertIsNotNone(labeler_attrs)
+        self.assertIsNotNone(no_labeler_attrs)
+        self.assertEqual(labeler_attrs["custom_attr"], "test_value")
+        self.assertNotIn("custom_attr", no_labeler_attrs)
+
     def test_flask_metrics_new_semconv(self):
         start = default_timer()
         self.client.get("/hello/123")
diff --git a/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py b/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py
index b30423d3bf..8d8d6c35d5 100644
--- a/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py
@@ -211,6 +211,58 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
 To record all of the names set the environment variable  ``OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS``
 to a value that evaluates to true, e.g. ``1``.
 
+Custom Metrics Attributes using Labeler
+***************************************
+The WSGI instrumentation reads custom attributes from the labeler (when present)
+and applies them to all HTTP server metric points emitted by the middleware:
+
+- Active requests counter (`http.server.active_requests`)
+- Duration histograms (`http.server.duration` and/or
+    `http.server.request.duration` depending on semantic convention mode)
+
+Labeler attributes are request-scoped and merged without overriding base metric
+attributes at the same keys.
+
+.. code-block:: python
+
+    import web
+    from cheroot import wsgi
+
+    from opentelemetry.instrumentation._labeler import get_labeler
+    from opentelemetry.instrumentation.wsgi import OpenTelemetryMiddleware
+
+    urls = (
+        '/', 'index',
+        '/users/(.+)/', 'user_profile'
+    )
+
+    class user_profile:
+        def GET(self, user_id):
+            # Get the labeler for the current request
+            labeler = get_labeler()
+
+            # Add custom attributes to WSGI instrumentation metrics
+            labeler.add("user_id", user_id)
+            labeler.add("user_type", "registered")
+
+            # Or, add multiple attributes at once
+            labeler.add_attributes({
+                "feature_flag": "new_ui",
+                "experiment_group": "control"
+            })
+            return f"User profile for {user_id}"
+
+    if __name__ == "__main__":
+        app = web.application(urls, globals())
+        func = app.wsgifunc()
+
+        func = OpenTelemetryMiddleware(func)
+
+        server = wsgi.WSGIServer(
+            ("localhost", 5100), func, server_name="localhost"
+        )
+        server.start()
+
 API
 ---
 """
@@ -223,6 +275,10 @@ def response_hook(span: Span, environ: WSGIEnvironment, status: str, response_he
 from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, TypeVar, cast
 
 from opentelemetry import context, trace
+from opentelemetry.instrumentation._labeler import (
+    enrich_metric_attributes,
+    get_labeler_attributes,
+)
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     _filter_semconv_active_request_count_attr,
@@ -668,6 +724,8 @@ def _start_response(
 
     # pylint: disable=too-many-branches
     # pylint: disable=too-many-locals
+    # pylint: disable=too-many-public-methods
+    # pylint: disable=too-many-statements
     def __call__(
         self, environ: WSGIEnvironment, start_response: StartResponse
     ):
@@ -684,6 +742,9 @@ def __call__(
             req_attrs,
             self._sem_conv_opt_in_mode,
         )
+        active_requests_count_attrs = enrich_metric_attributes(
+            active_requests_count_attrs
+        )
 
         span, token = _start_internal_or_server_span(
             tracer=self.tracer,
@@ -708,6 +769,8 @@ def __call__(
             response_hook = functools.partial(response_hook, span, environ)
 
         start = default_timer()
+        labeler_metric_attributes = {}
+        detach_token_in_finally = False
         self.active_requests_counter.add(1, active_requests_count_attrs)
         try:
             with trace.use_span(span):
@@ -718,7 +781,12 @@ def __call__(
                     req_attrs,
                     self._sem_conv_opt_in_mode,
                 )
-                iterable = self.wsgi(environ, start_response)
+                try:
+                    iterable = self.wsgi(environ, start_response)
+                except Exception:
+                    labeler_metric_attributes.update(get_labeler_attributes())
+                    raise
+                labeler_metric_attributes.update(get_labeler_attributes())
                 return _end_span_after_iterating(iterable, span, token)
         except Exception as ex:
             if _report_new(self._sem_conv_opt_in_mode):
@@ -727,8 +795,7 @@ def __call__(
                     span.set_attribute(ERROR_TYPE, type(ex).__qualname__)
                 span.set_status(Status(StatusCode.ERROR, str(ex)))
             span.end()
-            if token is not None:
-                context.detach(token)
+            detach_token_in_finally = True
             raise
         finally:
             duration_s = default_timer() - start
@@ -737,6 +804,12 @@ def __call__(
                 duration_attrs_old = _parse_duration_attrs(
                     req_attrs, _StabilityMode.DEFAULT
                 )
+                duration_attrs_old = enrich_metric_attributes(
+                    duration_attrs_old
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_old:
+                        duration_attrs_old[key] = value
                 self.duration_histogram_old.record(
                     max(round(duration_s * 1000), 0),
                     duration_attrs_old,
@@ -746,12 +819,20 @@ def __call__(
                 duration_attrs_new = _parse_duration_attrs(
                     req_attrs, _StabilityMode.HTTP
                 )
+                duration_attrs_new = enrich_metric_attributes(
+                    duration_attrs_new
+                )
+                for key, value in labeler_metric_attributes.items():
+                    if key not in duration_attrs_new:
+                        duration_attrs_new[key] = value
                 self.duration_histogram_new.record(
                     max(duration_s, 0),
                     duration_attrs_new,
                     context=active_metric_ctx,
                 )
             self.active_requests_counter.add(-1, active_requests_count_attrs)
+            if detach_token_in_finally and token is not None:
+                context.detach(token)
 
 
 # Put this in a subfunction to not delay the call to the wrapped
diff --git a/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py b/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py
index edb6655c5c..c8b3c53f91 100644
--- a/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py
+++ b/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py
@@ -22,6 +22,7 @@
 
 import opentelemetry.instrumentation.wsgi as otel_wsgi
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._labeler import clear_labeler, get_labeler
 from opentelemetry.instrumentation._semconv import (
     HTTP_DURATION_HISTOGRAM_BUCKETS_NEW,
     OTEL_SEMCONV_STABILITY_OPT_IN,
@@ -142,6 +143,30 @@ def error_wsgi_unhandled(environ, start_response):
     raise ValueError
 
 
+def error_wsgi_unhandled_custom_attrs(environ, start_response):
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add_attributes({"endpoint_type": "test", "feature_flag": True})
+    assert isinstance(environ, dict)
+    raise ValueError
+
+
+def error_wsgi_unhandled_override_attrs(environ, start_response):
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.method", "POST")
+    assert isinstance(environ, dict)
+    raise ValueError
+
+
+def error_wsgi_unhandled_override_attrs_new_semconv(environ, start_response):
+    labeler = get_labeler()
+    labeler.add("custom_attr", "test_value")
+    labeler.add("http.request.method", "POST")
+    assert isinstance(environ, dict)
+    raise ValueError
+
+
 def wsgi_with_custom_response_headers(environ, start_response):
     assert isinstance(environ, dict)
     start_response(
@@ -207,9 +232,11 @@ def wsgi_with_repeat_custom_response_headers(environ, start_response):
 SCOPE = "opentelemetry.instrumentation.wsgi"
 
 
+# pylint: disable=too-many-public-methods
 class TestWsgiApplication(WsgiTestBase):
     def setUp(self):
         super().setUp()
+        clear_labeler()
 
         test_name = ""
         if hasattr(self, "_testMethodName"):
@@ -451,6 +478,130 @@ def test_wsgi_metrics(self):
                     )
         self.assertTrue(number_data_point_seen and histogram_data_point_seen)
 
+    def test_wsgi_metrics_custom_attributes_skip_override_old_semconv(self):
+        labeler = get_labeler()
+        labeler.add("custom_attr", "test_value")
+        labeler.add("http.method", "POST")
+
+        app = otel_wsgi.OpenTelemetryMiddleware(error_wsgi_unhandled)
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        active_requests_point_seen = False
+        histogram_point_seen = False
+
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+                self.assertEqual(point.attributes["custom_attr"], "test_value")
+                active_requests_point_seen = True
+                continue
+
+            if metric.name != "http.server.duration":
+                continue
+
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+            self.assertNotIn("custom_attr", point.attributes)
+            histogram_point_seen = True
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertTrue(histogram_point_seen)
+
+    def test_wsgi_metrics_custom_attributes_skip_override_new_semconv(self):
+        labeler = get_labeler()
+        labeler.add("custom_attr", "test_value")
+        labeler.add("http.request.method", "POST")
+
+        app = otel_wsgi.OpenTelemetryMiddleware(error_wsgi_unhandled)
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        active_requests_point_seen = False
+        histogram_point_seen = False
+
+        for metric in metrics:
+            if metric.name == "http.server.active_requests":
+                data_points = list(metric.data.data_points)
+                self.assertEqual(len(data_points), 1)
+                point = data_points[0]
+                self.assertIsInstance(point, NumberDataPoint)
+                self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+                self.assertEqual(point.attributes["custom_attr"], "test_value")
+                active_requests_point_seen = True
+                continue
+
+            if metric.name != "http.server.request.duration":
+                continue
+
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+            self.assertNotIn("custom_attr", point.attributes)
+            histogram_point_seen = True
+
+        self.assertTrue(active_requests_point_seen)
+        self.assertTrue(histogram_point_seen)
+
+    def test_wsgi_duration_metrics_custom_attributes_skip_override_old_semconv(
+        self,
+    ):
+        app = otel_wsgi.OpenTelemetryMiddleware(
+            error_wsgi_unhandled_override_attrs
+        )
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        histogram_point_seen = False
+
+        for metric in metrics:
+            if metric.name != "http.server.duration":
+                continue
+
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            histogram_point_seen = True
+
+        self.assertTrue(histogram_point_seen)
+
+    def test_wsgi_duration_metrics_custom_attributes_skip_override_new_semconv(
+        self,
+    ):
+        app = otel_wsgi.OpenTelemetryMiddleware(
+            error_wsgi_unhandled_override_attrs_new_semconv
+        )
+        self.assertRaises(ValueError, app, self.environ, self.start_response)
+
+        metrics = self.get_sorted_metrics(SCOPE)
+        histogram_point_seen = False
+
+        for metric in metrics:
+            if metric.name != "http.server.request.duration":
+                continue
+
+            data_points = list(metric.data.data_points)
+            self.assertEqual(len(data_points), 1)
+            point = data_points[0]
+            self.assertIsInstance(point, HistogramDataPoint)
+            self.assertEqual(point.attributes[HTTP_REQUEST_METHOD], "GET")
+            self.assertEqual(point.attributes["custom_attr"], "test_value")
+            histogram_point_seen = True
+
+        self.assertTrue(histogram_point_seen)
+
     def test_wsgi_metrics_exemplars_expected_old_semconv(self):  # type: ignore[func-returns-value]
         """Failing test asserting exemplars should be present for duration histogram (old semconv)."""
         app = otel_wsgi.OpenTelemetryMiddleware(simple_wsgi)
diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py
new file mode 100644
index 0000000000..e36fe18733
--- /dev/null
+++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py
@@ -0,0 +1,98 @@
+# Copyright The OpenTelemetry Authors
+#
+# 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.
+
+"""
+OpenTelemetry Labeler
+=====================
+
+The labeler utility provides a way to add custom attributes to metrics.
+
+This was inspired by OpenTelemetry Go's net/http instrumentation Labeler
+https://github.com/open-telemetry/opentelemetry-go-contrib/pull/306
+
+Usage
+-----
+
+The labeler is typically used within the context of an instrumented request
+or operation. Use ``get_labeler`` to obtain a labeler instance for the
+current context, then add attributes using the ``add`` or
+``add_attributes`` methods.
+
+Example
+-------
+
+Here's a framework-agnostic example showing manual use of the labeler:
+
+.. code-block:: python
+
+    from opentelemetry.instrumentation._labeler import (
+        enrich_metric_attributes,
+        get_labeler,
+    )
+    from opentelemetry.metrics import get_meter
+
+    meter = get_meter("example.manual")
+    duration_histogram = meter.create_histogram(
+        name="http.server.request.duration",
+        unit="s",
+        description="Duration of HTTP server requests.",
+    )
+
+    def record_request(user_id: str, duration_s: float) -> None:
+        labeler = get_labeler()
+        labeler.add("user_id", user_id)
+        labeler.add_attributes(
+            {
+                "has_premium": user_id in ["123", "456"],
+                "experiment_group": "control",
+                "feature_enabled": True,
+                "user_segment": "active",
+            }
+        )
+
+        base_attributes = {
+            "http.request.method": "GET",
+            "http.response.status_code": 200,
+        }
+        duration_histogram.record(
+            max(duration_s, 0),
+            enrich_metric_attributes(base_attributes),
+        )
+
+This package introduces the shared Labeler API and helper utilities.
+Framework-specific integration points that call
+``enrich_metric_attributes`` (for example before ``Histogram.record``)
+can be added by individual instrumentors.
+
+When instrumentors use ``enrich_metric_attributes``, it does not
+overwrite base attributes that exist at the same keys.
+"""
+
+from opentelemetry.instrumentation._labeler._internal import (
+    Labeler,
+    clear_labeler,
+    enrich_metric_attributes,
+    get_labeler,
+    get_labeler_attributes,
+    set_labeler,
+)
+
+__all__ = [
+    "Labeler",
+    "get_labeler",
+    "set_labeler",
+    "clear_labeler",
+    "get_labeler_attributes",
+    "enrich_metric_attributes",
+]
diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py
new file mode 100644
index 0000000000..0230d6de0a
--- /dev/null
+++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py
@@ -0,0 +1,279 @@
+# Copyright The OpenTelemetry Authors
+#
+# 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.
+
+import logging
+import threading
+from types import MappingProxyType
+from typing import Any, Dict, Mapping, Optional, Union
+
+from opentelemetry.context import attach, create_key, get_value, set_value
+from opentelemetry.util.types import AttributeValue
+
+LABELER_CONTEXT_KEY = create_key("otel_labeler")
+
+_logger = logging.getLogger(__name__)
+
+
+class Labeler:
+    """
+    Stores custom attributes for the current OTel context.
+
+    This feature is experimental and unstable.
+    """
+
+    def __init__(
+        self, max_custom_attrs: int = 20, max_attr_value_length: int = 100
+    ):
+        """
+        Initialize a new Labeler instance.
+
+        Args:
+            max_custom_attrs: Maximum number of custom attributes to store.
+                When this limit is reached, new attributes will be ignored;
+                existing attributes can still be updated.
+            max_attr_value_length: Maximum length for string attribute values.
+                String values exceeding this length will be truncated.
+        """
+        self._lock = threading.Lock()
+        self._attributes: Dict[str, Union[str, int, float, bool]] = {}
+        self._max_custom_attrs = max_custom_attrs
+        self._max_attr_value_length = max_attr_value_length
+
+    def add(self, key: str, value: Any) -> None:
+        """
+        Add a single attribute to the labeler, subject to the labeler's limits:
+        - If max_custom_attrs limit is reached and this is a new key, the attribute is ignored
+        - String values exceeding max_attr_value_length are truncated
+
+        Args:
+            key: attribute key
+            value: attribute value, must be a primitive type: str, int, float, or bool
+        """
+        if not isinstance(value, (str, int, float, bool)):
+            _logger.warning(
+                "Skipping attribute '%s': value must be str, int, float, or bool, got %s",
+                key,
+                type(value).__name__,
+            )
+            return
+
+        with self._lock:
+            if (
+                len(self._attributes) >= self._max_custom_attrs
+                and key not in self._attributes
+            ):
+                return
+
+            if (
+                isinstance(value, str)
+                and len(value) > self._max_attr_value_length
+            ):
+                value = value[: self._max_attr_value_length]
+
+            self._attributes[key] = value
+
+    def add_attributes(self, attributes: Dict[str, Any]) -> None:
+        """
+        Add multiple attributes to the labeler, subject to the labeler's limits:
+        - If max_custom_attrs limit is reached and this is a new key, the attribute is ignored
+        - Existing attributes can still be updated
+        - String values exceeding max_attr_value_length are truncated
+
+        Args:
+            attributes: Dictionary of attributes to add. Values must be primitive types
+                (str, int, float, or bool)
+        """
+        with self._lock:
+            for key, value in attributes.items():
+                if not isinstance(value, (str, int, float, bool)):
+                    _logger.warning(
+                        "Skipping attribute '%s': value must be str, int, float, or bool, got %s",
+                        key,
+                        type(value).__name__,
+                    )
+                    continue
+
+                if (
+                    len(self._attributes) >= self._max_custom_attrs
+                    and key not in self._attributes
+                ):
+                    continue
+
+                if (
+                    isinstance(value, str)
+                    and len(value) > self._max_attr_value_length
+                ):
+                    value = value[: self._max_attr_value_length]
+
+                self._attributes[key] = value
+
+    def get_attributes(self) -> Mapping[str, Union[str, int, float, bool]]:
+        """
+        Return a read-only mapping view of attributes in this labeler.
+        """
+        with self._lock:
+            return MappingProxyType(self._attributes)
+
+    def clear(self) -> None:
+        with self._lock:
+            self._attributes.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._attributes)
+
+
+def _attach_context_value(value: Optional[Labeler]) -> None:
+    """
+    Attach a new OpenTelemetry context containing the given labeler value.
+
+    This helper is fail-safe: context attach errors are suppressed and
+    logged at debug level.
+
+    Args:
+        value: Labeler instance to store in context, or ``None`` to clear it.
+    """
+    try:
+        updated_context = set_value(LABELER_CONTEXT_KEY, value)
+        attach(updated_context)
+    except Exception:  # pylint: disable=broad-exception-caught
+        _logger.debug("Failed to attach labeler context", exc_info=True)
+
+
+def get_labeler() -> Labeler:
+    """
+    Get the Labeler instance for the current OTel context.
+
+    If no Labeler exists in the current context, a new one is created
+    and stored in the context.
+
+    Returns:
+        Labeler instance for the current OTel context, or a new empty Labeler
+        if no Labeler is currently stored in context.
+    """
+    try:
+        current_value = get_value(LABELER_CONTEXT_KEY)
+    except Exception:  # pylint: disable=broad-exception-caught
+        _logger.debug("Failed to read labeler from context", exc_info=True)
+        current_value = None
+
+    if isinstance(current_value, Labeler):
+        return current_value
+
+    labeler = Labeler()
+    _attach_context_value(labeler)
+    return labeler
+
+
+def set_labeler(labeler: Any) -> None:
+    """
+    Set the Labeler instance for the current OTel context.
+
+    Args:
+        labeler: The Labeler instance to set
+    """
+    if not isinstance(labeler, Labeler):
+        _logger.warning(
+            "Skipping set_labeler: value must be Labeler, got %s",
+            type(labeler).__name__,
+        )
+        return
+    _attach_context_value(labeler)
+
+
+def clear_labeler() -> None:
+    """
+    Clear the Labeler instance from the current OTel context.
+
+    This is primarily intended for test isolation or manual context-lifecycle
+    management. In typical framework-instrumented request handling,
+    applications generally should not need to call this directly.
+    """
+    _attach_context_value(None)
+
+
+def get_labeler_attributes() -> Mapping[str, Union[str, int, float, bool]]:
+    """
+    Get attributes from the current labeler, if any.
+
+    Returns:
+        Read-only mapping of custom attributes, or an empty read-only mapping
+        if no labeler exists.
+    """
+    empty_attributes: Dict[str, Union[str, int, float, bool]] = {}
+    try:
+        current_value = get_value(LABELER_CONTEXT_KEY)
+    except Exception:  # pylint: disable=broad-exception-caught
+        _logger.debug(
+            "Failed to read labeler attributes from context", exc_info=True
+        )
+        return MappingProxyType(empty_attributes)
+
+    if not isinstance(current_value, Labeler):
+        return MappingProxyType(empty_attributes)
+    return current_value.get_attributes()
+
+
+def enrich_metric_attributes(
+    base_attributes: Dict[str, Any],
+    enrich_enabled: bool = True,
+) -> Dict[str, AttributeValue]:
+    """
+    Combines base_attributes with custom attributes from the current labeler,
+    returning a new dictionary of attributes according to the labeler configuration:
+    - Attributes that would override base_attributes are skipped
+    - If max_custom_attrs limit is reached and this is a new key, the attribute is ignored
+    - String values exceeding max_attr_value_length are truncated
+
+    Args:
+        base_attributes: The base attributes for the metric
+        enrich_enabled: Whether to include custom labeler attributes
+
+    Returns:
+        Dictionary combining base and custom attributes. If no custom attributes,
+        returns a copy of the original base attributes.
+    """
+    if not enrich_enabled:
+        return base_attributes.copy()
+
+    labeler_attributes = get_labeler_attributes()
+    if not labeler_attributes:
+        return base_attributes.copy()
+
+    try:
+        labeler = get_value(LABELER_CONTEXT_KEY)
+    except Exception:  # pylint: disable=broad-exception-caught
+        labeler = None
+
+    if not isinstance(labeler, Labeler):
+        return base_attributes.copy()
+
+    enriched_attributes = base_attributes.copy()
+    added_count = 0
+    for key, value in labeler_attributes.items():
+        if added_count >= labeler._max_custom_attrs:
+            break
+        if key in base_attributes:
+            continue
+
+        if (
+            isinstance(value, str)
+            and len(value) > labeler._max_attr_value_length
+        ):
+            value = value[: labeler._max_attr_value_length]
+
+        enriched_attributes[key] = value
+        added_count += 1
+
+    return enriched_attributes
diff --git a/opentelemetry-instrumentation/tests/test_labeler.py b/opentelemetry-instrumentation/tests/test_labeler.py
new file mode 100644
index 0000000000..673cba1cc0
--- /dev/null
+++ b/opentelemetry-instrumentation/tests/test_labeler.py
@@ -0,0 +1,210 @@
+# Copyright The OpenTelemetry Authors
+#
+# 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.
+
+import contextvars
+import threading
+import unittest
+from unittest.mock import patch
+
+from opentelemetry.instrumentation._labeler import (
+    Labeler,
+    clear_labeler,
+    enrich_metric_attributes,
+    get_labeler,
+    get_labeler_attributes,
+    set_labeler,
+)
+from opentelemetry.instrumentation._labeler._internal import (
+    LABELER_CONTEXT_KEY,
+)
+
+
+class TestLabeler(unittest.TestCase):
+    def setUp(self):
+        clear_labeler()
+
+    def test_labeler_init(self):
+        labeler = Labeler()
+        self.assertEqual(labeler.get_attributes(), {})
+        self.assertEqual(len(labeler), 0)
+
+    def test_add_single_attribute(self):
+        labeler = Labeler()
+        labeler.add("test_key", "test_value")
+        self.assertEqual(labeler.get_attributes(), {"test_key": "test_value"})
+
+    def test_add_attributes_dict(self):
+        labeler = Labeler()
+        attrs = {"key1": "value1", "key2": 42, "key3": False}
+        labeler.add_attributes(attrs)
+        self.assertEqual(labeler.get_attributes(), attrs)
+
+    def test_overwrite_attribute(self):
+        labeler = Labeler()
+        labeler.add("key1", "original")
+        labeler.add("key1", "updated")
+        self.assertEqual(labeler.get_attributes(), {"key1": "updated"})
+
+    def test_clear_attributes(self):
+        labeler = Labeler()
+        labeler.add("key1", "value1")
+        labeler.add("key2", "value2")
+        labeler.clear()
+        self.assertEqual(labeler.get_attributes(), {})
+
+    def test_add_invalid_types_logs_warning_and_skips(self):
+        labeler = Labeler()
+        with patch(
+            "opentelemetry.instrumentation._labeler._internal._logger.warning"
+        ) as mock_warning:
+            labeler.add("valid", "value")
+            labeler.add("dict_key", {"nested": "dict"})
+            labeler.add("list_key", [1, 2, 3])
+            labeler.add("none_key", None)
+            labeler.add("another_valid", 123)
+
+        self.assertEqual(mock_warning.call_count, 3)
+        self.assertEqual(
+            labeler.get_attributes(), {"valid": "value", "another_valid": 123}
+        )
+
+    def test_limit_and_truncation(self):
+        labeler = Labeler(max_custom_attrs=2, max_attr_value_length=5)
+        labeler.add("a", "1234567")
+        labeler.add("b", "ok")
+        labeler.add("c", "ignored")
+        self.assertEqual(labeler.get_attributes(), {"a": "12345", "b": "ok"})
+
+    def test_enrich_metric_attributes_skips_base_overrides(self):
+        base_attributes = {"http.method": "GET", "http.status_code": 200}
+        labeler = get_labeler()
+        labeler.add("http.method", "POST")
+        labeler.add("custom_attr", "test-value")
+
+        enriched = enrich_metric_attributes(base_attributes)
+        self.assertEqual(enriched["http.method"], "GET")
+        self.assertEqual(enriched["custom_attr"], "test-value")
+        self.assertEqual(base_attributes["http.method"], "GET")
+
+    def test_thread_safety(self):
+        labeler = Labeler(max_custom_attrs=1100)
+        num_threads = 5
+        num_ops = 100
+
+        def worker(thread_id):
+            for index in range(num_ops):
+                labeler.add(f"thread_{thread_id}_{index}", f"v_{index}")
+
+        threads = [
+            threading.Thread(target=worker, args=(i,))
+            for i in range(num_threads)
+        ]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        self.assertEqual(len(labeler.get_attributes()), num_threads * num_ops)
+
+
+class TestLabelerContext(unittest.TestCase):
+    def setUp(self):
+        clear_labeler()
+
+    def test_get_labeler_creates_new(self):
+        labeler = get_labeler()
+        self.assertIsInstance(labeler, Labeler)
+
+    def test_get_labeler_returns_same_instance(self):
+        labeler1 = get_labeler()
+        labeler1.add("test", "value")
+        labeler2 = get_labeler()
+        self.assertIs(labeler1, labeler2)
+
+    def test_set_labeler(self):
+        custom_labeler = Labeler()
+        custom_labeler.add("custom", "value")
+        set_labeler(custom_labeler)
+        self.assertIs(get_labeler(), custom_labeler)
+
+    def test_set_labeler_invalid_type_is_ignored(self):
+        with patch(
+            "opentelemetry.instrumentation._labeler._internal._logger.warning"
+        ) as mock_warning:
+            set_labeler("bad")  # type: ignore[arg-type]
+        self.assertEqual(mock_warning.call_count, 1)
+
+    def test_clear_labeler(self):
+        labeler = get_labeler()
+        labeler.add("test", "value")
+        clear_labeler()
+        new_labeler = get_labeler()
+        self.assertIsNot(new_labeler, labeler)
+        self.assertEqual(new_labeler.get_attributes(), {})
+
+    def test_get_labeler_attributes(self):
+        clear_labeler()
+        self.assertEqual(get_labeler_attributes(), {})
+        labeler = get_labeler()
+        labeler.add("test", "value")
+        self.assertEqual(get_labeler_attributes(), {"test": "value"})
+
+    def test_context_isolation(self):
+        def context_worker(context_id, results):
+            labeler = get_labeler()
+            labeler.add("context_id", context_id)
+            results[context_id] = dict(labeler.get_attributes())
+
+        results = {}
+        for operation in range(3):
+            ctx = contextvars.copy_context()
+            ctx.run(context_worker, operation, results)
+
+        for operation in range(3):
+            self.assertEqual(results[operation], {"context_id": operation})
+
+
+class TestLabelerFailSafe(unittest.TestCase):
+    def setUp(self):
+        clear_labeler()
+
+    def test_get_labeler_failsafe_on_get_value_error(self):
+        with patch(
+            "opentelemetry.instrumentation._labeler._internal.get_value",
+            side_effect=RuntimeError("boom"),
+        ):
+            labeler = get_labeler()
+        self.assertIsInstance(labeler, Labeler)
+
+    def test_set_and_clear_failsafe_on_attach_error(self):
+        labeler = Labeler()
+        with patch(
+            "opentelemetry.instrumentation._labeler._internal.attach",
+            side_effect=RuntimeError("boom"),
+        ):
+            set_labeler(labeler)
+            clear_labeler()
+
+        self.assertIsInstance(get_labeler(), Labeler)
+
+    def test_get_labeler_attributes_failsafe(self):
+        with patch(
+            "opentelemetry.instrumentation._labeler._internal.get_value",
+            side_effect=RuntimeError("boom"),
+        ):
+            attrs = get_labeler_attributes()
+        self.assertEqual(attrs, {})
+
+    def test_context_key_constant(self):
+        self.assertTrue(isinstance(LABELER_CONTEXT_KEY, str))
diff --git a/pyproject.toml b/pyproject.toml
index 92cfe8b5d2..46a3d7d423 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -212,6 +212,7 @@ include = [
   "exporter/opentelemetry-exporter-credential-provider-gcp",
   "instrumentation/opentelemetry-instrumentation-aiohttp-client",
   "opamp/opentelemetry-opamp-client",
+  "opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler",
 ]
 # We should also add type hints to the test suite - It helps on finding bugs.
 # We are excluding for now because it's easier, and more important to add to the instrumentation packages.