feat: isolate provider MCP registrations from gateway execution

docs/plans/provider-executed-mcp.md

@@ -1,6 +1,6 @@
 # Provider-executed MCP registration
 
-Status: schema approved, database changes prepared for local diff review
+Status: database slice approved and committed, runtime isolation slice prepared for local diff review
 
 ## Implementation plan
 
@@ -9,7 +9,7 @@ Work through these layers in order on `litellm_provider_mcp_registration`. Updat
 | Layer | Scope | Status |
 | --- | --- | --- |
 | 1 | Settle registration fields, values, storage, and validation | Approved |
-| 2 | Implement persistence and execution-mode isolation before enabling provider registration | Database schema/migration ready for review; runtime models, persistence validation, and isolation pending |
+| 2 | Implement persistence and execution-mode isolation before enabling provider registration | Database slice committed; typed mode and gateway isolation ready for review; persistence validation pending |
 | 3 | Management UI/API, stored tool options, and existing key/team/toolset selectors | Pending |
 | 4 | Authorize effective OpenAI Responses and Anthropic Messages MCP attachments through existing MCP permissions | Pending |
 | 5 | Live provider tests, streaming admission tests, UI tests, and gateway regressions | Pending |
@@ -118,7 +118,7 @@ Management reads return the stored field values, including `execution_mode`, `tr
 | Omitted `execution_mode` on create/config | Preserve existing LiteLLM execution and transport defaults |
 | Unknown or null `execution_mode` | Reject |
 | `execution_mode` omitted on update | Preserve stored mode |
-| Change an existing registration's mode | Reject in v1; create a separate registration |
+| `execution_mode` supplied on update | Reject in v1; this field is immutable and must be omitted from updates |
 | Missing or null provider `url` | Reject |
 | Provider URL | Require HTTPS; reject URL userinfo and fragments; lowercase scheme/hostname and omit default port 443; preserve path, trailing slash, and query identity |
 | Provider `transport` omitted or null | Persist null explicitly, rather than applying the database's legacy SSE default |
@@ -199,6 +199,14 @@ Existing `metadata.allowed_tools` remains an independent compatibility restricti
 | API coverage | OpenAI Responses and both Anthropic Messages formats; request-supplied and model-configured attachments; streaming and non-streaming |
 | Live evidence | Real providers execute authorized read-only MCP tools; denied requests produce zero provider dispatches; existing gateway execution still works |
 
-The first implementation slice contains the database schema and migration only. Provider registration is not enabled through management APIs or config, and executable models, connection isolation, and authorization integration remain pending. This is an intermediate review slice, not a complete or deployable provider-registration feature
+The approved database slice was committed as `fd0ac3ca76`. The next slice adds shared execution-mode types, stored-record transport validation, and gateway isolation. Stored records can represent both modes, while the connection-capable `MCPServer` type permits only `litellm`
+
+Provider records are excluded during incremental add/update and database catalog reload, before cached gateway entries can be reused. Direct table-to-runtime conversion and both shared client-creation boundaries reject provider mode before OAuth discovery, credential resolution, or MCP connection. Config loading currently rejects provider mode before discovery. The create request schema currently accepts only `litellm`, so `execution_mode=provider` cannot be silently ignored. Update requests reject any supplied execution mode
+
+This is an intermediate review slice. Management API/config creation of provider registrations remains disabled. Canonical URL and allowlist validation, merged-record update validation, provider inventory loading, and management tool selection are still pending. Later work will enable creation after those contracts are wired in. Native LLM admission authorization is unchanged
 
 Database-slice validation passed with Prisma schema validation, identical schema copies, the repository's no-data-rewriting migration check, and six checks on a disposable PostgreSQL 16 database built from the pre-change schema. Those checks cover existing rows and duplicate gateway URLs, new legacy defaults, provider records with null connection fields and exact tool lists, duplicate provider URL rejection, distinct provider URLs with empty tool lists, and invalid/null mode rejection. Application-level URL canonicalization, field validation, and authorization are not covered by this database-only proof
+
+Runtime isolation was exercised against a real local MCP server. The existing gateway successfully initialized, listed tools, and called an echo tool, producing 11 HTTP requests. Provider-mode table conversion, config loading, manager client creation, and direct upstream client creation were rejected with zero additional HTTP requests. Provider rows produced zero executable catalog entries. This proves the local execution boundary, not native-provider request authorization
+
+The runtime slice passes 369 focused model, request, catalog, and MCP utility tests, along with Ruff and the test lint rules. A type-check comparison of the six changed runtime files found no new diagnostics against the approved commit, and strict Ruff rule counts did not increase. These are focused slice checks, not a full repository CI run

litellm/models/mcp_server.py

@@ -11,10 +11,11 @@ from datetime import datetime
 from types import MappingProxyType
 from typing import Literal
 
-from pydantic import Field, ValidationInfo, field_validator
+from pydantic import Field, ValidationInfo, field_validator, model_validator
+from typing_extensions import Self
 
 from litellm.types.llms.base import LiteLLMPydanticObjectBase
-from litellm.types.mcp import MCPAuthType, MCPCredentials, MCPTransportType
+from litellm.types.mcp import MCPAuthType, MCPCredentials, MCPExecutionMode, MCPTransportType
 from litellm.types.mcp_server.mcp_server_manager import MCPInfo, PinnedMCPTool, parse_pinned_tools
 
 
@@ -56,7 +57,8 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
     description: str | None = None
     url: str | None = None
     spec_path: str | None = None
-    transport: MCPTransportType
+    execution_mode: MCPExecutionMode = MCPExecutionMode.litellm
+    transport: MCPTransportType | None
     auth_type: MCPAuthType | None = None
     credentials: MCPCredentials | None = None
     instructions: str | None = None
@@ -121,6 +123,14 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
     reviewed_at: datetime | None = None
     review_notes: str | None = None
 
+    @model_validator(mode="after")
+    def validate_execution_transport(self) -> Self:
+        if self.execution_mode == MCPExecutionMode.provider and self.transport is not None:
+            raise ValueError("Provider-executed MCP registrations require transport=null")
+        if self.execution_mode == MCPExecutionMode.litellm and self.transport is None:
+            raise ValueError("LiteLLM-executed MCP registrations require a transport")
+        return self
+
     @field_validator("pinned_tools", mode="before")
     @classmethod
     def decode_stored_pinned_tools(cls, value: object) -> dict[str, PinnedMCPTool] | None:

litellm/proxy/_types.py

@@ -63,6 +63,8 @@ from litellm.types.mcp import (
     MCPAuthType,
     MCPCredentials,
     MCPDisabledCapabilities,
+    MCPExecutionMode,
+    MCPGatewayExecutionMode,
     MCPTransport,
     MCPTransportType,
     MCPUpstreamProtocol,
@@ -1376,6 +1378,7 @@ def _validate_mcp_transport_fields(values: object) -> None:
 
 
 class NewMCPServerRequest(LiteLLMPydanticObjectBase):
+    execution_mode: MCPGatewayExecutionMode = MCPExecutionMode.litellm
     server_id: str | None = None
     server_name: str | None = None
     alias: str | None = None
@@ -1533,6 +1536,13 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
     max_concurrent_requests: int | None = None
     rpm: int | None = Field(default=None, ge=0)
 
+    @model_validator(mode="before")
+    @classmethod
+    def reject_execution_mode_changes(cls, values: object) -> object:
+        if isinstance(values, dict) and "execution_mode" in values:
+            raise ValueError("execution_mode is immutable; omit it when updating an MCP registration")
+        return values
+
     @model_validator(mode="after")
     def validate_protocol_transport(self) -> "UpdateMCPServerRequest":
         if not {"transport", "mcp_info"}.issubset(self.model_fields_set):

litellm/proxy/mcp_server/catalog.py

@@ -24,6 +24,7 @@ from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
 from litellm._logging import verbose_logger
 from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.proxy.mcp_server.result_conversion import age_freshness, aggregate_freshness
+from litellm.types.mcp import MCPExecutionMode
 
 if TYPE_CHECKING:
     from mcp.types import ListToolsResult, PaginatedRequestParams, PaginatedResult
@@ -415,6 +416,8 @@ class CatalogSnapshots:
         for row in rows:
             try:
                 server = LiteLLM_MCPServerTable.model_validate(row.model_dump())
+                if server.execution_mode == MCPExecutionMode.provider:
+                    continue
                 existing_server = previous_registry.get(server.server_id)
 
                 if (

litellm/proxy/mcp_server/mcp_server_manager.py

@@ -230,8 +230,10 @@ from litellm.types.llms.custom_http import httpxSpecialProvider
 from litellm.types.mcp import (
     DEFAULT_SUBJECT_TOKEN_TYPE,
     MCPAuth,
+    MCPExecutionMode,
     MCPTokenEndpointAuthMethod,
     MCPUpstreamProtocol,
+    validate_mcp_gateway_execution,
     without_header,
 )
 from litellm.types.mcp_server.mcp_server_manager import (
@@ -405,6 +407,7 @@ class MCPServerConfig(TypedDict, total=False):
     whatever the admin wrote, and each read applies its own default."""
 
     server_id: ReadOnly[str]
+    execution_mode: ReadOnly[MCPExecutionMode]
     protocol_version: ReadOnly[MCPUpstreamProtocol]
     alias: str
     description: str
@@ -2583,6 +2586,7 @@ class MCPServerManager:
 
         for server_name, raw_server_config in mcp_servers_config.items():
             server_config: MCPServerConfig = raw_server_config
+            validate_mcp_gateway_execution(server_config.get("execution_mode", MCPExecutionMode.litellm))
             _mcp_info: MCPInfo = server_config.get("mcp_info", None) or {}
             # Preserve all custom fields from config while setting defaults for core fields
             mcp_info: MCPInfo = _mcp_info.copy()
@@ -3219,6 +3223,7 @@ class MCPServerManager:
         env_vars_are_encrypted: bool | None = None,
         register_oauth_discovery: bool = True,
     ) -> MCPServer:
+        validate_mcp_gateway_execution(mcp_server.execution_mode)
         _mcp_info: Final[MCPInfo] = mcp_server.mcp_info or {}
         env_dict: Final = _deserialize_json_dict(getattr(mcp_server, "env", None))
         static_headers_dict: Final = _deserialize_json_dict(getattr(mcp_server, "static_headers", None))
@@ -3485,6 +3490,10 @@ class MCPServerManager:
                 self.initialize_tool_name_to_mcp_server_name_mapping()
 
     async def add_server(self, mcp_server: LiteLLM_MCPServerTable):
+        if mcp_server.execution_mode == MCPExecutionMode.provider:
+            if mcp_server.server_id in self.registry:
+                self.remove_server(mcp_server)
+            return
         # The runtime registry is the allowlist for tool calls and health
         # probes (which spawn the underlying transport, including stdio
         # subprocesses). Match the eligibility set used by the bulk DB
@@ -3518,7 +3527,10 @@ class MCPServerManager:
         # If a previously-active server has been moved out of the active
         # state, evict any stale registry entry so subsequent tool calls and
         # health probes can't reach it.
-        if mcp_server.approval_status not in (None, "active", "approved"):
+        if (
+            mcp_server.execution_mode == MCPExecutionMode.provider
+            or mcp_server.approval_status not in (None, "active", "approved")
+        ):
             evicted = self.registry.pop(mcp_server.server_id, None)
             if evicted is None and mcp_server.server_name:
                 evicted = self.registry.pop(mcp_server.server_name, None)
@@ -4388,6 +4400,7 @@ class MCPServerManager:
         protocol_version_override: MCPUpstreamProtocol | None = None,
         for_tool_listing: bool = False,
     ) -> MCPClient:
+        validate_mcp_gateway_execution(server.execution_mode)
         self.catalog.assert_current(server)
         record_auth_resolution(server.server_id, AuthResolution.unresolved)
         resolved_server: Final = await self.ensure_oauth_metadata_discovered(server)

litellm/proxy/mcp_server/upstream.py

@@ -40,6 +40,7 @@ from litellm.types.mcp import (
     MCPTransportType,
     MCPUpstreamProtocol,
     has_header,
+    validate_mcp_gateway_execution,
     without_header,
 )
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
@@ -210,6 +211,7 @@ async def prepare_upstream_client(
     upstream_transport: httpx2.AsyncBaseTransport | None = None,
     for_tool_listing: bool = False,
 ) -> MCPClient:
+    validate_mcp_gateway_execution(server.execution_mode)
     transport: Final[MCPTransportType] = server.transport or MCPTransport.sse
     server_spec: Final = None if transport == MCPTransport.stdio else to_server_spec_fail_closed(server)
     spec: Final = (

litellm/types/mcp.py

@@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, TypeAlias
 from urllib.parse import urlsplit
 
 import httpx
-from pydantic import ConfigDict, Field
+from pydantic import ConfigDict, Field, TypeAdapter
 from typing_extensions import ReadOnly, TypedDict
 
 from litellm.types.llms.base import HiddenParams, LiteLLMBaseModel
@@ -24,6 +24,19 @@ else:
     MCPTextContent = Any
 
 
+class MCPExecutionMode(str, enum.Enum):
+    litellm = "litellm"
+    provider = "provider"
+
+
+MCPGatewayExecutionMode: TypeAlias = Literal[MCPExecutionMode.litellm]
+_GATEWAY_EXECUTION_MODE: Final = TypeAdapter[MCPGatewayExecutionMode](MCPGatewayExecutionMode)
+
+
+def validate_mcp_gateway_execution(value: object) -> MCPGatewayExecutionMode:
+    return _GATEWAY_EXECUTION_MODE.validate_python(value)
+
+
 class MCPTransport(str, enum.Enum):
     sse = "sse"
     http = "http"

litellm/types/mcp_server/mcp_server_manager.py

@@ -11,6 +11,8 @@ from litellm.types.mcp import (
     DEFAULT_SUBJECT_TOKEN_TYPE,
     MCPAuth,
     MCPAuthType,
+    MCPExecutionMode,
+    MCPGatewayExecutionMode,
     MCPTokenEndpointAuthMethod,
     MCPTransportType,
     MCPUpstreamProtocol,
@@ -94,6 +96,7 @@ def parse_pinned_tools(value: object) -> dict[str, PinnedMCPTool] | None:
 class MCPServer(LiteLLMBaseModel):
     server_id: str
     name: str
+    execution_mode: MCPGatewayExecutionMode = MCPExecutionMode.litellm
     alias: str | None = None
     server_name: str | None = None
     url: str | None = None

tests/unit/models/test_models.py

@@ -546,6 +546,26 @@ class TestMCPServerTable:
         with pytest.raises(ValidationError):
             LiteLLM_MCPServerTable(server_id="s1")
 
+    def test_provider_registration_round_trips_without_connection_fields(self) -> None:
+        record: Final = LiteLLM_MCPServerTable(
+            server_id="provider-docs",
+            execution_mode="provider",
+            transport=None,
+            credentials=None,
+            url="https://docs.example/mcp",
+            allowed_tools=["search_docs"],
+        )
+        restored: Final = LiteLLM_MCPServerTable.model_validate_json(record.model_dump_json())
+        assert restored.execution_mode == "provider"
+        assert restored.transport is None
+        assert restored.credentials is None
+        assert restored.allowed_tools == ["search_docs"]
+
+    @pytest.mark.parametrize("mode,transport", [("provider", "http"), ("litellm", None)])
+    def test_execution_mode_requires_the_matching_transport_shape(self, mode: str, transport: str | None) -> None:
+        with pytest.raises(ValidationError, match="transport"):
+            LiteLLM_MCPServerTable(server_id="s1", execution_mode=mode, transport=transport)
+
 
 class TestSpendLogs:
     def test_spend_logs_creation(self):

tests/unit/proxy/mcp_server/test_catalog.py

@@ -772,3 +772,82 @@ def test_partial_optional_catalog_never_advertises_freshness() -> None:
     assert result.ttl_ms == 0
     assert result.cache_scope == "private"
 
+
+
+@pytest.mark.asyncio
+async def test_provider_records_do_not_enter_or_survive_in_the_gateway_catalog() -> None:
+    from datetime import datetime, timezone
+
+    from litellm.models.mcp_server import LiteLLM_MCPServerTable
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    manager: Final = MCPServerManager()
+    updated_at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    record: Final = LiteLLM_MCPServerTable(
+        server_id="docs", execution_mode="provider", transport=None, url="https://docs.example/mcp", updated_at=updated_at
+    )
+    await manager.add_server(record)
+    assert manager.get_mcp_server_by_id("docs") is None
+
+    legacy: Final = MCPServer(server_id="docs", name="docs", transport="http", url=record.url, updated_at=updated_at)
+    manager.registry[legacy.server_id] = legacy
+    assert await manager.catalog._stage_servers((record,), reuse_unchanged=True) == {}
+    await manager.update_server(record)
+    assert manager.get_mcp_server_by_id("docs") is None
+    manager.registry[legacy.server_id] = legacy
+    await manager.add_server(record)
+    assert manager.get_mcp_server_by_id("docs") is None
+
+
+@pytest.mark.asyncio
+async def test_provider_table_cannot_be_converted_to_a_gateway_client() -> None:
+    from pydantic import ValidationError
+
+    from litellm.models.mcp_server import LiteLLM_MCPServerTable
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    manager: Final = MCPServerManager()
+    record: Final = LiteLLM_MCPServerTable(
+        server_id="docs", execution_mode="provider", transport=None, url="https://docs.example/mcp"
+    )
+    with pytest.raises(ValidationError):
+        await manager.build_mcp_server_from_table(record)
+
+
+@pytest.mark.asyncio
+async def test_config_cannot_silently_load_provider_mode_as_a_gateway_server() -> None:
+    from pydantic import ValidationError
+
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+    from litellm.types.mcp import MCPAuth, MCPExecutionMode
+
+    manager: Final = MCPServerManager()
+    with pytest.raises(ValidationError):
+        await manager.load_servers_from_config(
+            {"docs": {"execution_mode": MCPExecutionMode.provider, "url": "https://docs.example/mcp", "auth_type": MCPAuth.oauth2}}
+        )
+    assert manager.get_registry() == {}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("boundary", ["manager", "upstream"])
+async def test_provider_mode_is_rejected_before_upstream_auth_or_connection(boundary: str) -> None:
+    from pydantic import ValidationError
+
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+    from litellm.proxy.mcp_server.upstream import prepare_upstream_client
+    from litellm.types.mcp import MCPExecutionMode
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    manager: Final = MCPServerManager()
+    server: Final = MCPServer(server_id="docs", name="docs", transport="http", auth_type="oauth2").model_copy(
+        update={"execution_mode": MCPExecutionMode.provider, "url": "https://docs.example/mcp"}
+    )
+    attempt: Final = (
+        manager.create_mcp_client(server)
+        if boundary == "manager"
+        else prepare_upstream_client(server, provider=manager._cred_provider, root_path="", protocol_version="auto")
+    )
+    with pytest.raises(ValidationError):
+        await attempt

tests/unit/proxy/test__types.py

@@ -50,6 +50,20 @@ SERVER_ONLY_MARKERS = (
 )
 
 
+@pytest.mark.parametrize("request_type", [NewMCPServerRequest, UpdateMCPServerRequest])
+def test_gateway_management_requests_do_not_ignore_provider_execution_mode(
+    request_type: type[NewMCPServerRequest] | type[UpdateMCPServerRequest],
+) -> None:
+    with pytest.raises(ValidationError, match="execution_mode"):
+        request_type(server_id="docs", url="https://docs.example/mcp", execution_mode="provider")
+
+
+@pytest.mark.parametrize("mode", ["litellm", "provider", None])
+def test_update_requests_cannot_replace_a_registrations_execution_mode(mode: str | None) -> None:
+    with pytest.raises(ValidationError, match="execution_mode is immutable"):
+        UpdateMCPServerRequest(server_id="docs", execution_mode=mode)
+
+
 def _server_only_marker_value(marker: str, value: str) -> str | tuple[str, ...]:
     return (value,) if marker == "mcp_session_resource_server_ids" else value