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 @@
         "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 @@ async def load_servers_from_config(
                     "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 @@ async def load_servers_from_config(
                 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 @@ async def build_mcp_server_from_table(
             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 @@ async def _noop(session):
             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 @@ def _build_mcp_server_table(self, server: MCPServer) -> LiteLLM_MCPServerTable:
             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 @@
     ResponsesAPIResponse,
 )
 from litellm.types.mcp import (
+    MCPAuth,
     MCPAuthType,
     MCPCredentials,
     MCPTransport,
@@ -1229,6 +1230,14 @@ class MCPApprovalStatus(str, enum.Enum):
 
 
 # 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 @@ def validate_credentials_requirements(cls, values):
         """
         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 @@ def validate_transport_fields(cls, values):
                     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

@@ -2325,13 +2325,35 @@ async def edit_mcp_server(
         # warning instead of failing the edit, whose primary job is the update itself.
         try:
             old_server_record = await get_mcp_server(prisma_client, payload.server_id)
+            old_server_record_read_failed = False
         except Exception as exc:  # noqa: BLE001 - advisory read; invalidation is best-effort end-to-end
             verbose_logger.warning(
                 "MCP server %s: could not snapshot the pre-update record; skipping the stale-token check: %s",
                 payload.server_id,
                 exc,
             )
             old_server_record = None
+            old_server_record_read_failed = True
+
+        if (
+            payload.dcr_bridge
+            and payload.auth_type is None
+            and (old_server_record is not None or old_server_record_read_failed)
+        ):
+            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(

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 @@ def is_oauth_delegate(self) -> bool:
         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 @@ async def test_load_servers_from_config_non_oauth2_needs_no_flow(self):
         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."""