Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_fix/auth-module

# Conflicts:
#	uv.lock

.circleci/config.yml

@@ -1029,6 +1029,8 @@ jobs:
       - *python312_image
     working_directory: ~/project
     resource_class: large
+    environment:
+      REQUEST_TIMEOUT: "180"
 
     steps:
       - checkout
@@ -1058,7 +1060,8 @@ jobs:
                 -v -x \
                 --junitxml=test-results/junit.xml \
                 --durations=5 \
-                -n 8"
+                -n 8 \
+                --reruns 1 --only-rerun Timeout"
           no_output_timeout: 15m
 
       # Store test results
@@ -1610,14 +1613,14 @@ jobs:
       - run:
           name: Run helm lint
           command: |
-            helm lint ./deploy/charts/litellm-helm
+            helm lint ./helm/litellm-helm
 
       # Run helm tests
       - run:
           name: Run helm tests
           command: |
             IMAGE_TAG=${CIRCLE_SHA1:-ci}
-            helm install litellm ./deploy/charts/litellm-helm -f ./deploy/charts/litellm-helm/ci/test-values.yaml \
+            helm install litellm ./helm/litellm-helm -f ./helm/litellm-helm/ci/test-values.yaml \
               --set image.repository=litellm-ci \
               --set image.tag=${IMAGE_TAG} \
               --set image.pullPolicy=Never

.github/workflows/codspeed.yml

@@ -4,9 +4,11 @@ on:
   push:
     branches:
       - main
+      - litellm_internal_staging
   pull_request:
     branches:
       - main
+      - litellm_internal_staging
   # Allow CodSpeed to trigger backtest performance analysis
   # in order to generate initial data
   workflow_dispatch:
@@ -22,7 +24,7 @@ concurrency:
 jobs:
   benchmarks:
     runs-on: ubuntu-24.04
-    timeout-minutes: 15
+    timeout-minutes: 60
 
     steps:
       - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0

.github/workflows/helm_unit_test.yml

@@ -39,5 +39,5 @@ jobs:
 
       - name: Run unit tests
         run: |
-          helm unittest -f 'tests/*.yaml' deploy/charts/litellm-helm
+          helm unittest -f 'tests/*.yaml' helm/litellm-helm
           helm unittest -f 'tests/*.yaml' helm/litellm

.github/workflows/test-terraform-provider.yml

@@ -0,0 +1,113 @@
+name: Terraform Provider
+
+on:
+  push:
+    paths:
+      - "terraform/provider/**"
+      - ".github/workflows/test-terraform-provider.yml"
+  pull_request:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_staging
+      - "litellm_**"
+    paths:
+      - "terraform/provider/**"
+      - "litellm/proxy/**"
+      - ".github/workflows/test-terraform-provider.yml"
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  provider-checks:
+    name: gofmt, vet, build, test
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    defaults:
+      run:
+        working-directory: terraform/provider
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - uses: actions/setup-go@7a3fe6cf4cb3a834922a1244abfce67bcef6a0c5 # v6.2.0
+        with:
+          go-version-file: terraform/provider/go.mod
+          cache: true
+          cache-dependency-path: terraform/provider/go.sum
+
+      - name: gofmt
+        run: |
+          UNFORMATTED=$(gofmt -l .)
+          if [ -n "${UNFORMATTED}" ]; then
+            echo "::error::gofmt required for: ${UNFORMATTED}"
+            exit 1
+          fi
+
+      - name: go vet
+        run: go vet ./...
+
+      - name: Build
+        run: go build ./...
+
+      - name: Test
+        run: go test -timeout 120s ./...
+
+  endpoint-drift:
+    name: Provider endpoints vs proxy OpenAPI schema
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.12"
+
+      - name: Set up uv
+        uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7
+        with:
+          version: "0.10.9"
+
+      - name: Cache uv dependencies
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+        with:
+          path: |
+            ~/.cache/uv
+            .venv
+          key: ${{ runner.os }}-uv-${{ hashFiles('uv.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-uv-
+
+      - name: Install dependencies
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra google --extra proxy --extra semantic-router
+
+      - name: Generate Prisma client
+        env:
+          PRISMA_BINARY_CACHE_DIR: ${{ runner.temp }}/prisma-cache
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Generate proxy OpenAPI schema
+        run: |
+          uv run --no-sync python terraform/provider/tools/dump_openapi.py "${RUNNER_TEMP}/openapi.json"
+
+      - uses: actions/setup-go@7a3fe6cf4cb3a834922a1244abfce67bcef6a0c5 # v6.2.0
+        with:
+          go-version-file: terraform/provider/go.mod
+          cache: true
+          cache-dependency-path: terraform/provider/go.sum
+
+      - name: Audit provider endpoints against the schema
+        working-directory: terraform/provider
+        run: go run ./tools/endpointaudit -provider-dir ./litellm -spec "${RUNNER_TEMP}/openapi.json"

.github/workflows/test-unit-core-utils.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-documentation.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-enterprise-routing.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-integrations.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-llm-providers.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-misc.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-proxy-auth.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-proxy-db.yml

@@ -5,6 +5,10 @@ on:
     branches:
       - main
       - litellm_internal_staging
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-proxy-endpoints.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
   workflow_dispatch:
 
 permissions:

.github/workflows/test-unit-proxy-infra.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-proxy-legacy.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test-unit-responses-caching-types.yml

@@ -7,6 +7,10 @@ on:
       - litellm_internal_staging
       - litellm_oss_staging
       - "litellm_**"
+    paths-ignore:
+      - "ui/**"
+      - "**.md"
+      - "**.mdx"
 
 permissions:
   contents: read

.github/workflows/test_server_root_path.yml

@@ -16,6 +16,7 @@ jobs:
     timeout-minutes: 30
 
     strategy:
+      fail-fast: false
       matrix:
         root_path: ["/api/v1", "/llmproxy"]
 
@@ -108,8 +109,26 @@ jobs:
       - name: Install UI deps and Chromium
         working-directory: ui/litellm-dashboard
         run: |
-          npm ci
-          npx playwright install --with-deps chromium
+          retry() {
+            local attempt=1
+            local max_attempts=4
+            until "$@"; do
+              if [ "$attempt" -ge "$max_attempts" ]; then
+                echo "Command failed after $attempt attempts: $*"
+                return 1
+              fi
+              echo "Attempt $attempt failed: $*. Retrying in $((attempt * 15))s..."
+              sleep $((attempt * 15))
+              attempt=$((attempt + 1))
+            done
+          }
+
+          npm config set fetch-retries 5
+          npm config set fetch-retry-mintimeout 20000
+          npm config set fetch-retry-maxtimeout 120000
+
+          retry npm ci
+          retry npx playwright install --with-deps chromium
 
       - name: Run SERVER_ROOT_PATH redirect e2e
         working-directory: ui/litellm-dashboard

.gitignore

@@ -52,9 +52,8 @@ ui/litellm-dashboard/node_modules
 ui/litellm-dashboard/next-env.d.ts
 ui/litellm-dashboard/package.json
 ui/litellm-dashboard/package-lock.json
-deploy/charts/litellm/*.tgz
-deploy/charts/litellm/charts/*
-deploy/charts/*.tgz
+helm/litellm-helm/*.tgz
+helm/*.tgz
 litellm/proxy/vertex_key.json
 **/.vim/
 **/node_modules

CLAUDE.md

@@ -21,11 +21,11 @@ End-to-end tests belong in `tests/e2e/` and must follow the harness conventions
 
 When creating PRs, don't set base to `main`. `litellm_internal_staging` serves that purpose
 
-When writing a PR body, treat the comments and imperative instructions inside @.github/pull_request_template.md as rules to follow, not just layout
+When writing a PR body, treat the comments and imperative instructions inside @.github/pull_request_template.md as rules to follow, not just layout. Agent harnesses may strip HTML comments from copies of that file injected into context, so read .github/pull_request_template.md from disk before writing a PR body to make sure you see every comment rule
 
 If you're resolving a linear ticket, in the "## Linear ticket" section of the PR, say "Resolves LIT-1234", replacing "LIT-1234" with the actual ticket id that you're resolving. If you don't have the ticket id, don't make one up or search for it. Just leave the section blank
 
-Never use `pytest` commands or the like as "Screenshots / Proof of Fix". We prefer curl'ing a live proxy instance running on localhost:4000 (I like to run it with `python litellm/proxy/proxy_cli.py --config litellm/proxy/dev_config.yaml --detailed_debug --reload --use_v2_migration_resolver 2>&1 | tee litellm.log`) and showing both the command run and the output. Also, it should hit real LLM provider APIs, not mocks, and cost real $$$ because that is the most realistic test. The proof of fix should be exactly what the end user / customer would see / do. The run logs in PR #27703 is a prime example of how to do it (not a huge fan of using a python test script that future me and the team will have no visibility into; I prefer just curl commands or a short list of bash commands (e.g., using `for`)). If it's a UI thing, just tell me which URLs to go to (e.g., http://localhost:4000/ui/?page=logs), where to click, what fields to fill out, etc. along with the other commands to run in an ordered list, and I'll do it myself and post the screenshots after you make the PR
+Never use `pytest` commands or the like as "Screenshots / Proof of Fix". We prefer curl'ing a live proxy instance running on localhost:4000 (I like to run it with `python litellm/proxy/proxy_cli.py --config litellm/proxy/dev_config.yaml --detailed_debug --reload --use_v2_migration_resolver 2>&1 | tee litellm.log`; the Admin UI dev server is `npm run dev` in `ui/litellm-dashboard`, served on port 3000) and showing both the command run and the output. Also, it should hit real LLM provider APIs, not mocks, and cost real $$$ because that is the most realistic test. The proof of fix should be exactly what the end user / customer would see / do. The run logs in PR #27703 is a prime example of how to do it (not a huge fan of using a python test script that future me and the team will have no visibility into; I prefer just curl commands or a short list of bash commands (e.g., using `for`)). If it's a UI thing, just tell me which URLs to go to (e.g., http://localhost:4000/ui/?page=logs), where to click, what fields to fill out, etc. along with the other commands to run in an ordered list, and I'll do it myself and post the screenshots after you make the PR
 
 If you ever make public-facing PR descriptions, comments, issues, commit messages, etc., always follow these guidelines to sound less AI-y:
 - don't use emojis

Makefile

@@ -265,7 +265,7 @@ test-integration: install-test-deps
 	$(UV_RUN) pytest tests/ -k "not test_litellm"
 
 test-unit-helm: install-helm-unittest
-	helm unittest -f 'tests/*.yaml' deploy/charts/litellm-helm
+	helm unittest -f 'tests/*.yaml' helm/litellm-helm
 
 # LLM Translation testing targets
 test-llm-translation: install-test-deps

deploy/azure_resource_manager/azure_marketplace/createUiDefinition.json

@@ -1,15 +0,0 @@
-{
-    "$schema": "https://schema.management.azure.com/schemas/0.1.2-preview/CreateUIDefinition.MultiVm.json#",
-    "handler": "Microsoft.Azure.CreateUIDef",
-    "version": "0.1.2-preview",
-    "parameters": {
-        "config": {
-            "isWizard": false,
-            "basics": { }
-        },
-        "basics": [ ],
-        "steps": [ ],
-        "outputs": { },
-        "resourceTypes": [ ]
-    }
-}

deploy/azure_resource_manager/azure_marketplace/mainTemplate.json

@@ -1,63 +0,0 @@
-{
-    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
-    "contentVersion": "1.0.0.0",
-    "parameters": {
-      "imageName": {
-        "type": "string",
-        "defaultValue": "ghcr.io/berriai/litellm:main-latest"
-      },
-      "containerName": {
-        "type": "string",
-        "defaultValue": "litellm-container"
-      },
-      "dnsLabelName": {
-        "type": "string",
-        "defaultValue": "litellm"
-      },
-      "portNumber": {
-        "type": "int",
-        "defaultValue": 4000
-      }
-    },
-    "resources": [
-      {
-        "type": "Microsoft.ContainerInstance/containerGroups",
-        "apiVersion": "2021-03-01",
-        "name": "[parameters('containerName')]",
-        "location": "[resourceGroup().location]",
-        "properties": {
-          "containers": [
-            {
-              "name": "[parameters('containerName')]",
-              "properties": {
-                "image": "[parameters('imageName')]",
-                "resources": {
-                  "requests": {
-                    "cpu": 1,
-                    "memoryInGB": 2
-                  }
-                },
-                "ports": [
-                  {
-                    "port": "[parameters('portNumber')]"
-                  }
-                ]
-              }
-            }
-          ],
-          "osType": "Linux",
-          "restartPolicy": "Always",
-          "ipAddress": {
-            "type": "Public",
-            "ports": [
-              {
-                "protocol": "tcp",
-                "port": "[parameters('portNumber')]"
-              }
-            ],
-            "dnsNameLabel": "[parameters('dnsLabelName')]"
-          }
-        }
-      }
-    ]
-  }

deploy/azure_resource_manager/main.bicep

@@ -1,42 +0,0 @@
-param imageName string = 'ghcr.io/berriai/litellm:main-latest'
-param containerName string = 'litellm-container'
-param dnsLabelName string = 'litellm'
-param portNumber int = 4000
-
-resource containerGroupName 'Microsoft.ContainerInstance/containerGroups@2021-03-01' = {
-  name: containerName
-  location: resourceGroup().location
-  properties: {
-    containers: [
-      {
-        name: containerName
-        properties: {
-          image: imageName
-          resources: {
-            requests: {
-              cpu: 1
-              memoryInGB: 2
-            }
-          }
-          ports: [
-            {
-              port: portNumber
-            }
-          ]
-        }
-      }
-    ]
-    osType: 'Linux'
-    restartPolicy: 'Always'
-    ipAddress: {
-      type: 'Public'
-      ports: [
-        {
-          protocol: 'tcp'
-          port: portNumber
-        }
-      ]
-      dnsNameLabel: dnsLabelName
-    }
-  }
-}

enterprise/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "litellm-enterprise"
-version = "0.1.47"
+version = "0.1.48"
 description = "Package for LiteLLM Enterprise features"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -26,7 +26,7 @@ required-version = ">=0.10.9"
 module-root = ""
 
 [tool.commitizen]
-version = "0.1.47"
+version = "0.1.48"
 version_files = [
     "pyproject.toml:^version",
     "../pyproject.toml:litellm-enterprise==",

gateway/main.py

@@ -25,17 +25,25 @@ DatabaseURLSettings.from_env().apply_to_env()
 
 from litellm.proxy.proxy_server import app
 
-from gateway.routes.allowlist import GATEWAY_EXACT_PATHS, GATEWAY_PATH_PREFIXES
+from gateway.routes.allowlist import (
+    GATEWAY_EXACT_PATHS,
+    GATEWAY_MOUNT_PATHS,
+    GATEWAY_PATH_PREFIXES,
+)
 
 
 def _is_gateway_route(route) -> bool:
-    """Keep the route on the gateway if its path is in the LLM data-plane surface."""
+    """Keep the route on the gateway if its path is in the LLM data-plane surface.
+
+    Prometheus registers /metrics as a Mount (``app.mount("/metrics", make_asgi_app())``),
+    so Mounts are matched against GATEWAY_MOUNT_PATHS instead of being dropped with
+    the UI static mounts.
+    """
     path = getattr(route, "path", None)
     if path is None:
         return False
     if isinstance(route, Mount):
-        # Gateway never serves the static UI or its asset bundles.
-        return False
+        return path in GATEWAY_MOUNT_PATHS
     if path in GATEWAY_EXACT_PATHS:
         return True
     return any(path.startswith(prefix) for prefix in GATEWAY_PATH_PREFIXES)

gateway/routes/allowlist.py

@@ -106,7 +106,7 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
     # Health & ops
     "/health",
     "/metrics",
-    "/watsonx"
+    "/watsonx",
 )
 
 GATEWAY_EXACT_PATHS: frozenset[str] = frozenset(
@@ -120,3 +120,9 @@ GATEWAY_EXACT_PATHS: frozenset[str] = frozenset(
         "/test",
     }
 )
+
+GATEWAY_MOUNT_PATHS: frozenset[str] = frozenset(
+    {
+        "/metrics",
+    }
+)

litellm-proxy-extras/litellm_proxy_extras/migrations/20260630120000_add_token_exchange_to_mcp_servers/migration.sql

@@ -0,0 +1,8 @@
+-- Timestamp sorts before some already-applied migrations; this is safe: the
+-- runner is `prisma migrate deploy`, which applies every pending migration
+-- regardless of name order (utils.py has an informational check for exactly
+-- this), and IF NOT EXISTS keeps a re-apply idempotent.
+-- AlterTable
+ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN IF NOT EXISTS "token_exchange_endpoint" TEXT;
+ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN IF NOT EXISTS "audience" TEXT;
+ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN IF NOT EXISTS "subject_token_type" TEXT;

litellm-proxy-extras/litellm_proxy_extras/migrations/20260703120000_add_token_exchange_profile_to_mcp_servers/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN IF NOT EXISTS "token_exchange_profile" TEXT;

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -329,6 +329,12 @@ model LiteLLM_MCPServerTable {
   token_url          String?
   registration_url   String?
   oauth2_flow        String?
+  token_exchange_endpoint String?
+  // Named for the RFC 8693 "audience" token-exchange request parameter (that flow only).
+  // RFC 8707 resource indicators are a separate concept, named "resource" in the v2 egress types.
+  audience           String?
+  subject_token_type String?
+  token_exchange_profile String?
   allow_all_keys Boolean @default(false)
   available_on_public_internet Boolean @default(true)
   delegate_auth_to_upstream Boolean @default(false)

litellm/__init__.py

@@ -379,6 +379,7 @@ budget_duration: Optional[str] = (
     None  # proxy only - resets budget after fixed duration. You can set duration as seconds ("30s"), minutes ("30m"), hours ("30h"), days ("30d").
 )
 default_soft_budget: float = DEFAULT_SOFT_BUDGET  # by default all litellm proxy keys have a soft budget of 50.0
+budget_exceeded_throttle_percentage: Optional[float] = None
 forward_traceparent_to_llm_provider: bool = False
 
 

litellm/a2a_protocol/streaming_iterator.py

@@ -11,7 +11,6 @@ from litellm._logging import verbose_logger
 from litellm.a2a_protocol.cost_calculator import A2ACostCalculator
 from litellm.a2a_protocol.utils import A2ARequestUtils
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
-from litellm.litellm_core_utils.thread_pool_executor import executor
 
 if TYPE_CHECKING:
     from a2a.types import SendStreamingMessageRequest, SendStreamingMessageResponse
@@ -128,22 +127,15 @@ class A2AStreamingIterator:
 
             # Call success handlers - they will build standard_logging_object
             asyncio.create_task(
-                self.logging_obj.async_success_handler(
-                    result=result,
+                self.logging_obj.dispatch_success_handlers(
+                    result,
                     start_time=self.start_time,
                     end_time=end_time,
                     cache_hit=None,
+                    prefer_async_handlers=True,
                 )
             )
 
-            executor.submit(
-                self.logging_obj.success_handler,
-                result=result,
-                cache_hit=None,
-                start_time=self.start_time,
-                end_time=end_time,
-            )
-
             verbose_logger.info(
                 f"A2A streaming completed: prompt_tokens={prompt_tokens}, "
                 f"completion_tokens={completion_tokens}, total_tokens={total_tokens}, "

litellm/constants.py

@@ -1504,6 +1504,7 @@ LITELLM_SETTINGS_SAFE_DB_OVERRIDES = [
     "public_model_groups_links",
     "cost_discount_config",
     "cost_margin_config",
+    "budget_exceeded_throttle_percentage",
 ]
 SPECIAL_LITELLM_AUTH_TOKEN = ["ui-token"]
 DEFAULT_MANAGEMENT_OBJECT_IN_MEMORY_CACHE_TTL = int(os.getenv("DEFAULT_MANAGEMENT_OBJECT_IN_MEMORY_CACHE_TTL", 60))

litellm/integrations/otel/logger.py

@@ -368,7 +368,11 @@ class OpenTelemetryV2(CustomLogger):
                 # it (named provisionally) so it isn't leaked as an open span.
                 carrier.span.end(end_time=to_ns(end_time))
             return None
-        data = LLMCallSpanData.from_standard_logging_payload(payload, capture_content=self.config.capture_span_content)
+        data = LLMCallSpanData.from_standard_logging_payload(
+            payload,
+            capture_content=self.config.capture_span_content,
+            time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
+        )
         end_time_ns = to_ns(end_time)
         if carrier.span is not None:
             # Born at the boundary: stamp attributes from the typed payload, set

litellm/integrations/otel/mappers/genai.py

@@ -55,6 +55,7 @@ class GenAIMapper:
         GenAI.RESPONSE_MODEL: lambda d: d.response_model,
         GenAI.RESPONSE_ID: lambda d: d.response_id,
         GenAI.RESPONSE_FINISH_REASONS: lambda d: list(d.finish_reasons) if d.finish_reasons else None,
+        GenAI.RESPONSE_TIME_TO_FIRST_CHUNK: lambda d: d.time_to_first_chunk_seconds,
         GenAI.USAGE_INPUT_TOKENS: lambda d: d.usage.input_tokens,
         GenAI.USAGE_OUTPUT_TOKENS: lambda d: d.usage.output_tokens,
         Error.TYPE: lambda d: d.error.error_type if d.error else None,

litellm/integrations/otel/model/metadata.py

@@ -41,7 +41,7 @@ from typing import TYPE_CHECKING, Any, Mapping, cast
 
 from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL
 from litellm.integrations.otel.model.semconv import resolve_operation
-from litellm.integrations.otel.model.utils import as_str
+from litellm.integrations.otel.model.utils import as_str, to_seconds
 
 if TYPE_CHECKING:
     from litellm.types.utils import StandardLoggingPayload
@@ -201,6 +201,7 @@ class LLMCallEvent:
     # span is renamed from the typed payload at close (``finish_span``); this only
     # needs to be reasonable for a span that never gets closed (a leak).
     provisional_span_name: str
+    time_to_first_chunk_seconds: float | None
 
     @classmethod
     def from_dict(cls, kwargs: Mapping[str, Any]) -> "LLMCallEvent":
@@ -214,9 +215,25 @@ class LLMCallEvent:
             dynamic_params=kwargs.get("standard_callback_dynamic_params"),
             is_no_upstream_call=bool(kwargs.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL)),
             provisional_span_name=f"{operation.value} {model}".strip(),
+            time_to_first_chunk_seconds=time_to_first_chunk_seconds(kwargs),
         )
 
 
+def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
+    """Seconds from the upstream request being issued (``api_call_start_time``)
+    to the first streamed chunk (``completion_start_time``); ``None`` for
+    non-streaming calls, where ``completion_start_time`` is backfilled with the
+    end time and would not measure first-chunk latency."""
+    optional_params = cast(Mapping[str, Any], kwargs.get("optional_params") or {})
+    if not optional_params.get("stream"):
+        return None
+    api_call_start = to_seconds(kwargs.get("api_call_start_time"))
+    completion_start = to_seconds(kwargs.get("completion_start_time"))
+    if api_call_start is None or completion_start is None:
+        return None
+    return completion_start - api_call_start
+
+
 def _call_id(payload: "StandardLoggingPayload | None", kwargs: Mapping[str, Any]) -> str | None:
     """The call id from the payload (when closed) or the bare kwargs (at pre_call)."""
     if payload is not None:

litellm/integrations/otel/model/payloads.py

@@ -305,10 +305,14 @@ class LLMCallSpanData:
     messages_in: tuple[Mapping[str, object], ...] = ()
     choices_out: tuple[Mapping[str, object], ...] = ()
     system_fingerprint: str | None = None
+    time_to_first_chunk_seconds: float | None = None
 
     @classmethod
     def from_standard_logging_payload(
-        cls, payload: "StandardLoggingPayload", capture_content: bool = False
+        cls,
+        payload: "StandardLoggingPayload",
+        capture_content: bool = False,
+        time_to_first_chunk_seconds: float | None = None,
     ) -> "LLMCallSpanData":
         params = cast(Mapping[str, object], payload.get("model_parameters") or {})
         # The single parse of the request's metadata — the request-vs-provider
@@ -349,6 +353,7 @@ class LLMCallSpanData:
             messages_in=_dicts(payload.get("messages")) if capture_content else (),
             choices_out=choices_out if capture_content else (),
             system_fingerprint=as_str(response.get("system_fingerprint")),
+            time_to_first_chunk_seconds=time_to_first_chunk_seconds,
         )
 
 

litellm/integrations/otel/model/semconv.py

@@ -69,6 +69,7 @@ class GenAI:
     RESPONSE_ID: Final = "gen_ai.response.id"
     RESPONSE_MODEL: Final = "gen_ai.response.model"
     RESPONSE_FINISH_REASONS: Final = "gen_ai.response.finish_reasons"
+    RESPONSE_TIME_TO_FIRST_CHUNK: Final = "gen_ai.response.time_to_first_chunk"
     # usage
     USAGE_INPUT_TOKENS: Final = "gen_ai.usage.input_tokens"
     USAGE_OUTPUT_TOKENS: Final = "gen_ai.usage.output_tokens"

litellm/integrations/otel/plumbing/metrics.py

@@ -21,6 +21,7 @@ from litellm.integrations.opentelemetry import (
     _build_metric_attribute_filter,
     _resolve_metric_attribute_filter,
 )
+from litellm.integrations.otel.model.metadata import time_to_first_chunk_seconds
 from litellm.integrations.otel.model.semconv import Metric, resolve_operation
 from litellm.integrations.otel.model.utils import to_seconds
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
@@ -181,13 +182,10 @@ class GenAIMetricRecorder:
         self._metrics.token_usage.record(usage.get("completion_tokens", 0), attributes=out_attrs)
 
     def _record_time_to_first_token(self, kwargs: Mapping[str, Any], common_attrs: dict) -> None:
-        if not kwargs.get("optional_params", {}).get("stream", False):
+        time_to_first_chunk = time_to_first_chunk_seconds(kwargs)
+        if time_to_first_chunk is None:
             return
-        api_call_start = to_seconds(kwargs.get("api_call_start_time"))
-        completion_start = to_seconds(kwargs.get("completion_start_time"))
-        if api_call_start is None or completion_start is None:
-            return
-        self._metrics.time_to_first_token.record(completion_start - api_call_start, attributes=common_attrs)
+        self._metrics.time_to_first_token.record(time_to_first_chunk, attributes=common_attrs)
 
     def _record_time_per_output_token(
         self,

litellm/interactions/streaming_iterator.py

@@ -174,22 +174,15 @@ class InteractionsAPIStreamingIterator(BaseInteractionsAPIStreamingIterator):
         logging_response = copy.deepcopy(self.completed_response)
 
         asyncio.create_task(
-            self.logging_obj.async_success_handler(
-                result=logging_response,
+            self.logging_obj.dispatch_success_handlers(
+                logging_response,
                 start_time=self.start_time,
                 end_time=datetime.now(),
                 cache_hit=None,
+                prefer_async_handlers=True,
             )
         )
 
-        executor.submit(
-            self.logging_obj.success_handler,
-            result=logging_response,
-            cache_hit=None,
-            start_time=self.start_time,
-            end_time=datetime.now(),
-        )
-
 
 class SyncInteractionsAPIStreamingIterator(BaseInteractionsAPIStreamingIterator):
     """

litellm/litellm_core_utils/realtime_streaming.py

@@ -1,5 +1,4 @@
 import asyncio
-import concurrent.futures
 import json
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol, Union, cast
 
@@ -25,9 +24,6 @@ if TYPE_CHECKING:
 else:
     CLIENT_CONNECTION_CLASS = Any
 
-# Create a thread pool with a maximum of 10 threads
-executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)
-
 
 class RealtimeEventNormalizer(Protocol):
     def should_drop(self, event: object) -> bool: ...
@@ -315,13 +311,12 @@ class RealTimeStreaming:
             if self.session_tools or self.tool_calls:
                 self.logging_obj.model_call_details["realtime_tools"] = self.session_tools
                 self.logging_obj.model_call_details["realtime_tool_calls"] = self.tool_calls
-            ## ASYNC LOGGING
             # Route through the bounded logging worker (per-coroutine timeout +
             # concurrency cap) instead of a bare create_task, so a slow callback
             # can't leave suspended tasks pinning each call's response in memory.
-            GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(self.logging_obj.async_success_handler(self.messages))
-            ## SYNC LOGGING
-            executor.submit(self.logging_obj.success_handler(self.messages))
+            GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
+                self.logging_obj.dispatch_success_handlers(self.messages, prefer_async_handlers=True)
+            )
 
     async def _send_to_backend(self, message: str) -> bool:
         """Send a message to the backend WebSocket.

litellm/llms/bedrock/base_aws_llm.py

@@ -50,6 +50,8 @@ _STS_REGION_FROM_ENDPOINT_PATTERN = re.compile(
     r"(?:^|\.)sts(?:-fips)?\.([a-z0-9-]+)\.(?:amazonaws\.com(?:\.cn)?|vpce\.amazonaws\.com)"
 )
 
+SIGV4_COMPUTED_HEADERS = frozenset({"authorization", "x-amz-date", "x-amz-security-token", "date"})
+
 
 class Boto3CredentialsInfo(BaseModel):
     credentials: Credentials
@@ -1400,11 +1402,13 @@ class BaseAWSLLM:
 
             # Add back all original headers (including forwarded ones) after signature calculation
             for header_name, header_value in headers.items():
-                if header_value is not None:
+                if header_value is not None and header_name.lower() not in SIGV4_COMPUTED_HEADERS:
                     request.headers[header_name] = header_value
 
             if (
-                extra_headers is not None and "Authorization" in extra_headers
+                extra_headers is not None
+                and "Authorization" in extra_headers
+                and not extra_headers["Authorization"].startswith("AWS4-HMAC-SHA256")
             ):  # prevent sigv4 from overwriting the auth header
                 request.headers["Authorization"] = extra_headers["Authorization"]
         prepped = request.prepare()
@@ -1527,9 +1531,15 @@ class BaseAWSLLM:
         # Add back original headers after signing. Only headers in SignedHeaders
         # are integrity-protected; forwarded headers (x-forwarded-*) must remain unsigned.
         for header_name, header_value in headers.items():
-            if header_value is not None:
+            if header_value is not None and header_name.lower() not in SIGV4_COMPUTED_HEADERS:
                 request_headers_dict[header_name] = header_value
-        if headers is not None and "Authorization" in headers:  # prevent sigv4 from overwriting the auth header
-            request_headers_dict["Authorization"] = headers["Authorization"]
+        incoming_authorization = next(
+            (value for name, value in headers.items() if name.lower() == "authorization" and value is not None),
+            None,
+        )
+        if incoming_authorization is not None and not incoming_authorization.startswith(
+            "AWS4-HMAC-SHA256"
+        ):  # prevent sigv4 from overwriting the auth header
+            request_headers_dict["Authorization"] = incoming_authorization
 
         return request_headers_dict, request.body

litellm/llms/bedrock/chat/mantle/transformation.py

@@ -7,13 +7,14 @@ The bedrock-mantle endpoint uses the Anthropic Messages API format but is served
 at a different endpoint (bedrock-mantle.{region}.api.aws) with AWS SigV4 auth.
 """
 
-from typing import TYPE_CHECKING, Any, List, Optional
+from typing import TYPE_CHECKING, Any, AsyncIterator, Iterator, List, Optional
 
 from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
     AmazonAnthropicClaudeConfig,
 )
 from litellm.llms.bedrock.common_utils import build_mantle_messages_url
 from litellm.types.llms.openai import AllMessageValues
+from litellm.types.utils import ModelResponse
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
@@ -91,10 +92,14 @@ class AmazonMantleConfig(AmazonAnthropicClaudeConfig):
             litellm_params=litellm_params,
             headers=headers,
         )
-        # The parent strips "model" from the body (Invoke API puts it in URL).
-        # The mantle endpoint (Messages API) requires "model" in the body.
-        request["model"] = model_id
-        return request
+        # The parent strips "model" and "stream" from the body (Invoke API puts
+        # the model in the URL and streams via a dedicated endpoint). The mantle
+        # endpoint (Messages API) requires both in the body.
+        return self._restore_mantle_body_fields(
+            request=request,
+            model_id=model_id,
+            optional_params=optional_params,
+        )
 
     async def async_transform_request(
         self,
@@ -114,5 +119,31 @@ class AmazonMantleConfig(AmazonAnthropicClaudeConfig):
             headers=headers,
         )
         await self._async_convert_document_url_sources_to_base64(request)
-        request["model"] = model_id
-        return request
+        return self._restore_mantle_body_fields(
+            request=request,
+            model_id=model_id,
+            optional_params=optional_params,
+        )
+
+    @staticmethod
+    def _restore_mantle_body_fields(request: dict, model_id: str, optional_params: dict) -> dict:
+        stream_fields: dict = {"stream": True} if optional_params.get("stream") is True else {}
+        return {**request, "model": model_id, **stream_fields}
+
+    @property
+    def has_custom_stream_wrapper(self) -> bool:
+        return False
+
+    def get_model_response_iterator(
+        self,
+        streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
+        sync_stream: bool,
+        json_mode: Optional[bool] = False,
+    ) -> Any:
+        from litellm.llms.anthropic.chat.handler import ModelResponseIterator
+
+        return ModelResponseIterator(
+            streaming_response=streaming_response,
+            sync_stream=sync_stream,
+            json_mode=json_mode,
+        )

litellm/llms/bedrock/messages/mantle_transformation.py

@@ -6,8 +6,13 @@ AmazonAnthropicClaudeMessagesConfig. Overrides only the URL and model-prefix
 stripping that are specific to the bedrock-mantle endpoint.
 """
 
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, List, Optional, Tuple
 
+import httpx
+
+from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
+    AnthropicMessagesConfig,
+)
 from litellm.llms.bedrock.common_utils import build_mantle_messages_url
 from litellm.llms.bedrock.messages.invoke_transformations.anthropic_claude3_transformation import (
     AmazonAnthropicClaudeMessagesConfig,
@@ -89,8 +94,26 @@ class AmazonMantleMessagesConfig(AmazonAnthropicClaudeMessagesConfig):
             headers=headers,
         )
 
-        # Parent (AmazonAnthropicClaudeMessagesConfig) removes "model" from the
-        # body (Bedrock Invoke puts model in the URL). The mantle endpoint
-        # (Messages API) requires "model" in the request body.
-        request["model"] = model_id
-        return request
+        # Parent (AmazonAnthropicClaudeMessagesConfig) removes "model" and
+        # "stream" from the body (Bedrock Invoke puts the model in the URL and
+        # streams via a dedicated endpoint). The mantle endpoint (Messages API)
+        # requires both in the request body.
+        stream_fields: dict[str, bool] = (
+            {"stream": True} if anthropic_messages_optional_request_params.get("stream") is True else {}
+        )
+        return {**request, "model": model_id, **stream_fields}
+
+    def get_async_streaming_response_iterator(
+        self,
+        model: str,
+        httpx_response: httpx.Response,
+        request_body: dict,
+        litellm_logging_obj: LiteLLMLoggingObj,
+    ) -> AsyncIterator:
+        return AnthropicMessagesConfig.get_async_streaming_response_iterator(
+            self,
+            model=model,
+            httpx_response=httpx_response,
+            request_body=request_body,
+            litellm_logging_obj=litellm_logging_obj,
+        )

litellm/llms/vertex_ai/vertex_llm_base.py

@@ -9,6 +9,7 @@ import json
 import os
 import threading
 from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Tuple
+from urllib.parse import urlparse
 
 import litellm
 from litellm._logging import verbose_logger
@@ -315,6 +316,9 @@ class VertexBase:
             api_base=api_base,
         )
 
+        if partner == VertexPartnerProvider.llama:
+            return default_api_base
+
         if len(default_api_base.split(":")) > 1:
             endpoint = default_api_base.split(":")[-1]
         else:
@@ -615,7 +619,8 @@ class VertexBase:
 
         Handles custom api_base for:
         1. Gemini (Google AI Studio) - constructs /models/{model}:{endpoint}
-        2. Vertex AI with standard proxies - constructs {api_base}:{endpoint}
+        2. Vertex AI with standard proxies - constructs {api_base}:{endpoint};
+           if api_base has no path (bare host), grafts the default vertex URL path onto it
         3. Vertex AI with PSC endpoints - constructs full path structure
            {api_base}/v1/projects/{project}/locations/{location}/endpoints/{model}:{endpoint}
            (only when use_psc_endpoint_format=True)
@@ -660,8 +665,9 @@ class VertexBase:
                         model_for_url,
                         endpoint,
                     )
+                elif urlparse(api_base).path in ("", "/"):
+                    url = api_base.rstrip("/") + urlparse(url).path
                 else:
-                    # Fallback to simple format if we don't have all parameters
                     url = "{}:{}".format(api_base, endpoint)
             if stream is True:
                 url = url + "?alt=sse"

litellm/main.py

@@ -1082,6 +1082,54 @@ def _build_custom_pricing_entry(
     return entry
 
 
+def _get_router_deployment_id(kwargs: dict) -> Optional[str]:
+    for metadata_key in ("litellm_metadata", "metadata"):
+        metadata = kwargs.get(metadata_key) or {}
+        if not isinstance(metadata, dict):
+            continue
+        deployment_model_info = metadata.get("model_info") or {}
+        if not isinstance(deployment_model_info, dict):
+            continue
+        deployment_id = deployment_model_info.get("id")
+        if deployment_id is not None:
+            return str(deployment_id)
+    return None
+
+
+def _register_custom_pricing_for_request(
+    model: str,
+    custom_llm_provider: str,
+    kwargs: dict,
+    model_info: Optional[dict],
+) -> None:
+    """Register per-request custom pricing in litellm.model_cost.
+
+    Router-originated requests (identified by the deployment id the router puts
+    in metadata) get their full pricing registered under that unique id only;
+    the shared ``{provider}/{model}`` key receives the entry with pricing fields
+    stripped, mirroring Router._create_deployment. This keeps one deployment's
+    pricing overrides (e.g. a zero-cost wildcard) from clobbering built-in
+    pricing used by sibling deployments of the same backend model. Direct SDK
+    calls keep the legacy behavior of registering the shared key with pricing.
+    """
+    entry = _build_custom_pricing_entry(
+        custom_llm_provider=custom_llm_provider,
+        kwargs=kwargs,
+        model_info=model_info,
+    )
+    shared_key = f"{custom_llm_provider}/{model}"
+    deployment_id = _get_router_deployment_id(kwargs)
+    if deployment_id is None:
+        litellm.register_model({shared_key: entry})
+        return
+    litellm.register_model(
+        {
+            deployment_id: entry,
+            shared_key: CustomPricingLiteLLMParams.strip_custom_pricing_fields(entry),
+        }
+    )
+
+
 def _complete_azure(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
     _azure_detection_model = ctx._azure_detection_model
     acompletion = ctx.acompletion
@@ -5108,14 +5156,11 @@ def completion(  # type: ignore
         if (
             input_cost_per_token is not None and output_cost_per_token is not None
         ) or input_cost_per_second is not None:
-            litellm.register_model(
-                {
-                    f"{custom_llm_provider}/{model}": _build_custom_pricing_entry(
-                        custom_llm_provider=custom_llm_provider,
-                        kwargs=kwargs,
-                        model_info=model_info,
-                    )
-                }
+            _register_custom_pricing_for_request(
+                model=model,
+                custom_llm_provider=custom_llm_provider,
+                kwargs=kwargs,
+                model_info=model_info,
             )
         ### BUILD CUSTOM PROMPT TEMPLATE -- IF GIVEN ###
         custom_prompt_dict = {}  # type: ignore
@@ -5959,14 +6004,11 @@ def embedding(
 
     ### REGISTER CUSTOM MODEL PRICING -- IF GIVEN ###
     if (input_cost_per_token is not None and output_cost_per_token is not None) or input_cost_per_second is not None:
-        litellm.register_model(
-            {
-                f"{custom_llm_provider}/{model}": _build_custom_pricing_entry(
-                    custom_llm_provider=custom_llm_provider,
-                    kwargs=kwargs,
-                    model_info=kwargs.get("model_info"),
-                )
-            }
+        _register_custom_pricing_for_request(
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            kwargs=kwargs,
+            model_info=kwargs.get("model_info"),
         )
 
     litellm_params_dict = get_litellm_params(**kwargs)

litellm/model_prices_and_context_window_backup.json

@@ -23503,6 +23503,76 @@
         "supports_system_messages": true,
         "supports_tool_choice": true
     },
+    "gpt-realtime-2.1": {
+        "cache_creation_input_audio_token_cost": 4e-07,
+        "cache_read_input_audio_token_cost": 4e-07,
+        "cache_read_input_token_cost": 4e-07,
+        "input_cost_per_audio_token": 3.2e-05,
+        "input_cost_per_image": 5e-06,
+        "input_cost_per_token": 4e-06,
+        "litellm_provider": "openai",
+        "max_input_tokens": 128000,
+        "max_output_tokens": 32000,
+        "max_tokens": 32000,
+        "mode": "chat",
+        "output_cost_per_audio_token": 6.4e-05,
+        "output_cost_per_token": 2.4e-05,
+        "regional_processing_uplift_multiplier_eu": 1.1,
+        "regional_processing_uplift_multiplier_us": 1.1,
+        "supported_endpoints": [
+            "/v1/realtime"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio"
+        ],
+        "supported_output_modalities": [
+            "text",
+            "audio"
+        ],
+        "supports_audio_input": true,
+        "supports_audio_output": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true
+    },
+    "gpt-realtime-2.1-mini": {
+        "cache_creation_input_audio_token_cost": 3e-07,
+        "cache_read_input_audio_token_cost": 3e-07,
+        "cache_read_input_token_cost": 6e-08,
+        "input_cost_per_audio_token": 1e-05,
+        "input_cost_per_image": 8e-07,
+        "input_cost_per_token": 6e-07,
+        "litellm_provider": "openai",
+        "max_input_tokens": 128000,
+        "max_output_tokens": 4096,
+        "max_tokens": 4096,
+        "mode": "chat",
+        "output_cost_per_audio_token": 2e-05,
+        "output_cost_per_token": 2.4e-06,
+        "regional_processing_uplift_multiplier_eu": 1.1,
+        "regional_processing_uplift_multiplier_us": 1.1,
+        "supported_endpoints": [
+            "/v1/realtime"
+        ],
+        "supported_modalities": [
+            "text",
+            "image",
+            "audio"
+        ],
+        "supported_output_modalities": [
+            "text",
+            "audio"
+        ],
+        "supports_audio_input": true,
+        "supports_audio_output": true,
+        "supports_function_calling": true,
+        "supports_parallel_function_calling": true,
+        "supports_system_messages": true,
+        "supports_tool_choice": true
+    },
     "gpt-realtime-mini": {
         "cache_creation_input_audio_token_cost": 3e-07,
         "cache_read_input_audio_token_cost": 3e-07,

litellm/models/mcp_server.py

@@ -83,6 +83,15 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
     token_url: Optional[str] = None
     registration_url: Optional[str] = None
     oauth2_flow: Optional[Literal["client_credentials", "authorization_code"]] = None
+    # Token Exchange (OBO) fields — RFC 8693. ``audience`` is named for the RFC's
+    # request parameter (token-exchange only); RFC 8707 resource indicators are a
+    # separate concept named ``resource`` in the v2 egress types. A null
+    # ``subject_token_type`` means DEFAULT_SUBJECT_TOKEN_TYPE (litellm.types.mcp),
+    # applied at the egress build sites.
+    token_exchange_endpoint: Optional[str] = None
+    audience: Optional[str] = None
+    subject_token_type: Optional[str] = None
+    token_exchange_profile: Optional[str] = None
     allow_all_keys: bool = False
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False

litellm/proxy/_experimental/mcp_server/auth/token_exchange.py

@@ -28,6 +28,7 @@ from litellm.proxy._experimental.mcp_server.auth.token_endpoint_auth import (
     build_token_endpoint_client_auth,
 )
 from litellm.types.llms.custom_http import httpxSpecialProvider
+from litellm.types.mcp import DEFAULT_SUBJECT_TOKEN_TYPE
 
 if TYPE_CHECKING:
     from litellm.types.mcp_server.mcp_server_manager import MCPServer
@@ -35,8 +36,6 @@ if TYPE_CHECKING:
 # RFC 8693 grant type constant
 TOKEN_EXCHANGE_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange"
 
-DEFAULT_SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token"
-
 
 class TokenExchangeHandler:
     """Handles OAuth 2.0 Token Exchange (RFC 8693) for MCP servers.

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -8,6 +8,7 @@ from starlette.types import Scope
 
 from litellm._logging import verbose_logger
 from litellm.proxy._types import (
+    UI_TEAM_ID,
     LiteLLM_TeamTable,
     ProxyException,
     SpecialHeaders,
@@ -357,6 +358,7 @@ class MCPRequestHandler:
         # Inline imports avoid a circular dependency: mcp_server_manager imports
         # from this module.
         from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            MCPServerManager,
             global_mcp_server_manager,
         )
         from litellm.types.mcp import MCPAuth
@@ -382,7 +384,18 @@ class MCPRequestHandler:
             # fetches the upstream token automatically using stored credentials,
             # so allowing anonymous bypass would let any external caller invoke
             # tools authenticated as LiteLLM's service account.
-            if server.has_client_credentials:
+            #
+            # Resolve the flow rather than reading has_client_credentials directly:
+            # this is a security gate, and a legacy row whose oauth2_flow was never
+            # stamped still carries the M2M credential shape (client_id/secret +
+            # token_url, no authorization_url). Treating an unstamped-but-M2M-shaped
+            # row as non-M2M here would reopen the anonymous bypass the explicit
+            # column no longer closes on its own. Shares the one resolution helper
+            # with the egress backstop and the anonymous-delegate allowlist; all fail
+            # closed on the ambiguous shape and are removed together once no null rows
+            # remain. A pure-PKCE delegate server (no stored credentials) resolves to a
+            # non-M2M flow and keeps its bypass.
+            if MCPServerManager.effective_oauth2_flow(server) == "client_credentials":
                 return False
         return True
 
@@ -726,6 +739,9 @@ class MCPRequestHandler:
         if not user_api_key_auth or not user_api_key_auth.team_id or not prisma_client:
             return None
 
+        if user_api_key_auth.team_id == UI_TEAM_ID:
+            return None
+
         # Get the team object (which has object_permission already loaded)
         team_obj: Optional[LiteLLM_TeamTable] = await get_team_object(
             team_id=user_api_key_auth.team_id,
@@ -1021,6 +1037,9 @@ class MCPRequestHandler:
             if user_api_key_auth is None or not user_api_key_auth.team_id or prisma_client is None:
                 return []
 
+            if user_api_key_auth.team_id == UI_TEAM_ID:
+                return []
+
             team_obj: Optional[LiteLLM_TeamTable] = await get_team_object(
                 team_id=user_api_key_auth.team_id,
                 prisma_client=prisma_client,
@@ -1503,6 +1522,9 @@ class MCPRequestHandler:
             verbose_logger.debug("prisma_client is None")
             return []
 
+        if user_api_key_auth.team_id == UI_TEAM_ID:
+            return []
+
         try:
             team_obj: Optional[LiteLLM_TeamTable] = await get_team_object(
                 team_id=user_api_key_auth.team_id,

litellm/proxy/_experimental/mcp_server/db.py

@@ -46,6 +46,35 @@ from litellm.types.mcp import MCPCredentials
 if TYPE_CHECKING:
     from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
+_AUTH_FLOW_SCOPED_FIELDS: frozenset = frozenset(
+    {
+        "authorization_url",
+        "token_url",
+        "registration_url",
+        "oauth2_flow",
+        "token_exchange_endpoint",
+        "audience",
+        "subject_token_type",
+        "token_exchange_profile",
+    }
+)
+
+# Token-exchange settings with dedicated columns that also exist on
+# ``MCPCredentials`` as a legacy shape (rows and REST callers that predate the
+# columns). Every write lifts blob values into the columns and strips them from
+# the stored blob, so the read-time ``column or blob`` fallback only serves rows
+# the current code has never written — a cleared column can then never be
+# silently resurrected by a stale blob copy. These keys are stored plaintext
+# (endpoints/identifiers, not secrets), so values lift as-is.
+_TOKEN_EXCHANGE_COLUMN_FIELDS: frozenset = frozenset(
+    {
+        "token_exchange_endpoint",
+        "audience",
+        "subject_token_type",
+        "token_exchange_profile",
+    }
+)
+
 
 def _is_global_env_var_scope(scope: Any) -> bool:
     """``scope="user"`` entries are placeholders the user fills in; everything
@@ -241,6 +270,14 @@ def _prepare_mcp_server_data(
     # Handle credentials serialization
     credentials = data_dict.get("credentials")
     if credentials is not None:
+        # Lift legacy blob-shaped token-exchange settings into their dedicated
+        # columns (an explicit top-level value wins, including an explicit
+        # null) and strip them from the blob so it never seeds the read-time
+        # fallback for rows written by current code.
+        for te_field in _TOKEN_EXCHANGE_COLUMN_FIELDS:
+            blob_value = credentials.pop(te_field, None)
+            if blob_value is not None and te_field not in data_dict:
+                data_dict[te_field] = blob_value
         data_dict["credentials"] = encrypt_credentials(credentials=credentials, encryption_key=_get_salt_key())
         data_dict["credentials"] = safe_dumps(data_dict["credentials"])
 
@@ -603,19 +640,41 @@ async def update_mcp_server(
     # Pre-fetch existing record once if we need it for auth_type or credential logic
     existing = None
     has_credentials = "credentials" in data_dict and data_dict["credentials"] is not None
-    if data.auth_type or has_credentials:
+    # An explicit token-exchange column write (set or clear) also migrates the
+    # legacy blob copies below, so the existing row is needed for those updates.
+    explicit_te_write = bool(_TOKEN_EXCHANGE_COLUMN_FIELDS & data_dict.keys())
+    if data.auth_type or has_credentials or explicit_te_write:
         existing = await MCPServerRepository(prisma_client).table.find_unique(where={"server_id": data.server_id})
 
+    auth_type_changed = bool(
+        data.auth_type and existing and existing.auth_type is not None and existing.auth_type != data.auth_type
+    )
+
     # Clear stale credentials when auth_type changes but no new credentials provided
-    if (
-        data.auth_type
-        and "credentials" not in data_dict
-        and existing
-        and existing.auth_type is not None
-        and existing.auth_type != data.auth_type
-    ):
+    if auth_type_changed and "credentials" not in data_dict:
         data_dict["credentials"] = None
 
+    if auth_type_changed:
+        data_dict.update({field: None for field in _AUTH_FLOW_SCOPED_FIELDS if field not in data_dict})
+
+    # An explicit column write that does not touch credentials must still migrate
+    # the row's legacy blob copies: lift values for columns the caller left
+    # untouched, strip every copy from the blob. Without this, clearing a column
+    # (e.g. to re-enable RFC 9728/8414 discovery) would leave the blob copy in
+    # place, and the next credentials update's migrate-on-write would silently
+    # repopulate the column the admin just cleared. (When credentials ARE in the
+    # update, the merge below performs the same migration.)
+    if explicit_te_write and "credentials" not in data_dict and existing is not None and existing.credentials:
+        existing_creds = (
+            json.loads(existing.credentials) if isinstance(existing.credentials, str) else dict(existing.credentials)
+        )
+        if _TOKEN_EXCHANGE_COLUMN_FIELDS & existing_creds.keys():
+            for te_field in _TOKEN_EXCHANGE_COLUMN_FIELDS:
+                legacy_value = existing_creds.pop(te_field, None)
+                if legacy_value is not None and te_field not in data_dict and getattr(existing, te_field, None) is None:
+                    data_dict[te_field] = legacy_value
+            data_dict["credentials"] = safe_dumps(existing_creds)
+
     # Merge credentials: preserve existing fields not present in the update.
     # Without this, a partial credential update (e.g. changing only region)
     # would wipe encrypted secrets that the UI cannot display back.
@@ -638,6 +697,19 @@ async def update_mcp_server(
                 )
                 # New values override existing; existing keys not in update are preserved
                 merged = {**existing_creds, **new_creds}
+                # Migrate-on-write for legacy rows: token-exchange settings the
+                # old blob shape carried move to their dedicated columns (unless
+                # the caller set the column this update, or the row already has
+                # one) and are never re-persisted in the blob. Stored plaintext,
+                # so the merged value lifts as-is.
+                for te_field in _TOKEN_EXCHANGE_COLUMN_FIELDS:
+                    legacy_value = merged.pop(te_field, None)
+                    if (
+                        legacy_value is not None
+                        and te_field not in data_dict
+                        and getattr(existing, te_field, None) is None
+                    ):
+                        data_dict[te_field] = legacy_value
                 data_dict["credentials"] = safe_dumps(merged)
 
     # Add audit fields

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -115,7 +115,7 @@ from litellm.proxy.common_utils.user_api_key_cache import get_management_object_
 from litellm.proxy.utils import ProxyLogging, get_server_root_path
 from litellm.repositories.table_repositories import MCPServerRepository
 from litellm.types.llms.custom_http import httpxSpecialProvider
-from litellm.types.mcp import MCPAuth, MCPStdioConfig
+from litellm.types.mcp import DEFAULT_SUBJECT_TOKEN_TYPE, MCPAuth, MCPStdioConfig
 from litellm.types.mcp_server.mcp_server_manager import (
     MCPInfo,
     MCPOAuthMetadata,
@@ -253,6 +253,76 @@ def _without_authorization(
     return filtered or None
 
 
+def _format_byok_openapi_auth_header(mcp_server: MCPServer, mcp_auth_header: str) -> str:
+    """Format a raw BYOK credential for OpenAPI tool ``Authorization`` injection."""
+    if mcp_server.auth_type == MCPAuth.api_key:
+        return f"ApiKey {mcp_auth_header}"
+    if mcp_server.auth_type == MCPAuth.basic:
+        return f"Basic {mcp_auth_header}"
+    return f"Bearer {mcp_auth_header}"
+
+
+def _openapi_forwarded_extra_headers(
+    mcp_server: MCPServer,
+    raw_headers: Optional[dict[str, str]],
+    user_api_key_auth: Optional[UserAPIKeyAuth],
+) -> Optional[dict[str, str]]:
+    if not mcp_server.extra_headers or not raw_headers:
+        return None
+    normalized_raw = {str(k).lower(): v for k, v in raw_headers.items() if isinstance(k, str)}
+    skip_caller_authorization = _should_strip_caller_authorization(
+        mcp_server=mcp_server,
+        raw_headers=raw_headers,
+        user_api_key_auth=user_api_key_auth,
+    )
+    forwarded: dict[str, str] = {}
+    for header_name in mcp_server.extra_headers:
+        if not isinstance(header_name, str):
+            continue
+        if skip_caller_authorization and header_name.lower() == "authorization":
+            continue
+        value = normalized_raw.get(header_name.lower())
+        if value is not None:
+            forwarded[header_name] = value
+    return forwarded or None
+
+
+async def _resolve_byok_mcp_auth_header(
+    mcp_server: MCPServer,
+    user_api_key_auth: Optional[UserAPIKeyAuth],
+    mcp_auth_header: Optional[str],
+) -> Optional[str]:
+    """Resolve BYOK credential for tool calls that bypass ``execute_mcp_tool``."""
+    if not mcp_server.is_byok:
+        return mcp_auth_header
+
+    from litellm.proxy._experimental.mcp_server.server import (
+        _check_byok_credential,
+        _get_byok_credential,
+    )
+
+    if not mcp_auth_header:
+        byok_cred = await _get_byok_credential(mcp_server, user_api_key_auth)
+        if byok_cred is None:
+            raise HTTPException(
+                status_code=401,
+                detail={
+                    "error": "byok_auth_required",
+                    "server_id": mcp_server.server_id,
+                    "server_name": mcp_server.server_name or mcp_server.name,
+                    "message": (
+                        "No stored credential found for this BYOK server. "
+                        "Complete the OAuth authorization flow to provide your API key."
+                    ),
+                },
+                headers={"WWW-Authenticate": 'Bearer resource_metadata="/.well-known/oauth-protected-resource"'},
+            )
+        return byok_cred
+
+    await _check_byok_credential(mcp_server, user_api_key_auth)
+    return mcp_auth_header
+
+
 def _extract_upstream_auth_failure(
     exc: BaseException,
 ) -> Optional[tuple[int, Optional[str]]]:
@@ -511,6 +581,21 @@ def _create_elicitation_callback():
 class MCPServerManager:
     _STDIO_ENV_TEMPLATE_PATTERN = re.compile(r"^\$\{(X-[^}]+)\}$")
 
+    @staticmethod
+    def _explicit_oauth2_flow(
+        oauth2_flow: Optional[str],
+    ) -> Optional[Literal["client_credentials", "authorization_code"]]:
+        """DB rows persist their flow (write-time stamps plus the startup backfill) and
+        config servers must declare it (validated at load), so both builds read the
+        value verbatim: unknown or null resolves to None, which
+        ``needs_user_oauth_token`` already treats as interactive. Field-shape inference
+        survives only in the request-time security helpers (``effective_oauth2_flow`` /
+        ``resolve_oauth2_flow_for_request``).
+        """
+        if oauth2_flow in ("client_credentials", "authorization_code"):
+            return cast(Literal["client_credentials", "authorization_code"], oauth2_flow)
+        return None
+
     @staticmethod
     def _resolve_oauth2_flow(
         *,
@@ -521,11 +606,18 @@ class MCPServerManager:
         client_id: Optional[str],
         client_secret: Optional[str],
     ) -> Optional[Literal["client_credentials", "authorization_code"]]:
-        """Infer oauth2_flow for legacy records that omit the field.
+        """Infer oauth2_flow from field shape when the value is omitted.
 
-        DB rows created before oauth2_flow support may have OAuth2 client
-        credentials + token_url but a null oauth2_flow. Treat these as M2M,
-        unless authorization_url is present (interactive OAuth).
+        SECURITY-SENSITIVE: this is the shape-inference engine both request-time security
+        helpers delegate to, so it is what decides M2M-vs-interactive for an unstamped row.
+        Always access it through ``effective_oauth2_flow`` (boolean/enum decisions) or
+        ``resolve_oauth2_flow_for_request`` (the egress object backstop), which are the single
+        choke points for request-time resolution; do not call it directly from security sites
+        and do not weaken its M2M-shape branch without accounting for those callers. DB rows
+        are stamped at write time and by the startup backfill, config servers must declare
+        oauth2_flow (validated at load), and both builds read the value verbatim via
+        ``_explicit_oauth2_flow``. Delete this whole request-time layer only once the backstop
+        warning stays silent in production.
         """
         if oauth2_flow in ("client_credentials", "authorization_code"):
             return cast(Literal["client_credentials", "authorization_code"], oauth2_flow)
@@ -540,6 +632,51 @@ class MCPServerManager:
             return "client_credentials"
         return None
 
+    @staticmethod
+    def effective_oauth2_flow(server: "MCPServer") -> Optional[Literal["client_credentials", "authorization_code"]]:
+        """The oauth2_flow a security decision must use for ``server`` this request.
+
+        Column-first, shape-fallback: a stamped row returns its explicit value; an
+        unstamped (null) row whose fields carry the M2M shape resolves to
+        ``client_credentials`` so it is treated as M2M and fails closed. Every
+        security-sensitive reader (anonymous-delegate allowlist and gate, egress flow
+        resolution) goes through this one helper rather than reading the bare
+        ``has_client_credentials`` column, which is unreliable for null rows.
+        """
+        return MCPServerManager._resolve_oauth2_flow(
+            auth_type=server.auth_type,
+            oauth2_flow=server.oauth2_flow,
+            token_url=server.token_url,
+            authorization_url=server.authorization_url,
+            client_id=server.client_id,
+            client_secret=server.client_secret,
+        )
+
+    @staticmethod
+    def resolve_oauth2_flow_for_request(server: "MCPServer") -> "MCPServer":
+        """Return ``server`` with its effective oauth2_flow applied, for egress paths.
+
+        A stamped row is returned unchanged (its effective flow equals the stored value).
+        An unstamped M2M-shape row is returned as a per-request copy carrying
+        ``oauth2_flow=client_credentials`` so downstream ``has_client_credentials`` /
+        ``needs_user_oauth_token`` compute correctly and the stored client credentials are
+        used instead of forwarding the caller's Authorization. Use this at every point that
+        resolves an allowed server id into an ``MCPServer`` for a tool call or listing.
+        """
+        effective = MCPServerManager.effective_oauth2_flow(server)
+        if effective is None or effective == server.oauth2_flow:
+            return server
+        verbose_logger.warning(
+            "MCP server %s has no persisted oauth2_flow but matches the %s shape; using the "
+            "inferred flow for this request. The startup backfill leaves this ambiguous M2M "
+            "shape unstamped on purpose, so it will NOT self-heal: set oauth2_flow explicitly "
+            "in the dashboard or via PUT /v1/mcp/server (client_credentials for M2M, or "
+            "authorization_code after an interactive sign-in).",
+            server.server_id,
+            effective,
+        )
+        return server.model_copy(update={"oauth2_flow": effective})
+
     @staticmethod
     def _obo_needs_endpoint_discovery(
         auth_type: Optional[MCPAuthType],
@@ -578,8 +715,10 @@ class MCPServerManager:
         # Per-server outbound tool-call concurrency limiters, lazily created from
         # each server's max_concurrent_requests. Keyed by server_id so the cap
         # survives the registry atomic-swap on config reload; a missing key means
-        # the server has no configured limit.
-        self._server_call_semaphores: dict[str, asyncio.Semaphore] = {}
+        # the server has no configured limit. The limit is cached alongside the
+        # semaphore so an edited limit rebuilds it instead of keeping the old cap
+        # until restart.
+        self._server_call_semaphores: dict[str, tuple[int, asyncio.Semaphore]] = {}
         self.tool_name_to_mcp_server_name_mapping: dict[str, str] = {}
         """
         {
@@ -772,6 +911,20 @@ class MCPServerManager:
                 mcp_oauth_metadata.registration_url if mcp_oauth_metadata else None
             )
 
+            config_oauth2_flow = server_config.get("oauth2_flow", None)
+            if auth_type == MCPAuth.oauth2 and config_oauth2_flow not in (
+                "client_credentials",
+                "authorization_code",
+            ):
+                raise ValueError(
+                    f"Invalid config for MCP server '{server_name or server_id}': auth_type oauth2 "
+                    f"requires an explicit oauth2_flow (got {config_oauth2_flow!r}). Set "
+                    "oauth2_flow: client_credentials for machine-to-machine servers (the proxy mints "
+                    "a shared token at token_url using client_id/client_secret, no user interaction) "
+                    "or oauth2_flow: authorization_code for interactive servers (per-user tokens via "
+                    "browser sign-in, including delegate_auth_to_upstream)."
+                )
+
             new_server = MCPServer(
                 server_id=server_id,
                 name=name_for_prefix,
@@ -785,14 +938,7 @@ class MCPServerManager:
                 # oauth specific fields
                 client_id=server_config.get("client_id", None),
                 client_secret=server_config.get("client_secret", None),
-                oauth2_flow=self._resolve_oauth2_flow(
-                    auth_type=auth_type,
-                    oauth2_flow=server_config.get("oauth2_flow", None),
-                    token_url=resolved_token_url,
-                    authorization_url=resolved_authorization_url,
-                    client_id=server_config.get("client_id", None),
-                    client_secret=server_config.get("client_secret", None),
-                ),
+                oauth2_flow=self._explicit_oauth2_flow(config_oauth2_flow),
                 scopes=resolved_scopes,
                 authorization_url=resolved_authorization_url,
                 token_url=resolved_token_url,
@@ -828,7 +974,7 @@ class MCPServerManager:
                 audience=server_config.get("audience", None),
                 subject_token_type=server_config.get(
                     "subject_token_type",
-                    "urn:ietf:params:oauth:token-type:access_token",
+                    DEFAULT_SUBJECT_TOKEN_TYPE,
                 ),
                 token_exchange_profile=server_config.get("token_exchange_profile", "rfc8693"),
                 allow_sampling=bool(server_config.get("allow_sampling", False)),
@@ -1139,7 +1285,8 @@ class MCPServerManager:
             (auth_type == MCPAuth.oauth2 and not mcp_server.authorization_url)
             or self._obo_needs_endpoint_discovery(
                 auth_type,
-                credentials_dict.get("token_exchange_endpoint") if credentials_dict else None,
+                mcp_server.token_exchange_endpoint
+                or (credentials_dict.get("token_exchange_endpoint") if credentials_dict else None),
                 mcp_server.token_url,
             )
         )
@@ -1170,15 +1317,7 @@ class MCPServerManager:
             env_vars=env_vars_list,
             client_id=client_id_value or getattr(mcp_server, "client_id", None),
             client_secret=client_secret_value or getattr(mcp_server, "client_secret", None),
-            oauth2_flow=self._resolve_oauth2_flow(
-                auth_type=auth_type,
-                oauth2_flow=getattr(mcp_server, "oauth2_flow", None),
-                token_url=mcp_server.token_url or getattr(mcp_oauth_metadata, "token_url", None),
-                authorization_url=mcp_server.authorization_url
-                or getattr(mcp_oauth_metadata, "authorization_url", None),
-                client_id=client_id_value or getattr(mcp_server, "client_id", None),
-                client_secret=client_secret_value or getattr(mcp_server, "client_secret", None),
-            ),
+            oauth2_flow=self._explicit_oauth2_flow(getattr(mcp_server, "oauth2_flow", None)),
             scopes=resolved_scopes,
             authorization_url=mcp_server.authorization_url or getattr(mcp_oauth_metadata, "authorization_url", None),
             token_url=mcp_server.token_url or getattr(mcp_oauth_metadata, "token_url", None),
@@ -1213,12 +1352,16 @@ class MCPServerManager:
             aws_role_name=aws_creds.get("aws_role_name"),
             aws_session_name=aws_creds.get("aws_session_name"),
             instructions=mcp_server.instructions,
-            # Token Exchange (OBO) fields — read from credentials JSON blob
-            token_exchange_endpoint=(credentials_dict.get("token_exchange_endpoint") if credentials_dict else None),
-            audience=(credentials_dict.get("audience") if credentials_dict else None),
-            subject_token_type=(credentials_dict.get("subject_token_type") if credentials_dict else None)
-            or "urn:ietf:params:oauth:token-type:access_token",
-            token_exchange_profile=(credentials_dict.get("token_exchange_profile") if credentials_dict else None)
+            # Token exchange (OBO) fields: dedicated columns, with the credentials blob as a
+            # back-compat fallback for servers persisted before the columns existed.
+            token_exchange_endpoint=mcp_server.token_exchange_endpoint
+            or (credentials_dict.get("token_exchange_endpoint") if credentials_dict else None),
+            audience=mcp_server.audience or (credentials_dict.get("audience") if credentials_dict else None),
+            subject_token_type=mcp_server.subject_token_type
+            or (credentials_dict.get("subject_token_type") if credentials_dict else None)
+            or DEFAULT_SUBJECT_TOKEN_TYPE,
+            token_exchange_profile=mcp_server.token_exchange_profile
+            or (credentials_dict.get("token_exchange_profile") if credentials_dict else None)
             or "rfc8693",
             timeout=getattr(mcp_server, "timeout", None),
             max_concurrent_requests=getattr(mcp_server, "max_concurrent_requests", None),
@@ -1486,8 +1629,11 @@ class MCPServerManager:
                     and getattr(server, "delegate_auth_to_upstream", False) is True
                     # M2M servers must not be exposed anonymously: an
                     # unauthenticated caller would get LiteLLM to proxy tool
-                    # calls using its stored client_credentials.
-                    and not server.has_client_credentials
+                    # calls using its stored client_credentials. Resolve the flow
+                    # rather than reading has_client_credentials so an unstamped
+                    # M2M-shape row (null column, verbatim-read as non-M2M) still
+                    # fails closed here, matching the anonymous-delegate auth gate.
+                    and MCPServerManager.effective_oauth2_flow(server) != "client_credentials"
                 ]
                 combined_servers.update(delegate_server_ids)
 
@@ -3450,10 +3596,11 @@ class MCPServerManager:
         limit = mcp_server.max_concurrent_requests
         if limit is None or limit <= 0:
             return None
-        semaphore = self._server_call_semaphores.get(mcp_server.server_id)
-        if semaphore is None:
-            semaphore = asyncio.Semaphore(limit)
-            self._server_call_semaphores[mcp_server.server_id] = semaphore
+        cached = self._server_call_semaphores.get(mcp_server.server_id)
+        if cached is not None and cached[0] == limit:
+            return cached[1]
+        semaphore = asyncio.Semaphore(limit)
+        self._server_call_semaphores[mcp_server.server_id] = (limit, semaphore)
         return semaphore
 
     @asynccontextmanager
@@ -3667,17 +3814,21 @@ class MCPServerManager:
             # OBO: the exchanged token may have been revoked/rotated upstream since it was cached, so
             # an upstream 401 gets one re-mint + retry. Gated to this mode; all others keep the plain
             # single call below.
-            tool_call_coro = self._obo_call_tool_with_retry(
-                client=client,
-                call_tool_params=call_tool_params,
-                host_progress_callback=host_progress_callback,
-                mcp_server=mcp_server,
-                server_auth_header=server_auth_header,
-                extra_headers=extra_headers,
-                stdio_env=stdio_env,
-                subject_token=subject_token,
-                user_api_key_auth=user_api_key_auth,
-            )
+            async def _obo_call_tool_limited():
+                async with self._limit_outbound_concurrency(mcp_server):
+                    return await self._obo_call_tool_with_retry(
+                        client=client,
+                        call_tool_params=call_tool_params,
+                        host_progress_callback=host_progress_callback,
+                        mcp_server=mcp_server,
+                        server_auth_header=server_auth_header,
+                        extra_headers=extra_headers,
+                        stdio_env=stdio_env,
+                        subject_token=subject_token,
+                        user_api_key_auth=user_api_key_auth,
+                    )
+
+            tool_call_coro = _obo_call_tool_limited()
         else:
 
             async def _call_tool_via_client(client, params):
@@ -3861,6 +4012,15 @@ class MCPServerManager:
         start_time = datetime.datetime.now()
         mcp_server = self._resolve_mcp_server_for_tool_call(server_name, name)
 
+        # Resolved before any hook runs so a missing BYOK credential (401) never
+        # leaves during-hook side effects (audit logging, rate-limit bookkeeping)
+        # recorded against a call that ultimately fails.
+        mcp_auth_header = await _resolve_byok_mcp_auth_header(
+            mcp_server,
+            user_api_key_auth,
+            mcp_auth_header,
+        )
+
         #########################################################
         # Pre MCP Tool Call Hook
         # Allow validation and modification of tool calls before execution
@@ -3907,9 +4067,25 @@ class MCPServerManager:
                     server_name,
                 )
 
+            auth_header_value = (
+                _format_byok_openapi_auth_header(mcp_server, mcp_auth_header) if mcp_auth_header else None
+            )
+            forwarded_headers = _openapi_forwarded_extra_headers(mcp_server, raw_headers, user_api_key_auth)
+
             async def _call_openapi_via_handler():
-                async with self._limit_outbound_concurrency(mcp_server):
-                    return await self._call_openapi_tool_handler(mcp_server, name, arguments)
+                from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import (
+                    _request_auth_header,
+                    _request_extra_headers,
+                )
+
+                auth_token = _request_auth_header.set(auth_header_value)
+                extra_token = _request_extra_headers.set(forwarded_headers)
+                try:
+                    async with self._limit_outbound_concurrency(mcp_server):
+                        return await self._call_openapi_tool_handler(mcp_server, name, arguments)
+                finally:
+                    _request_auth_header.reset(auth_token)
+                    _request_extra_headers.reset(extra_token)
 
             tasks.append(asyncio.create_task(_call_openapi_via_handler()))
         else:
@@ -4465,6 +4641,11 @@ class MCPServerManager:
             authorization_url=server.authorization_url,
             token_url=server.token_url,
             registration_url=server.registration_url,
+            oauth2_flow=server.oauth2_flow,
+            token_exchange_endpoint=server.token_exchange_endpoint,
+            audience=server.audience,
+            subject_token_type=server.subject_token_type,
+            token_exchange_profile=server.token_exchange_profile,
             allow_all_keys=server.allow_all_keys,
             instructions=server.instructions,
             timeout=server.timeout,
@@ -4553,6 +4734,8 @@ class MCPServerManager:
             teams=[],
             mcp_access_groups=server.access_groups or [],
             allowed_tools=server.allowed_tools or [],
+            tool_name_to_display_name=server.tool_name_to_display_name,
+            tool_name_to_description=server.tool_name_to_description,
             extra_headers=server.extra_headers or [],
             mcp_info=server.mcp_info,
             static_headers=server.static_headers,
@@ -4566,6 +4749,11 @@ class MCPServerManager:
             authorization_url=server.authorization_url,
             token_url=server.token_url,
             registration_url=server.registration_url,
+            oauth2_flow=server.oauth2_flow,
+            token_exchange_endpoint=server.token_exchange_endpoint,
+            audience=server.audience,
+            subject_token_type=server.subject_token_type,
+            token_exchange_profile=server.token_exchange_profile,
             allow_all_keys=server.allow_all_keys,
             available_on_public_internet=server.available_on_public_internet,
             delegate_auth_to_upstream=server.delegate_auth_to_upstream,

litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py

@@ -28,7 +28,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.types import (
     Subject,
     TokenExchangeConfig,
 )
-from litellm.types.mcp import MCPAuth
+from litellm.types.mcp import DEFAULT_SUBJECT_TOKEN_TYPE, MCPAuth
 
 if TYPE_CHECKING:
     from litellm.proxy._types import UserAPIKeyAuth
@@ -124,7 +124,7 @@ def _token_exchange_spec(server: MCPServer, resource: str) -> Optional[ServerSpe
         resource=resource,
         config=TokenExchangeConfig(
             profile=profile,
-            subject_token_type=server.subject_token_type or "urn:ietf:params:oauth:token-type:access_token",
+            subject_token_type=server.subject_token_type or DEFAULT_SUBJECT_TOKEN_TYPE,
             token_exchange_endpoint=endpoint,
             audience=server.audience,
             client_id=server.client_id,

litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py

@@ -39,6 +39,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.result import (
     Ok,
     Result,
 )
+from litellm.types.mcp import DEFAULT_SUBJECT_TOKEN_TYPE
 
 
 class AuthSpecKind(str, Enum):
@@ -215,7 +216,7 @@ class TokenExchangeConfig(BaseModel):
     model_config = ConfigDict(frozen=True)
     kind: Literal[AuthSpecKind.token_exchange] = AuthSpecKind.token_exchange
     profile: Literal["rfc8693", "entra_obo"] = "rfc8693"
-    subject_token_type: str = "urn:ietf:params:oauth:token-type:access_token"
+    subject_token_type: str = DEFAULT_SUBJECT_TOKEN_TYPE
     token_exchange_endpoint: str | None = None
     audience: str | None = None
     client_id: str | None = None

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -77,6 +77,7 @@ if MCP_AVAILABLE:
         ListMCPToolsRestAPIResponseObject,
         MCPInfo,
         MCPServer,
+        _apply_toolset_scope,
         _fire_mcp_success_logging,
         _tool_name_matches,
         execute_mcp_tool,
@@ -541,10 +542,37 @@ if MCP_AVAILABLE:
             "message": "Successfully retrieved tools",
         }
 
+    def _as_query_str(value: Any) -> Optional[str]:
+        """Coerce an Optional[str] Query param to str|None, dropping unresolved FastAPI defaults."""
+        return value if isinstance(value, str) else None
+
+    async def _resolve_toolset_scope(
+        toolset_name: Optional[str],
+        user_api_key_dict: UserAPIKeyAuth,
+    ) -> UserAPIKeyAuth:
+        """Resolve ``toolset_name`` to its scoped ``UserAPIKeyAuth``, or return unchanged."""
+        if not toolset_name:
+            return user_api_key_dict
+
+        from litellm.proxy.utils import get_prisma_client_or_throw
+
+        prisma_client = get_prisma_client_or_throw("Database not available. Connect a database to your proxy")
+        toolset = await global_mcp_server_manager.get_toolset_by_name_cached(prisma_client, toolset_name)
+        if toolset is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Toolset '{toolset_name}' not found",
+            )
+        return await _apply_toolset_scope(user_api_key_dict, toolset.toolset_id)
+
     @router.get("/tools/list", dependencies=[Depends(user_api_key_auth)])
     async def list_tool_rest_api(
         request: Request,
         server_id: Optional[str] = Query(None, description="The server id to list tools for"),
+        mcp_server_name: Optional[str] = Query(
+            None, description="Filter tools to a single MCP server by name or alias"
+        ),
+        toolset_name: Optional[str] = Query(None, description="Filter tools to a single toolset by name"),
         include_disabled_tools: bool = Query(
             False,
             description=(
@@ -582,16 +610,29 @@ if MCP_AVAILABLE:
         )
 
         try:
+            mcp_server_name = _as_query_str(mcp_server_name)
+            toolset_name = _as_query_str(toolset_name)
+
             # The full catalog (allowlist filter skipped) is admin-only so the
             # REST endpoint can't be used to enumerate deliberately-disabled tools.
             apply_tool_filters = not (
                 include_disabled_tools and user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
             )
 
-            if apply_tool_filters and getattr(
-                getattr(user_api_key_dict, "object_permission", None),
-                "mcp_tool_search_enabled",
-                False,
+            user_api_key_dict = await _resolve_toolset_scope(toolset_name, user_api_key_dict)
+
+            if server_id is None:
+                server_id = mcp_server_name
+
+            if (
+                apply_tool_filters
+                and server_id is None
+                and toolset_name is None
+                and getattr(
+                    getattr(user_api_key_dict, "object_permission", None),
+                    "mcp_tool_search_enabled",
+                    False,
+                )
             ):
                 from litellm.proxy._experimental.mcp_server.tool_search import (
                     get_virtual_tool_definitions,
@@ -719,6 +760,8 @@ if MCP_AVAILABLE:
                 request_path=request.scope.get("_original_path") or request.url.path,
             )
         except HTTPException as http_exc:
+            if http_exc.status_code == status.HTTP_404_NOT_FOUND:
+                raise
             # Internal access/IP 403s keep the legacy error-dict response shape
             # so the existing contract stays intact.
             verbose_logger.exception("HTTPException in list_tool_rest_api: %s", str(http_exc))

litellm/proxy/_experimental/mcp_server/server.py

@@ -716,7 +716,7 @@ if MCP_AVAILABLE:
         if not (host_ctx and hasattr(host_ctx, "meta") and host_ctx.meta):
             return None
         host_token = getattr(host_ctx.meta, "progressToken", None)
-        if not (host_token and hasattr(host_ctx, "session") and host_ctx.session):
+        if host_token is None or not (hasattr(host_ctx, "session") and host_ctx.session):
             return None
         host_session = host_ctx.session
 
@@ -732,7 +732,7 @@ if MCP_AVAILABLE:
             except Exception as e:
                 verbose_logger.error(f"Failed to forward progress to Host: {e}")
 
-        verbose_logger.debug(f"Host progressToken captured: {host_token[:8]}...")
+        verbose_logger.debug(f"Host progressToken captured: {str(host_token)[:8]}...")
         return forward_progress
 
     async def _build_virtual_call_logging_obj(
@@ -1427,18 +1427,8 @@ if MCP_AVAILABLE:
         for allowed_mcp_server_id in allowed_mcp_server_ids:
             mcp_server = global_mcp_server_manager.get_mcp_server_by_id(allowed_mcp_server_id)
             if mcp_server is not None:
-                # Apply oauth2_flow resolution for legacy DB rows where it may be NULL
-                resolved_flow = MCPServerManager._resolve_oauth2_flow(
-                    auth_type=mcp_server.auth_type,
-                    oauth2_flow=mcp_server.oauth2_flow,
-                    token_url=mcp_server.token_url,
-                    authorization_url=mcp_server.authorization_url,
-                    client_id=mcp_server.client_id,
-                    client_secret=mcp_server.client_secret,
-                )
-                if resolved_flow and resolved_flow != mcp_server.oauth2_flow:
-                    # Create a new instance with the resolved flow for this request
-                    mcp_server = mcp_server.model_copy(update={"oauth2_flow": resolved_flow})
+                # Apply the request-time oauth2_flow backstop for legacy null rows.
+                mcp_server = MCPServerManager.resolve_oauth2_flow_for_request(mcp_server)
                 allowed_mcp_servers.append(mcp_server)
 
         if mcp_servers is not None:
@@ -2800,6 +2790,9 @@ if MCP_AVAILABLE:
             for allowed_mcp_server_id in allowed_mcp_server_ids:
                 allowed_server = global_mcp_server_manager.get_mcp_server_by_id(allowed_mcp_server_id)
                 if allowed_server is not None:
+                    # Same request-time oauth2_flow backstop the listing path applies,
+                    # so a null-flow M2M-shape row is treated as M2M on tool calls too.
+                    allowed_server = MCPServerManager.resolve_oauth2_flow_for_request(allowed_server)
                     allowed_mcp_servers.append(allowed_server)
 
             allowed_mcp_servers = await _get_allowed_mcp_servers_from_mcp_server_names(

litellm/proxy/_experimental/mcp_server/utils.py

@@ -214,12 +214,14 @@ def server_applies_tool_allowlist(mcp_server: Any) -> bool:
 
 def validate_and_normalize_mcp_server_payload(payload: Any) -> None:
     """
-    Validate and normalize MCP server payload fields (server_name and alias).
+    Validate and normalize MCP server payload fields (server_name, alias, and
+    tool_name_to_display_name).
 
     This function:
     1. Validates that server_name and alias don't contain the MCP_TOOL_PREFIX_SEPARATOR
-    2. Normalizes alias by replacing spaces with underscores
-    3. Sets default alias if not provided (using server_name as base)
+    2. Validates that tool_name_to_display_name values satisfy Bedrock's tool-name pattern
+    3. Normalizes alias by replacing spaces with underscores
+    4. Sets default alias if not provided (using server_name as base)
 
     Args:
         payload: The payload object containing server_name and alias fields
@@ -235,6 +237,10 @@ def validate_and_normalize_mcp_server_payload(payload: Any) -> None:
     if hasattr(payload, "alias") and payload.alias:
         validate_mcp_server_name(payload.alias, raise_http_exception=True)
 
+    # Tool display name validation: must satisfy Bedrock's tool-name pattern
+    if hasattr(payload, "tool_name_to_display_name") and payload.tool_name_to_display_name:
+        validate_tool_display_names(payload.tool_name_to_display_name)
+
     # Alias normalization and defaulting
     alias = getattr(payload, "alias", None)
     server_name = getattr(payload, "server_name", None)
@@ -409,6 +415,42 @@ def validate_mcp_server_name(server_name: str, raise_http_exception: bool = Fals
             raise Exception(error_message)
 
 
+TOOL_DISPLAY_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
+
+
+def validate_tool_display_names(tool_name_to_display_name: Optional[Mapping[str, str]]) -> None:
+    """
+    Validate tool display name overrides against Bedrock's tool-name constraint.
+
+    A display name replaces the tool name sent to the LLM provider, so it must
+    satisfy the strictest provider requirement in use (Bedrock's
+    ``[a-zA-Z0-9_-]+``); a name with spaces or other characters saves
+    successfully but fails every subsequent Bedrock tool call.
+
+    Raises:
+        HTTPException: If any display name fails the pattern.
+    """
+    if not tool_name_to_display_name:
+        return
+
+    for original_name, display_name in tool_name_to_display_name.items():
+        if display_name and not TOOL_DISPLAY_NAME_PATTERN.match(display_name):
+            from fastapi import HTTPException
+            from starlette import status
+
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail={
+                    "error": (
+                        f"Invalid display name '{display_name}' for tool '{original_name}'. "
+                        "Display names may only contain letters, digits, underscores, and "
+                        "hyphens (no spaces or other special characters), since they replace "
+                        "the tool name sent to the LLM provider."
+                    )
+                },
+            )
+
+
 class MCPMissingUserEnvVarsError(Exception):
     """Raised when an MCP request can't be built because the calling user has
     not supplied one or more required per-user environment variables.

litellm/proxy/_types.py

@@ -1070,6 +1070,7 @@ class KeyRequestBase(GenerateRequestBase):
     budget_id: Optional[str] = None
     tags: Optional[List[str]] = None
     disable_global_guardrails: Optional[bool] = None
+    throttle_on_budget_exceeded: Optional[bool] = None
     enforced_params: Optional[List[str]] = None
     allowed_routes: Optional[list] = []
     allowed_passthrough_routes: Optional[list] = None
@@ -1254,6 +1255,14 @@ class NewMCPServerRequest(LiteLLMPydanticObjectBase):
     token_url: Optional[str] = None
     registration_url: Optional[str] = None
     oauth2_flow: Optional[Literal["client_credentials", "authorization_code"]] = None
+    # Token Exchange (OBO) fields — RFC 8693. These top-level fields are the
+    # canonical shape; the same keys inside ``credentials`` are the legacy
+    # pre-column REST shape and are lifted into these columns on write (an
+    # explicit top-level value wins) and stripped from the stored blob.
+    token_exchange_endpoint: Optional[str] = None
+    audience: Optional[str] = None
+    subject_token_type: Optional[str] = None
+    token_exchange_profile: Optional[str] = None
     allow_all_keys: bool = False
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False
@@ -1340,6 +1349,14 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
     token_url: Optional[str] = None
     registration_url: Optional[str] = None
     oauth2_flow: Optional[Literal["client_credentials", "authorization_code"]] = None
+    # Token Exchange (OBO) fields — RFC 8693. These top-level fields are the
+    # canonical shape; the same keys inside ``credentials`` are the legacy
+    # pre-column REST shape and are lifted into these columns on write (an
+    # explicit top-level value wins) and stripped from the stored blob.
+    token_exchange_endpoint: Optional[str] = None
+    audience: Optional[str] = None
+    subject_token_type: Optional[str] = None
+    token_exchange_profile: Optional[str] = None
     allow_all_keys: bool = False
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False
@@ -2322,6 +2339,28 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
             "is active as a reminder that hard enforcement is relaxed."
         ),
     )
+    user_url_validation: Optional[bool] = Field(
+        None,
+        description=(
+            "Master switch for the SSRF guard applied to user-supplied URLs "
+            "(image_url, file_url, MCP/OpenAPI spec URLs, etc). Defaults to True. "
+            "Set to False to disable DNS/IP validation entirely (not recommended)."
+        ),
+    )
+    user_url_allowed_hosts: Optional[list[str]] = Field(
+        None,
+        description=(
+            "SSRF allowlist for user-supplied URLs. Entries are `hostname` or "
+            "`hostname:port` (bracketed for IPv6, e.g. `[::1]:8080`). Allowlisted "
+            "hosts skip the blocked-network check in validate_url() but still "
+            "resolve DNS. Use this to permit legitimate internal targets, e.g. "
+            "an internal OpenAPI/MCP server."
+        ),
+    )
+    provider_url_destination_allowed_hosts: Optional[list[str]] = Field(
+        None,
+        description="Allowlist of hosts a request may redirect a provider call's destination URL to.",
+    )
 
 
 class ConfigYAML(LiteLLMPydanticObjectBase):
@@ -2447,6 +2486,7 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView):  # the expected response ob
     request_route: Optional[str] = None
     is_session_token: bool = False
     budget_reservation: Optional[Dict[str, Any]] = Field(default=None, exclude=True)
+    budget_throttle_pct: Optional[float] = Field(default=None, exclude=True)
     user: Optional[Any] = None  # Expanded user object when expand=user is used
     created_by_user: Optional[Any] = None  # Expanded created_by user when expand=user is used
     end_user_object_permission: Optional[LiteLLM_ObjectPermissionTable] = None
@@ -3837,6 +3877,7 @@ LiteLLM_ManagementEndpoint_MetadataFields = [
     "allowed_vector_store_indexes",
     "enforced_batch_output_expires_after",
     "enforced_file_expires_after",
+    "throttle_on_budget_exceeded",
 ]
 
 LiteLLM_ManagementEndpoint_MetadataFields_Premium = [

litellm/proxy/auth/auth_checks.py

@@ -60,6 +60,10 @@ from litellm.proxy._types import (
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.proxy.auth.budget_throttle import (
+    budget_throttle_percentage,
+    should_throttle_budget_exceeded,
+)
 from litellm.proxy.spend_tracking.budget_reservation import get_budget_window_start
 from litellm.proxy.common_utils.cache_pydantic_utils import CacheCodec
 from litellm.proxy.common_utils.http_parsing_utils import (
@@ -2496,6 +2500,7 @@ def _copy_user_api_key_auth_for_cache(
 ) -> UserAPIKeyAuth:
     copied_key_obj = user_api_key_obj.model_copy()
     copied_key_obj.budget_reservation = None
+    copied_key_obj.budget_throttle_pct = None
     copied_key_obj.parent_otel_span = None
     copied_key_obj.request_route = None
     return copied_key_obj
@@ -3428,6 +3433,24 @@ async def is_valid_fallback_model(
     return True
 
 
+def _apply_budget_exceeded_throttle(valid_token: UserAPIKeyAuth) -> bool:
+    """
+    Throttle an over-budget key instead of blocking it, when the key opted in
+    via `throttle_on_budget_exceeded` and a global percentage is configured.
+
+    Records the percentage on the request-scoped `budget_throttle_pct` so the
+    rate limiter scales the key's TPM/RPM down to it; the persistent limits are
+    left untouched so the throttle never compounds across requests. Returns True
+    when the key was throttled (caller skips raising), False when it should still
+    be hard-blocked.
+    """
+    pct = budget_throttle_percentage()
+    if pct is None or not should_throttle_budget_exceeded(valid_token):
+        return False
+    valid_token.budget_throttle_pct = pct
+    return True
+
+
 async def _virtual_key_max_budget_check(
     valid_token: UserAPIKeyAuth,
     proxy_logging_obj: ProxyLogging,
@@ -3488,6 +3511,8 @@ async def _virtual_key_max_budget_check(
         # so a NaN max_budget would silently disable enforcement.  Treat a
         # non-finite max_budget as "no configured limit" rather than as a bypass.
         if math.isfinite(valid_token.max_budget) and spend >= valid_token.max_budget:
+            if _apply_budget_exceeded_throttle(valid_token):
+                return
             # name the key in the error so operators don't have to reverse-map
             # spend back to a key; key_name is the masked form (last 4 chars)
             key_label = valid_token.key_alias or "key"

litellm/proxy/auth/budget_throttle.py

@@ -0,0 +1,56 @@
+"""
+Throttle a key after it exceeds its own ``max_budget`` instead of blocking it.
+
+When a key opts in via ``throttle_on_budget_exceeded`` and a global
+``budget_exceeded_throttle_percentage`` is configured, an over-budget key keeps
+serving requests but at a reduced TPM/RPM (the configured percentage of its
+configured limits). The decision (over budget + opted in) is made once during
+auth; the scaling is recomputed from the key's original limits on every request
+so it never compounds across requests.
+"""
+
+import math
+from typing import Optional
+
+import litellm
+from litellm.proxy._types import UserAPIKeyAuth
+
+
+def budget_throttle_percentage() -> Optional[float]:
+    """
+    The global throttle percentage, or None when throttling is disabled /
+    misconfigured (in which case an over-budget key is hard-blocked, the safe
+    default).
+    """
+    pct = litellm.budget_exceeded_throttle_percentage
+    if not isinstance(pct, (int, float)) or isinstance(pct, bool):
+        return None
+    if not 0 < pct <= 1:
+        return None
+    return float(pct)
+
+
+def should_throttle_budget_exceeded(valid_token: UserAPIKeyAuth) -> bool:
+    """
+    True when a key that exceeded its own ``max_budget`` should be throttled
+    rather than blocked: it opted in, a valid global percentage is set, and the
+    key has a TPM or RPM limit to scale down. A key with neither limit has
+    nothing to throttle, so it stays hard-blocked (the safe default) rather than
+    serving unlimited requests past its budget.
+    """
+    if (valid_token.metadata or {}).get("throttle_on_budget_exceeded") is not True:
+        return False
+    if valid_token.tpm_limit is None and valid_token.rpm_limit is None:
+        return False
+    return budget_throttle_percentage() is not None
+
+
+def throttled_limit(limit: Optional[int], pct: Optional[float]) -> Optional[int]:
+    """
+    Scale a TPM/RPM limit to ``pct`` of its value, keeping a trickle of at least
+    1 so a throttled key is slowed rather than fully locked out. An unset limit
+    or unset percentage leaves the limit unchanged.
+    """
+    if limit is None or pct is None:
+        return limit
+    return max(1, math.floor(limit * pct))

litellm/proxy/auth/user_api_key_auth.py

@@ -2004,6 +2004,11 @@ async def _user_api_key_auth_builder(
 
             valid_token_dict = valid_token.model_dump(exclude_none=True)
             valid_token_dict.pop("token", None)
+            # budget_throttle_pct is excluded from model_dump (it must not leak
+            # into serialized responses), so carry the request-scoped decision
+            # forward by hand to the auth object the rate limiter receives.
+            if valid_token.budget_throttle_pct is not None:
+                valid_token_dict["budget_throttle_pct"] = valid_token.budget_throttle_pct
 
             if _end_user_object is not None:
                 valid_token_dict.update(end_user_params)

litellm/proxy/db/prisma_client.py

@@ -137,8 +137,17 @@ class PrismaWrapper:
         self.on_engine_replaced: Callable[[], None] | None = None
 
     def _get_engine_pid(self) -> int:
-        """Get the PID of the current Prisma engine subprocess, or 0 if unavailable."""
+        """Get the PID of the current Prisma engine subprocess, or 0 if unavailable.
+
+        Must never raise: it runs inside the reconnect path, where the client
+        may be in any broken state. Prisma's ``_engine`` is a property that
+        raises ``ClientNotConnectedError`` on a disconnected client; if that
+        escaped here, ``recreate_prisma_client`` would fail before it could
+        build a replacement client and the reconnect loop could never recover.
+        """
         try:
+            if self._original_prisma.is_connected() is not True:
+                return 0
             engine = self._original_prisma._engine
             process = getattr(engine, "process", None) if engine is not None else None
             if process is not None:

litellm/proxy/hooks/mcp_semantic_filter/hook.py

@@ -129,6 +129,27 @@ class SemanticToolFilterHook(CustomLogger):
 
         return openai_tools_as_dicts
 
+    async def _filter_expanded_tools(
+        self,
+        data: dict,
+        expanded_tools: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """
+        Apply the semantic filter to expanded MCP tool definitions.
+
+        Expanded tools are flat OpenAI function dicts with a top-level
+        "name" (see transform_mcp_tool_to_openai_responses_api_tool), so
+        filter_tools can name-match them against the semantic router.
+        """
+        raw_messages = data.get("messages") or data.get("input") or []
+        messages = [{"role": "user", "content": raw_messages}] if isinstance(raw_messages, str) else raw_messages
+        user_query = self.filter.extract_user_query(messages)
+        if not user_query:
+            verbose_proxy_logger.debug("No user query found, skipping semantic filter on expanded MCP tools")
+            return expanded_tools
+
+        return await self.filter.filter_tools(query=user_query, available_tools=expanded_tools)
+
     def _is_mcp_tool(self, tool: object) -> bool:
         """
         Check whether *tool* is registered in the MCP semantic router.
@@ -184,6 +205,32 @@ class SemanticToolFilterHook(CustomLogger):
                 f"Semantic tool filter: all {len(native_tools)} tools are native, no MCP filtering applied"
             )
 
+    def _emit_filter_metadata_safe(
+        self,
+        data: dict,
+        mcp_tools: list[object],
+        filtered_mcp_tools: list[object],
+        native_tools: list[object],
+        filtered_tools: list[object],
+    ) -> None:
+        """
+        Emit filter metadata without letting an emission failure abort the
+        already-filtered request.
+        """
+        try:
+            self._emit_filter_metadata(
+                data=data,
+                mcp_tools=mcp_tools,
+                filtered_mcp_tools=filtered_mcp_tools,
+                native_tools=native_tools,
+                filtered_tools=filtered_tools,
+            )
+        except Exception as e:
+            verbose_proxy_logger.warning(
+                f"Failed to emit semantic filter metadata: {e}",
+                exc_info=True,
+            )
+
     async def async_pre_call_hook(
         self,
         user_api_key_dict: "UserAPIKeyAuth",
@@ -206,9 +253,6 @@ class SemanticToolFilterHook(CustomLogger):
             verbose_proxy_logger.debug("No tools in request, skipping semantic filter")
             return None
 
-        # Expanded MCP tools are in OpenAI nested format which
-        # filter_tools/_extract_tool_info cannot name-match, so we skip
-        # semantic filtering and return early.
         if self._should_expand_mcp_tools(tools):
             verbose_proxy_logger.debug("Detected litellm_proxy MCP references, expanding before semantic filtering")
 
@@ -227,11 +271,26 @@ class SemanticToolFilterHook(CustomLogger):
                     verbose_proxy_logger.warning("No tools expanded from MCP references")
                     return None
 
-                data["tools"] = native_tools_before_expand + expanded_tools
+                if not self.filter.enabled:
+                    data["tools"] = native_tools_before_expand + expanded_tools
+                    verbose_proxy_logger.debug("Semantic filter disabled, forwarding expanded MCP tools unfiltered")
+                    return data
+
+                filtered_expanded_tools = await self._filter_expanded_tools(data=data, expanded_tools=expanded_tools)
+
+                combined_tools = native_tools_before_expand + filtered_expanded_tools
+                data["tools"] = combined_tools
+                self._emit_filter_metadata_safe(
+                    data=data,
+                    mcp_tools=expanded_tools,
+                    filtered_mcp_tools=filtered_expanded_tools,
+                    native_tools=native_tools_before_expand,
+                    filtered_tools=combined_tools,
+                )
                 verbose_proxy_logger.info(
                     f"Expanded MCP references to {len(expanded_tools)} tools "
                     f"({len(native_tools_before_expand)} native preserved), "
-                    f"skipping semantic filter (OpenAI nested format)"
+                    f"semantic filter selected {len(filtered_expanded_tools)}"
                 )
                 return data
 
@@ -297,19 +356,13 @@ class SemanticToolFilterHook(CustomLogger):
 
             data["tools"] = filtered_tools
 
-            try:
-                self._emit_filter_metadata(
-                    data=data,
-                    mcp_tools=mcp_tools,
-                    filtered_mcp_tools=filtered_mcp_tools,
-                    native_tools=native_tools,
-                    filtered_tools=filtered_tools,
-                )
-            except Exception as e:
-                verbose_proxy_logger.warning(
-                    f"Failed to emit semantic filter metadata: {e}",
-                    exc_info=True,
-                )
+            self._emit_filter_metadata_safe(
+                data=data,
+                mcp_tools=mcp_tools,
+                filtered_mcp_tools=filtered_mcp_tools,
+                native_tools=native_tools,
+                filtered_tools=filtered_tools,
+            )
 
             return data
 

litellm/proxy/hooks/parallel_request_limiter.py

@@ -17,6 +17,7 @@ from litellm.proxy.auth.auth_utils import (
     get_key_model_rpm_limit,
     get_key_model_tpm_limit,
 )
+from litellm.proxy.auth.budget_throttle import throttled_limit
 from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
 from litellm.proxy.hooks.rate_limiter_utils import resolve_llm_provider_for_rate_limit
 
@@ -248,10 +249,11 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
         if data is None:
             data = {}
         global_max_parallel_requests = data.get("metadata", {}).get("global_max_parallel_requests", None)
-        tpm_limit = getattr(user_api_key_dict, "tpm_limit", sys.maxsize)
+        throttle_pct = getattr(user_api_key_dict, "budget_throttle_pct", None)
+        tpm_limit = throttled_limit(getattr(user_api_key_dict, "tpm_limit", sys.maxsize), throttle_pct)
         if tpm_limit is None:
             tpm_limit = sys.maxsize
-        rpm_limit = getattr(user_api_key_dict, "rpm_limit", sys.maxsize)
+        rpm_limit = throttled_limit(getattr(user_api_key_dict, "rpm_limit", sys.maxsize), throttle_pct)
         if rpm_limit is None:
             rpm_limit = sys.maxsize
 

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -32,6 +32,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
 )
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.auth_utils import get_model_rate_limit_from_metadata
+from litellm.proxy.auth.budget_throttle import throttled_limit
 from litellm.proxy.common_utils.proxy_rate_limit_error import (
     ProxyRateLimitError,
     map_v3_rate_limit_type,
@@ -1549,18 +1550,19 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
             or user_api_key_dict.tpm_limit is not None
             or user_api_key_dict.max_parallel_requests is not None
         ):
+            throttle_pct = user_api_key_dict.budget_throttle_pct
             descriptors.append(
                 RateLimitDescriptor(
                     key="api_key",
                     value=user_api_key_dict.api_key,
                     rate_limit={
                         "requests_per_unit": self._get_enforced_limit(
-                            limit_value=user_api_key_dict.rpm_limit,
+                            limit_value=throttled_limit(user_api_key_dict.rpm_limit, throttle_pct),
                             limit_type=rpm_limit_type,
                             model_has_failures=model_has_failures,
                         ),
                         "tokens_per_unit": self._get_enforced_limit(
-                            limit_value=user_api_key_dict.tpm_limit,
+                            limit_value=throttled_limit(user_api_key_dict.tpm_limit, throttle_pct),
                             limit_type=tpm_limit_type,
                             model_has_failures=model_has_failures,
                         ),

litellm/proxy/management_endpoints/cache_settings_endpoints.py

@@ -38,13 +38,35 @@ from litellm.types.management_endpoints import (
 router = APIRouter()
 
 # Cache fields holding credentials. Masked on read so plaintext Redis /
-# Sentinel passwords never leave the server in a GET response.
-_CACHE_SENSITIVE_FIELDS: set = {"password", "sentinel_password"}
+# Sentinel passwords never leave the server in a GET response. `url` is here
+# because a Redis/Valkey URL can embed a password inline
+# (e.g. redis://:secret@host:6379/1).
+_CACHE_SENSITIVE_FIELDS: set = {"password", "sentinel_password", "url"}
 
 
 _REDACTED_VALUE = "***REDACTED***"
 
 
+_URL_OVERRIDDEN_CONNECTION_FIELDS: frozenset = frozenset({"host", "port", "db", "password", "username"})
+
+
+def _resolve_cache_url_precedence(settings: Mapping[str, Any]) -> dict[str, Any]:
+    """Return cache settings with the url-vs-discrete-fields ambiguity resolved.
+
+    When a full ``url`` is supplied it wins: the discrete
+    host/port/db/password/username fields are dropped so the persisted config
+    is unambiguous and matches runtime resolution in ``litellm._redis``
+    (``redis.Redis.from_url`` ignores them). Cluster mode
+    (``redis_startup_nodes``) is exempt because it authenticates via the
+    discrete fields rather than a url.
+    """
+    url = settings.get("url")
+    has_url = isinstance(url, str) and url.strip() != ""
+    if not has_url or settings.get("redis_startup_nodes"):
+        return dict(settings)
+    return {k: v for k, v in settings.items() if k not in _URL_OVERRIDDEN_CONNECTION_FIELDS}
+
+
 def _redact_settings(settings: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
     """Replace every value in a settings map with a fixed marker.
 
@@ -311,7 +333,7 @@ async def test_cache_connection(
     from litellm import Cache
 
     try:
-        cache_settings = request.cache_settings.copy()
+        cache_settings = _resolve_cache_url_precedence(request.cache_settings)
         verbose_proxy_logger.debug("Testing cache connection with settings: %s", cache_settings)
 
         # Only support Redis for now
@@ -378,7 +400,7 @@ async def update_cache_settings(
         )
 
     try:
-        cache_settings = request.cache_settings.copy()
+        cache_settings = _resolve_cache_url_precedence(request.cache_settings)
 
         # Snapshot the prior settings (key set only — values get redacted in
         # the audit row) so the audit-log entry shows which fields changed.