Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_fix_responses_reasoning_items

2026-09-09 22:31:41 +00:00 · 2026-07-15 19:14:31 +00:00 · 2026-07-15 19:14:31 +00:00 · 9bf8504055
commit 9bf8504055
parent e8c014362c a643dd0820
29 changed files with 2703 additions and 18 deletions
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@ -1,2 +1,3 @@
 /ui/ @yuneng-jiang @ryan-crabbe-berri
 /litellm/proxy/_experimental/out/ @yuneng-jiang @ryan-crabbe-berri
+/ui/litellm-dashboard/src/lib/http/schema.d.ts
--- a/helm/litellm-helm/README.md
+++ b/helm/litellm-helm/README.md
@ -54,6 +54,12 @@ If `db.useStackgresOperator` is used (not yet implemented):
 | `pdb.annotations`           | Extra metadata annotations to add to the PDB                                                                                                                                                                                                  | `{}`                      |
 | `pdb.labels`                | Extra metadata labels to add to the PDB                                                                                                                                                                                                       | `{}`                      |

+| `billingMetrics.enabled`    | Enable enterprise billable-request metering. Requires an enterprise license.                                                                                                                                                                  | `false`                   |
+| `billingMetrics.endpoint`   | Collector that the billable-request counter is pushed to.                                                                                                                                                                                    | `https://telemetry.litellm.ai` |
+| `billingMetrics.secretName` | Name of an existing Secret holding the mTLS client certificate, under the keys `tls.crt` and `tls.key`.                                                                                                                                       | `litellm-billing-metrics-mtls` |
+| `billingMetrics.caSecretName` | Name of an existing Secret holding a CA bundle under the key `ca.crt`. Only needed for a private or test collector whose server certificate is not on the public web PKI.                                                                    | `""`                      |
+| `billingMetrics.exportIntervalMs` | How often the counter is pushed, in milliseconds. The proxy defaults to `60000` when unset.                                                                                                                                             | `""`                      |
+
 #### Example `proxy_config` ConfigMap from values (default):

 ```
@ -94,6 +100,21 @@ data:
 type: Opaque
 ```

+#### Enterprise billable-request metering
+
+Enterprise licenses meter billable requests by pushing a counter to LiteLLM's collector over mutual TLS. The chart does not create the client certificate; it mounts one you already hold, read-only, so the private key is never exposed through the environment. Create the Secret under the name the chart expects, then turn the block on:
+
+```
+kubectl create secret tls litellm-billing-metrics-mtls --cert=client.crt --key=client.key
+```
+
+```
+billingMetrics:
+  enabled: true
+```
+
+Set `billingMetrics.caSecretName` only when the collector is a private or test one whose server certificate is not on the public web PKI; the production collector needs no CA override. The chart fails the render rather than deploying a proxy that silently never exports, so a missing `secretName` or an emptied `endpoint` surfaces at `helm install` time.
+
 ### Database Settings

 | Name                      | Description                                                                                                                                                                                                                                                                               | Value                                                                                      |
--- a/helm/litellm-helm/templates/_helpers.tpl
+++ b/helm/litellm-helm/templates/_helpers.tpl
@ -50,6 +50,53 @@ app.kubernetes.io/name: {{ include "litellm.name" . }}
 app.kubernetes.io/instance: {{ .Release.Name }}
 {{- end }}

+{{/*
+Enterprise billable-request metering. The client certificate identifies the
+deployment to LiteLLM's collector, so it is mounted read-only from an existing
+Secret rather than passed through the environment.
+*/}}
+{{- define "litellm.billingMetrics.certDir" -}}/etc/litellm/billing-mtls{{- end -}}
+{{- define "litellm.billingMetrics.caDir" -}}/etc/litellm/billing-mtls-ca{{- end -}}
+
+{{- define "litellm.billingMetricsEnv" -}}
+- name: LITELLM_BILLING_METRICS_ENDPOINT
+  value: {{ required "billingMetrics.endpoint is required when billingMetrics.enabled is true" .Values.billingMetrics.endpoint | quote }}
+- name: LITELLM_BILLING_METRICS_CLIENT_CERT
+  value: {{ printf "%s/tls.crt" (include "litellm.billingMetrics.certDir" .) | quote }}
+- name: LITELLM_BILLING_METRICS_CLIENT_KEY
+  value: {{ printf "%s/tls.key" (include "litellm.billingMetrics.certDir" .) | quote }}
+{{- if .Values.billingMetrics.caSecretName }}
+- name: LITELLM_BILLING_METRICS_CA_CERT
+  value: {{ printf "%s/ca.crt" (include "litellm.billingMetrics.caDir" .) | quote }}
+{{- end }}
+{{- with .Values.billingMetrics.exportIntervalMs }}
+- name: LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS
+  value: {{ . | quote }}
+{{- end }}
+{{- end -}}
+
+{{- define "litellm.billingMetricsVolumes" -}}
+- name: billing-metrics-mtls
+  secret:
+    secretName: {{ required "billingMetrics.secretName is required when billingMetrics.enabled is true (an existing Secret with tls.crt and tls.key)" .Values.billingMetrics.secretName }}
+{{- if .Values.billingMetrics.caSecretName }}
+- name: billing-metrics-mtls-ca
+  secret:
+    secretName: {{ .Values.billingMetrics.caSecretName }}
+{{- end }}
+{{- end -}}
+
+{{- define "litellm.billingMetricsVolumeMounts" -}}
+- name: billing-metrics-mtls
+  mountPath: {{ include "litellm.billingMetrics.certDir" . }}
+  readOnly: true
+{{- if .Values.billingMetrics.caSecretName }}
+- name: billing-metrics-mtls-ca
+  mountPath: {{ include "litellm.billingMetrics.caDir" . }}
+  readOnly: true
+{{- end }}
+{{- end -}}
+
 {{/*
 Create the name of the service account to use
 */}}
--- a/helm/litellm-helm/templates/deployment.yaml
+++ b/helm/litellm-helm/templates/deployment.yaml
@ -142,6 +142,9 @@ spec:
            {{- with .Values.extraEnvVars }}
              {{- toYaml . | nindent 12 }}
            {{- end }}
+            {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsEnv" . | nindent 12 }}
+            {{- end }}
            {{- if .Values.migrationJob.enabled }}
            # Schema updates are owned by the dedicated migrations Job; skip
            # the proxy's startup `prisma db push` so N replicas don't race
@ -220,6 +223,9 @@ spec:
            - name: npm
              mountPath: /.npm
          {{- end }}
+          {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsVolumeMounts" . | nindent 12 }}
+          {{- end }}
          {{- with .Values.volumeMounts }}
            {{- toYaml . | nindent 12 }}
          {{- end }}
@ -252,6 +258,9 @@ spec:
            items:
              - key: {{ .Values.proxyConfigMap.key | default "config.yaml" }}
                path: "config.yaml"
+        {{- if .Values.billingMetrics.enabled }}
+        {{- include "litellm.billingMetricsVolumes" . | nindent 8 }}
+        {{- end }}
      {{- with .Values.volumes }}
        {{- toYaml . | nindent 8 }}
      {{- end }}
--- a/helm/litellm-helm/tests/billing_metrics_tests.yaml
+++ b/helm/litellm-helm/tests/billing_metrics_tests.yaml
@ -0,0 +1,297 @@
+suite: test billingMetrics wiring on the proxy deployment
+templates:
+  - deployment.yaml
+  - configmap-litellm.yaml
+  - migrations-job.yaml
+tests:
+  - it: is off by default, adding no env, volume, or mount
+    template: deployment.yaml
+    asserts:
+      - notContains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: litellm-billing-metrics-mtls
+      - notContains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls
+            mountPath: /etc/litellm/billing-mtls
+            readOnly: true
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+
+  - it: renders the endpoint and the mounted cert paths when enabled
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CLIENT_CERT
+            value: /etc/litellm/billing-mtls/tls.crt
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CLIENT_KEY
+            value: /etc/litellm/billing-mtls/tls.key
+
+  # The conventional Secret name is the default, so enabling the block is enough.
+  - it: mounts the default cert secret read-only alongside the config volume
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: litellm-billing-metrics-mtls
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls
+            mountPath: /etc/litellm/billing-mtls
+            readOnly: true
+
+  - it: honours a secretName override
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: my-billing-mtls
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: my-billing-mtls
+      - notContains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: litellm-billing-metrics-mtls
+
+  - it: honours an endpoint override
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        endpoint: https://collector.internal:4318
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://collector.internal:4318
+
+  # The production collector presents a public web-PKI certificate, so the CA
+  # override must stay absent unless a private collector is configured.
+  - it: omits the CA env, volume, and mount when no caSecretName is set
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - notContains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls-ca
+            secret:
+              secretName: billing-ca
+      - notContains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls-ca
+            mountPath: /etc/litellm/billing-mtls-ca
+            readOnly: true
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CA_CERT
+            value: /etc/litellm/billing-mtls-ca/ca.crt
+
+  - it: mounts the CA secret when caSecretName is set
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        caSecretName: billing-ca
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CA_CERT
+            value: /etc/litellm/billing-mtls-ca/ca.crt
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls-ca
+            secret:
+              secretName: billing-ca
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls-ca
+            mountPath: /etc/litellm/billing-mtls-ca
+            readOnly: true
+
+  - it: passes the export interval through only when set
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        exportIntervalMs: 5000
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS
+            value: "5000"
+
+  - it: omits the export interval when unset
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS
+            value: "60000"
+
+  # Kubernetes resolves duplicate env names last-wins, so the chart-owned billing
+  # entries must render after .Values.envVars or a user could silently redirect
+  # the metering export. The three billing entries are the last ones emitted here
+  # (migrationJob, which appends DISABLE_SCHEMA_UPDATE, is off for this case).
+  - it: renders the billing endpoint after envVars so it cannot be shadowed
+    template: deployment.yaml
+    set:
+      migrationJob:
+        enabled: false
+      billingMetrics:
+        enabled: true
+      envVars:
+        LITELLM_BILLING_METRICS_ENDPOINT: https://shadowed.example
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://shadowed.example
+      - equal:
+          path: spec.template.spec.containers[0].env[-3]
+          value:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - equal:
+          path: spec.template.spec.containers[0].env[-2].name
+          value: LITELLM_BILLING_METRICS_CLIENT_CERT
+      - equal:
+          path: spec.template.spec.containers[0].env[-1].name
+          value: LITELLM_BILLING_METRICS_CLIENT_KEY
+
+  - it: keeps user-supplied volumes and mounts alongside the billing secret
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+      volumes:
+        - name: custom-callbacks
+          configMap:
+            name: my-callbacks
+      volumeMounts:
+        - name: custom-callbacks
+          mountPath: /app/callbacks
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: custom-callbacks
+            configMap:
+              name: my-callbacks
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: litellm-billing-metrics-mtls
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: custom-callbacks
+            mountPath: /app/callbacks
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls
+            mountPath: /etc/litellm/billing-mtls
+            readOnly: true
+
+  - it: still mounts the proxy config when enabled
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: litellm-config
+            mountPath: /etc/litellm/config.yaml
+            subPath: config.yaml
+
+  # Only the proxy serves billable traffic. The migrations Job must never mount
+  # the client certificate, and it renders its own env and volumes, so nothing
+  # stops a future edit from wiring the billing include into it by mistake.
+  - it: does not touch the migrations job when enabled
+    template: migrations-job.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - notExists:
+          path: spec.template.spec.containers[0].volumeMounts
+      - notExists:
+          path: spec.template.spec.volumes
+
+  - it: fails loudly when enabled with an emptied secretName
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: ""
+    asserts:
+      - failedTemplate:
+          errorMessage: billingMetrics.secretName is required when billingMetrics.enabled is true (an existing Secret with tls.crt and tls.key)
+
+  - it: fails loudly when enabled without an endpoint
+    template: deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        endpoint: ""
+    asserts:
+      - failedTemplate:
+          errorMessage: billingMetrics.endpoint is required when billingMetrics.enabled is true
--- a/helm/litellm-helm/values.yaml
+++ b/helm/litellm-helm/values.yaml
@ -139,6 +139,20 @@ masterkeySecretName: ""
 # if set, use this secret key for the master key; otherwise, use the default key
 masterkeySecretKey: ""

+# Optional: enterprise billable-request metering. When enabled, the proxy counts
+# successful requests to inference, MCP, and A2A endpoints and pushes them to
+# LiteLLM's collector over mutual TLS. Requires an enterprise license.
+# The client certificate identifies the deployment, so it is mounted read-only
+# from an existing Secret and never passed through the environment.
+billingMetrics:
+  enabled: false
+  endpoint: https://telemetry.litellm.ai   # collector to push the counter to
+  secretName: litellm-billing-metrics-mtls # existing Secret holding tls.crt and tls.key
+  # Only for private or test collectors whose server certificate is not on the
+  # public web PKI. The production collector needs no CA override.
+  caSecretName: ""                         # existing Secret holding ca.crt
+  exportIntervalMs: ""                     # push cadence; the proxy defaults to 60000
+
 proxyConfigMap:
  # when true, creates a new configmap
  create: true
--- a/helm/litellm/templates/_helpers.tpl
+++ b/helm/litellm/templates/_helpers.tpl
@ -34,6 +34,57 @@ app.kubernetes.io/managed-by: {{ .Release.Service }}
 helm.sh/chart: {{ printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" }}
 {{- end -}}

+{{/*
+Enterprise billable-request metering. Wired into gateway and backend, not the
+migrations job. The gateway serves nearly all billable traffic, but the backend
+keeps the named-server MCP transport (/{mcp_server_name}/mcp), which writes a
+SpendLogs row, so metering only the gateway would silently drop that traffic.
+The client certificate identifies the deployment to LiteLLM's collector, so it is
+mounted read-only from an existing Secret rather than passed through the
+environment.
+*/}}
+{{- define "litellm.billingMetrics.certDir" -}}/etc/litellm/billing-mtls{{- end -}}
+{{- define "litellm.billingMetrics.caDir" -}}/etc/litellm/billing-mtls-ca{{- end -}}
+
+{{- define "litellm.billingMetricsEnv" -}}
+- name: LITELLM_BILLING_METRICS_ENDPOINT
+  value: {{ required "billingMetrics.endpoint is required when billingMetrics.enabled is true" .Values.billingMetrics.endpoint | quote }}
+- name: LITELLM_BILLING_METRICS_CLIENT_CERT
+  value: {{ printf "%s/tls.crt" (include "litellm.billingMetrics.certDir" .) | quote }}
+- name: LITELLM_BILLING_METRICS_CLIENT_KEY
+  value: {{ printf "%s/tls.key" (include "litellm.billingMetrics.certDir" .) | quote }}
+{{- if .Values.billingMetrics.caSecretName }}
+- name: LITELLM_BILLING_METRICS_CA_CERT
+  value: {{ printf "%s/ca.crt" (include "litellm.billingMetrics.caDir" .) | quote }}
+{{- end }}
+{{- with .Values.billingMetrics.exportIntervalMs }}
+- name: LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS
+  value: {{ . | quote }}
+{{- end }}
+{{- end -}}
+
+{{- define "litellm.billingMetricsVolumes" -}}
+- name: billing-metrics-mtls
+  secret:
+    secretName: {{ required "billingMetrics.secretName is required when billingMetrics.enabled is true (an existing Secret with tls.crt and tls.key)" .Values.billingMetrics.secretName }}
+{{- if .Values.billingMetrics.caSecretName }}
+- name: billing-metrics-mtls-ca
+  secret:
+    secretName: {{ .Values.billingMetrics.caSecretName }}
+{{- end }}
+{{- end -}}
+
+{{- define "litellm.billingMetricsVolumeMounts" -}}
+- name: billing-metrics-mtls
+  mountPath: {{ include "litellm.billingMetrics.certDir" . }}
+  readOnly: true
+{{- if .Values.billingMetrics.caSecretName }}
+- name: billing-metrics-mtls-ca
+  mountPath: {{ include "litellm.billingMetrics.caDir" . }}
+  readOnly: true
+{{- end }}
+{{- end -}}
+
 {{/*
 Per-component selector labels — used in both Service selectors and Deployment matchLabels.
 */}}
--- a/helm/litellm/templates/backend/deployment.yaml
+++ b/helm/litellm/templates/backend/deployment.yaml
@ -44,14 +44,20 @@ spec:
            - name: CONFIG_FILE_PATH
              value: /app/config/config.yaml
            {{- end }}
+            {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsEnv" . | nindent 12 }}
+            {{- end }}
          {{- include "litellm.envFrom" .Values.backend | nindent 10 }}
-          {{- if or .Values.gateway.config.create .Values.backend.volumeMounts }}
+          {{- if or .Values.gateway.config.create .Values.backend.volumeMounts .Values.billingMetrics.enabled }}
          volumeMounts:
            {{- if .Values.gateway.config.create }}
            - name: gateway-config
              mountPath: /app/config/config.yaml
              subPath: config.yaml
            {{- end }}
+            {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsVolumeMounts" . | nindent 12 }}
+            {{- end }}
            {{- with .Values.backend.volumeMounts }}
            {{- toYaml . | nindent 12 }}
            {{- end }}
@ -66,13 +72,16 @@ spec:
          {{- end }}
          resources:
            {{- toYaml .Values.backend.resources | nindent 12 }}
-      {{- if or .Values.gateway.config.create .Values.backend.volumes }}
+      {{- if or .Values.gateway.config.create .Values.backend.volumes .Values.billingMetrics.enabled }}
      volumes:
        {{- if .Values.gateway.config.create }}
        - name: gateway-config
          configMap:
            name: {{ include "litellm.gateway.fullname" . }}-config
        {{- end }}
+        {{- if .Values.billingMetrics.enabled }}
+        {{- include "litellm.billingMetricsVolumes" . | nindent 8 }}
+        {{- end }}
        {{- with .Values.backend.volumes }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
--- a/helm/litellm/templates/gateway/deployment.yaml
+++ b/helm/litellm/templates/gateway/deployment.yaml
@ -46,14 +46,20 @@ spec:
            - name: NUM_WORKERS
              value: {{ .Values.gateway.numWorkers | quote }}
            {{- end }}
+            {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsEnv" . | nindent 12 }}
+            {{- end }}
          {{- include "litellm.envFrom" .Values.gateway | nindent 10 }}
-          {{- if or .Values.gateway.config.create .Values.gateway.volumeMounts }}
+          {{- if or .Values.gateway.config.create .Values.gateway.volumeMounts .Values.billingMetrics.enabled }}
          volumeMounts:
            {{- if .Values.gateway.config.create }}
            - name: gateway-config
              mountPath: /app/config/config.yaml
              subPath: config.yaml
            {{- end }}
+            {{- if .Values.billingMetrics.enabled }}
+            {{- include "litellm.billingMetricsVolumeMounts" . | nindent 12 }}
+            {{- end }}
            {{- with .Values.gateway.volumeMounts }}
            {{- toYaml . | nindent 12 }}
            {{- end }}
@ -68,13 +74,16 @@ spec:
          {{- end }}
          resources:
            {{- toYaml .Values.gateway.resources | nindent 12 }}
-      {{- if or .Values.gateway.config.create .Values.gateway.volumes }}
+      {{- if or .Values.gateway.config.create .Values.gateway.volumes .Values.billingMetrics.enabled }}
      volumes:
        {{- if .Values.gateway.config.create }}
        - name: gateway-config
          configMap:
            name: {{ include "litellm.gateway.fullname" . }}-config
        {{- end }}
+        {{- if .Values.billingMetrics.enabled }}
+        {{- include "litellm.billingMetricsVolumes" . | nindent 8 }}
+        {{- end }}
        {{- with .Values.gateway.volumes }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
--- a/helm/litellm/tests/billing_metrics_tests.yaml
+++ b/helm/litellm/tests/billing_metrics_tests.yaml
@ -0,0 +1,249 @@
+suite: test billingMetrics wiring on gateway and backend
+templates:
+  - gateway/deployment.yaml
+  - gateway/configmap.yaml
+  - backend/deployment.yaml
+  - migrations-job.yaml
+values:
+  - ./values/required.yaml
+tests:
+  - it: is off by default, adding no env, volume, or mount
+    template: gateway/deployment.yaml
+    asserts:
+      - notContains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: billing-mtls
+      - equal:
+          path: spec.template.spec.containers[0].volumeMounts
+          value:
+            - name: gateway-config
+              mountPath: /app/config/config.yaml
+              subPath: config.yaml
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+
+  - it: renders the endpoint and the mounted cert paths when enabled
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CLIENT_CERT
+            value: /etc/litellm/billing-mtls/tls.crt
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CLIENT_KEY
+            value: /etc/litellm/billing-mtls/tls.key
+
+  - it: mounts the cert secret read-only alongside the config volume
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: billing-mtls
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls
+            mountPath: /etc/litellm/billing-mtls
+            readOnly: true
+
+  # The production collector presents a public web-PKI certificate, so the CA
+  # override must stay absent unless a private collector is configured.
+  - it: omits the CA env, volume, and mount when no caSecretName is set
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+    asserts:
+      - notContains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls-ca
+            secret:
+              secretName: billing-ca
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CA_CERT
+            value: /etc/litellm/billing-mtls-ca/ca.crt
+
+  - it: mounts the CA secret when caSecretName is set
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+        caSecretName: billing-ca
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_CA_CERT
+            value: /etc/litellm/billing-mtls-ca/ca.crt
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls-ca
+            secret:
+              secretName: billing-ca
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls-ca
+            mountPath: /etc/litellm/billing-mtls-ca
+            readOnly: true
+
+  - it: passes the export interval through only when set
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+        exportIntervalMs: 5000
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS
+            value: "5000"
+
+  - it: keeps user-supplied gateway volumes alongside the billing secret
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+      gateway.volumes:
+        - name: custom-callbacks
+          configMap:
+            name: my-callbacks
+      gateway.volumeMounts:
+        - name: custom-callbacks
+          mountPath: /app/callbacks
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: custom-callbacks
+            configMap:
+              name: my-callbacks
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: billing-mtls
+
+  # The backend keeps the named-server MCP transport (/{mcp_server_name}/mcp),
+  # which writes a SpendLogs row, so it must meter too or that traffic is lost.
+  - it: meters the backend as well, since it serves the MCP transport
+    template: backend/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+    asserts:
+      - contains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - contains:
+          path: spec.template.spec.containers[0].volumeMounts
+          content:
+            name: billing-metrics-mtls
+            mountPath: /etc/litellm/billing-mtls
+            readOnly: true
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: billing-mtls
+
+  - it: leaves the backend alone when metering is off
+    template: backend/deployment.yaml
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+
+  # The migrations job runs prisma and serves no traffic; it must never receive
+  # the client key.
+  - it: never mounts the billing cert on the migrations job
+    template: migrations-job.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: billing-mtls
+    asserts:
+      - notContains:
+          path: spec.template.spec.containers[0].env
+          content:
+            name: LITELLM_BILLING_METRICS_ENDPOINT
+            value: https://telemetry.litellm.ai
+      - isNull:
+          path: spec.template.spec.volumes
+
+  # The conventional Secret name is the default, so enabling metering needs no
+  # secretName at all; the guard below only fires on an explicitly blanked one.
+  - it: uses the conventional secret name by default
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+    asserts:
+      - contains:
+          path: spec.template.spec.volumes
+          content:
+            name: billing-metrics-mtls
+            secret:
+              secretName: litellm-billing-metrics-mtls
+
+  - it: fails loudly when the secretName is explicitly blanked
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        secretName: ""
+    asserts:
+      - failedTemplate:
+          errorMessage: billingMetrics.secretName is required when billingMetrics.enabled is true (an existing Secret with tls.crt and tls.key)
+
+  - it: fails loudly when enabled without an endpoint
+    template: gateway/deployment.yaml
+    set:
+      billingMetrics:
+        enabled: true
+        endpoint: ""
+        secretName: billing-mtls
+    asserts:
+      - failedTemplate:
+          errorMessage: billingMetrics.endpoint is required when billingMetrics.enabled is true
--- a/helm/litellm/values.yaml
+++ b/helm/litellm/values.yaml
@ -73,6 +73,25 @@ masterKey:
  secretName: litellm-master-key-secret              # name of a Secret containing the master key
  secretKey: master-key

+# Optional: enterprise billable-request metering. When enabled, the gateway and
+# backend count successful requests to inference, MCP, and A2A endpoints and push
+# them to LiteLLM's collector over mutual TLS. Both components serve billable
+# routes: the backend keeps the named-server MCP transport. Requires an
+# enterprise license. The client certificate identifies the deployment, so it is
+# mounted read-only from an existing Secret and never passed through the env.
+billingMetrics:
+  enabled: false
+  endpoint: https://telemetry.litellm.ai              # collector to push the counter to
+  # An existing Secret holding the client certificate under tls.crt and its key
+  # under tls.key, usually created from the onboarding artifact. The default is
+  # the conventional name, so the common path is to create that Secret and set
+  # enabled: true. Override only if yours is named differently.
+  secretName: litellm-billing-metrics-mtls
+  # Only for private or test collectors whose server certificate is not on the
+  # public web PKI. The production collector needs no CA override.
+  caSecretName: ""                                    # existing Secret holding ca.crt
+  exportIntervalMs: ""                                # push cadence; the proxy defaults to 60000
+
 # External Postgres connection. 
 database:
  writer:
--- a/litellm/proxy/enterprise_billing/init.py
+++ b/litellm/proxy/enterprise_billing/init.py
--- a/litellm/proxy/enterprise_billing/billing_metrics.py
+++ b/litellm/proxy/enterprise_billing/billing_metrics.py
@ -0,0 +1,323 @@
+"""
+Push-based OTLP metering for enterprise litellm deployments.
+
+Owns a dedicated OpenTelemetry meter provider and an OTLP/HTTP exporter
+authenticated to our global collector with a TLS client certificate. The
+collector front end terminates mutual TLS: the client certificate presented
+here is validated against our CA at the edge, and the verified subject is
+what identifies the deployment. It is intentionally isolated from the global
+meter provider so the customer's own OTEL metrics are untouched and ours
+never leak into their backend.
+
+The deployment's identity rides on the TLS client certificate, not on the
+payload; the secret license key is never sent as an attribute or header.
+"""
+
+import os
+import tempfile
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Optional, Union
+
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
+from opentelemetry.metrics import Counter
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import Resource
+
+from litellm._logging import verbose_proxy_logger
+from litellm.proxy.middleware.billable_request_metrics_middleware import (
+    BillableCategory,
+)
+
+if TYPE_CHECKING:
+    from litellm.proxy._types import EnterpriseLicenseData
+
+ENDPOINT_ENV = "LITELLM_BILLING_METRICS_ENDPOINT"
+CLIENT_CERT_ENV = "LITELLM_BILLING_METRICS_CLIENT_CERT"
+CLIENT_KEY_ENV = "LITELLM_BILLING_METRICS_CLIENT_KEY"
+CA_CERT_ENV = "LITELLM_BILLING_METRICS_CA_CERT"
+EXPORT_INTERVAL_ENV = "LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS"
+DEFAULT_EXPORT_INTERVAL_MS = 60_000
+SHUTDOWN_FLUSH_TIMEOUT_MS = 5_000
+_METRICS_PATH = "/v1/metrics"
+
+# The cert env vars take a path or the PEM itself. Secret stores that inject
+# values as env content cannot mount them as files, so inline PEM is written out.
+_PEM_PREFIX = "-----BEGIN"
+_PEM_DIR_PREFIX = "litellm-billing-mtls-"
+_PEM_FILE_MODE = 0o600
+_CLIENT_CERT_FILENAME = "client.crt"
+_CLIENT_KEY_FILENAME = "client.key"
+_CA_CERT_FILENAME = "ca.crt"
+
+METRIC_NAME = "litellm.enterprise.billable_requests"
+METER_NAME = "litellm.enterprise.billing"
+
+AttributeValue = Union[str, int]
+
+
+@dataclass(frozen=True, slots=True)
+class BillingMetricsConfig:
+    endpoint: str
+    client_cert_path: str
+    client_key_path: str
+    ca_cert_path: Optional[str]
+    export_interval_ms: int
+    litellm_version: str
+    license_id: Optional[str]
+
+
+def _metrics_endpoint(endpoint: str) -> str:
+    """The OTLP/HTTP metric exporter wants the full URL including the signal path."""
+    trimmed = endpoint.rstrip("/")
+    return trimmed if trimmed.endswith(_METRICS_PATH) else f"{trimmed}{_METRICS_PATH}"
+
+
+def _resource_attributes(config: BillingMetricsConfig) -> dict[str, AttributeValue]:
+    base: dict[str, AttributeValue] = {
+        "service.name": "litellm-proxy",
+        "litellm.version": config.litellm_version,
+    }
+    license_attr: dict[str, AttributeValue] = {"litellm.license.id": config.license_id} if config.license_id else {}
+    return {**base, **license_attr}
+
+
+def _billable_attributes(
+    category: BillableCategory, route: str, status_code: int, model_id: Optional[str]
+) -> dict[str, AttributeValue]:
+    base: dict[str, AttributeValue] = {
+        "litellm.endpoint.category": category.value,
+        "http.route": route,
+        "http.response.status_code": status_code,
+    }
+    model_attr: dict[str, AttributeValue] = {"litellm.model_id": model_id} if model_id else {}
+    return {**base, **model_attr}
+
+
+def build_mtls_meter_provider(config: BillingMetricsConfig) -> MeterProvider:
+    """OTLP/HTTP exporter presenting a TLS client certificate.
+
+    The collector's load balancer terminates mutual TLS and validates the client
+    certificate against our CA. Server verification uses the system trust store
+    (the collector presents a public web-PKI certificate); ca_cert_path overrides
+    it only for private/test collectors.
+    """
+    exporter = OTLPMetricExporter(
+        endpoint=_metrics_endpoint(config.endpoint),
+        # None -> exporter falls back to the system trust store.
+        certificate_file=config.ca_cert_path,
+        client_certificate_file=config.client_cert_path,
+        client_key_file=config.client_key_path,
+    )
+    reader = PeriodicExportingMetricReader(exporter, export_interval_millis=config.export_interval_ms)
+    return MeterProvider(metric_readers=[reader], resource=Resource.create(_resource_attributes(config)))
+
+
+class BillingMetricsRecorder:
+    """Increments one OTLP counter per billable request. The meter provider is injected (see the factory)."""
+
+    def __init__(self, provider: MeterProvider) -> None:
+        self._provider = provider
+        self._counter: Counter = provider.get_meter(METER_NAME).create_counter(
+            name=METRIC_NAME,
+            unit="{request}",
+            description="Count of 2xx HTTP requests to billable LLM/MCP/A2A endpoints",
+        )
+
+    def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: Optional[str]) -> None:
+        self._counter.add(1, _billable_attributes(category, route, status_code, model_id))
+
+    def shutdown(self) -> None:
+        """Final flush + exporter-thread stop. Without this, up to one export
+        interval of billable counts is dropped on every proxy restart."""
+        self._provider.shutdown(timeout_millis=SHUTDOWN_FLUSH_TIMEOUT_MS)
+
+
+def _export_interval_ms() -> int:
+    raw = os.getenv(EXPORT_INTERVAL_ENV)
+    if raw is None:
+        return DEFAULT_EXPORT_INTERVAL_MS
+    try:
+        return int(raw)
+    except ValueError:
+        verbose_proxy_logger.warning(
+            "Invalid %s=%r, falling back to %d ms", EXPORT_INTERVAL_ENV, raw, DEFAULT_EXPORT_INTERVAL_MS
+        )
+        return DEFAULT_EXPORT_INTERVAL_MS
+
+
+@dataclass(frozen=True, slots=True)
+class _CredentialPaths:
+    client_cert_path: str
+    client_key_path: str
+    ca_cert_path: Optional[str]
+
+
+def _is_pem_content(value: str) -> bool:
+    return value.lstrip().startswith(_PEM_PREFIX)
+
+
+def _write_pem(directory: str, filename: str, pem: str) -> str:
+    path = os.path.join(directory, filename)
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(pem if pem.endswith("\n") else f"{pem}\n")
+    os.chmod(path, _PEM_FILE_MODE)
+    return path
+
+
+def _resolve_credential_paths(*, client_cert: str, client_key: str, ca_cert: Optional[str]) -> _CredentialPaths:
+    """
+    Accept either a filesystem path or inline PEM content for each credential.
+
+    Secret stores that inject values as environment content rather than mounted
+    files (ECS tasks reading AWS Secrets Manager, Cloud Run reading Secret
+    Manager) can only deliver the certificate as a string. The OTLP exporter
+    takes paths, so inline PEM is written to a private directory once, when the
+    recorder is built. Raises OSError if that write fails; the caller disables
+    metering rather than propagating.
+    """
+    inline = tuple(value for value in (client_cert, client_key, ca_cert) if value and _is_pem_content(value))
+    if not inline:
+        return _CredentialPaths(client_cert, client_key, ca_cert)
+
+    # mkdtemp is 0o700, so the 0o600 key file it holds is unreachable by other users.
+    directory = tempfile.mkdtemp(prefix=_PEM_DIR_PREFIX)
+    return _CredentialPaths(
+        client_cert_path=(
+            _write_pem(directory, _CLIENT_CERT_FILENAME, client_cert) if _is_pem_content(client_cert) else client_cert
+        ),
+        client_key_path=(
+            _write_pem(directory, _CLIENT_KEY_FILENAME, client_key) if _is_pem_content(client_key) else client_key
+        ),
+        ca_cert_path=(
+            _write_pem(directory, _CA_CERT_FILENAME, ca_cert) if ca_cert and _is_pem_content(ca_cert) else ca_cert
+        ),
+    )
+
+
+def load_billing_metrics_config(
+    *, license_data: Optional["EnterpriseLicenseData"], litellm_version: str
+) -> Optional[BillingMetricsConfig]:
+    endpoint = os.getenv(ENDPOINT_ENV)
+    client_cert = os.getenv(CLIENT_CERT_ENV)
+    client_key = os.getenv(CLIENT_KEY_ENV)
+    # Optional: only for private/test collectors whose server cert is not on the
+    # public web PKI. The production collector needs no CA override.
+    ca_cert = os.getenv(CA_CERT_ENV)
+
+    missing = [
+        name
+        for name, value in (
+            (ENDPOINT_ENV, endpoint),
+            (CLIENT_CERT_ENV, client_cert),
+            (CLIENT_KEY_ENV, client_key),
+        )
+        if not value
+    ]
+    if not endpoint or not client_cert or not client_key:
+        verbose_proxy_logger.warning(
+            "Enterprise billing metrics disabled: licensed deployment missing config (%s)",
+            ", ".join(missing),
+        )
+        return None
+
+    try:
+        paths = _resolve_credential_paths(client_cert=client_cert, client_key=client_key, ca_cert=ca_cert)
+    except OSError as exc:
+        verbose_proxy_logger.warning(
+            "Enterprise billing metrics disabled: could not write inline certificate content to disk: %s", exc
+        )
+        return None
+
+    # Report the variable names, never their values. A value that is neither a
+    # readable path nor recognizable PEM is still secret material, and this
+    # warning would otherwise copy a client key straight into the proxy logs.
+    unreadable = [
+        env_name
+        for env_name, path in (
+            (CLIENT_CERT_ENV, paths.client_cert_path),
+            (CLIENT_KEY_ENV, paths.client_key_path),
+            (CA_CERT_ENV, paths.ca_cert_path),
+        )
+        if path and not os.path.isfile(path)
+    ]
+    if unreadable:
+        verbose_proxy_logger.warning(
+            "Enterprise billing metrics disabled: %s did not resolve to a readable certificate file. "
+            "Set each to a file path, or to inline PEM content beginning with '%s'.",
+            ", ".join(unreadable),
+            _PEM_PREFIX,
+        )
+        return None
+
+    return BillingMetricsConfig(
+        endpoint=endpoint,
+        client_cert_path=paths.client_cert_path,
+        client_key_path=paths.client_key_path,
+        ca_cert_path=paths.ca_cert_path,
+        export_interval_ms=_export_interval_ms(),
+        litellm_version=litellm_version,
+        license_id=(license_data or {}).get("user_id"),
+    )
+
+
+class _ActiveRecorderRegistry:
+    """One-slot registry linking the factory-built recorder to the shutdown
+    hook; the middleware instance holding the recorder is not reachable from
+    proxy_shutdown_event."""
+
+    def __init__(self) -> None:
+        self._recorder: Optional[BillingMetricsRecorder] = None
+
+    def set(self, recorder: BillingMetricsRecorder) -> None:
+        self._recorder = recorder
+
+    def pop(self) -> Optional[BillingMetricsRecorder]:
+        recorder = self._recorder
+        self._recorder = None
+        return recorder
+
+
+_ACTIVE_RECORDER = _ActiveRecorderRegistry()
+
+
+def build_billing_metrics_recorder(
+    *, premium: bool, license_data: Optional["EnterpriseLicenseData"], litellm_version: str
+) -> Optional[BillingMetricsRecorder]:
+    """Build the recorder, or None when the deployment is not licensed or metering is unconfigured."""
+    if not premium:
+        # Debug, not warning: unlicensed is the common case and a warning here
+        # would be noise on every OSS proxy. Every other disable path warns.
+        verbose_proxy_logger.debug("Enterprise billing metrics disabled: deployment is not licensed")
+        return None
+
+    config = load_billing_metrics_config(license_data=license_data, litellm_version=litellm_version)
+    if config is None:
+        return None
+
+    try:
+        recorder = BillingMetricsRecorder(build_mtls_meter_provider(config))
+    except Exception as exc:  # noqa: BLE001 -- metering must never break proxy startup
+        verbose_proxy_logger.warning("Enterprise billing metrics disabled: failed to initialize exporter: %s", exc)
+        return None
+    _ACTIVE_RECORDER.set(recorder)
+    # The only positive signal that this component meters. Without it, a silent
+    # return above is indistinguishable from a working exporter in the logs, and
+    # a component that carries the cert but no license would look healthy.
+    verbose_proxy_logger.info(
+        "Enterprise billing metrics enabled: exporting to %s every %d ms",
+        config.endpoint,
+        config.export_interval_ms,
+    )
+    return recorder
+
+
+def shutdown_billing_metrics_recorder() -> None:
+    """Flush and stop the active recorder, if any. Idempotent; never raises."""
+    recorder = _ACTIVE_RECORDER.pop()
+    if recorder is None:
+        return
+    try:
+        recorder.shutdown()
+    except Exception as exc:  # noqa: BLE001 -- shutdown must never block or fail proxy exit
+        verbose_proxy_logger.warning("Enterprise billing metrics: final flush failed: %s", exc)
--- a/litellm/proxy/management_endpoints/team_endpoints.py
+++ b/litellm/proxy/management_endpoints/team_endpoints.py
@ -1621,6 +1621,7 @@ async def update_team(
    - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team.
    - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit per model for this team. Example: {"gpt-4": 100, "gpt-3.5-turbo": 200}
    - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit per model for this team. Example: {"gpt-4": 10000, "gpt-3.5-turbo": 20000}
+    - mcp_rpm_limit: Optional[Dict[str, int]] - Per-MCP-server RPM limit for this team, keyed by MCP server name (alias if set, else the configured name). Example: {"github": 100, "slack": 200}. Applied across all keys for this team.
    Example - update team TPM Limit
    - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint.
    - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview)
--- a/litellm/proxy/middleware/billable_request_metrics_middleware.py
+++ b/litellm/proxy/middleware/billable_request_metrics_middleware.py
@ -0,0 +1,234 @@
+"""
+Counts billable HTTP requests on enterprise deployments.
+
+A billable request is an inbound request to an LLM inference, MCP, or A2A
+endpoint that returns a 2xx status. The actual export happens in an injected
+recorder (see litellm.proxy.enterprise_billing.billing_metrics); when no
+recorder is injected (non-enterprise, or metering misconfigured) this
+middleware is a transparent pass-through.
+"""
+
+import re
+import threading
+from enum import Enum
+from typing import Callable, Optional, Protocol, Sequence, runtime_checkable
+
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from litellm._logging import verbose_proxy_logger
+from litellm.proxy._types import LiteLLMRoutes
+
+
+class BillableCategory(str, Enum):
+    LLM = "llm"
+    MCP = "mcp"
+    A2A = "a2a"
+
+
+@runtime_checkable
+class BillingRecorder(Protocol):
+    def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: Optional[str]) -> None: ...
+
+
+_MODEL_ID_HEADER = b"x-litellm-model-id"
+
+# Ordered: a longer suffix that shares an ending with a shorter one must come
+# first, e.g. "/chat/completions" before "/completions". This is the POST
+# inference surface that writes a SpendLogs row on success, so the exported
+# count lines up with the admin UI usage page for inference traffic. Billing
+# is a deliberate lower bound on SpendLogs rows: management writes that also
+# log (batch/file/fine-tuning creation, interaction cancel) and non-POST calls
+# that log (passthrough reads) never bill, so drift only ever undercounts.
+_LLM_ROUTE_SUFFIXES: tuple[str, ...] = (
+    "/chat/completions",
+    "/completions",
+    "/embeddings",
+    "/responses",
+    "/rerank",
+    "/moderations",
+    "/images/generations",
+    "/images/edits",
+    "/images/variations",
+    "/audio/transcriptions",
+    "/audio/translations",
+    "/audio/speech",
+    "/videos",  # create; GET list is excluded by the POST gate
+    "/remix",  # /v1/videos/{id}/remix
+    "/ocr",
+    "/search",  # /v1/search and /v1/vector_stores/{id}/search
+    "/rag/query",
+    "/rag/ingest",
+    ":generateContent",  # Gemini-native /v1beta/models/{model}:generateContent
+    ":streamGenerateContent",
+)
+
+# Exact paths only: a suffix match would also catch non-inference resources that
+# share the ending, e.g. the OpenAI Assistants route /v1/threads/{id}/messages
+# writes no SpendLogs row and must not bill, unlike Anthropic /v1/messages.
+_LLM_ROUTE_EXACT: tuple[str, ...] = (
+    "/v1/messages",
+    "/interactions",  # Google Interactions create; /{id} reads and /cancel do not match
+    "/v1beta/interactions",
+)
+
+# Provider passthrough prefixes (e.g. /bedrock/..., /vertex-ai/...) carry real
+# inference calls that write SpendLogs rows, so they bill. Anchored to the
+# routes enum so new providers are picked up without touching this module.
+# /langfuse forwards observability traffic, not inference: it writes no
+# SpendLogs row and must not bill.
+_NON_BILLABLE_PASSTHROUGH_PREFIXES = frozenset({"/langfuse"})
+_PASSTHROUGH_PREFIXES: tuple[str, ...] = tuple(
+    prefix
+    for prefix in LiteLLMRoutes.mapped_pass_through_routes.value
+    if prefix not in _NON_BILLABLE_PASSTHROUGH_PREFIXES
+)
+
+
+def _classify_llm_route(path: str) -> Optional[str]:
+    exact_match = next((route for route in _LLM_ROUTE_EXACT if path == route), None)
+    if exact_match is not None:
+        return exact_match
+    suffix_match = next((suffix for suffix in _LLM_ROUTE_SUFFIXES if path == suffix or path.endswith(suffix)), None)
+    if suffix_match is not None:
+        return suffix_match
+    # Deep passthrough paths only: the bare prefix itself is not an inference call.
+    return next((prefix for prefix in _PASSTHROUGH_PREFIXES if path.startswith(f"{prefix}/")), None)
+
+
+_MCP_MANAGEMENT_PREFIX = "/v1/mcp"
+_MCP_DYNAMIC_TRANSPORT = re.compile(r"/(?:toolset/)?[^/]+/mcp")
+# The REST wrapper's tool-call endpoint executes a tool and fires the same MCP
+# spend logging as the /mcp transport; its list/test siblings do not bill.
+_MCP_REST_TOOL_CALL = "/mcp-rest/tools/call"
+
+_A2A_INVOKE_SUFFIX = "/message/send"
+_A2A_TRANSPORT_PREFIXES: tuple[str, ...] = ("/v1/a2a/", "/a2a/")
+# Bare POST /a2a/{agent_id} carries the JSON-RPC method in the body, not the
+# path. Only message/send and message/stream write a SpendLogs row there; the
+# task RPCs (tasks/get, tasks/cancel, tasks/pushNotificationConfig/*, ...) are
+# forwarded upstream and write none. A path-only classifier cannot separate
+# them, so the bare route does not bill: counting a task RPC would overcount,
+# while missing a bare-path message/send only undercounts, and undercounting is
+# the sole direction this metric is allowed to drift. The /mcp transport is
+# method-agnostic by contrast because its list path logs a SpendLogs row too.
+
+
+def _classify_mcp_route(path: str) -> Optional[str]:
+    if path == _MCP_MANAGEMENT_PREFIX or path.startswith(f"{_MCP_MANAGEMENT_PREFIX}/"):
+        return None
+    if path == "/mcp" or path.startswith("/mcp/"):
+        return "/mcp"
+    if path == _MCP_REST_TOOL_CALL:
+        return "/mcp"
+    if _MCP_DYNAMIC_TRANSPORT.fullmatch(path) is not None:
+        return "/mcp"
+    return None
+
+
+def _classify_a2a_route(path: str) -> Optional[str]:
+    if path.endswith(_A2A_INVOKE_SUFFIX) and any(path.startswith(prefix) for prefix in _A2A_TRANSPORT_PREFIXES):
+        return "/a2a"
+    return None
+
+
+def classify_billable_request(path: str, method: str = "POST") -> Optional[tuple[BillableCategory, str]]:
+    """Map a request path to its (category, normalized route), or None if not billable."""
+    normalized = path.rstrip("/") or "/"
+
+    mcp_route = _classify_mcp_route(normalized)
+    if mcp_route is not None:
+        return (BillableCategory.MCP, mcp_route)
+
+    a2a_route = _classify_a2a_route(normalized)
+    if a2a_route is not None:
+        return (BillableCategory.A2A, a2a_route)
+
+    # POST-only is a conservative gate: non-POST calls can still write a
+    # SpendLogs row (passthrough reads, resource GETs) but must not bill, so
+    # any classifier-vs-dashboard mismatch is an undercount, never an overcount.
+    if method.upper() != "POST":
+        return None
+
+    llm_route = _classify_llm_route(normalized)
+    if llm_route is not None:
+        return (BillableCategory.LLM, llm_route)
+    return None
+
+
+def _extract_model_id(headers: Sequence[tuple[bytes, bytes]]) -> Optional[str]:
+    return next(
+        (value.decode("latin-1") for name, value in headers if name.lower() == _MODEL_ID_HEADER and value),
+        None,
+    )
+
+
+class BillableRequestMetricsMiddleware:
+    """
+    Pure ASGI middleware that records one billable request per 2xx response to a
+    billable endpoint. Modeled on InFlightRequestsMiddleware: it wraps `send`,
+    reads the final status and the x-litellm-model-id header off the
+    `http.response.start` message, and never blocks or fails the request path.
+    """
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        recorder: Optional[BillingRecorder] = None,
+        recorder_factory: Optional[Callable[[], Optional[BillingRecorder]]] = None,
+    ) -> None:
+        self.app = app
+        self.recorder = recorder
+        # The factory defers recorder construction to the first request, AFTER the
+        # startup event has loaded the YAML config's environment_variables (license
+        # and cert env vars). Building at import time captured recorder=None for
+        # deployments configured that way. Resolved exactly once; the result
+        # (including None) is cached.
+        self._recorder_factory = recorder_factory
+        self._resolved = recorder_factory is None
+        self._resolve_lock = threading.Lock()
+
+    def _resolve_recorder(self) -> Optional[BillingRecorder]:
+        if self._resolved:
+            return self.recorder
+        # The lock keeps concurrent first requests from each building their own
+        # MeterProvider (and leaking its background exporter thread).
+        with self._resolve_lock:
+            if not self._resolved:
+                factory = self._recorder_factory
+                self.recorder = factory() if factory is not None else self.recorder
+                self._resolved = True
+        return self.recorder
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        recorder = self._resolve_recorder()
+        if recorder is None:
+            await self.app(scope, receive, send)
+            return
+
+        classification = classify_billable_request(scope.get("path", ""), scope.get("method", "POST"))
+        if classification is None:
+            await self.app(scope, receive, send)
+            return
+
+        category, route = classification
+        status_code = 0
+        model_id: Optional[str] = None
+
+        async def send_wrapper(message: Message) -> None:
+            nonlocal status_code, model_id
+            if message["type"] == "http.response.start":
+                status_code = message["status"]
+                model_id = _extract_model_id(message.get("headers", []))
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)
+
+        if 200 <= status_code < 300:
+            try:
+                recorder.record(category=category, route=route, status_code=status_code, model_id=model_id)
+            except Exception:  # noqa: BLE001 -- metering must never fail a request that was already served
+                verbose_proxy_logger.warning("billable request metering failed for %s", route, exc_info=True)
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@ -438,10 +438,30 @@ from litellm.proxy.management_helpers.audit_logs import (
    create_object_audit_log,
 )
 from litellm.proxy.memory.memory_endpoints import router as memory_router
+from litellm.proxy.middleware.billable_request_metrics_middleware import (
+    BillableRequestMetricsMiddleware,
+    BillingRecorder,
+)
 from litellm.proxy.plugin_routes import (
-    router as plugin_router,
    register_plugins_from_config,
 )
+from litellm.proxy.plugin_routes import (
+    router as plugin_router,
+)
+
+try:
+    from litellm.proxy.enterprise_billing.billing_metrics import (
+        build_billing_metrics_recorder as _build_billing_metrics_recorder,
+    )
+    from litellm.proxy.enterprise_billing.billing_metrics import (
+        shutdown_billing_metrics_recorder as _shutdown_billing_metrics_recorder,
+    )
+
+    build_billing_metrics_recorder: Optional[Callable[..., Optional[BillingRecorder]]] = _build_billing_metrics_recorder
+    shutdown_billing_metrics_recorder: Optional[Callable[[], None]] = _shutdown_billing_metrics_recorder
+except ImportError:
+    build_billing_metrics_recorder = None
+    shutdown_billing_metrics_recorder = None
 from litellm.proxy.middleware.in_flight_requests_middleware import (
    InFlightRequestsMiddleware,
 )
@ -461,13 +481,11 @@ from litellm.proxy.openai_files_endpoints.files_endpoints import (
 )
 from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
    passthrough_endpoint_router,
+    vertex_ai_live_websocket_passthrough,
 )
 from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
    router as llm_passthrough_router,
 )
-from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
-    vertex_ai_live_websocket_passthrough,
-)
 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
    initialize_pass_through_endpoints,
 )
@ -552,22 +570,19 @@ from litellm.types.proxy.management_endpoints.ui_sso import (
 from litellm.types.realtime import RealtimeQueryParams
 from litellm.types.router import (
    DeploymentTypedDict,
-)
-from litellm.types.router import ModelInfo as RouterModelInfo
-from litellm.types.router import (
    RouterGeneralSettings,
    RoutingPlugin,
    SearchToolTypedDict,
    updateDeployment,
 )
+from litellm.types.router import ModelInfo as RouterModelInfo
 from litellm.types.scheduler import DefaultPriorities
 from litellm.types.secret_managers.main import (
    KeyManagementSettings,
    KeyManagementSystem,
 )
-from litellm.types.utils import CredentialItem, CustomHuggingfaceTokenizer
+from litellm.types.utils import CredentialItem, CustomHuggingfaceTokenizer, RawRequestTypedDict, StandardLoggingPayload
 from litellm.types.utils import ModelInfo as ModelMapInfo
-from litellm.types.utils import RawRequestTypedDict, StandardLoggingPayload
 from litellm.utils import _add_custom_logger_callback_to_specific_event

 try:
@ -768,6 +783,11 @@ async def proxy_shutdown_event():
    if db_writer_client is not None:
        await db_writer_client.close()  # type: ignore[reportGeneralTypeIssues]

+    # final flush of billable-request counts: without it, up to one export
+    # interval of enterprise billing data is dropped on every restart
+    if shutdown_billing_metrics_recorder is not None:
+        shutdown_billing_metrics_recorder()
+
    # flush remaining langfuse logs
    if "langfuse" in litellm.success_callback:
        try:
@ -973,11 +993,11 @@ async def proxy_startup_event(app: FastAPI):
        if is_otel_v2_enabled():
            from opentelemetry import trace as _otel_trace

-            from litellm.litellm_core_utils.litellm_logging import _in_memory_loggers
            from litellm.integrations.otel.logger import (
                OpenTelemetryV2,
                publish_global_otel_v2_provider,
            )
+            from litellm.litellm_core_utils.litellm_logging import _in_memory_loggers

            registered = open_telemetry_logger if isinstance(open_telemetry_logger, OpenTelemetryV2) else None
            publish_global_otel_v2_provider(
@ -1781,6 +1801,31 @@ app.add_middleware(
 )

 app.add_middleware(PrometheusAuthMiddleware)
+# Added before InFlightRequestsMiddleware so it nests *inside* it: Starlette
+# makes the last-added middleware outermost. The billable count is recorded
+# after the inner app returns, so if this sat outside the in-flight tracker a
+# request could be counted as drained while its record() had not yet run, and
+# proxy_shutdown_event could flush and stop the exporter underneath it.
+app.add_middleware(
+    BillableRequestMetricsMiddleware,
+    # Factory, not an instance: the recorder is resolved on the first request so
+    # it sees premium_user and the billing env vars AFTER proxy_startup_event has
+    # loaded the YAML config's environment_variables. Building it here at import
+    # time would permanently capture recorder=None for YAML-configured
+    # deployments. The lambda reads the module globals at call time.
+    recorder_factory=lambda: (
+        build_billing_metrics_recorder(
+            premium=premium_user,
+            # Read from the license check, not the premium_user_data module
+            # global: that global is bound once at import and goes stale when
+            # the license arrives via the YAML config's environment_variables.
+            license_data=_license_check.airgapped_license_data,
+            litellm_version=version,
+        )
+        if build_billing_metrics_recorder is not None
+        else None
+    ),
+)
 app.add_middleware(InFlightRequestsMiddleware)
 app.add_middleware(SecurityHeadersMiddleware)

--- a/terraform/litellm/aws/README.md
+++ b/terraform/litellm/aws/README.md
@ -158,6 +158,38 @@ AgentOps) live under `proxy_config.litellm_settings.callbacks` and are
 orthogonal to the OTLP variables above; their credentials still go in
 `*_extra_secrets`.

+### Enterprise billing metrics
+
+License-gated request metering is opt-in and gated entirely on
+`billing_metrics_endpoint`. Empty (default) and no billing env is added to
+the container, so existing deployments are unchanged. Set it and both
+gateway and backend export billable-request counts over OTLP/HTTP,
+authenticating to the collector with the mTLS client certificate issued for
+your deployment.
+
+The proxy accepts the certificate, key, and CA bundle as either a file path
+or literal PEM content. This stack takes the PEM, writes each one to its own
+Secrets Manager entry, grants the task-execution role
+`secretsmanager:GetSecretValue` on them, and injects them as
+`LITELLM_BILLING_METRICS_CLIENT_CERT` / `_CLIENT_KEY` (and `_CA_CERT` when
+set), so no volume mount is needed on Fargate.
+
+```hcl
+billing_metrics_endpoint = "https://telemetry.litellm.ai/v1/metrics"
+```
+
+```bash
+export TF_VAR_billing_metrics_client_cert_pem="$(cat client.crt)"
+export TF_VAR_billing_metrics_client_key_pem="$(cat client.key)"
+```
+
+`billing_metrics_ca_cert_pem` is only for private or test collectors whose
+CA is not in the system trust store; leave it empty against
+`telemetry.litellm.ai`. Metering requires an enterprise license, so pair
+this with `litellm_license`. To tune the export cadence, set
+`LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS` through `gateway_extra_env` /
+`backend_extra_env`
+
 ## Tenant deployment

 Every resource the stack creates is named `${tenant}-litellm-${env}` (or
--- a/terraform/litellm/aws/ecs.tf
+++ b/terraform/litellm/aws/ecs.tf
@ -76,6 +76,33 @@ locals {
    { name = "OTEL_HEADERS", valueFrom = var.otel_headers_secret_arn },
  ] : []

+  # Enterprise request metering, gated on billing_metrics_endpoint. The
+  # endpoint rides in as a plain env var; the mTLS material is stored in
+  # Secrets Manager (secrets.tf) and injected as PEM-valued env vars, which
+  # the proxy accepts in place of file paths. Each PEM is wired only when the
+  # operator supplied it, so an empty ca_cert_pem falls back to the system
+  # trust store.
+  billing_metrics_enabled             = var.billing_metrics_endpoint != ""
+  billing_metrics_client_cert_enabled = local.billing_metrics_enabled && var.billing_metrics_client_cert_pem != ""
+  billing_metrics_client_key_enabled  = local.billing_metrics_enabled && var.billing_metrics_client_key_pem != ""
+  billing_metrics_ca_cert_enabled     = local.billing_metrics_enabled && var.billing_metrics_ca_cert_pem != ""
+
+  billing_metrics_env = local.billing_metrics_enabled ? [
+    { name = "LITELLM_BILLING_METRICS_ENDPOINT", value = var.billing_metrics_endpoint },
+  ] : []
+
+  billing_metrics_secrets = concat(
+    local.billing_metrics_client_cert_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CLIENT_CERT", valueFrom = aws_secretsmanager_secret.billing_metrics_client_cert[0].arn },
+    ] : [],
+    local.billing_metrics_client_key_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CLIENT_KEY", valueFrom = aws_secretsmanager_secret.billing_metrics_client_key[0].arn },
+    ] : [],
+    local.billing_metrics_ca_cert_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CA_CERT", valueFrom = aws_secretsmanager_secret.billing_metrics_ca_cert[0].arn },
+    ] : [],
+  )
+
  shared_env = [
    { name = "IAM_TOKEN_DB_AUTH", value = "true" },
    { name = "DATABASE_HOST", value = aws_rds_cluster.this.endpoint },
@ -108,6 +135,7 @@ locals {
      { name = "LITELLM_LICENSE", valueFrom = aws_secretsmanager_secret.license[0].arn },
    ],
    local.otel_secrets,
+    local.billing_metrics_secrets,
  )

  # Backend-only managed secrets. UI_PASSWORD is consumed by the management
@ -179,6 +207,30 @@ locals {

 # ---------- Gateway ----------
 resource "aws_ecs_task_definition" "gateway" {
+  # Metering needs a client certificate AND its key. Each secret is created only
+  # when its own PEM is supplied, so an endpoint set with a missing key would
+  # otherwise apply cleanly and leave the proxy logging "missing config" and
+  # never exporting. ca_cert_pem stays optional: empty means fall back to the
+  # system trust store.
+  #
+  # The guard lives here, on an unconditional resource, rather than on the cert
+  # secret: that secret is count-gated on the cert itself, so it has zero
+  # instances in exactly the case this must catch. Adding count or for_each to
+  # this resource would silently stop the guard from evaluating.
+  #
+  #   endpoint  cert  key  -> result
+  #   ""        any   any  -> metering off, no secrets created
+  #   set       set   set  -> metering on
+  #   set       any-missing -> plan fails here
+  lifecycle {
+    precondition {
+      condition = var.billing_metrics_endpoint == "" || (
+        var.billing_metrics_client_cert_pem != "" && var.billing_metrics_client_key_pem != ""
+      )
+      error_message = "billing_metrics_client_cert_pem and billing_metrics_client_key_pem are both required when billing_metrics_endpoint is set."
+    }
+  }
+
  family                   = "${local.name}-gateway"
  network_mode             = "awsvpc"
  requires_compatibilities = ["FARGATE"]
@ -198,6 +250,7 @@ resource "aws_ecs_task_definition" "gateway" {
        environment = concat(
          local.shared_env,
          local.gateway_otel_env,
+          local.billing_metrics_env,
          local.gateway_extra_env_list,
          local.proxy_config_env,
        )
@ -264,6 +317,18 @@ resource "aws_ecs_service" "gateway" {

 # ---------- Backend ----------
 resource "aws_ecs_task_definition" "backend" {
+  # Same guard as the gateway: the backend meters too (it serves the named-server
+  # MCP transport), and a targeted apply of just this resource must not slip a
+  # billing endpoint through without the credentials to use it.
+  lifecycle {
+    precondition {
+      condition = var.billing_metrics_endpoint == "" || (
+        var.billing_metrics_client_cert_pem != "" && var.billing_metrics_client_key_pem != ""
+      )
+      error_message = "billing_metrics_client_cert_pem and billing_metrics_client_key_pem are both required when billing_metrics_endpoint is set."
+    }
+  }
+
  family                   = "${local.name}-backend"
  network_mode             = "awsvpc"
  requires_compatibilities = ["FARGATE"]
@ -284,6 +349,7 @@ resource "aws_ecs_task_definition" "backend" {
          local.shared_env,
          local.backend_default_env,
          local.backend_otel_env,
+          local.billing_metrics_env,
          local.backend_extra_env_list,
          local.proxy_config_env,
        )
--- a/terraform/litellm/aws/iam.tf
+++ b/terraform/litellm/aws/iam.tf
@ -53,6 +53,9 @@ data "aws_iam_policy_document" "secrets_access" {
      [aws_secretsmanager_secret.master_key.arn],
      aws_secretsmanager_secret.license[*].arn,
      aws_secretsmanager_secret.ui_password[*].arn,
+      aws_secretsmanager_secret.billing_metrics_client_cert[*].arn,
+      aws_secretsmanager_secret.billing_metrics_client_key[*].arn,
+      aws_secretsmanager_secret.billing_metrics_ca_cert[*].arn,
      local.extra_secret_arns,
      var.otel_headers_secret_arn == "" ? [] : [var.otel_headers_secret_arn],
    )
--- a/terraform/litellm/aws/secrets.tf
+++ b/terraform/litellm/aws/secrets.tf
@ -74,6 +74,61 @@ resource "aws_secretsmanager_secret_version" "ui_password" {
  secret_string = var.ui_password
 }

+# Billing-metrics mTLS material — only created when metering is enabled
+# (billing_metrics_endpoint non-empty) and the operator supplied the PEM.
+# The task-execution role gets GetSecretValue via iam.tf, and gateway +
+# backend pick the env vars up through shared_secrets in ecs.tf.
+resource "aws_secretsmanager_secret" "billing_metrics_client_cert" {
+  count = local.billing_metrics_client_cert_enabled ? 1 : 0
+
+  name                    = "${local.name}-billing-metrics-client-cert"
+  description             = "LITELLM_BILLING_METRICS_CLIENT_CERT for gateway + backend."
+  recovery_window_in_days = 0
+
+  tags = local.tags
+}
+
+resource "aws_secretsmanager_secret_version" "billing_metrics_client_cert" {
+  count = local.billing_metrics_client_cert_enabled ? 1 : 0
+
+  secret_id     = aws_secretsmanager_secret.billing_metrics_client_cert[0].id
+  secret_string = var.billing_metrics_client_cert_pem
+}
+
+resource "aws_secretsmanager_secret" "billing_metrics_client_key" {
+  count = local.billing_metrics_client_key_enabled ? 1 : 0
+
+  name                    = "${local.name}-billing-metrics-client-key"
+  description             = "LITELLM_BILLING_METRICS_CLIENT_KEY for gateway + backend."
+  recovery_window_in_days = 0
+
+  tags = local.tags
+}
+
+resource "aws_secretsmanager_secret_version" "billing_metrics_client_key" {
+  count = local.billing_metrics_client_key_enabled ? 1 : 0
+
+  secret_id     = aws_secretsmanager_secret.billing_metrics_client_key[0].id
+  secret_string = var.billing_metrics_client_key_pem
+}
+
+resource "aws_secretsmanager_secret" "billing_metrics_ca_cert" {
+  count = local.billing_metrics_ca_cert_enabled ? 1 : 0
+
+  name                    = "${local.name}-billing-metrics-ca-cert"
+  description             = "LITELLM_BILLING_METRICS_CA_CERT for gateway + backend."
+  recovery_window_in_days = 0
+
+  tags = local.tags
+}
+
+resource "aws_secretsmanager_secret_version" "billing_metrics_ca_cert" {
+  count = local.billing_metrics_ca_cert_enabled ? 1 : 0
+
+  secret_id     = aws_secretsmanager_secret.billing_metrics_ca_cert[0].id
+  secret_string = var.billing_metrics_ca_cert_pem
+}
+
 resource "aws_secretsmanager_secret" "db_master_password" {
  name                    = "${local.name}-db-master-password"
  description             = "Aurora master-user password - bootstrap only. Runtime auth is IAM-token."
--- a/terraform/litellm/aws/variables.tf
+++ b/terraform/litellm/aws/variables.tf
@ -533,3 +533,65 @@ variable "otel_headers_secret_arn" {
  type        = string
  default     = ""
 }
+
+# ---------- Enterprise billing metrics ----------
+#
+# License-gated request metering. Opt-in and gated entirely on
+# billing_metrics_endpoint: leave it empty (the default) and nothing
+# metering-related lands in the container env. Set it and gateway + backend
+# export billable-request counts over OTLP/HTTP, authenticating to the
+# collector with an mTLS client cert. The proxy accepts the cert, key, and CA
+# as either a file path or literal PEM content, so on Fargate they are
+# injected straight from Secrets Manager as env vars and no volume is needed.
+
+variable "billing_metrics_endpoint" {
+  description = <<-EOT
+    OTLP/HTTP endpoint for enterprise billing metrics (sets
+    LITELLM_BILLING_METRICS_ENDPOINT). Non-empty enables request metering;
+    empty (default) disables it and adds no billing env to the container.
+    Requires an enterprise license. Example:
+    "https://telemetry.litellm.ai/v1/metrics"
+  EOT
+  type        = string
+  default     = ""
+}
+
+variable "billing_metrics_client_cert_pem" {
+  description = <<-EOT
+    PEM content of the mTLS client certificate issued for this deployment.
+    When billing_metrics_endpoint is set, the stack stores this in a
+    `<tenant>-litellm-<env>-billing-metrics-client-cert` Secrets Manager
+    entry, grants the task-execution role GetSecretValue on it, and exposes
+    it to gateway + backend as LITELLM_BILLING_METRICS_CLIENT_CERT. Required
+    whenever metering is enabled.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
+
+variable "billing_metrics_client_key_pem" {
+  description = <<-EOT
+    PEM content of the private key matching
+    billing_metrics_client_cert_pem. Stored in a
+    `<tenant>-litellm-<env>-billing-metrics-client-key` Secrets Manager
+    entry and exposed as LITELLM_BILLING_METRICS_CLIENT_KEY. Required
+    whenever metering is enabled.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
+
+variable "billing_metrics_ca_cert_pem" {
+  description = <<-EOT
+    PEM content of the CA bundle used to verify the metering collector.
+    Only needed for private or test collectors whose CA is not in the
+    system trust store; telemetry.litellm.ai is publicly trusted, so leave
+    this empty for production. When set, it is exposed as
+    LITELLM_BILLING_METRICS_CA_CERT.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
--- a/terraform/litellm/gcp/README.md
+++ b/terraform/litellm/gcp/README.md
@ -204,6 +204,40 @@ Behavior matches the AWS stack 1:1; the only naming differences are
 `otel_headers_secret` (a Secret Manager resource ID) vs AWS's
 `otel_headers_secret_arn` (a Secrets Manager ARN).

+### Enterprise billing metrics
+
+License-gated request metering is opt-in and gated entirely on
+`billing_metrics_endpoint`. Empty (default) and no billing env is added to
+the container, so existing deployments are unchanged. Set it and both
+gateway and backend export billable-request counts over OTLP/HTTP,
+authenticating to the collector with the mTLS client certificate issued for
+your deployment.
+
+The proxy accepts the certificate, key, and CA bundle as either a file path
+or literal PEM content. This stack takes the PEM, writes each one to its own
+Secret Manager entry, grants the runtime service account
+`roles/secretmanager.secretAccessor` on them, and injects them as Cloud Run
+secret env vars `LITELLM_BILLING_METRICS_CLIENT_CERT` / `_CLIENT_KEY` (and
+`_CA_CERT` when set), so no volume mount is needed.
+
+```hcl
+billing_metrics_endpoint = "https://telemetry.litellm.ai/v1/metrics"
+```
+
+```bash
+export TF_VAR_billing_metrics_client_cert_pem="$(cat client.crt)"
+export TF_VAR_billing_metrics_client_key_pem="$(cat client.key)"
+```
+
+`billing_metrics_ca_cert_pem` is only for private or test collectors whose
+CA is not in the system trust store; leave it empty against
+`telemetry.litellm.ai`. Metering requires an enterprise license, so pair
+this with `litellm_license`. To tune the export cadence, set
+`LITELLM_BILLING_METRICS_EXPORT_INTERVAL_MS` through `gateway_extra_env` /
+`backend_extra_env`
+
+Behavior matches the AWS stack 1:1; the variable names are identical
+
 ## Tenant deployment

 Every resource the stack creates is named `${tenant}-litellm-${env}` (or
--- a/terraform/litellm/gcp/cloudrun.tf
+++ b/terraform/litellm/gcp/cloudrun.tf
@ -59,6 +59,33 @@ locals {
    { name = "OTEL_HEADERS", secret = var.otel_headers_secret, version = "latest" },
  ] : []

+  # Enterprise request metering, gated on billing_metrics_endpoint. The
+  # endpoint rides in as a plain env var; the mTLS material lives in Secret
+  # Manager (secrets.tf) and is injected as PEM-valued env vars, which the
+  # proxy accepts in place of file paths. Each PEM is wired only when the
+  # operator supplied it, so an empty ca_cert_pem falls back to the system
+  # trust store.
+  billing_metrics_enabled             = var.billing_metrics_endpoint != ""
+  billing_metrics_client_cert_enabled = local.billing_metrics_enabled && var.billing_metrics_client_cert_pem != ""
+  billing_metrics_client_key_enabled  = local.billing_metrics_enabled && var.billing_metrics_client_key_pem != ""
+  billing_metrics_ca_cert_enabled     = local.billing_metrics_enabled && var.billing_metrics_ca_cert_pem != ""
+
+  billing_metrics_env_kv = local.billing_metrics_enabled ? [
+    { name = "LITELLM_BILLING_METRICS_ENDPOINT", value = var.billing_metrics_endpoint },
+  ] : []
+
+  billing_metrics_env_secrets = concat(
+    local.billing_metrics_client_cert_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CLIENT_CERT", secret = google_secret_manager_secret.billing_metrics_client_cert[0].id, version = "latest" },
+    ] : [],
+    local.billing_metrics_client_key_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CLIENT_KEY", secret = google_secret_manager_secret.billing_metrics_client_key[0].id, version = "latest" },
+    ] : [],
+    local.billing_metrics_ca_cert_enabled ? [
+      { name = "LITELLM_BILLING_METRICS_CA_CERT", secret = google_secret_manager_secret.billing_metrics_ca_cert[0].id, version = "latest" },
+    ] : [],
+  )
+
  # Cloud Run v2 secret env vars use value_source.secret_key_ref pointing at a
  # secret resource ID. Shared between gateway and backend (the migrations
  # job has its own narrower env list — see migrations_env_secrets below).
@ -138,6 +165,30 @@ locals {

 # ---------- Gateway ----------
 resource "google_cloud_run_v2_service" "gateway" {
+  # Metering needs a client certificate AND its key. Each secret is created only
+  # when its own PEM is supplied, so an endpoint set with a missing key would
+  # otherwise apply cleanly and leave the proxy logging "missing config" and
+  # never exporting. ca_cert_pem stays optional: empty means fall back to the
+  # system trust store.
+  #
+  # The guard lives here, on an unconditional resource, rather than on the cert
+  # secret: that secret is count-gated on the cert itself, so it has zero
+  # instances in exactly the case this must catch. Adding count or for_each to
+  # this resource would silently stop the guard from evaluating.
+  #
+  #   endpoint  cert  key  -> result
+  #   ""        any   any  -> metering off, no secrets created
+  #   set       set   set  -> metering on
+  #   set       any-missing -> plan fails here
+  lifecycle {
+    precondition {
+      condition = var.billing_metrics_endpoint == "" || (
+        var.billing_metrics_client_cert_pem != "" && var.billing_metrics_client_key_pem != ""
+      )
+      error_message = "billing_metrics_client_cert_pem and billing_metrics_client_key_pem are both required when billing_metrics_endpoint is set."
+    }
+  }
+
  name                = "${local.name}-gateway"
  location            = var.region
  ingress             = "INGRESS_TRAFFIC_INTERNAL_LOAD_BALANCER"
@ -175,7 +226,7 @@ resource "google_cloud_run_v2_service" "gateway" {
      }

      dynamic "env" {
-        for_each = concat(local.shared_env_kv, local.gateway_otel_env_kv, local.gateway_extra_env_kv, local.proxy_config_env)
+        for_each = concat(local.shared_env_kv, local.gateway_otel_env_kv, local.billing_metrics_env_kv, local.gateway_extra_env_kv, local.proxy_config_env)
        content {
          name  = env.value.name
          value = env.value.value
@ -183,7 +234,7 @@ resource "google_cloud_run_v2_service" "gateway" {
      }

      dynamic "env" {
-        for_each = concat(local.shared_env_secrets, local.otel_env_secrets, local.gateway_extra_secret_kv)
+        for_each = concat(local.shared_env_secrets, local.otel_env_secrets, local.billing_metrics_env_secrets, local.gateway_extra_secret_kv)
        content {
          name = env.value.name
          value_source {
@ -242,6 +293,9 @@ resource "google_cloud_run_v2_service" "gateway" {
    google_secret_manager_secret_iam_member.license,
    google_secret_manager_secret_iam_member.extras,
    google_secret_manager_secret_iam_member.otel_headers,
+    google_secret_manager_secret_iam_member.billing_metrics_client_cert,
+    google_secret_manager_secret_iam_member.billing_metrics_client_key,
+    google_secret_manager_secret_iam_member.billing_metrics_ca_cert,
    google_storage_bucket_iam_member.proxy_config_runtime,
    google_sql_user.app,
    # Don't go live until the schema is migrated; otherwise the proxy boots,
@ -252,6 +306,18 @@ resource "google_cloud_run_v2_service" "gateway" {

 # ---------- Backend ----------
 resource "google_cloud_run_v2_service" "backend" {
+  # Same guard as the gateway: the backend meters too (it serves the named-server
+  # MCP transport), and a targeted apply of just this resource must not slip a
+  # billing endpoint through without the credentials to use it.
+  lifecycle {
+    precondition {
+      condition = var.billing_metrics_endpoint == "" || (
+        var.billing_metrics_client_cert_pem != "" && var.billing_metrics_client_key_pem != ""
+      )
+      error_message = "billing_metrics_client_cert_pem and billing_metrics_client_key_pem are both required when billing_metrics_endpoint is set."
+    }
+  }
+
  name                = "${local.name}-backend"
  location            = var.region
  ingress             = "INGRESS_TRAFFIC_INTERNAL_LOAD_BALANCER"
@ -289,7 +355,7 @@ resource "google_cloud_run_v2_service" "backend" {
      }

      dynamic "env" {
-        for_each = concat(local.shared_env_kv, local.backend_default_env_kv, local.backend_otel_env_kv, local.backend_extra_env_kv, local.proxy_config_env)
+        for_each = concat(local.shared_env_kv, local.backend_default_env_kv, local.backend_otel_env_kv, local.billing_metrics_env_kv, local.backend_extra_env_kv, local.proxy_config_env)
        content {
          name  = env.value.name
          value = env.value.value
@ -297,7 +363,7 @@ resource "google_cloud_run_v2_service" "backend" {
      }

      dynamic "env" {
-        for_each = concat(local.shared_env_secrets, local.backend_managed_env_secrets, local.otel_env_secrets, local.backend_extra_secret_kv)
+        for_each = concat(local.shared_env_secrets, local.backend_managed_env_secrets, local.otel_env_secrets, local.billing_metrics_env_secrets, local.backend_extra_secret_kv)
        content {
          name = env.value.name
          value_source {
@ -357,6 +423,9 @@ resource "google_cloud_run_v2_service" "backend" {
    google_secret_manager_secret_iam_member.ui_password,
    google_secret_manager_secret_iam_member.extras,
    google_secret_manager_secret_iam_member.otel_headers,
+    google_secret_manager_secret_iam_member.billing_metrics_client_cert,
+    google_secret_manager_secret_iam_member.billing_metrics_client_key,
+    google_secret_manager_secret_iam_member.billing_metrics_ca_cert,
    google_storage_bucket_iam_member.proxy_config_runtime,
    google_sql_user.app,
    terraform_data.migration,
--- a/terraform/litellm/gcp/iam.tf
+++ b/terraform/litellm/gcp/iam.tf
@ -79,3 +79,29 @@ resource "google_secret_manager_secret_iam_member" "otel_headers" {
  role      = "roles/secretmanager.secretAccessor"
  member    = "serviceAccount:${google_service_account.runtime.email}"
 }
+
+# Billing-metrics mTLS accessors — only created when request metering is
+# enabled and the matching PEM was supplied.
+resource "google_secret_manager_secret_iam_member" "billing_metrics_client_cert" {
+  count = local.billing_metrics_client_cert_enabled ? 1 : 0
+
+  secret_id = google_secret_manager_secret.billing_metrics_client_cert[0].id
+  role      = "roles/secretmanager.secretAccessor"
+  member    = "serviceAccount:${google_service_account.runtime.email}"
+}
+
+resource "google_secret_manager_secret_iam_member" "billing_metrics_client_key" {
+  count = local.billing_metrics_client_key_enabled ? 1 : 0
+
+  secret_id = google_secret_manager_secret.billing_metrics_client_key[0].id
+  role      = "roles/secretmanager.secretAccessor"
+  member    = "serviceAccount:${google_service_account.runtime.email}"
+}
+
+resource "google_secret_manager_secret_iam_member" "billing_metrics_ca_cert" {
+  count = local.billing_metrics_ca_cert_enabled ? 1 : 0
+
+  secret_id = google_secret_manager_secret.billing_metrics_ca_cert[0].id
+  role      = "roles/secretmanager.secretAccessor"
+  member    = "serviceAccount:${google_service_account.runtime.email}"
+}
--- a/terraform/litellm/gcp/secrets.tf
+++ b/terraform/litellm/gcp/secrets.tf
@ -63,3 +63,58 @@ resource "google_secret_manager_secret_version" "ui_password" {
  secret      = google_secret_manager_secret.ui_password[0].id
  secret_data = var.ui_password
 }
+
+# Billing-metrics mTLS material — only created when metering is enabled
+# (billing_metrics_endpoint non-empty) and the operator supplied the PEM.
+# The runtime SA gets accessor permission via iam.tf, and gateway + backend
+# pick the env vars up through billing_metrics_env_secrets in cloudrun.tf.
+resource "google_secret_manager_secret" "billing_metrics_client_cert" {
+  count = local.billing_metrics_client_cert_enabled ? 1 : 0
+
+  secret_id = "${local.name}-billing-metrics-client-cert"
+  labels    = local.labels
+  replication {
+    auto {}
+  }
+}
+
+resource "google_secret_manager_secret_version" "billing_metrics_client_cert" {
+  count = local.billing_metrics_client_cert_enabled ? 1 : 0
+
+  secret      = google_secret_manager_secret.billing_metrics_client_cert[0].id
+  secret_data = var.billing_metrics_client_cert_pem
+}
+
+resource "google_secret_manager_secret" "billing_metrics_client_key" {
+  count = local.billing_metrics_client_key_enabled ? 1 : 0
+
+  secret_id = "${local.name}-billing-metrics-client-key"
+  labels    = local.labels
+  replication {
+    auto {}
+  }
+}
+
+resource "google_secret_manager_secret_version" "billing_metrics_client_key" {
+  count = local.billing_metrics_client_key_enabled ? 1 : 0
+
+  secret      = google_secret_manager_secret.billing_metrics_client_key[0].id
+  secret_data = var.billing_metrics_client_key_pem
+}
+
+resource "google_secret_manager_secret" "billing_metrics_ca_cert" {
+  count = local.billing_metrics_ca_cert_enabled ? 1 : 0
+
+  secret_id = "${local.name}-billing-metrics-ca-cert"
+  labels    = local.labels
+  replication {
+    auto {}
+  }
+}
+
+resource "google_secret_manager_secret_version" "billing_metrics_ca_cert" {
+  count = local.billing_metrics_ca_cert_enabled ? 1 : 0
+
+  secret      = google_secret_manager_secret.billing_metrics_ca_cert[0].id
+  secret_data = var.billing_metrics_ca_cert_pem
+}
--- a/terraform/litellm/gcp/variables.tf
+++ b/terraform/litellm/gcp/variables.tf
@ -490,3 +490,66 @@ variable "otel_capture_message_content" {
    error_message = "otel_capture_message_content must be one of: no_content, prompt_and_completion."
  }
 }
+
+# ---------- Enterprise billing metrics ----------
+#
+# License-gated request metering. Opt-in and gated entirely on
+# billing_metrics_endpoint: leave it empty (the default) and nothing
+# metering-related is added to the container env. Set it and gateway +
+# backend export billable-request counts over OTLP/HTTP, authenticating to
+# the collector with an mTLS client cert. The proxy accepts the cert, key,
+# and CA as either a file path or literal PEM content, so on Cloud Run they
+# are injected straight from Secret Manager as env vars and no volume is
+# needed.
+
+variable "billing_metrics_endpoint" {
+  description = <<-EOT
+    OTLP/HTTP endpoint for enterprise billing metrics (sets
+    LITELLM_BILLING_METRICS_ENDPOINT). Non-empty enables request metering;
+    empty (default) disables it and adds no billing env to the container.
+    Requires an enterprise license. Example:
+    "https://telemetry.litellm.ai/v1/metrics"
+  EOT
+  type        = string
+  default     = ""
+}
+
+variable "billing_metrics_client_cert_pem" {
+  description = <<-EOT
+    PEM content of the mTLS client certificate issued for this deployment.
+    When billing_metrics_endpoint is set, the stack stores this in a
+    `<tenant>-litellm-<env>-billing-metrics-client-cert` Secret Manager
+    entry, grants the runtime SA accessor on it, and exposes it to gateway +
+    backend as LITELLM_BILLING_METRICS_CLIENT_CERT. Required whenever
+    metering is enabled.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
+
+variable "billing_metrics_client_key_pem" {
+  description = <<-EOT
+    PEM content of the private key matching
+    billing_metrics_client_cert_pem. Stored in a
+    `<tenant>-litellm-<env>-billing-metrics-client-key` Secret Manager entry
+    and exposed as LITELLM_BILLING_METRICS_CLIENT_KEY. Required whenever
+    metering is enabled.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
+
+variable "billing_metrics_ca_cert_pem" {
+  description = <<-EOT
+    PEM content of the CA bundle used to verify the metering collector.
+    Only needed for private or test collectors whose CA is not in the
+    system trust store; telemetry.litellm.ai is publicly trusted, so leave
+    this empty for production. When set, it is exposed as
+    LITELLM_BILLING_METRICS_CA_CERT.
+  EOT
+  type        = string
+  default     = ""
+  sensitive   = true
+}
--- a/tests/test_litellm/proxy/enterprise_billing/test_billing_metrics.py
+++ b/tests/test_litellm/proxy/enterprise_billing/test_billing_metrics.py
@ -0,0 +1,432 @@
+"""
+Tests for the enterprise billing-metrics recorder and its factory.
+
+These verify the license gate, the missing-config and missing-cert disable
+paths, the OTLP/HTTP exporter wiring (client cert+key authenticate us to the
+collector's mTLS-terminating front end; CA override optional for private
+collectors), the metric attribute mapping, and that recording produces the
+expected OTLP counter via an in-memory reader.
+"""
+
+import os
+import socket
+import stat
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import pytest
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+
+from litellm.proxy.enterprise_billing import billing_metrics as bm
+from litellm.proxy.middleware.billable_request_metrics_middleware import BillableCategory
+
+_ENV_VARS = (
+    bm.ENDPOINT_ENV,
+    bm.CLIENT_CERT_ENV,
+    bm.CLIENT_KEY_ENV,
+    bm.CA_CERT_ENV,
+    bm.EXPORT_INTERVAL_ENV,
+)
+
+
+@pytest.fixture(autouse=True)
+def clear_env(monkeypatch):
+    for name in _ENV_VARS:
+        monkeypatch.delenv(name, raising=False)
+    yield
+    bm.shutdown_billing_metrics_recorder()
+
+
+def _write_certs(tmp_path: Path) -> Dict[str, str]:
+    files = {
+        bm.CA_CERT_ENV: ("ca.pem", b"ca-bytes"),
+        bm.CLIENT_CERT_ENV: ("client.pem", b"client-cert-bytes"),
+        bm.CLIENT_KEY_ENV: ("client.key", b"client-key-bytes"),
+    }
+    paths = {}
+    for env_name, (filename, content) in files.items():
+        path = tmp_path / filename
+        path.write_bytes(content)
+        paths[env_name] = str(path)
+    return paths
+
+
+def _set_full_env(monkeypatch, tmp_path: Path) -> Dict[str, str]:
+    paths = _write_certs(tmp_path)
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CA_CERT_ENV, paths[bm.CA_CERT_ENV])
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, paths[bm.CLIENT_CERT_ENV])
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, paths[bm.CLIENT_KEY_ENV])
+    return paths
+
+
+def _config(tmp_path: Path, license_id: Optional[str] = "org-1") -> bm.BillingMetricsConfig:
+    paths = _write_certs(tmp_path)
+    return bm.BillingMetricsConfig(
+        endpoint="https://collector.example:4317",
+        client_cert_path=paths[bm.CLIENT_CERT_ENV],
+        client_key_path=paths[bm.CLIENT_KEY_ENV],
+        ca_cert_path=paths[bm.CA_CERT_ENV],
+        export_interval_ms=60_000,
+        litellm_version="1.2.3",
+        license_id=license_id,
+    )
+
+
+# ── Factory gating ────────────────────────────────────────────────────────────
+
+
+def test_not_premium_returns_none(tmp_path, monkeypatch):
+    _set_full_env(monkeypatch, tmp_path)
+    assert bm.build_billing_metrics_recorder(premium=False, license_data=None, litellm_version="1.0") is None
+
+
+def test_premium_without_config_returns_none(monkeypatch):
+    assert bm.build_billing_metrics_recorder(premium=True, license_data={"user_id": "x"}, litellm_version="1.0") is None
+
+
+def test_premium_with_missing_cert_files_returns_none(monkeypatch, tmp_path):
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CA_CERT_ENV, str(tmp_path / "missing-ca.pem"))
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, str(tmp_path / "missing-cert.pem"))
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, str(tmp_path / "missing-key.pem"))
+    assert bm.build_billing_metrics_recorder(premium=True, license_data=None, litellm_version="1.0") is None
+
+
+def test_premium_with_full_config_builds_recorder(monkeypatch, tmp_path):
+    """Builds a real MeterProvider, so the exporter is stubbed: the live one
+    resolves the collector and opens a TLS connection during the shutdown flush.
+    The getaddrinfo spy keeps that stub from being quietly dropped later."""
+    _set_full_env(monkeypatch, tmp_path)
+    monkeypatch.setattr(bm, "OTLPMetricExporter", _fake_exporter_class({}))
+
+    resolved: List[str] = []
+    real_getaddrinfo = socket.getaddrinfo
+
+    def _spy_getaddrinfo(host, port, *args, **kwargs):
+        resolved.append(str(host))
+        return real_getaddrinfo(host, port, *args, **kwargs)
+
+    monkeypatch.setattr(socket, "getaddrinfo", _spy_getaddrinfo)
+
+    recorder = bm.build_billing_metrics_recorder(
+        premium=True, license_data={"user_id": "org-1"}, litellm_version="1.0"
+    )
+    assert isinstance(recorder, bm.BillingMetricsRecorder)
+    recorder.record(category=BillableCategory.LLM, route="/chat/completions", status_code=200, model_id=None)
+    bm.shutdown_billing_metrics_recorder()
+
+    assert [host for host in resolved if "collector.example" in host] == []
+
+
+def test_building_the_recorder_logs_an_affirmative_line(monkeypatch, tmp_path):
+    """
+    Every disable path logs; a successful build must log too. Otherwise an
+    operator cannot tell a metering component from one that silently returned
+    None, which is how an unlicensed component looks healthy while exporting
+    nothing.
+    """
+    _set_full_env(monkeypatch, tmp_path)
+    monkeypatch.setenv(bm.EXPORT_INTERVAL_ENV, "5000")
+    monkeypatch.setattr(bm, "OTLPMetricExporter", _fake_exporter_class({}))
+
+    infos: List[str] = []
+    monkeypatch.setattr(bm.verbose_proxy_logger, "info", lambda msg, *args: infos.append(msg % args if args else msg))
+
+    recorder = bm.build_billing_metrics_recorder(premium=True, license_data={"user_id": "org-1"}, litellm_version="1.0")
+
+    assert recorder is not None
+    joined = "\n".join(infos)
+    assert "https://collector.example:4317" in joined
+    assert "5000" in joined
+
+
+def test_unlicensed_build_does_not_warn(monkeypatch, tmp_path):
+    """Unlicensed is the common OSS case; warning there would be pure noise."""
+    _set_full_env(monkeypatch, tmp_path)
+
+    warnings: List[str] = []
+    monkeypatch.setattr(bm.verbose_proxy_logger, "warning", lambda msg, *args: warnings.append(str(msg)))
+
+    assert bm.build_billing_metrics_recorder(premium=False, license_data=None, litellm_version="1.0") is None
+    assert warnings == []
+
+
+def test_shutdown_flushes_active_recorder_once(monkeypatch, tmp_path):
+    """The shutdown hook must flush the recorder the factory built (buffered
+    counts are lost on restart otherwise) and be idempotent for repeat calls."""
+    _set_full_env(monkeypatch, tmp_path)
+    shutdowns = []
+
+    class _SpyProvider:
+        def get_meter(self, name):
+            return MeterProvider().get_meter(name)
+
+        def shutdown(self, timeout_millis=None):
+            shutdowns.append(timeout_millis)
+
+    monkeypatch.setattr(bm, "build_mtls_meter_provider", lambda config: _SpyProvider())
+    recorder = bm.build_billing_metrics_recorder(premium=True, license_data=None, litellm_version="1.0")
+    assert recorder is not None
+
+    bm.shutdown_billing_metrics_recorder()
+    bm.shutdown_billing_metrics_recorder()
+    assert shutdowns == [bm.SHUTDOWN_FLUSH_TIMEOUT_MS]
+
+
+def test_shutdown_without_active_recorder_is_noop():
+    bm.shutdown_billing_metrics_recorder()
+
+
+# ── Config loading ────────────────────────────────────────────────────────────
+
+
+def test_load_config_carries_license_id(monkeypatch, tmp_path):
+    _set_full_env(monkeypatch, tmp_path)
+    config = bm.load_billing_metrics_config(license_data={"user_id": "org-42"}, litellm_version="9.9")
+    assert config is not None and config.license_id == "org-42" and config.litellm_version == "9.9"
+
+
+def test_load_config_with_empty_string_env_is_disabled(monkeypatch, tmp_path):
+    """An env var set to the empty string is as unusable as an unset one and
+    must disable metering rather than produce a config with a blank endpoint."""
+    paths = _write_certs(tmp_path)
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, paths[bm.CLIENT_CERT_ENV])
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, paths[bm.CLIENT_KEY_ENV])
+    assert bm.load_billing_metrics_config(license_data=None, litellm_version="1.0") is None
+
+
+_CLIENT_CERT_PEM = "-----BEGIN CERTIFICATE-----\nclient-cert-body\n-----END CERTIFICATE-----"
+_CLIENT_KEY_PEM = "-----BEGIN PRIVATE KEY-----\nclient-key-body\n-----END PRIVATE KEY-----"
+_CA_CERT_PEM = "-----BEGIN CERTIFICATE-----\nca-body\n-----END CERTIFICATE-----"
+
+
+def test_load_config_materializes_inline_pem_content(monkeypatch):
+    """
+    ECS and Cloud Run inject secrets as env content, not as mounted files, so the
+    cert env vars must accept PEM directly. The exporter takes paths, so the PEM
+    is written to disk and the config points at those files.
+    """
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, _CLIENT_CERT_PEM)
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, _CLIENT_KEY_PEM)
+    monkeypatch.setenv(bm.CA_CERT_ENV, _CA_CERT_PEM)
+
+    config = bm.load_billing_metrics_config(license_data=None, litellm_version="1.0")
+
+    assert config is not None
+    assert config.ca_cert_path is not None
+    written = {
+        config.client_cert_path: _CLIENT_CERT_PEM,
+        config.client_key_path: _CLIENT_KEY_PEM,
+        config.ca_cert_path: _CA_CERT_PEM,
+    }
+    for path, pem in written.items():
+        assert path != pem, "config must carry a file path, not the PEM itself"
+        assert os.path.isfile(path)
+        assert Path(path).read_text(encoding="utf-8") == f"{pem}\n"
+
+    # The private key must not be world- or group-readable.
+    assert stat.S_IMODE(os.stat(config.client_key_path).st_mode) == 0o600
+
+
+def test_load_config_accepts_a_mix_of_pem_content_and_file_paths(monkeypatch, tmp_path):
+    """A deployment may mount the CA but inject the client credentials inline."""
+    paths = _write_certs(tmp_path)
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, _CLIENT_CERT_PEM)
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, _CLIENT_KEY_PEM)
+    monkeypatch.setenv(bm.CA_CERT_ENV, paths[bm.CA_CERT_ENV])
+
+    config = bm.load_billing_metrics_config(license_data=None, litellm_version="1.0")
+
+    assert config is not None
+    assert config.ca_cert_path == paths[bm.CA_CERT_ENV]
+    assert Path(config.client_cert_path).read_text(encoding="utf-8") == f"{_CLIENT_CERT_PEM}\n"
+
+
+def test_load_config_leaves_file_paths_untouched(monkeypatch, tmp_path):
+    """Path-valued env vars keep working; nothing is copied or rewritten."""
+    paths = _set_full_env(monkeypatch, tmp_path)
+
+    config = bm.load_billing_metrics_config(license_data=None, litellm_version="1.0")
+
+    assert config is not None
+    assert config.client_cert_path == paths[bm.CLIENT_CERT_ENV]
+    assert config.client_key_path == paths[bm.CLIENT_KEY_ENV]
+    assert config.ca_cert_path == paths[bm.CA_CERT_ENV]
+
+
+def test_load_config_with_inline_pem_disabled_when_unwritable(monkeypatch):
+    """A failure to materialize the PEM disables metering instead of raising."""
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, _CLIENT_CERT_PEM)
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, _CLIENT_KEY_PEM)
+
+    def _explode(prefix=None):
+        raise OSError("read-only filesystem")
+
+    monkeypatch.setattr(bm.tempfile, "mkdtemp", _explode)
+
+    assert bm.load_billing_metrics_config(license_data=None, litellm_version="1.0") is None
+
+
+def test_load_config_never_logs_credential_values(monkeypatch):
+    """
+    A value that is neither a readable path nor `-----BEGIN`-prefixed PEM is
+    still secret material. The disable warning must name the env vars, never
+    echo their contents, or a malformed key lands in the proxy logs.
+    """
+    secret_material = "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQ-not-pem-prefixed"
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, secret_material)
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, secret_material)
+
+    logged: List[str] = []
+
+    def _capture(msg, *args):
+        logged.append(msg % args if args else msg)
+
+    monkeypatch.setattr(bm.verbose_proxy_logger, "warning", _capture)
+
+    assert bm.load_billing_metrics_config(license_data=None, litellm_version="1.0") is None
+
+    joined = "\n".join(logged)
+    assert secret_material not in joined
+    assert bm.CLIENT_CERT_ENV in joined and bm.CLIENT_KEY_ENV in joined
+
+
+def test_load_config_with_empty_pem_env_is_disabled(monkeypatch):
+    """Empty stays empty: an unset secret must not be mistaken for inline PEM."""
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://collector.example:4317")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, "")
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, "")
+
+    assert bm.load_billing_metrics_config(license_data=None, litellm_version="1.0") is None
+
+
+def test_export_interval_default_and_override(monkeypatch):
+    assert bm._export_interval_ms() == bm.DEFAULT_EXPORT_INTERVAL_MS
+    monkeypatch.setenv(bm.EXPORT_INTERVAL_ENV, "5000")
+    assert bm._export_interval_ms() == 5000
+    monkeypatch.setenv(bm.EXPORT_INTERVAL_ENV, "not-a-number")
+    assert bm._export_interval_ms() == bm.DEFAULT_EXPORT_INTERVAL_MS
+
+
+# ── OTLP/HTTP exporter wiring ─────────────────────────────────────────────────
+
+
+def test_metrics_endpoint_appends_signal_path():
+    assert bm._metrics_endpoint("https://telemetry.example.com") == "https://telemetry.example.com/v1/metrics"
+    assert bm._metrics_endpoint("https://telemetry.example.com/") == "https://telemetry.example.com/v1/metrics"
+    assert bm._metrics_endpoint("https://telemetry.example.com/v1/metrics") == "https://telemetry.example.com/v1/metrics"
+
+
+def _fake_exporter_class(captured: Dict[str, object]) -> type:
+    """A no-network stand-in for OTLPMetricExporter. Tests that build a real
+    MeterProvider must install this: the real exporter resolves the collector
+    host and opens a TLS connection on the reader's first export and on the
+    shutdown flush."""
+
+    class _FakeExporter:
+        # PeriodicExportingMetricReader probes these on the exporter it wraps.
+        _preferred_temporality: dict = {}
+        _preferred_aggregation: dict = {}
+
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+
+        def export(self, *args, **kwargs):
+            return None
+
+        def shutdown(self, *args, **kwargs):
+            return None
+
+        def force_flush(self, *args, **kwargs):
+            return True
+
+    return _FakeExporter
+
+
+def test_meter_provider_wires_client_cert_into_http_exporter(tmp_path, monkeypatch):
+    """Client cert+key authenticate us at the collector's mTLS front end; CA override rides certificate_file."""
+    captured: Dict[str, object] = {}
+
+    monkeypatch.setattr(bm, "OTLPMetricExporter", _fake_exporter_class(captured))
+    config = _config(tmp_path)
+    provider = bm.build_mtls_meter_provider(config)
+    provider.shutdown()
+
+    assert captured["endpoint"] == "https://collector.example:4317/v1/metrics"
+    assert captured["client_certificate_file"] == config.client_cert_path
+    assert captured["client_key_file"] == config.client_key_path
+    assert captured["certificate_file"] == config.ca_cert_path
+
+
+def test_load_config_without_ca_is_valid(monkeypatch, tmp_path):
+    """The production collector presents a public web-PKI cert: no CA override required."""
+    paths = _write_certs(tmp_path)
+    monkeypatch.setenv(bm.ENDPOINT_ENV, "https://telemetry.example.com")
+    monkeypatch.setenv(bm.CLIENT_CERT_ENV, paths[bm.CLIENT_CERT_ENV])
+    monkeypatch.setenv(bm.CLIENT_KEY_ENV, paths[bm.CLIENT_KEY_ENV])
+    config = bm.load_billing_metrics_config(license_data=None, litellm_version="1.0")
+    assert config is not None and config.ca_cert_path is None
+
+
+# ── Resource and metric attributes ────────────────────────────────────────────
+
+
+def test_resource_attributes_include_license_id(tmp_path):
+    attrs = bm._resource_attributes(_config(tmp_path, license_id="org-7"))
+    assert attrs["service.name"] == "litellm-proxy"
+    assert attrs["litellm.version"] == "1.2.3"
+    assert attrs["litellm.license.id"] == "org-7"
+
+
+def test_resource_attributes_omit_license_id_when_absent(tmp_path):
+    attrs = bm._resource_attributes(_config(tmp_path, license_id=None))
+    assert "litellm.license.id" not in attrs
+
+
+def test_billable_attributes_with_model_id():
+    attrs = bm._billable_attributes(BillableCategory.LLM, "/chat/completions", 200, "deploy-3")
+    assert attrs == {
+        "litellm.endpoint.category": "llm",
+        "http.route": "/chat/completions",
+        "http.response.status_code": 200,
+        "litellm.model_id": "deploy-3",
+    }
+
+
+def test_billable_attributes_omit_model_id_when_none():
+    attrs = bm._billable_attributes(BillableCategory.MCP, "/mcp", 200, None)
+    assert "litellm.model_id" not in attrs
+
+
+# ── End-to-end recording via in-memory reader ─────────────────────────────────
+
+
+def _counter_points(reader: InMemoryMetricReader):
+    data = reader.get_metrics_data()
+    for resource_metric in data.resource_metrics:
+        for scope_metric in resource_metric.scope_metrics:
+            for metric in scope_metric.metrics:
+                if metric.name == bm.METRIC_NAME:
+                    return list(metric.data.data_points)
+    return []
+
+
+def test_record_increments_counter_with_attributes():
+    reader = InMemoryMetricReader()
+    recorder = bm.BillingMetricsRecorder(MeterProvider(metric_readers=[reader]))
+
+    recorder.record(category=BillableCategory.LLM, route="/chat/completions", status_code=200, model_id="m1")
+    recorder.record(category=BillableCategory.LLM, route="/chat/completions", status_code=200, model_id="m1")
+    recorder.record(category=BillableCategory.MCP, route="/mcp", status_code=200, model_id=None)
+
+    points = _counter_points(reader)
+    by_category = {point.attributes["litellm.endpoint.category"]: point.value for point in points}
+    assert by_category["llm"] == 2
+    assert by_category["mcp"] == 1
--- a/tests/test_litellm/proxy/middleware/test_billable_request_metrics_middleware.py
+++ b/tests/test_litellm/proxy/middleware/test_billable_request_metrics_middleware.py
@ -0,0 +1,458 @@
+"""
+Tests for BillableRequestMetricsMiddleware and route classification.
+
+These verify the metering gate (records only on 2xx to a billable endpoint),
+correct category/route classification, model-id extraction, and that the
+middleware is a transparent pass-through when no recorder is injected.
+"""
+
+import asyncio
+import threading
+from typing import List, Optional, Tuple
+
+import pytest
+from starlette.applications import Starlette
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from litellm.proxy.middleware.billable_request_metrics_middleware import (
+    BillableCategory,
+    BillableRequestMetricsMiddleware,
+    _extract_model_id,
+    classify_billable_request,
+)
+from litellm.proxy.middleware.in_flight_requests_middleware import (
+    InFlightRequestsMiddleware,
+)
+
+
+class FakeRecorder:
+    def __init__(self) -> None:
+        self.calls: List[dict] = []
+
+    def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: Optional[str]) -> None:
+        self.calls.append(
+            {"category": category, "route": route, "status_code": status_code, "model_id": model_id}
+        )
+
+
+def _make_app(recorder: Optional[FakeRecorder], status_code: int = 200, model_id: Optional[str] = None) -> Starlette:
+    async def handler(request: Request) -> Response:
+        headers = {"x-litellm-model-id": model_id} if model_id else {}
+        return JSONResponse({}, status_code=status_code, headers=headers)
+
+    paths = [
+        "/v1/chat/completions",
+        "/chat/completions",
+        "/v1/embeddings",
+        "/v1/completions",
+        "/mcp",
+        "/github/mcp",
+        "/toolset/my-tools/mcp",
+        "/v1/mcp/tools",
+        "/v1/mcp/server",
+        "/a2a/agent-1/message/send",
+        "/v1/a2a/discover",
+        "/health",
+        "/ui",
+    ]
+    app = Starlette(routes=[Route(p, handler, methods=["GET", "POST"]) for p in paths])
+    app.add_middleware(BillableRequestMetricsMiddleware, recorder=recorder)
+    return app
+
+
+# ── Structure ───────────────────────────────────────────────────────────────
+
+
+def test_is_pure_asgi_not_base_http_middleware():
+    assert not issubclass(BillableRequestMetricsMiddleware, BaseHTTPMiddleware)
+
+
+# ── classify_billable_request ─────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "path,expected",
+    [
+        ("/v1/chat/completions", (BillableCategory.LLM, "/chat/completions")),
+        ("/chat/completions", (BillableCategory.LLM, "/chat/completions")),
+        ("/openai/deployments/gpt-4o/chat/completions", (BillableCategory.LLM, "/chat/completions")),
+        ("/engines/gpt-4o/chat/completions", (BillableCategory.LLM, "/chat/completions")),
+        ("/v1/completions", (BillableCategory.LLM, "/completions")),
+        ("/completions", (BillableCategory.LLM, "/completions")),
+        ("/v1/embeddings", (BillableCategory.LLM, "/embeddings")),
+        ("/v1/responses", (BillableCategory.LLM, "/responses")),
+        ("/v1/rerank", (BillableCategory.LLM, "/rerank")),
+        ("/v1/audio/transcriptions", (BillableCategory.LLM, "/audio/transcriptions")),
+        # Routes from the metering-bypass finding: authenticated inference
+        # endpoints that must bill and previously classified as None.
+        ("/v1/images/edits", (BillableCategory.LLM, "/images/edits")),
+        ("/images/edits", (BillableCategory.LLM, "/images/edits")),
+        ("/openai/deployments/dall-e/images/edits", (BillableCategory.LLM, "/images/edits")),
+        ("/v1/images/variations", (BillableCategory.LLM, "/images/variations")),
+        ("/v1/messages", (BillableCategory.LLM, "/v1/messages")),
+        ("/interactions", (BillableCategory.LLM, "/interactions")),
+        ("/v1beta/interactions", (BillableCategory.LLM, "/v1beta/interactions")),
+        ("/v1/videos", (BillableCategory.LLM, "/videos")),
+        ("/v1/videos/video_123/remix", (BillableCategory.LLM, "/remix")),
+        ("/v1/ocr", (BillableCategory.LLM, "/ocr")),
+        ("/v1beta/models/gemini-2.5-pro:generateContent", (BillableCategory.LLM, ":generateContent")),
+        ("/v1beta/models/gemini-2.5-pro:streamGenerateContent", (BillableCategory.LLM, ":streamGenerateContent")),
+        # SpendLogs-producing routes surfaced by the route-inventory audit
+        ("/v1/search", (BillableCategory.LLM, "/search")),
+        ("/v1/vector_stores/vs_1/search", (BillableCategory.LLM, "/search")),
+        ("/v1/rag/query", (BillableCategory.LLM, "/rag/query")),
+        ("/rag/ingest", (BillableCategory.LLM, "/rag/ingest")),
+        # Provider passthrough carries real inference and writes SpendLogs
+        ("/bedrock/model/anthropic.claude-v2/invoke", (BillableCategory.LLM, "/bedrock")),
+        ("/vertex-ai/publishers/google/models/gemini:predict", (BillableCategory.LLM, "/vertex-ai")),
+        ("/cohere/v2/chat", (BillableCategory.LLM, "/cohere")),
+        # Passthrough inference bills under its provider prefix
+        ("/anthropic/v1/messages", (BillableCategory.LLM, "/anthropic")),
+        ("/mcp", (BillableCategory.MCP, "/mcp")),
+        ("/mcp/", (BillableCategory.MCP, "/mcp")),
+        ("/mcp/tools/list", (BillableCategory.MCP, "/mcp")),
+        ("/github/mcp", (BillableCategory.MCP, "/mcp")),
+        ("/github/mcp/", (BillableCategory.MCP, "/mcp")),
+        ("/toolset/my-tools/mcp", (BillableCategory.MCP, "/mcp")),
+        ("/github,slack/mcp", (BillableCategory.MCP, "/mcp")),
+        # REST wrapper tool execution fires the same MCP spend logging as /mcp
+        ("/mcp-rest/tools/call", (BillableCategory.MCP, "/mcp")),
+        ("/a2a/agent-1/message/send", (BillableCategory.A2A, "/a2a")),
+        ("/v1/a2a/agent-9/message/send", (BillableCategory.A2A, "/a2a")),
+    ],
+)
+def test_classify_billable(path: str, expected: Tuple[BillableCategory, str]):
+    assert classify_billable_request(path) == expected
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/health",
+        "/health/readiness",
+        "/metrics",
+        "/ui",
+        "/",
+        "/v1/models",
+        "/key/generate",
+        "/v1/files",
+        # tokenization helper, not an inference call
+        "/v1/messages/count_tokens",
+        # OpenAI Assistants thread messages write no SpendLogs row
+        "/v1/threads/thread_abc123/messages",
+        "/threads/thread_abc123/messages",
+        # Google Interactions reads and cancel are not inference calls
+        "/interactions/int_123",
+        "/v1beta/interactions/int_123",
+        "/interactions/int_123/cancel",
+        "/v1beta/interactions/int_123/cancel",
+        # observability passthrough writes no SpendLogs row
+        "/langfuse/api/public/ingestion",
+        # a bare provider prefix is not an inference call
+        "/bedrock",
+        "/v1/mcp",
+        "/v1/mcp/tools",
+        "/v1/mcp/server",
+        "/v1/mcp/server/health",
+        "/v1/mcp/server/some-id",
+        "/v1/mcp/server/register",
+        "/v1/mcp/oauth/some-id/authorize",
+        "/a2a/agent-1/.well-known/agent-card.json",
+        "/v1/a2a/discover",
+        "/.well-known/oauth-protected-resource/github/mcp",
+        "/mcp-rest/tools/list",
+        "/mcp-rest/test/connection",
+        "/mcp-rest/test/tools/list",
+    ],
+)
+def test_classify_non_billable_returns_none(path: str):
+    assert classify_billable_request(path) is None
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/v1/mcp/tools",
+        "/v1/mcp/server",
+        "/v1/mcp/server/register",
+        "/v1/a2a/discover",
+    ],
+)
+def test_classify_management_writes_are_not_billable(path: str):
+    assert classify_billable_request(path, "POST") is None
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/a2a/agent-1",
+        "/a2a/agent-1/",
+        "/v1/a2a/agent-1",
+    ],
+)
+def test_classify_bare_a2a_route_is_not_billable(path: str):
+    """
+    The bare A2A route multiplexes JSON-RPC methods off the request body. Only
+    message/send and message/stream write a SpendLogs row; tasks/get,
+    tasks/cancel and the pushNotificationConfig RPCs are forwarded upstream and
+    write none. Billing the path would count those task RPCs and push the metric
+    above the dashboard's successful-request count, so it must stay unbilled.
+    """
+    assert classify_billable_request(path, "POST") is None
+
+
+@pytest.mark.parametrize(
+    "path",
+    ["/v1/videos", "/v1/responses", "/v1/chat/completions", "/v1/messages"],
+)
+def test_classify_get_reads_are_not_billable(path: str):
+    """GETs on inference resources (list videos, fetch a response) write no
+    SpendLogs row and must not bill; only POST inference calls count."""
+    assert classify_billable_request(path, "GET") is None
+
+
+def test_classify_mcp_not_method_gated():
+    assert classify_billable_request("/mcp/tools/list", "GET") == (BillableCategory.MCP, "/mcp")
+
+
+def test_chat_completions_not_misclassified_as_plain_completions():
+    """The /chat/completions suffix must win over /completions so the route label is correct."""
+    category, route = classify_billable_request("/v1/chat/completions")
+    assert route == "/chat/completions"
+
+
+# ── _extract_model_id ─────────────────────────────────────────────────────────
+
+
+def test_extract_model_id_present():
+    headers = [(b"content-type", b"application/json"), (b"x-litellm-model-id", b"deploy-123")]
+    assert _extract_model_id(headers) == "deploy-123"
+
+
+def test_extract_model_id_case_insensitive():
+    assert _extract_model_id([(b"X-LiteLLM-Model-Id", b"deploy-9")]) == "deploy-9"
+
+
+def test_extract_model_id_absent():
+    assert _extract_model_id([(b"content-type", b"application/json")]) is None
+
+
+# ── Middleware recording behaviour ────────────────────────────────────────────
+
+
+def test_records_once_on_2xx_llm_with_model_id():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder, status_code=200, model_id="deploy-7")).post("/v1/chat/completions")
+    assert recorder.calls == [
+        {"category": BillableCategory.LLM, "route": "/chat/completions", "status_code": 200, "model_id": "deploy-7"}
+    ]
+
+
+def test_records_mcp_category():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder)).post("/github/mcp")
+    assert len(recorder.calls) == 1 and recorder.calls[0]["category"] == BillableCategory.MCP
+
+
+def test_records_a2a_category():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder)).post("/a2a/agent-1/message/send")
+    assert len(recorder.calls) == 1 and recorder.calls[0]["category"] == BillableCategory.A2A
+
+
+def test_does_not_record_mcp_management_read():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder)).get("/v1/mcp/tools")
+    assert recorder.calls == []
+
+
+def test_does_not_record_mcp_management_write():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder)).post("/v1/mcp/server")
+    assert recorder.calls == []
+
+
+def test_does_not_record_a2a_discovery():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder)).post("/v1/a2a/discover")
+    assert recorder.calls == []
+
+
+def test_does_not_record_on_4xx():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder, status_code=404)).post("/v1/chat/completions")
+    assert recorder.calls == []
+
+
+def test_does_not_record_on_5xx():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder, status_code=503)).post("/v1/chat/completions")
+    assert recorder.calls == []
+
+
+def test_does_not_record_non_billable_path():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder, status_code=200)).get("/health")
+    assert recorder.calls == []
+
+
+def test_no_model_id_when_header_absent():
+    recorder = FakeRecorder()
+    TestClient(_make_app(recorder, status_code=200, model_id=None)).post("/github/mcp")
+    assert recorder.calls[0]["model_id"] is None
+
+
+def test_passthrough_when_recorder_is_none():
+    """Non-enterprise: middleware records nothing and does not break the response."""
+    response = TestClient(_make_app(None, status_code=200)).post("/v1/chat/completions")
+    assert response.status_code == 200
+
+
+def test_record_raising_does_not_fail_the_request():
+    """A broken exporter must never surface to the client: the response was
+    already served when record() runs, so exceptions are swallowed and logged."""
+
+    class ExplodingRecorder:
+        def record(self, *, category, route, status_code, model_id):
+            raise RuntimeError("exporter down")
+
+    app = _make_app(None, status_code=200)
+    app.user_middleware.clear()
+    app.add_middleware(BillableRequestMetricsMiddleware, recorder=ExplodingRecorder())
+    response = TestClient(app).post("/v1/chat/completions")
+    assert response.status_code == 200
+
+
+def test_non_http_scope_is_ignored():
+    recorder = FakeRecorder()
+
+    class _Inner:
+        async def __call__(self, scope, receive, send):
+            return None
+
+    mw = BillableRequestMetricsMiddleware(_Inner(), recorder=recorder)
+    asyncio.run(mw({"type": "lifespan"}, None, None))  # type: ignore[arg-type]
+    assert recorder.calls == []
+
+
+# ── lazy recorder factory ─────────────────────────────────────────────────────
+
+
+def test_recorder_factory_not_called_at_init():
+    """The factory must run on the first request, not at middleware construction:
+    building at import time captured recorder=None before the YAML config's
+    environment_variables loaded the license and cert env vars."""
+    calls = []
+
+    def factory():
+        calls.append(1)
+        return FakeRecorder()
+
+    class _Inner:
+        async def __call__(self, scope, receive, send):
+            return None
+
+    BillableRequestMetricsMiddleware(_Inner(), recorder_factory=factory)
+    assert calls == []
+
+
+def test_recorder_factory_resolved_once_on_first_request():
+    recorder = FakeRecorder()
+    calls = []
+
+    def factory():
+        calls.append(1)
+        return recorder
+
+    client = TestClient(_make_app_with_factory(factory, status_code=200))
+    client.post("/v1/chat/completions")
+    client.post("/v1/chat/completions")
+    assert calls == [1]
+    assert len(recorder.calls) == 2
+
+
+def test_recorder_factory_returning_none_is_cached():
+    calls = []
+
+    def factory():
+        calls.append(1)
+        return None
+
+    client = TestClient(_make_app_with_factory(factory, status_code=200))
+    assert client.post("/v1/chat/completions").status_code == 200
+    assert client.post("/v1/chat/completions").status_code == 200
+    assert calls == [1]
+
+
+def test_recorder_factory_resolved_once_under_concurrency():
+    """Concurrent first requests must not each build a recorder: every extra
+    build leaks a MeterProvider and its background exporter thread."""
+    calls = []
+    release = threading.Event()
+
+    def slow_factory():
+        calls.append(1)
+        release.wait(timeout=2)
+        return FakeRecorder()
+
+    class _Inner:
+        async def __call__(self, scope, receive, send):
+            return None
+
+    mw = BillableRequestMetricsMiddleware(_Inner(), recorder_factory=slow_factory)
+    threads = [threading.Thread(target=mw._resolve_recorder) for _ in range(8)]
+    for t in threads:
+        t.start()
+    release.set()
+    for t in threads:
+        t.join(timeout=5)
+    assert calls == [1]
+
+
+def _make_app_with_factory(factory, status_code: int) -> Starlette:
+    async def handler(request: Request) -> Response:
+        return JSONResponse({}, status_code=status_code)
+
+    app = Starlette(routes=[Route("/v1/chat/completions", handler, methods=["POST"])])
+    app.add_middleware(BillableRequestMetricsMiddleware, recorder_factory=factory)
+    return app
+
+
+# ── Shutdown ordering ───────────────────────────────────────────────────────
+
+
+def test_record_runs_before_request_leaves_the_in_flight_tracker():
+    """
+    The count is recorded after the inner app returns. If this middleware sat
+    outside InFlightRequestsMiddleware, a request could be seen as drained while
+    its record() had not run, letting proxy_shutdown_event flush and stop the
+    exporter underneath it. Nested inside, the in-flight count still covers it.
+    """
+    observed: List[int] = []
+
+    class _CountingRecorder:
+        def record(self, *, category: BillableCategory, route: str, status_code: int, model_id: Optional[str]) -> None:
+            observed.append(InFlightRequestsMiddleware.get_count())
+
+    async def inner(scope, receive, send) -> None:
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"{}"})
+
+    stack = InFlightRequestsMiddleware(BillableRequestMetricsMiddleware(inner, recorder=_CountingRecorder()))
+    assert TestClient(stack).post("/v1/chat/completions").status_code == 200
+
+    assert observed == [1]
+    assert InFlightRequestsMiddleware.get_count() == 0
+
+
+def test_billable_middleware_is_registered_inside_the_in_flight_tracker():
+    """Starlette makes the last-added middleware outermost, so the in-flight
+    tracker must be registered after the billing middleware to wrap it."""
+    from litellm.proxy.proxy_server import app as proxy_app
+
+    classes = [middleware.cls for middleware in proxy_app.user_middleware]
+    assert classes.index(InFlightRequestsMiddleware) < classes.index(BillableRequestMetricsMiddleware)
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@ -13810,6 +13810,7 @@ export interface paths {
         *     - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team.
         *     - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit per model for this team. Example: {"gpt-4": 100, "gpt-3.5-turbo": 200}
         *     - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit per model for this team. Example: {"gpt-4": 10000, "gpt-3.5-turbo": 20000}
+         *     - mcp_rpm_limit: Optional[Dict[str, int]] - Per-MCP-server RPM limit for this team, keyed by MCP server name (alias if set, else the configured name). Example: {"github": 100, "slack": 200}. Applied across all keys for this team.
         *     Example - update team TPM Limit
         *     - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint.
         *     - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview)