diff --git a/pkg/collector/corechecks/containers/kubelet/provider/cadvisor/provider_test.go b/pkg/collector/corechecks/containers/kubelet/provider/cadvisor/provider_test.go
index dd71c04458e5..b8dc67e32189 100644
--- a/pkg/collector/corechecks/containers/kubelet/provider/cadvisor/provider_test.go
+++ b/pkg/collector/corechecks/containers/kubelet/provider/cadvisor/provider_test.go
@@ -321,10 +321,10 @@ func (suite *ProviderTestSuite) TestPrometheusFiltering() {
 				suite.T().Fatalf("error created kubelet mock: %v", err)
 			}
 
-			prometheus.ParseMetricsWithFilterFunc = func(data []byte, filter []string) ([]prom.MetricFamily, error) {
+			prometheus.ParseMetricsWithFilterFunc = func(data []byte, filter []string, contentType string) ([]prom.MetricFamily, error) {
 				// We are going to intercept the parsed prometheus metric family data to determine if the configured provider
 				// has the expected text blacklist settings by default, and that this functionality still works
-				metrics, err := prom.ParseMetricsWithFilter(data, filter)
+				metrics, err := prom.ParseMetricsWithFilter(data, filter, contentType)
 				var found bool
 				for _, metric := range metrics {
 					if metric.Name == "container_cpu_usage_seconds_total" {
diff --git a/pkg/collector/corechecks/containers/kubelet/provider/prometheus/provider.go b/pkg/collector/corechecks/containers/kubelet/provider/prometheus/provider.go
index 43a78b80ff6d..8094f86af03d 100644
--- a/pkg/collector/corechecks/containers/kubelet/provider/prometheus/provider.go
+++ b/pkg/collector/corechecks/containers/kubelet/provider/prometheus/provider.go
@@ -178,7 +178,7 @@ func (p *Provider) Provide(kc kubelet.KubeUtilInterface, sender sender.Sender) e
 		return nil
 	}
 
-	metrics, err := ParseMetricsWithFilterFunc(data, p.ScraperConfig.TextFilterBlacklist)
+	metrics, err := ParseMetricsWithFilterFunc(data, p.ScraperConfig.TextFilterBlacklist, "")
 	if err != nil {
 		return err
 	}
diff --git a/pkg/collector/python/BUILD.bazel b/pkg/collector/python/BUILD.bazel
index 5668eba3532b..fdf17783c4d9 100644
--- a/pkg/collector/python/BUILD.bazel
+++ b/pkg/collector/python/BUILD.bazel
@@ -162,6 +162,7 @@ go_library(
         "//pkg/util/kubernetes/kubelet",
         "//pkg/util/log",
         "//pkg/util/option",
+        "//pkg/util/prometheus",
         "//pkg/util/retry",
         "//pkg/version",
         "@com_github_datadog_agent_payload_v5//healthplatform",
diff --git a/pkg/collector/python/datadog_agent.go b/pkg/collector/python/datadog_agent.go
index 8ea9bca69d0e..110bda1dd3e6 100644
--- a/pkg/collector/python/datadog_agent.go
+++ b/pkg/collector/python/datadog_agent.go
@@ -32,6 +32,7 @@ import (
 	hostnameUtil "github.com/DataDog/datadog-agent/pkg/util/hostname"
 	"github.com/DataDog/datadog-agent/pkg/util/kubernetes/clustername"
 	"github.com/DataDog/datadog-agent/pkg/util/log"
+	promutil "github.com/DataDog/datadog-agent/pkg/util/prometheus"
 	"github.com/DataDog/datadog-agent/pkg/version"
 )
 
@@ -744,6 +745,34 @@ func ResolveIssue(issueID *C.char, errOut **C.char) {
 	hp.ResolveIssue(id)
 }
 
+// ParsePrometheusMetrics parses Prometheus/OpenMetrics text format using the Go parser
+// and returns the result as a JSON string.
+//
+//export ParsePrometheusMetrics
+func ParsePrometheusMetrics(rawText *C.char, contentType *C.char, errResult **C.char) *C.char {
+	data := []byte(C.GoString(rawText))
+	jsonResult, err := promutil.ParseMetricsToJSON(data, C.GoString(contentType))
+	if err != nil {
+		*errResult = TrackedCString(err.Error())
+		return nil
+	}
+	return TrackedCString(jsonResult)
+}
+
+// ProcessPrometheusMetrics parses Prometheus/OpenMetrics text format using the Go parser,
+// applies label/tag processing based on the provided config, and returns processed results as JSON.
+//
+//export ProcessPrometheusMetrics
+func ProcessPrometheusMetrics(rawText *C.char, contentType *C.char, configJSON *C.char, errResult **C.char) *C.char {
+	data := []byte(C.GoString(rawText))
+	jsonResult, err := promutil.ProcessMetricsToJSON(data, C.GoString(contentType), C.GoString(configJSON))
+	if err != nil {
+		*errResult = TrackedCString(err.Error())
+		return nil
+	}
+	return TrackedCString(jsonResult)
+}
+
 // httpHeaders returns a http headers including various basic information (User-Agent, Content-Type...).
 func httpHeaders() map[string]string {
 	av, _ := version.Agent()
diff --git a/pkg/collector/python/datadog_agent_test.go b/pkg/collector/python/datadog_agent_test.go
index 3b2e509a51b8..7f227df72bb3 100644
--- a/pkg/collector/python/datadog_agent_test.go
+++ b/pkg/collector/python/datadog_agent_test.go
@@ -39,6 +39,18 @@ func TestEmitAgentTelemetry(t *testing.T) {
 	testEmitAgentTelemetry(t)
 }
 
+func TestParsePrometheusMetrics(t *testing.T) {
+	testParsePrometheusMetrics(t)
+}
+
+func TestParsePrometheusMetricsError(t *testing.T) {
+	testParsePrometheusMetricsError(t)
+}
+
+func TestProcessPrometheusMetrics(t *testing.T) {
+	testProcessPrometheusMetrics(t)
+}
+
 func TestObfuscaterConfig(t *testing.T) {
 	testObfuscaterConfig(t)
 }
diff --git a/pkg/collector/python/init.go b/pkg/collector/python/init.go
index 211321157bb8..65d8e72ee1b7 100644
--- a/pkg/collector/python/init.go
+++ b/pkg/collector/python/init.go
@@ -99,6 +99,8 @@ char* ObfuscateMongoDBString(char *, char **);
 void EmitAgentTelemetry(char *, char *, double, char *);
 void ReportIssue(char *, char *, char **);
 void ResolveIssue(char *, char **);
+char* ParsePrometheusMetrics(char *, char *, char **);
+char* ProcessPrometheusMetrics(char *, char *, char *, char **);
 
 void initDatadogAgentModule(rtloader_t *rtloader) {
 	set_get_clustername_cb(rtloader, GetClusterName);
@@ -120,6 +122,8 @@ void initDatadogAgentModule(rtloader_t *rtloader) {
 	set_emit_agent_telemetry_cb(rtloader, EmitAgentTelemetry);
 	set_report_issue_cb(rtloader, ReportIssue);
 	set_resolve_issue_cb(rtloader, ResolveIssue);
+	set_parse_prometheus_metrics_cb(rtloader, ParsePrometheusMetrics);
+	set_process_prometheus_metrics_cb(rtloader, ProcessPrometheusMetrics);
 }
 
 //
diff --git a/pkg/collector/python/test_datadog_agent.go b/pkg/collector/python/test_datadog_agent.go
index c00e9706a5e2..45daaf0dc99c 100644
--- a/pkg/collector/python/test_datadog_agent.go
+++ b/pkg/collector/python/test_datadog_agent.go
@@ -128,6 +128,80 @@ func testEmitAgentTelemetry(t *testing.T) {
 	assert.True(t, true)
 }
 
+func testParsePrometheusMetrics(t *testing.T) {
+	rawText := C.CString(`# TYPE http_requests_total counter
+http_requests_total{method="GET",status="200"} 1234
+http_requests_total{method="POST",status="500"} 5`)
+
+	var errResult *C.char
+	result := ParsePrometheusMetrics(rawText, nil, &errResult)
+	require.Nil(t, errResult)
+	require.NotNil(t, result)
+
+	var families []struct {
+		Name    string `json:"name"`
+		Type    string `json:"type"`
+		Samples []struct {
+			Labels    map[string]string `json:"labels"`
+			Value     float64           `json:"value"`
+			Timestamp int64             `json:"timestamp"`
+		} `json:"samples"`
+	}
+	err := json.Unmarshal([]byte(C.GoString(result)), &families)
+	require.NoError(t, err)
+	require.Len(t, families, 1)
+	assert.Equal(t, "http_requests", families[0].Name) // _total stripped for counters
+	assert.Equal(t, "COUNTER", families[0].Type)
+	require.Len(t, families[0].Samples, 2)
+	assert.Equal(t, 1234.0, families[0].Samples[0].Value)
+	assert.Equal(t, "GET", families[0].Samples[0].Labels["method"])
+}
+
+func testParsePrometheusMetricsError(t *testing.T) {
+	// Invalid prometheus text should return an error
+	rawText := C.CString(`{invalid`)
+
+	var errResult *C.char
+	result := ParsePrometheusMetrics(rawText, nil, &errResult)
+	assert.Nil(t, result)
+	assert.NotNil(t, errResult)
+}
+
+func testProcessPrometheusMetrics(t *testing.T) {
+	rawText := C.CString(`# TYPE http_requests_total counter
+http_requests_total{method="GET",status="200"} 1234
+http_requests_total{method="POST",status="500"} 5`)
+
+	configJSON := C.CString(`{"static_tags":["env:test"],"exclude_labels":["__name__"],"rename_labels":{"method":"http_method"}}`)
+
+	var errResult *C.char
+	result := ProcessPrometheusMetrics(rawText, nil, configJSON, &errResult)
+	require.Nil(t, errResult)
+	require.NotNil(t, result)
+
+	var processResult struct {
+		Families []struct {
+			Name    string `json:"name"`
+			Type    string `json:"type"`
+			Samples []struct {
+				SampleName string   `json:"sample_name"`
+				Value      float64  `json:"value"`
+				Tags       []string `json:"tags"`
+				Hostname   string   `json:"hostname"`
+			} `json:"samples"`
+		} `json:"families"`
+	}
+	err := json.Unmarshal([]byte(C.GoString(result)), &processResult)
+	require.NoError(t, err)
+	require.Len(t, processResult.Families, 1)
+	assert.Equal(t, "http_requests", processResult.Families[0].Name) // _total stripped for counters
+	assert.Equal(t, "COUNTER", processResult.Families[0].Type)
+	require.Len(t, processResult.Families[0].Samples, 2)
+	assert.Equal(t, 1234.0, processResult.Families[0].Samples[0].Value)
+	assert.Contains(t, processResult.Families[0].Samples[0].Tags, "env:test")
+	assert.Contains(t, processResult.Families[0].Samples[0].Tags, "http_method:GET")
+}
+
 func testObfuscaterConfig(t *testing.T) {
 	pkgconfigmodel.CleanOverride(t)
 	_ = pkgconfigmock.New(t)
diff --git a/pkg/util/prometheus/BUILD.bazel b/pkg/util/prometheus/BUILD.bazel
index 624f07b40328..8e8a2d5d31ec 100644
--- a/pkg/util/prometheus/BUILD.bazel
+++ b/pkg/util/prometheus/BUILD.bazel
@@ -3,7 +3,10 @@ load("//bazel/rules/go:dd_agent_go_test.bzl", "dd_agent_go_test")
 
 go_library(
     name = "prometheus",
-    srcs = ["parse.go"],
+    srcs = [
+        "parse.go",
+        "process.go",
+    ],
     importpath = "github.com/DataDog/datadog-agent/pkg/util/prometheus",
     visibility = ["//visibility:public"],
     deps = [
@@ -15,7 +18,10 @@ go_library(
 
 dd_agent_go_test(
     name = "prometheus_test",
-    srcs = ["parse_test.go"],
+    srcs = [
+        "parse_test.go",
+        "process_test.go",
+    ],
     embed = [":prometheus"],
     deps = [
         "@com_github_stretchr_testify//assert",
diff --git a/pkg/util/prometheus/parse.go b/pkg/util/prometheus/parse.go
index 34191f25a907..06d0e674594c 100644
--- a/pkg/util/prometheus/parse.go
+++ b/pkg/util/prometheus/parse.go
@@ -10,8 +10,10 @@ package prometheus
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"io"
+	"math"
 	"strings"
 
 	"github.com/prometheus/common/model"
@@ -24,16 +26,24 @@ type Metric map[string]string
 
 // Sample represents a single metric data point.
 type Sample struct {
-	Metric    Metric
-	Value     float64
-	Timestamp int64 // milliseconds since epoch, 0 if not set
+	Metric    Metric  `json:"labels"`
+	Value     float64 `json:"value"`
+	Timestamp int64   `json:"timestamp"` // milliseconds since epoch, 0 if not set
 }
 
 // MetricFamily represents a metric family that is returned by a prometheus endpoint.
 type MetricFamily struct {
-	Name    string
-	Type    string
-	Samples []Sample
+	Name    string   `json:"name"`
+	Type    string   `json:"type"`
+	Samples []Sample `json:"samples"`
+}
+
+// trimCounterSuffix removes the OpenMetrics counter suffix (_total).
+func trimCounterSuffix(name string) string {
+	if trimmed, ok := strings.CutSuffix(name, "_total"); ok {
+		return trimmed
+	}
+	return name
 }
 
 // trimHistogramSuffix removes histogram-specific suffixes (_bucket, _sum, _count).
@@ -78,12 +88,18 @@ func preprocessData(data []byte, filter []string) []byte {
 }
 
 // ParseMetricsWithFilter parses prometheus-formatted metrics from the input data, ignoring lines which contain
-// text that matches the passed in filter.
-func ParseMetricsWithFilter(data []byte, filter []string) ([]MetricFamily, error) {
+// text that matches the passed in filter. The contentType selects the parser: "application/openmetrics-text"
+// uses the OpenMetrics parser, anything else uses the Prometheus text parser.
+func ParseMetricsWithFilter(data []byte, filter []string, contentType string) ([]MetricFamily, error) {
 	data = preprocessData(data, filter)
 
 	st := labels.NewSymbolTable()
-	parser := textparse.NewPromParser(data, st, false)
+	var parser textparse.Parser
+	if strings.HasPrefix(contentType, "application/openmetrics-text") {
+		parser = textparse.NewOpenMetricsParser(data, st)
+	} else {
+		parser = textparse.NewPromParser(data, st, false)
+	}
 
 	var result []MetricFamily
 	var lbls labels.Labels
@@ -112,16 +128,27 @@ func ParseMetricsWithFilter(data []byte, filter []string) ([]MetricFamily, error
 
 		case textparse.EntrySeries:
 			_, ts, value := parser.Series()
+			// Skip NaN/Inf values — they can't be JSON-encoded and the Python
+			// scraper already drops them, so omitting them here is safe.
+			if math.IsNaN(value) || math.IsInf(value, 0) {
+				continue
+			}
 			parser.Labels(&lbls)
 
 			rawName := lbls.Get(model.MetricNameLabel)
 
 			// Fast path: check if raw name matches current family (common for COUNTER/GAUGE)
 			if len(result) == 0 || result[len(result)-1].Name != rawName {
-				// Slow path: try trimming suffix based on current family type
+				// Slow path: try trimming suffix based on current family type.
+				// For COUNTER this handles OpenMetrics format where the TYPE line uses
+				// the base name (e.g. "foo") but series are named "foo_total".
+				// For HISTOGRAM/SUMMARY, sub-series (_bucket, _sum, _count) must be
+				// mapped back to the base family name.
 				name := rawName
 				if len(result) > 0 {
 					switch result[len(result)-1].Type {
+					case "COUNTER":
+						name = trimCounterSuffix(rawName)
 					case "HISTOGRAM":
 						name = trimHistogramSuffix(rawName)
 					case "SUMMARY":
@@ -172,5 +199,26 @@ func ParseMetricsWithFilter(data []byte, filter []string) ([]MetricFamily, error
 
 // ParseMetrics parses prometheus-formatted metrics from the input data.
 func ParseMetrics(data []byte) ([]MetricFamily, error) {
-	return ParseMetricsWithFilter(data, nil)
+	return ParseMetricsWithFilter(data, nil, "")
+}
+
+// ParseMetricsToJSON parses prometheus-formatted metrics and returns the result as a JSON string.
+// This is used by the Python check bridge to avoid Python-side parsing overhead.
+// Counter family names have their _total suffix stripped to match Python prometheus_client (>= 0.14).
+func ParseMetricsToJSON(data []byte, contentType string) (string, error) {
+	families, err := ParseMetricsWithFilter(data, nil, contentType)
+	if err != nil {
+		return "", err
+	}
+	// Strip _total suffix from counter family names to match Python prometheus_client behavior.
+	for i := range families {
+		if families[i].Type == "COUNTER" {
+			families[i].Name = trimCounterSuffix(families[i].Name)
+		}
+	}
+	out, err := json.Marshal(families)
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
 }
diff --git a/pkg/util/prometheus/parse_test.go b/pkg/util/prometheus/parse_test.go
index 6ec5dd81d910..32448aedca14 100644
--- a/pkg/util/prometheus/parse_test.go
+++ b/pkg/util/prometheus/parse_test.go
@@ -6,6 +6,7 @@
 package prometheus
 
 import (
+	"encoding/json"
 	"fmt"
 	"strings"
 	"testing"
@@ -47,7 +48,7 @@ container_memory_usage_bytes{pod="",container="empty"} 500
 container_memory_usage_bytes{pod="other-pod",container="sidecar"} 750`
 
 	t.Run("filter pod_name empty", func(t *testing.T) {
-		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod_name=""`})
+		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod_name=""`}, "")
 		require.NoError(t, err)
 
 		cpuFamily := findFamily(metrics, "container_cpu_usage_seconds_total")
@@ -56,7 +57,7 @@ container_memory_usage_bytes{pod="other-pod",container="sidecar"} 750`
 	})
 
 	t.Run("filter pod empty", func(t *testing.T) {
-		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod=""`})
+		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod=""`}, "")
 		require.NoError(t, err)
 
 		memFamily := findFamily(metrics, "container_memory_usage_bytes")
@@ -65,7 +66,7 @@ container_memory_usage_bytes{pod="other-pod",container="sidecar"} 750`
 	})
 
 	t.Run("filter both empty labels", func(t *testing.T) {
-		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod_name=""`, `pod=""`})
+		metrics, err := ParseMetricsWithFilter([]byte(testData), []string{`pod_name=""`, `pod=""`}, "")
 		require.NoError(t, err)
 
 		cpuFamily := findFamily(metrics, "container_cpu_usage_seconds_total")
@@ -78,7 +79,7 @@ container_memory_usage_bytes{pod="other-pod",container="sidecar"} 750`
 	})
 
 	t.Run("no filter", func(t *testing.T) {
-		metrics, err := ParseMetricsWithFilter([]byte(testData), nil)
+		metrics, err := ParseMetricsWithFilter([]byte(testData), nil, "")
 		require.NoError(t, err)
 
 		cpuFamily := findFamily(metrics, "container_cpu_usage_seconds_total")
@@ -128,18 +129,21 @@ func TestMetricTypeUppercase(t *testing.T) {
 		name         string
 		data         string
 		expectedType string
+		expectedName string
 	}{
 		{
 			name: "counter",
 			data: `# TYPE http_requests_total counter
 http_requests_total 100`,
 			expectedType: "COUNTER",
+			expectedName: "http_requests_total",
 		},
 		{
 			name: "gauge",
 			data: `# TYPE temperature gauge
 temperature 23.5`,
 			expectedType: "GAUGE",
+			expectedName: "temperature",
 		},
 		{
 			name: "histogram",
@@ -148,6 +152,7 @@ request_latency_bucket{le="0.1"} 10
 request_latency_sum 5.5
 request_latency_count 10`,
 			expectedType: "HISTOGRAM",
+			expectedName: "request_latency",
 		},
 		{
 			name: "summary",
@@ -156,11 +161,13 @@ latency{quantile="0.5"} 0.05
 latency_sum 100
 latency_count 200`,
 			expectedType: "SUMMARY",
+			expectedName: "latency",
 		},
 		{
 			name:         "untyped",
 			data:         `some_metric 42`,
 			expectedType: "UNTYPED",
+			expectedName: "some_metric",
 		},
 	}
 
@@ -170,6 +177,7 @@ latency_count 200`,
 			require.NoError(t, err)
 			require.Len(t, metrics, 1)
 			assert.Equal(t, tc.expectedType, metrics[0].Type)
+			assert.Equal(t, tc.expectedName, metrics[0].Name)
 		})
 	}
 }
@@ -241,6 +249,171 @@ func TestMetricsWithLeadingWhitespace(t *testing.T) {
 	}
 }
 
+func TestParseMetricsToJSON(t *testing.T) {
+	testData := `# TYPE http_requests_total counter
+http_requests_total{method="GET",status="200"} 1234
+http_requests_total{method="POST",status="500"} 5
+# TYPE temperature gauge
+temperature 23.5`
+
+	jsonStr, err := ParseMetricsToJSON([]byte(testData), "")
+	require.NoError(t, err)
+
+	var families []MetricFamily
+	err = json.Unmarshal([]byte(jsonStr), &families)
+	require.NoError(t, err)
+
+	require.Len(t, families, 2)
+	assert.Equal(t, "http_requests", families[0].Name) // _total stripped for counters
+	assert.Equal(t, "COUNTER", families[0].Type)
+	require.Len(t, families[0].Samples, 2)
+	assert.Equal(t, 1234.0, families[0].Samples[0].Value)
+	assert.Equal(t, "GET", families[0].Samples[0].Metric["method"])
+
+	assert.Equal(t, "temperature", families[1].Name)
+	assert.Equal(t, "GAUGE", families[1].Type)
+	require.Len(t, families[1].Samples, 1)
+	assert.Equal(t, 23.5, families[1].Samples[0].Value)
+}
+
+func TestParseMetricsToJSONEmpty(t *testing.T) {
+	jsonStr, err := ParseMetricsToJSON([]byte(""), "")
+	require.NoError(t, err)
+	assert.Equal(t, "null", jsonStr)
+}
+
+func TestOpenMetricsCounterTotal(t *testing.T) {
+	// OpenMetrics counters use _total suffix on series but the TYPE line uses the base name.
+	// The parser should group foo_total series under the "foo" family.
+	testData := `# TYPE foo counter
+foo_total 17.0
+foo_total{a="b"} 42.0
+# EOF
+`
+	metrics, err := ParseMetricsWithFilter([]byte(testData), nil, "application/openmetrics-text")
+	require.NoError(t, err)
+
+	fooFamily := findFamily(metrics, "foo")
+	require.NotNil(t, fooFamily, "should find family named 'foo'")
+	assert.Equal(t, "COUNTER", fooFamily.Type)
+	assert.Len(t, fooFamily.Samples, 2, "both foo_total series should be in the foo family")
+}
+
+func TestOpenMetricsGaugeAndHistogram(t *testing.T) {
+	testData := `# TYPE temperature gauge
+temperature 23.5
+# TYPE request_latency histogram
+request_latency_bucket{le="0.1"} 10
+request_latency_bucket{le="+Inf"} 35
+request_latency_sum 50.5
+request_latency_count 35
+# EOF
+`
+	metrics, err := ParseMetricsWithFilter([]byte(testData), nil, "application/openmetrics-text")
+	require.NoError(t, err)
+
+	tempFamily := findFamily(metrics, "temperature")
+	require.NotNil(t, tempFamily)
+	assert.Equal(t, "GAUGE", tempFamily.Type)
+	assert.Len(t, tempFamily.Samples, 1)
+
+	histFamily := findFamily(metrics, "request_latency")
+	require.NotNil(t, histFamily)
+	assert.Equal(t, "HISTOGRAM", histFamily.Type)
+	assert.Len(t, histFamily.Samples, 4)
+}
+
+func TestOpenMetricsContentTypeSelection(t *testing.T) {
+	// Same counter data, but with Prometheus content type should use PromParser.
+	// In Prometheus format, the TYPE line includes _total in the name; the raw
+	// family name is preserved here — stripping only happens in the Python bridge.
+	promData := `# TYPE http_requests_total counter
+http_requests_total{method="GET"} 100
+`
+	metrics, err := ParseMetricsWithFilter([]byte(promData), nil, "text/plain")
+	require.NoError(t, err)
+	require.Len(t, metrics, 1)
+	assert.Equal(t, "http_requests_total", metrics[0].Name)
+	assert.Equal(t, "COUNTER", metrics[0].Type)
+
+	// Empty content type should default to Prometheus parser
+	metrics2, err := ParseMetricsWithFilter([]byte(promData), nil, "")
+	require.NoError(t, err)
+	require.Len(t, metrics2, 1)
+	assert.Equal(t, "http_requests_total", metrics2[0].Name)
+}
+
+func TestParseMetricsToJSONOpenMetrics(t *testing.T) {
+	testData := `# TYPE http_requests counter
+http_requests_total{method="GET"} 100
+# EOF
+`
+	jsonStr, err := ParseMetricsToJSON([]byte(testData), "application/openmetrics-text; version=1.0.0")
+	require.NoError(t, err)
+
+	var families []MetricFamily
+	err = json.Unmarshal([]byte(jsonStr), &families)
+	require.NoError(t, err)
+	require.Len(t, families, 1)
+	assert.Equal(t, "http_requests", families[0].Name)
+	assert.Equal(t, "COUNTER", families[0].Type)
+}
+
+func TestParseMetricsToJSON_NaNInfFiltered(t *testing.T) {
+	testData := `# TYPE test gauge
+test{a="1"} 1.0
+test{a="2"} NaN
+test{a="3"} +Inf
+test{a="4"} -Inf
+test{a="5"} 2.0`
+
+	jsonStr, err := ParseMetricsToJSON([]byte(testData), "")
+	require.NoError(t, err)
+
+	var families []MetricFamily
+	err = json.Unmarshal([]byte(jsonStr), &families)
+	require.NoError(t, err)
+	require.Len(t, families, 1)
+	assert.Len(t, families[0].Samples, 2, "NaN and Inf samples should be filtered out")
+	assert.Equal(t, 1.0, families[0].Samples[0].Value)
+	assert.Equal(t, 2.0, families[0].Samples[1].Value)
+}
+
+func TestCounterTotalStripped(t *testing.T) {
+	// ParseMetrics preserves the raw Prometheus family name (including _total).
+	// _total stripping only happens in the Python bridge (ParseMetricsToJSON).
+	testData := `# TYPE go_memstats_frees_total counter
+go_memstats_frees_total 2.012275711e+09`
+
+	metrics, err := ParseMetrics([]byte(testData))
+	require.NoError(t, err)
+	require.Len(t, metrics, 1)
+	assert.Equal(t, "go_memstats_frees_total", metrics[0].Name, "raw family name should be preserved")
+	assert.Equal(t, "COUNTER", metrics[0].Type)
+	require.Len(t, metrics[0].Samples, 1)
+	assert.Equal(t, "go_memstats_frees_total", metrics[0].Samples[0].Metric["__name__"])
+
+	// ParseMetricsToJSON (Python bridge) strips _total to match Python prometheus_client behavior.
+	jsonStr, err := ParseMetricsToJSON([]byte(testData), "")
+	require.NoError(t, err)
+	var families []MetricFamily
+	require.NoError(t, json.Unmarshal([]byte(jsonStr), &families))
+	require.Len(t, families, 1)
+	assert.Equal(t, "go_memstats_frees", families[0].Name, "Python bridge should strip _total")
+}
+
+func TestCounterWithoutTotalSuffix(t *testing.T) {
+	// A counter whose TYPE line doesn't end in _total
+	testData := `# TYPE http_requests counter
+http_requests 100`
+
+	metrics, err := ParseMetrics([]byte(testData))
+	require.NoError(t, err)
+	require.Len(t, metrics, 1)
+	assert.Equal(t, "http_requests", metrics[0].Name, "name without _total should not be modified")
+	assert.Equal(t, "COUNTER", metrics[0].Type)
+}
+
 func findFamily(families []MetricFamily, name string) *MetricFamily {
 	for i := range families {
 		if families[i].Name == name {
@@ -281,7 +454,7 @@ func BenchmarkParseMetricsWithFilter(b *testing.B) {
 	b.ResetTimer()
 	b.ReportAllocs()
 	for b.Loop() {
-		metrics, err = ParseMetricsWithFilter(data, filter)
+		metrics, err = ParseMetricsWithFilter(data, filter, "")
 	}
 	b.StopTimer()
 
diff --git a/pkg/util/prometheus/process.go b/pkg/util/prometheus/process.go
new file mode 100644
index 000000000000..28ca84584316
--- /dev/null
+++ b/pkg/util/prometheus/process.go
@@ -0,0 +1,496 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package prometheus
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// ProcessConfig holds the label/tag processing configuration passed from Python.
+type ProcessConfig struct {
+	// ExcludeLabels is the set of label names to drop from tags.
+	ExcludeLabels []string `json:"exclude_labels"`
+	// IncludeLabels if non-empty, only these label names become tags.
+	IncludeLabels []string `json:"include_labels"`
+	// RenameLabels maps original label names to new tag names.
+	RenameLabels map[string]string `json:"rename_labels"`
+	// ExcludeMetrics is the set of exact metric names to skip.
+	ExcludeMetrics []string `json:"exclude_metrics"`
+	// ExcludeMetricsPatterns is a list of regex patterns for metric exclusion.
+	ExcludeMetricsPatterns []string `json:"exclude_metrics_patterns"`
+	// ExcludeMetricsByLabels maps label names to lists of regex patterns;
+	// samples matching any pattern are skipped. An empty list means skip any value.
+	ExcludeMetricsByLabels map[string][]string `json:"exclude_metrics_by_labels"`
+	// RawMetricPrefix is stripped from the beginning of all metric names.
+	RawMetricPrefix string `json:"raw_metric_prefix"`
+	// HostnameLabel is the label name whose value becomes the hostname.
+	HostnameLabel string `json:"hostname_label"`
+	// HostnameFormat is a template with <HOSTNAME> placeholder.
+	HostnameFormat string `json:"hostname_format"`
+	// StaticTags are appended to every sample's tags.
+	StaticTags []string `json:"static_tags"`
+	// ShareLabels configures label propagation from source metrics.
+	ShareLabels map[string]ShareLabelConfig `json:"share_labels"`
+}
+
+// ProcessResult wraps processed metric families returned by ProcessMetrics.
+type ProcessResult struct {
+	Families []ProcessedMetricFamily `json:"families"`
+}
+
+// ShareLabelConfig configures how labels from one metric are shared to others.
+type ShareLabelConfig struct {
+	// Match is the set of label names used for matching (join keys).
+	Match []string `json:"match"`
+	// Labels is the set of label names to propagate. Empty means all.
+	Labels []string `json:"labels"`
+	// Values restricts to samples whose value is in this set.
+	Values []float64 `json:"values"`
+}
+
+// ProcessedSample is a single metric sample with pre-built tags.
+type ProcessedSample struct {
+	SampleName string            `json:"sample_name"`
+	Value      float64           `json:"value"`
+	Tags       []string          `json:"tags"`
+	Hostname   string            `json:"hostname,omitempty"`
+	Labels     map[string]string `json:"labels,omitempty"`
+}
+
+// ProcessedMetricFamily is a metric family with processed samples.
+type ProcessedMetricFamily struct {
+	Name    string            `json:"name"`
+	Type    string            `json:"type"`
+	Samples []ProcessedSample `json:"samples"`
+}
+
+// compiledConfig holds pre-compiled versions of the processing config.
+type compiledConfig struct {
+	excludeLabels          map[string]struct{}
+	includeLabels          map[string]struct{}
+	renameLabels           map[string]string
+	excludeMetrics         map[string]struct{}
+	excludeMetricsPattern  *regexp.Regexp
+	excludeMetricsByLabels map[string]*regexp.Regexp // nil regexp means "any value"
+	rawMetricPrefix        string
+	hostnameLabel          string
+	hostnameFormat         string
+	staticTags             []string
+	shareLabels            map[string]compiledShareLabel
+}
+
+type compiledShareLabel struct {
+	match     map[string]struct{}
+	labels    map[string]struct{}
+	values    map[float64]struct{}
+	allLabels bool
+	anyValue  bool
+}
+
+func compileConfig(cfg *ProcessConfig) (*compiledConfig, error) {
+	cc := &compiledConfig{
+		excludeLabels:   make(map[string]struct{}, len(cfg.ExcludeLabels)),
+		includeLabels:   make(map[string]struct{}, len(cfg.IncludeLabels)),
+		renameLabels:    cfg.RenameLabels,
+		excludeMetrics:  make(map[string]struct{}, len(cfg.ExcludeMetrics)),
+		rawMetricPrefix: cfg.RawMetricPrefix,
+		hostnameLabel:   cfg.HostnameLabel,
+		hostnameFormat:  cfg.HostnameFormat,
+		staticTags:      cfg.StaticTags,
+		shareLabels:     make(map[string]compiledShareLabel, len(cfg.ShareLabels)),
+	}
+	if cc.renameLabels == nil {
+		cc.renameLabels = map[string]string{}
+	}
+	if cc.staticTags == nil {
+		cc.staticTags = []string{}
+	}
+
+	for _, l := range cfg.ExcludeLabels {
+		cc.excludeLabels[l] = struct{}{}
+	}
+	for _, l := range cfg.IncludeLabels {
+		cc.includeLabels[l] = struct{}{}
+	}
+	for _, m := range cfg.ExcludeMetrics {
+		cc.excludeMetrics[m] = struct{}{}
+	}
+
+	if len(cfg.ExcludeMetricsPatterns) > 0 {
+		combined := strings.Join(cfg.ExcludeMetricsPatterns, "|")
+		p, err := regexp.Compile(combined)
+		if err != nil {
+			return nil, fmt.Errorf("invalid exclude_metrics_patterns: %w", err)
+		}
+		cc.excludeMetricsPattern = p
+	}
+
+	cc.excludeMetricsByLabels = make(map[string]*regexp.Regexp, len(cfg.ExcludeMetricsByLabels))
+	for label, patterns := range cfg.ExcludeMetricsByLabels {
+		if len(patterns) == 0 {
+			// empty list means "any value"
+			cc.excludeMetricsByLabels[label] = nil
+			continue
+		}
+		combined := strings.Join(patterns, "|")
+		p, err := regexp.Compile(combined)
+		if err != nil {
+			return nil, fmt.Errorf("invalid exclude_metrics_by_labels pattern for %q: %w", label, err)
+		}
+		cc.excludeMetricsByLabels[label] = p
+	}
+
+	for name, slCfg := range cfg.ShareLabels {
+		csl := compiledShareLabel{
+			match:     make(map[string]struct{}, len(slCfg.Match)),
+			labels:    make(map[string]struct{}, len(slCfg.Labels)),
+			values:    make(map[float64]struct{}, len(slCfg.Values)),
+			allLabels: len(slCfg.Labels) == 0,
+			anyValue:  len(slCfg.Values) == 0,
+		}
+		for _, m := range slCfg.Match {
+			csl.match[m] = struct{}{}
+		}
+		for _, l := range slCfg.Labels {
+			csl.labels[l] = struct{}{}
+		}
+		for _, v := range slCfg.Values {
+			csl.values[v] = struct{}{}
+		}
+		cc.shareLabels[name] = csl
+	}
+
+	return cc, nil
+}
+
+// ProcessMetrics parses prometheus-formatted metrics and applies label/tag processing.
+//
+// When ShareLabels is configured, all source-metric labels are collected from the whole payload
+// first (batch mode), then applied to every family regardless of order.
+func ProcessMetrics(data []byte, contentType string, cfg *ProcessConfig) (ProcessResult, error) {
+	families, err := ParseMetricsWithFilter(data, nil, contentType)
+	if err != nil {
+		return ProcessResult{}, err
+	}
+
+	cc, err := compileConfig(cfg)
+	if err != nil {
+		return ProcessResult{}, err
+	}
+
+	// Strip _total suffix from counter family names to match Python prometheus_client behavior.
+	for i := range families {
+		if families[i].Type == "COUNTER" {
+			families[i].Name = trimCounterSuffix(families[i].Name)
+		}
+	}
+
+	// Strip raw_metric_prefix.
+	for i := range families {
+		if cc.rawMetricPrefix != "" && strings.HasPrefix(families[i].Name, cc.rawMetricPrefix) {
+			families[i].Name = families[i].Name[len(cc.rawMetricPrefix):]
+		}
+	}
+
+	result := make([]ProcessedMetricFamily, 0, len(families))
+
+	// Collect shared labels from the whole payload first (batch mode).
+	sharedState := collectSharedLabels(families, cc)
+
+	for _, fam := range families {
+		if _, excluded := cc.excludeMetrics[fam.Name]; excluded {
+			continue
+		}
+		if cc.excludeMetricsPattern != nil && cc.excludeMetricsPattern.MatchString(fam.Name) {
+			continue
+		}
+		processed := processFamily(fam, cc, sharedState)
+		if len(processed.Samples) > 0 {
+			result = append(result, processed)
+		}
+	}
+	return ProcessResult{Families: result}, nil
+}
+
+// ProcessMetricsToJSON processes metrics and returns a JSON-encoded ProcessResult.
+func ProcessMetricsToJSON(data []byte, contentType string, configJSON string) (string, error) {
+	var cfg ProcessConfig
+	if err := json.Unmarshal([]byte(configJSON), &cfg); err != nil {
+		return "", fmt.Errorf("invalid config: %w", err)
+	}
+
+	processResult, err := ProcessMetrics(data, contentType, &cfg)
+	if err != nil {
+		return "", err
+	}
+
+	out, err := json.Marshal(processResult)
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
+}
+
+// sharedLabelState holds collected labels from share_labels source metrics.
+type sharedLabelState struct {
+	// unconditional labels applied to all samples (no match key required)
+	unconditional map[string]string
+	// conditional labels: each entry is (matchSet, sharedLabels)
+	conditional []conditionalLabels
+}
+
+type conditionalLabels struct {
+	matchSet     map[labelPair]struct{}
+	sharedLabels map[string]string
+}
+
+type labelPair struct {
+	name  string
+	value string
+}
+
+func collectSharedLabels(families []MetricFamily, cc *compiledConfig) *sharedLabelState {
+	state := &sharedLabelState{
+		unconditional: make(map[string]string),
+	}
+
+	if len(cc.shareLabels) == 0 {
+		return state
+	}
+
+	for _, fam := range families {
+		slCfg, ok := cc.shareLabels[fam.Name]
+		if !ok {
+			continue
+		}
+
+		for _, sample := range fam.Samples {
+			// Check value restriction
+			if !slCfg.anyValue {
+				if _, allowed := slCfg.values[sample.Value]; !allowed {
+					continue
+				}
+			}
+
+			if len(slCfg.match) > 0 {
+				// Conditional: collect match keys and shared labels
+				matchSet := make(map[labelPair]struct{})
+				shared := make(map[string]string)
+
+				for labelName, labelValue := range sample.Metric {
+					if _, isMatch := slCfg.match[labelName]; isMatch {
+						matchSet[labelPair{labelName, labelValue}] = struct{}{}
+					}
+					if slCfg.allLabels || isInSet(labelName, slCfg.labels) {
+						shared[labelName] = labelValue
+					}
+				}
+				state.conditional = append(state.conditional, conditionalLabels{
+					matchSet:     matchSet,
+					sharedLabels: shared,
+				})
+			} else {
+				// Unconditional: apply to all samples
+				for labelName, labelValue := range sample.Metric {
+					if slCfg.allLabels || isInSet(labelName, slCfg.labels) {
+						state.unconditional[labelName] = labelValue
+					}
+				}
+			}
+		}
+	}
+
+	return state
+}
+
+func isInSet(key string, set map[string]struct{}) bool {
+	_, ok := set[key]
+	return ok
+}
+
+func processFamily(fam MetricFamily, cc *compiledConfig, shared *sharedLabelState) ProcessedMetricFamily {
+	result := ProcessedMetricFamily{
+		Name:    fam.Name,
+		Type:    fam.Type,
+		Samples: make([]ProcessedSample, 0, len(fam.Samples)),
+	}
+
+	famType := strings.ToLower(fam.Type)
+
+	for _, sample := range fam.Samples {
+		// Skip NaN/Inf
+		if math.IsNaN(sample.Value) || math.IsInf(sample.Value, 0) {
+			continue
+		}
+
+		// Build effective labels: start with sample labels, apply shared
+		labels := make(map[string]string, len(sample.Metric))
+		for k, v := range sample.Metric {
+			labels[k] = v
+		}
+
+		// Apply shared labels
+		applySharedLabels(labels, shared)
+
+		// Normalize histogram/summary labels
+		normalizeLabels(labels, famType)
+
+		// Check exclude_metrics_by_labels
+		if shouldExcludeByLabels(labels, cc) {
+			continue
+		}
+
+		// Build tags and extract hostname
+		tags, hostname := buildTags(labels, cc)
+
+		// Determine sample name from the __name__ label or metric name
+		sampleName := labels["__name__"]
+		if sampleName == "" {
+			sampleName = fam.Name
+		}
+
+		result.Samples = append(result.Samples, ProcessedSample{
+			SampleName: sampleName,
+			Value:      sample.Value,
+			Tags:       tags,
+			Hostname:   hostname,
+			Labels:     labels,
+		})
+	}
+
+	return result
+}
+
+func applySharedLabels(labels map[string]string, shared *sharedLabelState) {
+	// Apply unconditional labels first
+	for k, v := range shared.unconditional {
+		if _, exists := labels[k]; !exists {
+			labels[k] = v
+		}
+	}
+
+	// Apply conditional labels
+	for _, cond := range shared.conditional {
+		if matchesLabelSet(labels, cond.matchSet) {
+			for k, v := range cond.sharedLabels {
+				if _, exists := labels[k]; !exists {
+					labels[k] = v
+				}
+			}
+		}
+	}
+}
+
+func matchesLabelSet(labels map[string]string, matchSet map[labelPair]struct{}) bool {
+	for pair := range matchSet {
+		if labels[pair.name] != pair.value {
+			return false
+		}
+	}
+	return true
+}
+
+func normalizeLabels(labels map[string]string, metricType string) {
+	switch metricType {
+	case "histogram":
+		// Rename le → upper_bound with canonical numeric value
+		if le, ok := labels["le"]; ok {
+			delete(labels, "le")
+			labels["upper_bound"] = canonicalizeNumericLabel(le)
+		}
+	case "summary":
+		// Canonicalize quantile value
+		if q, ok := labels["quantile"]; ok {
+			labels["quantile"] = canonicalizeNumericLabel(q)
+		}
+	}
+}
+
+// canonicalizeNumericLabel converts a numeric label to its canonical string form.
+// This matches the Python canonicalize_numeric_label: float(label) or 0.
+func canonicalizeNumericLabel(s string) string {
+	f, err := strconv.ParseFloat(s, 64)
+	if err != nil {
+		return s
+	}
+	// Prevent -0.0
+	if f == 0 {
+		f = 0
+	}
+	return strconv.FormatFloat(f, 'f', -1, 64)
+}
+
+func shouldExcludeByLabels(labels map[string]string, cc *compiledConfig) bool {
+	for labelName, pattern := range cc.excludeMetricsByLabels {
+		labelValue, exists := labels[labelName]
+		if !exists {
+			continue
+		}
+		if pattern == nil {
+			// nil means "any value matches"
+			return true
+		}
+		if pattern.MatchString(labelValue) {
+			return true
+		}
+	}
+	return false
+}
+
+func buildTags(labels map[string]string, cc *compiledConfig) ([]string, string) {
+	hasIncludeFilter := len(cc.includeLabels) > 0
+	hostname := ""
+
+	// Pre-allocate tags: labels + static tags
+	tags := make([]string, 0, len(labels)+len(cc.staticTags))
+
+	for labelName, labelValue := range labels {
+		// Skip __name__ — it's metadata, not a tag
+		if labelName == "__name__" {
+			continue
+		}
+
+		// Check exclude
+		if _, excluded := cc.excludeLabels[labelName]; excluded {
+			continue
+		}
+
+		// Check include filter
+		if hasIncludeFilter {
+			if _, included := cc.includeLabels[labelName]; !included {
+				continue
+			}
+		}
+
+		// Apply rename
+		tagName := labelName
+		if renamed, ok := cc.renameLabels[labelName]; ok {
+			tagName = renamed
+		}
+
+		tags = append(tags, tagName+":"+labelValue)
+	}
+
+	// Append static tags
+	tags = append(tags, cc.staticTags...)
+
+	// Extract hostname
+	if cc.hostnameLabel != "" {
+		if h, ok := labels[cc.hostnameLabel]; ok {
+			hostname = h
+			if cc.hostnameFormat != "" {
+				hostname = strings.Replace(cc.hostnameFormat, "<HOSTNAME>", hostname, 1)
+			}
+		}
+	}
+
+	return tags, hostname
+}
diff --git a/pkg/util/prometheus/process_test.go b/pkg/util/prometheus/process_test.go
new file mode 100644
index 000000000000..8211a73855e0
--- /dev/null
+++ b/pkg/util/prometheus/process_test.go
@@ -0,0 +1,493 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+package prometheus
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// processMetrics is a test helper that calls ProcessMetrics and fails immediately on error.
+func processMetrics(t testing.TB, data []byte, contentType string, cfg *ProcessConfig) []ProcessedMetricFamily {
+	t.Helper()
+	pr, err := ProcessMetrics(data, contentType, cfg)
+	require.NoError(t, err)
+	return pr.Families
+}
+
+func TestProcessMetrics_BasicGauge(t *testing.T) {
+	data := []byte(`# TYPE temperature gauge
+temperature{location="us-east",host="web01"} 72.5
+temperature{location="us-west",host="web02"} 68.3`)
+
+	cfg := &ProcessConfig{
+		StaticTags: []string{"env:prod"},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "temperature", result[0].Name)
+	assert.Equal(t, "GAUGE", result[0].Type)
+	require.Len(t, result[0].Samples, 2)
+
+	// Check that tags are built and static tags appended
+	sample := result[0].Samples[0]
+	assert.Equal(t, 72.5, sample.Value)
+	assert.Contains(t, sample.Tags, "env:prod")
+	assert.Contains(t, sample.Tags, "location:us-east")
+	assert.Contains(t, sample.Tags, "host:web01")
+}
+
+func TestProcessMetrics_ExcludeLabels(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{host="web01",pod="abc",container_id="xyz"} 0.5`)
+
+	cfg := &ProcessConfig{
+		ExcludeLabels: []string{"container_id"},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	require.Len(t, result[0].Samples, 1)
+
+	tags := result[0].Samples[0].Tags
+	assert.Contains(t, tags, "host:web01")
+	assert.Contains(t, tags, "pod:abc")
+	for _, tag := range tags {
+		assert.NotContains(t, tag, "container_id")
+	}
+}
+
+func TestProcessMetrics_IncludeLabels(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{host="web01",pod="abc",container_id="xyz"} 0.5`)
+
+	cfg := &ProcessConfig{
+		IncludeLabels: []string{"host"},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result[0].Samples, 1)
+
+	tags := result[0].Samples[0].Tags
+	assert.Contains(t, tags, "host:web01")
+	assert.Len(t, tags, 1) // only "host" included, no static tags
+}
+
+func TestProcessMetrics_RenameLabels(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{old_name="value1"} 1.0`)
+
+	cfg := &ProcessConfig{
+		RenameLabels: map[string]string{"old_name": "new_name"},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+
+	tags := result[0].Samples[0].Tags
+	assert.Contains(t, tags, "new_name:value1")
+	for _, tag := range tags {
+		assert.NotContains(t, tag, "old_name:")
+	}
+}
+
+func TestProcessMetrics_ExcludeMetrics(t *testing.T) {
+	data := []byte(`# TYPE wanted gauge
+wanted{a="1"} 1.0
+# TYPE unwanted gauge
+unwanted{a="2"} 2.0
+# TYPE also_unwanted gauge
+also_unwanted{a="3"} 3.0`)
+
+	cfg := &ProcessConfig{
+		ExcludeMetrics:         []string{"unwanted"},
+		ExcludeMetricsPatterns: []string{"also_.*"},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "wanted", result[0].Name)
+}
+
+func TestProcessMetrics_ExcludeMetricsByLabels(t *testing.T) {
+	data := []byte(`# TYPE http_requests gauge
+http_requests{status="200"} 100
+http_requests{status="500"} 5
+http_requests{status="503"} 2`)
+
+	cfg := &ProcessConfig{
+		ExcludeMetricsByLabels: map[string][]string{
+			"status": {"5.."},
+		},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	require.Len(t, result[0].Samples, 1)
+	assert.Equal(t, 100.0, result[0].Samples[0].Value)
+}
+
+func TestProcessMetrics_ExcludeMetricsByLabelsAnyValue(t *testing.T) {
+	data := []byte(`# TYPE http_requests gauge
+http_requests{status="200"} 100
+http_requests{debug="true"} 5`)
+
+	cfg := &ProcessConfig{
+		ExcludeMetricsByLabels: map[string][]string{
+			"debug": {},
+		},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	require.Len(t, result[0].Samples, 1)
+	assert.Equal(t, 100.0, result[0].Samples[0].Value)
+}
+
+func TestProcessMetrics_RawMetricPrefix(t *testing.T) {
+	data := []byte(`# TYPE myapp_requests gauge
+myapp_requests{a="1"} 10`)
+
+	cfg := &ProcessConfig{
+		RawMetricPrefix: "myapp_",
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "requests", result[0].Name)
+}
+
+func TestProcessMetrics_HostnameLabel(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{node="web01.example.com",region="us-east"} 0.5`)
+
+	cfg := &ProcessConfig{
+		HostnameLabel: "node",
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	assert.Equal(t, "web01.example.com", result[0].Samples[0].Hostname)
+}
+
+func TestProcessMetrics_HostnameFormat(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{node="web01"} 0.5`)
+
+	cfg := &ProcessConfig{
+		HostnameLabel:  "node",
+		HostnameFormat: "<HOSTNAME>.example.com",
+	}
+
+	result := processMetrics(t, data, "", cfg)
+	assert.Equal(t, "web01.example.com", result[0].Samples[0].Hostname)
+}
+
+func TestProcessMetrics_HistogramNormalization(t *testing.T) {
+	data := []byte(`# TYPE http_duration histogram
+http_duration_bucket{le="0.1"} 10
+http_duration_bucket{le="0.5"} 50
+http_duration_bucket{le="+Inf"} 100
+http_duration_sum 35.2
+http_duration_count 100`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "HISTOGRAM", result[0].Type)
+
+	// le should be renamed to upper_bound
+	for _, s := range result[0].Samples {
+		for _, tag := range s.Tags {
+			assert.NotContains(t, tag, "le:")
+		}
+		if s.SampleName == "http_duration_bucket" {
+			hasUpperBound := false
+			for _, tag := range s.Tags {
+				if tag == "upper_bound:0.1" || tag == "upper_bound:0.5" || tag == "upper_bound:+Inf" {
+					hasUpperBound = true
+				}
+			}
+			assert.True(t, hasUpperBound, "bucket sample should have upper_bound tag, got: %v", s.Tags)
+		}
+	}
+}
+
+func TestProcessMetrics_SummaryNormalization(t *testing.T) {
+	data := []byte(`# TYPE rpc_duration summary
+rpc_duration{quantile="0.50"} 0.5
+rpc_duration{quantile="0.90"} 0.9
+rpc_duration{quantile="0.990"} 0.99
+rpc_duration_sum 100
+rpc_duration_count 200`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+
+	// quantile values should be canonicalized
+	for _, s := range result[0].Samples {
+		if s.SampleName == "rpc_duration" {
+			for _, tag := range s.Tags {
+				// "0.50" → "0.5", "0.990" → "0.99"
+				assert.NotEqual(t, "quantile:0.50", tag)
+				assert.NotEqual(t, "quantile:0.990", tag)
+			}
+		}
+	}
+}
+
+func TestProcessMetrics_ShareLabelsUnconditional(t *testing.T) {
+	data := []byte(`# TYPE kube_pod_info gauge
+kube_pod_info{pod="pod1",namespace="ns1",node="node1",host_ip="10.0.0.1"} 1
+# TYPE kube_pod_status gauge
+kube_pod_status{pod="pod1",namespace="ns1"} 1`)
+
+	cfg := &ProcessConfig{
+		ShareLabels: map[string]ShareLabelConfig{
+			"kube_pod_info": {
+				Labels: []string{"node", "host_ip"},
+				Values: []float64{1},
+			},
+		},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+
+	// Find kube_pod_status
+	var statusFamily *ProcessedMetricFamily
+	for i := range result {
+		if result[i].Name == "kube_pod_status" {
+			statusFamily = &result[i]
+			break
+		}
+	}
+	require.NotNil(t, statusFamily)
+	require.Len(t, statusFamily.Samples, 1)
+
+	tags := statusFamily.Samples[0].Tags
+	assert.Contains(t, tags, "node:node1")
+	assert.Contains(t, tags, "host_ip:10.0.0.1")
+}
+
+func TestProcessMetrics_ShareLabelsConditional(t *testing.T) {
+	data := []byte(`# TYPE kube_pod_info gauge
+kube_pod_info{pod="pod1",namespace="ns1",node="node1"} 1
+kube_pod_info{pod="pod2",namespace="ns2",node="node2"} 1
+# TYPE kube_pod_status gauge
+kube_pod_status{pod="pod1",namespace="ns1"} 1
+kube_pod_status{pod="pod2",namespace="ns2"} 2`)
+
+	cfg := &ProcessConfig{
+		ShareLabels: map[string]ShareLabelConfig{
+			"kube_pod_info": {
+				Match:  []string{"pod", "namespace"},
+				Labels: []string{"node"},
+				Values: []float64{1},
+			},
+		},
+	}
+
+	result := processMetrics(t, data, "", cfg)
+
+	var statusFamily *ProcessedMetricFamily
+	for i := range result {
+		if result[i].Name == "kube_pod_status" {
+			statusFamily = &result[i]
+			break
+		}
+	}
+	require.NotNil(t, statusFamily)
+	require.Len(t, statusFamily.Samples, 2)
+
+	// pod1 should get node1, pod2 should get node2
+	for _, s := range statusFamily.Samples {
+		hasPod1 := false
+		hasPod2 := false
+		for _, tag := range s.Tags {
+			if tag == "pod:pod1" {
+				hasPod1 = true
+			}
+			if tag == "pod:pod2" {
+				hasPod2 = true
+			}
+		}
+		if hasPod1 {
+			assert.Contains(t, s.Tags, "node:node1")
+		}
+		if hasPod2 {
+			assert.Contains(t, s.Tags, "node:node2")
+		}
+	}
+}
+
+func TestProcessMetrics_NaNInfSkipped(t *testing.T) {
+	data := []byte(`# TYPE test gauge
+test{a="1"} 1.0
+test{a="2"} NaN
+test{a="3"} +Inf`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	require.Len(t, result[0].Samples, 1)
+	assert.Equal(t, 1.0, result[0].Samples[0].Value)
+}
+
+func TestProcessMetrics_OpenMetricsContentType(t *testing.T) {
+	data := []byte(`# TYPE http_requests counter
+http_requests_total{method="GET"} 100
+# EOF
+`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "application/openmetrics-text; version=1.0.0", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "http_requests", result[0].Name)
+	assert.Equal(t, "COUNTER", result[0].Type)
+}
+
+func TestProcessMetrics_PrometheusCounterTotalStripped(t *testing.T) {
+	data := []byte(`# TYPE http_requests_total counter
+http_requests_total{method="GET"} 100`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "", cfg)
+	require.Len(t, result, 1)
+	assert.Equal(t, "http_requests", result[0].Name) // _total stripped for Prometheus counters too
+	assert.Equal(t, "COUNTER", result[0].Type)
+}
+
+func TestProcessMetrics_NameLabelExcluded(t *testing.T) {
+	data := []byte(`# TYPE test gauge
+test{a="1"} 1.0`)
+
+	cfg := &ProcessConfig{}
+
+	result := processMetrics(t, data, "", cfg)
+
+	// __name__ should not appear in tags
+	for _, tag := range result[0].Samples[0].Tags {
+		assert.NotContains(t, tag, "__name__")
+	}
+}
+
+func TestProcessMetricsToJSON(t *testing.T) {
+	data := []byte(`# TYPE cpu gauge
+cpu{host="web01"} 0.5`)
+
+	configJSON := `{"static_tags":["env:test"],"exclude_labels":["__name__"]}`
+
+	jsonResult, err := ProcessMetricsToJSON(data, "", configJSON)
+	require.NoError(t, err)
+
+	var processResult ProcessResult
+	err = json.Unmarshal([]byte(jsonResult), &processResult)
+	require.NoError(t, err)
+	require.Len(t, processResult.Families, 1)
+	assert.Equal(t, "cpu", processResult.Families[0].Name)
+	assert.Contains(t, processResult.Families[0].Samples[0].Tags, "env:test")
+	assert.Contains(t, processResult.Families[0].Samples[0].Tags, "host:web01")
+}
+
+func TestProcessMetricsToJSON_InvalidConfig(t *testing.T) {
+	_, err := ProcessMetricsToJSON([]byte(""), "", `{invalid`)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid config")
+}
+
+func TestCanonicalizeNumericLabel(t *testing.T) {
+	tests := []struct {
+		input    string
+		expected string
+	}{
+		{"0.5", "0.5"},
+		{"0.50", "0.5"},
+		{"0.990", "0.99"},
+		{"1", "1"},
+		{"0.0", "0"},
+		{"-0.0", "0"},
+		{"+Inf", "+Inf"},
+		{"-Inf", "-Inf"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.input, func(t *testing.T) {
+			assert.Equal(t, tt.expected, canonicalizeNumericLabel(tt.input))
+		})
+	}
+}
+
+// Benchmarks
+
+func BenchmarkProcessMetrics(b *testing.B) {
+	data := generateLargeMetricsData()
+	cfg := &ProcessConfig{
+		ExcludeLabels: []string{"container_id"},
+		RenameLabels:  map[string]string{"pod_name": "pod"},
+		StaticTags:    []string{"env:prod", "endpoint:http://localhost:9090/metrics"},
+	}
+
+	var pr ProcessResult
+	var err error
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for b.Loop() {
+		pr, err = ProcessMetrics(data, "", cfg)
+	}
+	b.StopTimer()
+
+	require.NoError(b, err)
+	require.NotEmpty(b, pr.Families)
+}
+
+func BenchmarkProcessMetricsWithShareLabels(b *testing.B) {
+	// Build data with a source metric and many target metrics
+	var lines []byte
+	lines = append(lines, []byte("# TYPE kube_pod_info gauge\n")...)
+	for i := 0; i < 100; i++ {
+		line := []byte(fmt.Sprintf("kube_pod_info{pod=\"pod-%d\",namespace=\"ns1\",node=\"node-%d\"} 1\n", i, i%10))
+		lines = append(lines, line...)
+	}
+	lines = append(lines, []byte("# TYPE kube_pod_status gauge\n")...)
+	for i := 0; i < 1000; i++ {
+		line := []byte(fmt.Sprintf("kube_pod_status{pod=\"pod-%d\",namespace=\"ns1\"} %d\n", i%100, i))
+		lines = append(lines, line...)
+	}
+
+	cfg := &ProcessConfig{
+		ShareLabels: map[string]ShareLabelConfig{
+			"kube_pod_info": {
+				Match:  []string{"pod", "namespace"},
+				Labels: []string{"node"},
+				Values: []float64{1},
+			},
+		},
+		StaticTags: []string{"env:prod"},
+	}
+
+	var pr ProcessResult
+	var err error
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for b.Loop() {
+		pr, err = ProcessMetrics(lines, "", cfg)
+	}
+	b.StopTimer()
+
+	require.NoError(b, err)
+	require.NotEmpty(b, pr.Families)
+}
diff --git a/release.json b/release.json
index 4ca074d92fd2..eebd7df26cb0 100644
--- a/release.json
+++ b/release.json
@@ -11,7 +11,7 @@
         "AGENT_DATA_PLANE_HASH_LINUX_ARM64": "95c3ce251e686f371d516930d96d5b9357463faaec15dc361240403767e63450",
         "AGENT_DATA_PLANE_HASH_WINDOWS_AMD64": "8c0370e37d1b544647dae9411955b1a6324ff23d485bf8be57ef4cf97cfe2b30",
         "AGENT_DATA_PLANE_VERSION": "1.3.1",
-        "INTEGRATIONS_CORE_VERSION": "7.82.1",
+        "INTEGRATIONS_CORE_VERSION": "fc18c967d5e0e26452e417f3c83792d854e63e6b",
         "INTEGRATIONS_WHEELS_STORAGE": "stable",
         "JMXFETCH_HASH": "3eb735171a3e41518c4101a6aee06fee8b93092d015769c2e470eecadb4bd11b",
         "JMXFETCH_VERSION": "0.52.0",
diff --git a/rtloader/common/builtins/datadog_agent.c b/rtloader/common/builtins/datadog_agent.c
index 2ad8668ebfad..2bef50e9c5f1 100644
--- a/rtloader/common/builtins/datadog_agent.c
+++ b/rtloader/common/builtins/datadog_agent.c
@@ -29,6 +29,8 @@ static cb_obfuscate_mongodb_string_t cb_obfuscate_mongodb_string = NULL;
 static cb_emit_agent_telemetry_t cb_emit_agent_telemetry = NULL;
 static cb_report_issue_t cb_report_issue = NULL;
 static cb_resolve_issue_t cb_resolve_issue = NULL;
+static cb_parse_prometheus_metrics_t cb_parse_prometheus_metrics = NULL;
+static cb_process_prometheus_metrics_t cb_process_prometheus_metrics = NULL;
 
 // forward declarations
 static PyObject *get_clustername(PyObject *self, PyObject *args);
@@ -51,6 +53,8 @@ static PyObject *obfuscate_mongodb_string(PyObject *self, PyObject *args, PyObje
 static PyObject *emit_agent_telemetry(PyObject *self, PyObject *args, PyObject *kwargs);
 static PyObject *report_issue(PyObject *self, PyObject *args, PyObject *kwargs);
 static PyObject *resolve_issue(PyObject *self, PyObject *args, PyObject *kwargs);
+static PyObject *parse_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs);
+static PyObject *process_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs);
 
 static PyMethodDef methods[] = {
     { "get_clustername", get_clustername, METH_NOARGS, "Get the cluster name." },
@@ -73,6 +77,8 @@ static PyMethodDef methods[] = {
     { "emit_agent_telemetry", (PyCFunction)emit_agent_telemetry, METH_VARARGS|METH_KEYWORDS, "Emit agent telemetry." },
     { "report_issue", (PyCFunction)report_issue, METH_VARARGS|METH_KEYWORDS, "Report a health platform issue." },
     { "resolve_issue", (PyCFunction)resolve_issue, METH_VARARGS|METH_KEYWORDS, "Resolve a health platform issue by issue id." },
+    { "parse_prometheus_metrics", (PyCFunction)parse_prometheus_metrics, METH_VARARGS|METH_KEYWORDS, "Parse Prometheus/OpenMetrics text using the Go parser." },
+    { "process_prometheus_metrics", (PyCFunction)process_prometheus_metrics, METH_VARARGS|METH_KEYWORDS, "Parse and process Prometheus/OpenMetrics text with label/tag processing using the Go parser." },
     { NULL, NULL } // guards
 };
 
@@ -176,6 +182,14 @@ void _set_resolve_issue_cb(cb_resolve_issue_t cb)
     cb_resolve_issue = cb;
 }
 
+void _set_parse_prometheus_metrics_cb(cb_parse_prometheus_metrics_t cb) {
+    cb_parse_prometheus_metrics = cb;
+}
+
+void _set_process_prometheus_metrics_cb(cb_process_prometheus_metrics_t cb) {
+    cb_process_prometheus_metrics = cb;
+}
+
 
 /*! \fn PyObject *get_version(PyObject *self, PyObject *args)
     \brief This function implements the `datadog-agent.get_version` method, collecting
@@ -1007,8 +1021,8 @@ static PyObject *report_issue(PyObject *self, PyObject *args, PyObject *kwargs)
 
     if (err != NULL) {
         PyErr_SetString(PyExc_RuntimeError, err);
-    }   
-    
+    }
+
     cgo_free(err);
     PyGILState_Release(gstate);
     // we need to return NULL to raise the exception set by PyErr_SetString
@@ -1042,7 +1056,7 @@ static PyObject *resolve_issue(PyObject *self, PyObject *args, PyObject *kwargs)
     if (err != NULL) {
         PyErr_SetString(PyExc_RuntimeError, err);
     }
-    
+
     cgo_free(err);
     PyGILState_Release(gstate);
     // we need to return NULL to raise the exception set by PyErr_SetString
@@ -1051,3 +1065,100 @@ static PyObject *resolve_issue(PyObject *self, PyObject *args, PyObject *kwargs)
     }
     Py_RETURN_NONE;
 }
+
+/*! \fn PyObject *parse_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs)
+    \brief This function implements the `datadog_agent.parse_prometheus_metrics` method, parsing
+    Prometheus/OpenMetrics text format metrics using the Go parser and returning the result as
+    a JSON string.
+    \param self A PyObject* pointer to the `datadog_agent` module.
+    \param args A PyObject* pointer to a tuple containing the raw metrics text.
+    \param kwargs A PyObject* pointer to a map of key value pairs (content_type).
+    \return A PyObject* pointer to a string containing the parsed metrics as JSON.
+
+    This function is callable as the `datadog_agent.parse_prometheus_metrics` Python method and
+    uses the `cb_parse_prometheus_metrics()` callback to parse the metrics using the Go parser
+    with CGO. If the callback has not been set `None` will be returned.
+*/
+static PyObject *parse_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+    // callback must be set
+    if (cb_parse_prometheus_metrics == NULL) {
+        Py_RETURN_NONE;
+    }
+
+    PyGILState_STATE gstate = PyGILState_Ensure();
+
+    char *raw_text = NULL;
+    char *content_type = NULL;
+    static char *kwlist[] = {"raw_text", "content_type", NULL};
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|s", kwlist, &raw_text, &content_type)) {
+        PyGILState_Release(gstate);
+        return NULL;
+    }
+
+    char *error_message = NULL;
+    char *json_result = cb_parse_prometheus_metrics(raw_text, content_type, &error_message);
+
+    PyObject *retval = NULL;
+    if (error_message != NULL) {
+        PyErr_SetString(PyExc_RuntimeError, error_message);
+    } else if (json_result == NULL) {
+        PyErr_SetString(PyExc_RuntimeError, "internal error: empty cb_parse_prometheus_metrics response");
+    } else {
+        retval = PyUnicode_FromString(json_result);
+    }
+
+    cgo_free(error_message);
+    cgo_free(json_result);
+    PyGILState_Release(gstate);
+    return retval;
+}
+
+/*! \fn PyObject *process_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs)
+    \brief This function implements the `datadog_agent.process_prometheus_metrics` method, parsing
+    and processing Prometheus/OpenMetrics text format metrics using the Go parser with label/tag
+    processing applied.
+    \param self A PyObject* pointer to the `datadog_agent` module.
+    \param args A PyObject* pointer to a tuple containing the raw metrics text.
+    \param kwargs A PyObject* pointer to a map of key value pairs (content_type, config).
+    \return A PyObject* pointer to a string containing the processed metrics as JSON.
+
+    This function is callable as the `datadog_agent.process_prometheus_metrics` Python method and
+    uses the `cb_process_prometheus_metrics()` callback to parse and process the metrics using the
+    Go parser with CGO. If the callback has not been set `None` will be returned.
+*/
+static PyObject *process_prometheus_metrics(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+    // callback must be set
+    if (cb_process_prometheus_metrics == NULL) {
+        Py_RETURN_NONE;
+    }
+
+    PyGILState_STATE gstate = PyGILState_Ensure();
+
+    char *raw_text = NULL;
+    char *content_type = NULL;
+    char *config = NULL;
+    static char *kwlist[] = {"raw_text", "config", "content_type", NULL};
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "ss|s", kwlist, &raw_text, &config, &content_type)) {
+        PyGILState_Release(gstate);
+        return NULL;
+    }
+
+    char *error_message = NULL;
+    char *json_result = cb_process_prometheus_metrics(raw_text, content_type, config, &error_message);
+
+    PyObject *retval = NULL;
+    if (error_message != NULL) {
+        PyErr_SetString(PyExc_RuntimeError, error_message);
+    } else if (json_result == NULL) {
+        PyErr_SetString(PyExc_RuntimeError, "internal error: empty cb_process_prometheus_metrics response");
+    } else {
+        retval = PyUnicode_FromString(json_result);
+    }
+
+    cgo_free(error_message);
+    cgo_free(json_result);
+    PyGILState_Release(gstate);
+    return retval;
+}
diff --git a/rtloader/common/builtins/datadog_agent.h b/rtloader/common/builtins/datadog_agent.h
index ecf9f64f2cbd..e2e7da08bc68 100644
--- a/rtloader/common/builtins/datadog_agent.h
+++ b/rtloader/common/builtins/datadog_agent.h
@@ -159,6 +159,8 @@ void _set_obfuscate_mongodb_string_cb(cb_obfuscate_mongodb_string_t);
 void _set_emit_agent_telemetry_cb(cb_emit_agent_telemetry_t);
 void _set_report_issue_cb(cb_report_issue_t);
 void _set_resolve_issue_cb(cb_resolve_issue_t);
+void _set_parse_prometheus_metrics_cb(cb_parse_prometheus_metrics_t);
+void _set_process_prometheus_metrics_cb(cb_process_prometheus_metrics_t);
 
 PyObject *_public_headers(PyObject *self, PyObject *args, PyObject *kwargs);
 
diff --git a/rtloader/include/datadog_agent_rtloader.h b/rtloader/include/datadog_agent_rtloader.h
index 2f0a898388b3..e0eece81f8b2 100644
--- a/rtloader/include/datadog_agent_rtloader.h
+++ b/rtloader/include/datadog_agent_rtloader.h
@@ -707,6 +707,27 @@ DATADOG_AGENT_RTLOADER_API void set_report_issue_cb(rtloader_t *, cb_report_issu
     The callback is expected to be provided by the rtloader caller - in go-context: CGO.
 */
 DATADOG_AGENT_RTLOADER_API void set_resolve_issue_cb(rtloader_t *, cb_resolve_issue_t);
+/*! \fn void set_parse_prometheus_metrics_cb(rtloader_t *, cb_parse_prometheus_metrics_t)
+    \brief Sets a callback to be used by rtloader to parse Prometheus/OpenMetrics text
+    format metrics using the Go parser.
+    \param rtloader_t A rtloader_t * pointer to the RtLoader instance.
+    \param object A function pointer with cb_parse_prometheus_metrics_t prototype to the callback
+    function.
+
+    The callback is expected to be provided by the rtloader caller - in go-context: CGO.
+*/
+DATADOG_AGENT_RTLOADER_API void set_parse_prometheus_metrics_cb(rtloader_t *, cb_parse_prometheus_metrics_t);
+
+/*! \fn void set_process_prometheus_metrics_cb(rtloader_t *, cb_process_prometheus_metrics_t)
+    \brief Sets a callback to be used by rtloader to parse and process Prometheus/OpenMetrics text
+    format metrics with label/tag processing using the Go parser.
+    \param rtloader_t A rtloader_t * pointer to the RtLoader instance.
+    \param object A function pointer with cb_process_prometheus_metrics_t prototype to the callback
+    function.
+
+    The callback is expected to be provided by the rtloader caller - in go-context: CGO.
+*/
+DATADOG_AGENT_RTLOADER_API void set_process_prometheus_metrics_cb(rtloader_t *, cb_process_prometheus_metrics_t);
 
 #ifdef __cplusplus
 }
diff --git a/rtloader/include/rtloader.h b/rtloader/include/rtloader.h
index 8e5c847a48ec..ae724c243c77 100644
--- a/rtloader/include/rtloader.h
+++ b/rtloader/include/rtloader.h
@@ -537,6 +537,23 @@ class RtLoader
       Marks a health platform issue resolved by IssueId from Python.
     */
     virtual void setResolveIssueCb(cb_resolve_issue_t) = 0;
+    //! setParsePrometheusMetricsCb member.
+    /*!
+      \param A cb_parse_prometheus_metrics_t function pointer to the CGO callback.
+
+      This allows us to set the relevant CGO callback that will allow parsing
+      Prometheus/OpenMetrics text format using the Go parser.
+    */
+    virtual void setParsePrometheusMetricsCb(cb_parse_prometheus_metrics_t) = 0;
+
+    //! setProcessPrometheusMetricsCb member.
+    /*!
+      \param A cb_process_prometheus_metrics_t function pointer to the CGO callback.
+
+      This allows us to set the relevant CGO callback that will allow parsing and
+      processing Prometheus/OpenMetrics text format with label/tag processing using the Go parser.
+    */
+    virtual void setProcessPrometheusMetricsCb(cb_process_prometheus_metrics_t) = 0;
 
 protected:
     //! _allocateInternalErrorDiagnoses member.
diff --git a/rtloader/include/rtloader_types.h b/rtloader/include/rtloader_types.h
index e8799c6f8a84..8634bd83d6bf 100644
--- a/rtloader/include/rtloader_types.h
+++ b/rtloader/include/rtloader_types.h
@@ -144,6 +144,10 @@ typedef void (*cb_emit_agent_telemetry_t)(char *, char *, double, char *);
 typedef void (*cb_report_issue_t)(char *, char *, char **);
 // (issue_id, error_message_out)
 typedef void (*cb_resolve_issue_t)(char *, char **);
+// (raw_text, content_type, error_message) -> json_result
+typedef char *(*cb_parse_prometheus_metrics_t)(char *, char *, char **);
+// (raw_text, content_type, config_json, error_message) -> json_result
+typedef char *(*cb_process_prometheus_metrics_t)(char *, char *, char *, char **);
 
 // _util
 // (argv, env, stdout, stderr, ret_code, exception)
diff --git a/rtloader/rtloader/api.cpp b/rtloader/rtloader/api.cpp
index 65407da5988d..21873c892206 100644
--- a/rtloader/rtloader/api.cpp
+++ b/rtloader/rtloader/api.cpp
@@ -607,6 +607,16 @@ void set_resolve_issue_cb(rtloader_t *rtloader, cb_resolve_issue_t cb)
     AS_TYPE(RtLoader, rtloader)->setResolveIssueCb(cb);
 }
 
+void set_parse_prometheus_metrics_cb(rtloader_t *rtloader, cb_parse_prometheus_metrics_t cb)
+{
+    AS_TYPE(RtLoader, rtloader)->setParsePrometheusMetricsCb(cb);
+}
+
+void set_process_prometheus_metrics_cb(rtloader_t *rtloader, cb_process_prometheus_metrics_t cb)
+{
+    AS_TYPE(RtLoader, rtloader)->setProcessPrometheusMetricsCb(cb);
+}
+
 /*
  * _util API
  */
diff --git a/rtloader/three/three.cpp b/rtloader/three/three.cpp
index f8d6efe789a5..c56d6a0e5ec0 100644
--- a/rtloader/three/three.cpp
+++ b/rtloader/three/three.cpp
@@ -1125,6 +1125,16 @@ void Three::setResolveIssueCb(cb_resolve_issue_t cb)
     _set_resolve_issue_cb(cb);
 }
 
+void Three::setParsePrometheusMetricsCb(cb_parse_prometheus_metrics_t cb)
+{
+    _set_parse_prometheus_metrics_cb(cb);
+}
+
+void Three::setProcessPrometheusMetricsCb(cb_process_prometheus_metrics_t cb)
+{
+    _set_process_prometheus_metrics_cb(cb);
+}
+
 // Python Helpers
 
 // get_integration_list return a list of every datadog's wheels installed.
diff --git a/rtloader/three/three.h b/rtloader/three/three.h
index 7da5bfbd16b9..1d8f17d6f2b7 100644
--- a/rtloader/three/three.h
+++ b/rtloader/three/three.h
@@ -113,6 +113,8 @@ class Three : public RtLoader
     void setEmitAgentTelemetryCb(cb_emit_agent_telemetry_t);
     void setReportIssueCb(cb_report_issue_t);
     void setResolveIssueCb(cb_resolve_issue_t);
+    void setParsePrometheusMetricsCb(cb_parse_prometheus_metrics_t);
+    void setProcessPrometheusMetricsCb(cb_process_prometheus_metrics_t);
 
     void initPymemStats();
     void getPymemStats(pymem_stats_t &);