feat(mcp): add dcr_bridge column and plumbing for client-forwarded auth modes

litellm-proxy-extras/litellm_proxy_extras/migrations/20260710000000_add_dcr_bridge_to_mcp_server_table/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN     "dcr_bridge" BOOLEAN;

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -339,6 +339,7 @@ model LiteLLM_MCPServerTable {
   available_on_public_internet Boolean @default(true)
   delegate_auth_to_upstream Boolean @default(false)
   oauth_passthrough Boolean @default(false)
+  dcr_bridge Boolean?
   is_byok               Boolean  @default(false)
   byok_description      String[] @default([])
   byok_api_key_help_url String?

litellm/models/mcp_server.py

@@ -96,6 +96,7 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False
     oauth_passthrough: bool = False
+    dcr_bridge: Optional[bool] = None
     is_byok: bool = False
     byok_description: List[str] = Field(default_factory=list)
     byok_api_key_help_url: Optional[str] = None

litellm/proxy/_experimental/mcp_server/db.py

@@ -52,6 +52,7 @@ _AUTH_FLOW_SCOPED_FIELDS: frozenset = frozenset(
         "token_url",
         "registration_url",
         "oauth2_flow",
+        "dcr_bridge",
         "token_exchange_endpoint",
         "audience",
         "subject_token_type",

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -1044,6 +1044,24 @@ class MCPServerManager:
                     "browser sign-in, including delegate_auth_to_upstream)."
                 )
 
+            config_dcr_bridge = server_config.get("dcr_bridge", None)
+            if config_dcr_bridge is not None and not isinstance(config_dcr_bridge, bool):
+                raise ValueError(
+                    f"Invalid config for MCP server '{server_name or server_id}': dcr_bridge "
+                    f"must be a boolean (got {config_dcr_bridge!r})."
+                )
+            if config_dcr_bridge and auth_type not in (
+                MCPAuth.true_passthrough,
+                MCPAuth.oauth_delegate,
+            ):
+                raise ValueError(
+                    f"Invalid config for MCP server '{server_name or server_id}': dcr_bridge is only "
+                    f"supported for auth_type true_passthrough or oauth_delegate (got {auth_type!r}). "
+                    "The DCR bridge serves gateway-hosted OAuth discovery for the client-forwarded "
+                    "token modes; interactive oauth2 servers already run the gateway "
+                    "authorization-code flow."
+                )
+
             new_server = MCPServer(
                 server_id=server_id,
                 name=name_for_prefix,
@@ -1079,6 +1097,7 @@ class MCPServerManager:
                 available_on_public_internet=bool(server_config.get("available_on_public_internet", True)),
                 delegate_auth_to_upstream=bool(server_config.get("delegate_auth_to_upstream", False)),
                 oauth_passthrough=bool(server_config.get("oauth_passthrough", False)),
+                dcr_bridge=config_dcr_bridge,
                 # AWS SigV4 fields
                 aws_access_key_id=server_config.get("aws_access_key_id", None),
                 aws_secret_access_key=server_config.get("aws_secret_access_key", None),
@@ -1454,6 +1473,7 @@ class MCPServerManager:
             available_on_public_internet=bool(getattr(mcp_server, "available_on_public_internet", True)),
             delegate_auth_to_upstream=bool(getattr(mcp_server, "delegate_auth_to_upstream", False)),
             oauth_passthrough=bool(getattr(mcp_server, "oauth_passthrough", False)),
+            dcr_bridge=getattr(mcp_server, "dcr_bridge", None),
             created_at=getattr(mcp_server, "created_at", None),
             updated_at=getattr(mcp_server, "updated_at", None),
             tool_name_to_display_name=_deserialize_json_dict(getattr(mcp_server, "tool_name_to_display_name", None)),
@@ -4800,6 +4820,7 @@ class MCPServerManager:
             token_url=server.token_url,
             registration_url=server.registration_url,
             oauth2_flow=server.oauth2_flow,
+            dcr_bridge=server.dcr_bridge,
             token_exchange_endpoint=server.token_exchange_endpoint,
             audience=server.audience,
             subject_token_type=server.subject_token_type,
@@ -4916,6 +4937,7 @@ class MCPServerManager:
             available_on_public_internet=server.available_on_public_internet,
             delegate_auth_to_upstream=server.delegate_auth_to_upstream,
             oauth_passthrough=getattr(server, "oauth_passthrough", False),
+            dcr_bridge=server.dcr_bridge,
             is_byok=server.is_byok,
             byok_description=server.byok_description,
             byok_api_key_help_url=server.byok_api_key_help_url,

litellm/proxy/_types.py

@@ -26,6 +26,7 @@ from litellm.types.llms.openai import (
     ResponsesAPIResponse,
 )
 from litellm.types.mcp import (
+    MCPAuth,
     MCPAuthType,
     MCPCredentials,
     MCPTransport,
@@ -1229,6 +1230,14 @@ from litellm.models.mcp_server import (  # noqa: E402
 
 
 # MCP Proxy Request Types
+def _dcr_bridge_auth_type_error(auth_type: object) -> ValueError:
+    return ValueError(
+        f"dcr_bridge is only supported for auth_type true_passthrough or oauth_delegate (got {auth_type!r}). "
+        "The DCR bridge serves gateway-hosted OAuth discovery for the client-forwarded token modes; "
+        "interactive oauth2 servers already run the gateway authorization-code flow."
+    )
+
+
 class NewMCPServerRequest(LiteLLMPydanticObjectBase):
     server_id: Optional[str] = None
     server_name: Optional[str] = None
@@ -1268,6 +1277,7 @@ class NewMCPServerRequest(LiteLLMPydanticObjectBase):
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False
     oauth_passthrough: bool = False
+    dcr_bridge: Optional[bool] = None
     is_byok: bool = False
     byok_description: List[str] = Field(default_factory=list)
     byok_api_key_help_url: Optional[str] = None
@@ -1322,6 +1332,16 @@ class NewMCPServerRequest(LiteLLMPydanticObjectBase):
         """
         return values
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_dcr_bridge_auth_type(cls, values):
+        if not isinstance(values, dict) or not values.get("dcr_bridge"):
+            return values
+        auth_type = values.get("auth_type")
+        if auth_type in (MCPAuth.true_passthrough, MCPAuth.oauth_delegate):
+            return values
+        raise _dcr_bridge_auth_type_error(auth_type)
+
 
 class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
     server_id: str
@@ -1362,6 +1382,7 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
     available_on_public_internet: bool = True
     delegate_auth_to_upstream: bool = False
     oauth_passthrough: bool = False
+    dcr_bridge: Optional[bool] = None
     is_byok: bool = False
     byok_description: List[str] = Field(default_factory=list)
     byok_api_key_help_url: Optional[str] = None
@@ -1391,6 +1412,21 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
                     raise ValueError("url or spec_path is required for HTTP/SSE transport")
         return values
 
+    @model_validator(mode="before")
+    @classmethod
+    def validate_dcr_bridge_auth_type(cls, values):
+        """Partial updates omit auth_type; that case is validated against the stored row by the
+        update endpoint, which can read the database. This validator covers payloads that carry
+        both fields."""
+        if not isinstance(values, dict) or not values.get("dcr_bridge"):
+            return values
+        if "auth_type" not in values:
+            return values
+        auth_type = values.get("auth_type")
+        if auth_type in (MCPAuth.true_passthrough, MCPAuth.oauth_delegate):
+            return values
+        raise _dcr_bridge_auth_type_error(auth_type)
+
 
 from litellm.models.mcp_server import (  # noqa: E402
     LiteLLM_MCPServerTable as LiteLLM_MCPServerTable,

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -2333,6 +2333,22 @@ if MCP_AVAILABLE:
             )
             old_server_record = None
 
+        if payload.dcr_bridge and payload.auth_type is None:
+            stored_auth_type = old_server_record.auth_type if old_server_record else None
+            stored_auth_type_name = getattr(stored_auth_type, "value", stored_auth_type)
+            if stored_auth_type not in (MCPAuth.true_passthrough, MCPAuth.oauth_delegate):
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail={
+                        "error": (
+                            "dcr_bridge is only supported for auth_type true_passthrough or "
+                            f"oauth_delegate (stored auth_type: {stored_auth_type_name!r}). Include "
+                            "the server's auth_type in the update payload or configure one of the "
+                            "client-forwarded token modes first."
+                        )
+                    },
+                )
+
         # try to update the mcp server
         mcp_server_record_updated = await update_mcp_server(
             prisma_client,

litellm/proxy/schema.prisma

@@ -339,6 +339,7 @@ model LiteLLM_MCPServerTable {
   available_on_public_internet Boolean @default(true)
   delegate_auth_to_upstream Boolean @default(false)
   oauth_passthrough Boolean @default(false)
+  dcr_bridge Boolean?
   is_byok               Boolean  @default(false)
   byok_description      String[] @default([])
   byok_api_key_help_url String?

litellm/types/mcp_server/mcp_server_manager.py

@@ -103,6 +103,7 @@ class MCPServer(BaseModel):
     # ``Authorization`` for non-OAuth reasons (e.g. static bearer tokens). Must
     # be set explicitly to avoid regressing servers that did not opt in.
     oauth_passthrough: bool = False
+    dcr_bridge: Optional[bool] = None
     is_byok: bool = False
     byok_description: List[str] = []
     byok_api_key_help_url: Optional[str] = None
@@ -164,6 +165,15 @@ class MCPServer(BaseModel):
         JWT) but forwards the caller's separate upstream ``Authorization`` unchanged, minting nothing."""
         return self.auth_type == MCPAuth.oauth_delegate
 
+    @property
+    def is_dcr_bridge(self) -> bool:
+        """True when this client-forwarded-token server serves the gateway-hosted DCR front door
+        (gateway-self protected-resource and authorization-server metadata plus the register,
+        authorize, and token relays) instead of relaying the upstream's own OAuth discovery
+        verbatim. ``dcr_bridge`` is rejected on every other auth type at create, update, and
+        config load, so the mode gate here only defends rows edited outside those paths."""
+        return bool(self.dcr_bridge) and (self.is_true_passthrough or self.is_oauth_delegate)
+
     @property
     def requires_per_user_auth(self) -> bool:
         """

schema.prisma

@@ -339,6 +339,7 @@ model LiteLLM_MCPServerTable {
   available_on_public_internet Boolean @default(true)
   delegate_auth_to_upstream Boolean @default(false)
   oauth_passthrough Boolean @default(false)
+  dcr_bridge Boolean?
   is_byok               Boolean  @default(false)
   byok_description      String[] @default([])
   byok_api_key_help_url String?

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_oauth_passthrough.py

@@ -110,6 +110,42 @@ def test_is_oauth_passthrough_false_without_authorization_header():
     assert server.is_oauth_passthrough is False
 
 
+@pytest.mark.parametrize("auth_type", [MCPAuth.true_passthrough, MCPAuth.oauth_delegate])
+def test_is_dcr_bridge_true_for_flagged_client_forwarded_modes(auth_type):
+    server = MCPServer(
+        server_id="s1",
+        name="s1",
+        transport=MCPTransport.http,
+        auth_type=auth_type,
+        dcr_bridge=True,
+    )
+    assert server.is_dcr_bridge is True
+
+
+@pytest.mark.parametrize("auth_type", [MCPAuth.true_passthrough, MCPAuth.oauth_delegate])
+def test_is_dcr_bridge_false_when_flag_unset(auth_type):
+    server = MCPServer(
+        server_id="s1",
+        name="s1",
+        transport=MCPTransport.http,
+        auth_type=auth_type,
+    )
+    assert server.dcr_bridge is None
+    assert server.is_dcr_bridge is False
+
+
+@pytest.mark.parametrize("auth_type", [MCPAuth.oauth2, MCPAuth.none, MCPAuth.api_key, None])
+def test_is_dcr_bridge_false_for_non_client_forwarded_auth_types(auth_type):
+    server = MCPServer(
+        server_id="s1",
+        name="s1",
+        transport=MCPTransport.http,
+        auth_type=auth_type,
+        dcr_bridge=True,
+    )
+    assert server.is_dcr_bridge is False
+
+
 def test_is_oauth_passthrough_false_without_extra_headers():
     server = MCPServer(
         server_id="s1",

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_partial_update.py

@@ -201,6 +201,7 @@ async def test_auth_type_switch_clears_stale_flow_scoped_fields():
         "token_url",
         "registration_url",
         "oauth2_flow",
+        "dcr_bridge",
         "token_exchange_endpoint",
         "audience",
         "subject_token_type",
@@ -225,6 +226,20 @@ async def test_auth_type_switch_keeps_explicitly_provided_flow_fields():
     assert data_dict["token_url"] is None
 
 
+@pytest.mark.asyncio
+async def test_auth_type_switch_to_client_forwarded_keeps_explicit_dcr_bridge():
+    data = UpdateMCPServerRequest(
+        server_id="my-test-server",
+        auth_type="true_passthrough",
+        dcr_bridge=True,
+    )
+
+    data_dict = await _run_update_with_existing(data, existing_auth_type="oauth2")
+
+    assert data_dict["dcr_bridge"] is True
+    assert data_dict["oauth2_flow"] is None
+
+
 @pytest.mark.asyncio
 async def test_auth_type_switch_back_to_oauth2_clears_token_exchange_fields():
     """The reverse switch must not leave token-exchange settings behind to
@@ -256,6 +271,7 @@ async def test_unchanged_auth_type_does_not_clear_flow_fields():
         "token_url",
         "registration_url",
         "oauth2_flow",
+        "dcr_bridge",
         "token_exchange_endpoint",
         "audience",
         "subject_token_type",

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -373,6 +373,66 @@ class TestMCPServerManager:
         server = next(iter(manager.config_mcp_servers.values()))
         assert server.oauth2_flow is None
 
+    def _client_forwarded_config(self, auth_type, **overrides):
+        base = {
+            "url": "https://example.com/mcp",
+            "transport": MCPTransport.http,
+            "auth_type": auth_type,
+        }
+        base.update(overrides)
+        return {"bridgeserver": base}
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_rejects_dcr_bridge_on_gateway_managed_auth_type(self):
+        manager = MCPServerManager()
+
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            pytest.raises(ValueError) as exc_info,
+        ):
+            await manager.load_servers_from_config(
+                self._oauth2_config(oauth2_flow="authorization_code", dcr_bridge=True)
+            )
+
+        assert "dcr_bridge is only supported" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_rejects_non_boolean_dcr_bridge(self):
+        manager = MCPServerManager()
+
+        with (
+            patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)),
+            pytest.raises(ValueError) as exc_info,
+        ):
+            await manager.load_servers_from_config(
+                self._client_forwarded_config(MCPAuth.true_passthrough, dcr_bridge="yes")
+            )
+
+        assert "must be a boolean" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("auth_type", [MCPAuth.true_passthrough, MCPAuth.oauth_delegate])
+    async def test_load_servers_from_config_accepts_dcr_bridge_on_client_forwarded_modes(self, auth_type):
+        manager = MCPServerManager()
+
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)):
+            await manager.load_servers_from_config(self._client_forwarded_config(auth_type, dcr_bridge=True))
+
+        server = next(iter(manager.config_mcp_servers.values()))
+        assert server.dcr_bridge is True
+        assert server.is_dcr_bridge is True
+
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_dcr_bridge_defaults_off(self):
+        manager = MCPServerManager()
+
+        with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)):
+            await manager.load_servers_from_config(self._client_forwarded_config(MCPAuth.true_passthrough))
+
+        server = next(iter(manager.config_mcp_servers.values()))
+        assert server.dcr_bridge is None
+        assert server.is_dcr_bridge is False
+
     @pytest.mark.asyncio
     async def test_load_servers_from_config_coerces_cost_string_to_float(self):
         """YAML 1.1 parses `7e-05` as a string; ingest must coerce it to float."""

tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -4900,6 +4900,138 @@ def test_oauth2_flow_defaults_to_none_when_omitted():
     assert LiteLLM_MCPServerTable(server_id="srv-1", transport="http").oauth2_flow is None
 
 
+def test_dcr_bridge_rejected_on_create_for_gateway_managed_auth_type():
+    from pydantic import ValidationError
+
+    from litellm.proxy._types import NewMCPServerRequest
+
+    with pytest.raises(ValidationError) as exc:
+        NewMCPServerRequest(
+            server_name="bridge-server",
+            url="https://example.com/mcp",
+            transport="http",
+            auth_type="oauth2",
+            oauth2_flow="authorization_code",
+            dcr_bridge=True,
+        )
+    assert "dcr_bridge is only supported" in str(exc.value)
+
+
+def test_dcr_bridge_rejected_on_create_when_auth_type_omitted():
+    from pydantic import ValidationError
+
+    from litellm.proxy._types import NewMCPServerRequest
+
+    with pytest.raises(ValidationError) as exc:
+        NewMCPServerRequest(
+            server_name="bridge-server",
+            url="https://example.com/mcp",
+            transport="http",
+            dcr_bridge=True,
+        )
+    assert "dcr_bridge is only supported" in str(exc.value)
+
+
+@pytest.mark.parametrize("auth_type", ["true_passthrough", "oauth_delegate"])
+def test_dcr_bridge_accepted_on_create_for_client_forwarded_modes(auth_type):
+    from litellm.proxy._experimental.mcp_server.db import _prepare_mcp_server_data
+    from litellm.proxy._types import NewMCPServerRequest
+
+    payload = NewMCPServerRequest(
+        server_name="bridge-server",
+        url="https://example.com/mcp",
+        transport="http",
+        auth_type=auth_type,
+        dcr_bridge=True,
+    )
+    data_dict = _prepare_mcp_server_data(payload)
+    assert data_dict["dcr_bridge"] is True
+
+
+def test_dcr_bridge_update_rejected_when_payload_auth_type_not_client_forwarded():
+    from pydantic import ValidationError
+
+    from litellm.proxy._types import UpdateMCPServerRequest
+
+    with pytest.raises(ValidationError) as exc:
+        UpdateMCPServerRequest(server_id="srv-1", auth_type="oauth2", dcr_bridge=True)
+    assert "dcr_bridge is only supported" in str(exc.value)
+
+
+def test_dcr_bridge_update_without_auth_type_defers_to_endpoint():
+    from litellm.proxy._types import UpdateMCPServerRequest
+
+    assert UpdateMCPServerRequest(server_id="srv-1", dcr_bridge=True).dcr_bridge is True
+
+
+def test_dcr_bridge_round_trips_on_response_model():
+    from litellm.proxy._types import LiteLLM_MCPServerTable
+
+    row = LiteLLM_MCPServerTable(server_id="srv-1", transport="http", dcr_bridge=True)
+    assert row.dcr_bridge is True
+    assert LiteLLM_MCPServerTable(server_id="srv-1", transport="http").dcr_bridge is None
+
+
+def _edit_endpoint_patches(old_record, update_mock):
+    return (
+        patch("litellm.proxy.management_endpoints.mcp_management_endpoints.MCP_AVAILABLE", True),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
+            return_value=MagicMock(),
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
+            AsyncMock(side_effect=old_record) if isinstance(old_record, Exception) else AsyncMock(return_value=old_record),
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.update_mcp_server",
+            update_mock,
+        ),
+        patch(
+            "litellm.proxy.management_endpoints.mcp_management_endpoints.validate_and_normalize_mcp_server_payload",
+            autospec=True,
+        ),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stored_auth_type", ["oauth2", "api_key", "none"])
+async def test_edit_mcp_server_rejects_dcr_bridge_when_stored_auth_type_not_client_forwarded(stored_auth_type):
+    from litellm.proxy._types import UpdateMCPServerRequest
+    from litellm.proxy.management_endpoints.mcp_management_endpoints import edit_mcp_server
+
+    old_record = MagicMock()
+    old_record.auth_type = stored_auth_type
+    update_mock = AsyncMock()
+    p1, p2, p3, p4, p5 = _edit_endpoint_patches(old_record, update_mock)
+    with p1, p2, p3, p4, p5:
+        payload = UpdateMCPServerRequest(server_id="srv-1", dcr_bridge=True)
+        user_auth = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
+        with pytest.raises(HTTPException) as exc:
+            await edit_mcp_server(payload=payload, user_api_key_dict=user_auth)
+
+    assert exc.value.status_code == 400
+    assert "dcr_bridge is only supported" in str(exc.value.detail)
+    update_mock.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_edit_mcp_server_rejects_dcr_bridge_when_stored_record_unreadable():
+    from litellm.proxy._types import UpdateMCPServerRequest
+    from litellm.proxy.management_endpoints.mcp_management_endpoints import edit_mcp_server
+
+    update_mock = AsyncMock()
+    p1, p2, p3, p4, p5 = _edit_endpoint_patches(RuntimeError("db down"), update_mock)
+    with p1, p2, p3, p4, p5:
+        payload = UpdateMCPServerRequest(server_id="srv-1", dcr_bridge=True)
+        user_auth = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
+        with pytest.raises(HTTPException) as exc:
+            await edit_mcp_server(payload=payload, user_api_key_dict=user_auth)
+
+    assert exc.value.status_code == 400
+    update_mock.assert_not_called()
+
+
 class TestPerUserCredentialConfigServerResolution:
     """Per-user credential and env-var endpoints must resolve config-defined MCP
     servers, which live only in the in-memory registry and never get a DB row, so