feat: support provider MCP registration in config

docs/plans/provider-executed-mcp.md

@@ -1,6 +1,6 @@
 # Provider-executed MCP registration
 
-Status: database, isolation, and admin registration API slices approved. Scoped retrieval and provider config loading are intentionally deferred
+Status: database, isolation, and admin registration API slices approved and pushed through `1ba99633e4`. Provider YAML registration and the explicit provider input model are reviewed and approved for commit and push. Permission integration and native-request enforcement are next
 
 ## Implementation plan
 
@@ -9,18 +9,22 @@ 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 and isolation committed; API persistence validation approved; YAML provider inventory pending |
+| 2 | Implement persistence and execution-mode isolation before enabling provider registration | Database, isolation, and admin API committed; YAML registration and provider input model reviewed and approved |
 | 3 | Management UI/API, stored tool options, and existing key/team/toolset selectors | Admin CRUD and inventory API approved; UI and scoped 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 |
 
+The next phase connects existing object permissions to provider registrations and enforces native attachments before provider dispatch. Existing permission storage can represent the IDs and tools, but listing a server in the admin catalog does not complete this integration. Assignment validation still resolves config entries through executable inventory, scoped lists still retrieve executable server objects, and tool pickers still call `/mcp-rest/tools/list`. Those paths must read registration identity and stored allowed tools from management inventory for provider mode
+
+Reuse the existing server/tool grant resolution functions, groups, toolsets, and explicit denies. Apply the agreed key/team/server ceiling to provider authorization. The current resolver also has legacy additive access-group grants and some permissive lookup-error handling, so calling it unchanged would not guarantee that contract. Native admission must resolve the canonical URL, require provider execution mode, and validate a finite requested tool scope after effective request configuration is assembled. Cover both native APIs and streaming, with denied requests producing no provider dispatch. Provider MCP discovery/execution stays outside LiteLLM
+
 V1 supports URL-based OpenAI Responses MCP and both Anthropic Messages connector formats. Request authorization applies to streaming and non-streaming calls before provider dispatch. Server restrictions, team permissions, and key permissions intersect. Existing applicable permission ceilings and explicit denies continue to apply
 
 Response/SSE inspection, per-invocation argument policy inside a provider's tool loop, automatic MCP discovery for provider registrations, local MCP references, provider connector IDs, and additional API bridges are outside v1
 
 ## Current review acceptance criteria
 
-This batch covers admin CRUD, input validation, cached management inventory, and execution isolation. Key/team-scoped retrieval, selectors, toolsets, and native-request admission remain later layers. Their absence is intentional for this batch
+The approved batch covered admin CRUD, input validation, cached management inventory, and execution isolation. The current batch adds YAML registration to that same catalog. Key/team-scoped retrieval, selectors, toolsets, and native-request admission remain later layers. Their absence is intentional for this batch
 
 Core provider fields remain `execution_mode`, `server_name`, `url`, `allowed_tools`, and optional `tool_name_to_description`, alongside the existing shared management fields. Neutral connection values remain accepted where the shared schema permits them, but never enable MCP execution. Active credentials, transports, OAuth settings, commands, and other connection configuration are rejected. No strict five-field schema rewrite is part of this revision
 
@@ -36,7 +40,41 @@ Provider updates keep read-merge-validation atomic through the existing writer.
 
 The later permissions layer will test matching list/detail visibility under key/team intersection, explicit denial, disjoint scopes, revocation after caching, and separation of read grants from mutation privileges. The admin-only live run is not evidence for that layer
 
-The next config slice should accept provider entries under the existing `mcp_servers` key, reuse registration validation, and publish read-only management records in the shared catalog without constructing gateway clients. Verify startup and reload, removal, stable IDs, conflicting config/database identities or URLs, and zero upstream activity. Existing LiteLLM config loading must remain unchanged
+The config slice accepts provider entries under the existing `mcp_servers` key, reuses registration validation, and publishes read-only management records in the shared catalog without constructing gateway clients. It covers startup and reload, removal, stable IDs, conflicting config/database identities or URLs, and zero upstream activity. Existing LiteLLM entries use the existing gateway loader
+
+```yaml
+mcp_servers:
+  docs:
+    server_id: docs-native
+    execution_mode: provider
+    url: https://docs.example/mcp
+    allowed_tools:
+      - search_docs
+      - fetch_doc
+    tool_name_to_description:
+      search_docs: Search documentation
+      fetch_doc: Read a documentation page
+```
+
+`server_id` is optional. Without it, the ID is derived from execution mode, config name, canonical URL, and alias. Tool-list edits keep the derived ID stable. Pin the ID when it must survive renaming or endpoint changes. Config registrations return `is_config=true`, appear in admin list/detail, and are edited through config rather than management CRUD
+
+Config registration review follows this control flow
+
+1. `ProxyConfig._init_non_llm_configs` passes nonempty `mcp_servers` to the manager. Removing the section clears provider config registrations without reloading unrelated gateway entries
+2. `MCPServerManager.load_servers_from_config` validates provider entries through `NewMCPServerRequest` and its `ProviderMCPRegistration` model, assigns identities, and checks conflicts before invoking the existing loader with gateway entries only. It publishes provider records into the catalog after validation
+3. `CatalogSnapshots` combines provider config records, database registrations, and gateway inventory into one management snapshot. Only gateway records enter executable inventory. Config publication invalidates the shared snapshot, while an in-flight operation retains its captured snapshot
+4. Admin list/detail read the catalog. Provider config detail works without a database and skips reachability probes. Config entries remain read-only through the management API
+5. API create/update reject collisions with config IDs, names, aliases, or canonical provider URLs. Database refresh applies the same conflict rule when config was loaded first. A gateway and provider registration may share a URL if their identities are distinct
+
+Validation passed 1,333 focused tests and 37 live proxy/PostgreSQL checks. A separate production-initializer run covered YAML reload, removal, stable IDs, request snapshot consistency, and rejected-input preservation with a real loopback recorder and no mocks. Both live runs recorded zero provider MCP connections. The initializer run fails on approved HEAD because provider config was previously rejected
+
+`ProviderMCPRegistration` now declares the supported fields and owns their validation. There is no connection-field blacklist to extend when gateway settings are added. Other active fields reject automatically, while neutral values are discarded. The existing create/update request models remain the shared API types. For updates, validation waits until the stored mode is known and the patch is merged under the existing row lock. The update model retains unknown active field names privately so Pydantic cannot silently discard them before provider validation. It retains no unknown field values and preserves legacy gateway handling
+
+Provider updates project stored data onto the existing request schema, excluding response-only audit/health fields before validation. Only supported supplied fields are written. Neutral connection settings do not cause credential or transport writes. Three regression cases for an unrecognized active setting failed before this revision and pass afterward across create validation, YAML loading, and updates
+
+One SSRF test was excluded from the focused pass count. It fails identically on approved HEAD at the initial `validate_url` assertion, before config loading. Changed-file Ruff and formatting passed, strict rule counts did not increase, and the type-check comparison introduced no diagnostics. These are focused local checks, not a full repository CI run
+
+The separate usage audit made eight real OpenAI Responses and Anthropic Messages requests, comparing control/native MCP and JSON/streaming. All four MCP cases executed successfully. Client usage, normalized token totals, and model costs matched upstream usage and registry pricing. Native server/tool spend attribution and some Anthropic normalized usage details remain follow-ups. No accounting changes are included in this slice
 
 ## Current invocation routing
 
@@ -221,11 +259,11 @@ Existing `metadata.allowed_tools` remains an independent compatibility restricti
 
 The approved database slice was committed as `fd0ac3ca76` and the approved isolation slice as `42ea534932`. Stored records can represent both modes, while the connection-capable `MCPServer` type permits only `litellm`. The redundant named gateway-mode alias is replaced with a literal annotation, leaving one named execution-mode enum
 
-Provider records remain 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. YAML loading still rejects provider mode until its inventory path is implemented
+Provider records remain 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. The current YAML slice adds provider registrations to management inventory while preserving that execution boundary
 
 The current review slice enables provider registration on the existing create API, canonicalizes HTTPS endpoint identity, requires an explicit finite tool list, and rejects active gateway connection settings. Create persists null transport/auth fields and an enforced allowlist. Update validates the merged record under a row lock, preserves omitted fields, rejects null tool lists, and rejects changes to execution mode. The existing provider-only unique index rejects duplicate endpoints with HTTP 409
 
-Admin management inventory includes active provider records. Detail reads return stored allowed tools without running health probes. The UI, scoped key/team selectors, toolset integration, YAML provider inventory, and native LLM admission authorization remain pending. Registration management alone does not yet enforce native-request permissions
+Admin management inventory includes active provider records. Detail reads return stored allowed tools without running health probes. The UI, scoped key/team selectors, toolset integration, and native LLM admission authorization remain pending. YAML provider inventory is implemented for review. Registration management alone does not yet enforce native-request permissions
 
 Management inventory and executable servers share one revision-based catalog snapshot. A refresh reads active rows once, retains provider registrations for management, and excludes them from runtime construction. Management listing does not issue a second database inventory query
 

litellm/proxy/_types.py

@@ -15,7 +15,9 @@ from pydantic import (
     Field,
     Json,
     JsonValue,
+    ModelWrapValidatorHandler,
     PositiveInt,
+    PrivateAttr,
     TypeAdapter,
     field_validator,
     model_validator,
@@ -70,7 +72,7 @@ from litellm.types.mcp import (
     validate_mcp_protocol_transport,
 )
 from litellm.types.mcp_server.mcp_server_manager import MCPInfo
-from litellm.types.mcp_server.provider_registration import normalize_provider_registration
+from litellm.types.mcp_server.provider_registration import active_registration_extras, normalize_provider_registration
 from litellm.types.proxy.agent_identity import ManagedAgentContext
 from litellm.types.proxy.auth.special_headers import (
     SpecialHeaders as SpecialHeaders,  # noqa: PLC0414  # public re-export
@@ -1499,6 +1501,8 @@ class NewMCPServerRequest(LiteLLMPydanticObjectBase):
 
 
 class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
+    _provider_extra_fields: tuple[str, ...] = PrivateAttr(default=())
+
     server_id: str
     server_name: str | None = None
     alias: str | None = None
@@ -1548,6 +1552,28 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
     max_concurrent_requests: int | None = None
     rpm: int | None = Field(default=None, ge=0)
 
+    @model_validator(mode="wrap")
+    @classmethod
+    def retain_extra_field_names(
+        cls,
+        values: object,
+        handler: ModelWrapValidatorHandler["UpdateMCPServerRequest"],
+    ) -> "UpdateMCPServerRequest":
+        validated: Final = handler(values)
+        if isinstance(values, Mapping):
+            validated._provider_extra_fields = active_registration_extras(
+                TypeAdapter(dict[str, object]).validate_python(values),
+                cls.model_fields.keys(),
+            )
+        return validated
+
+    def provider_registration_patch(self, fields_set: set[str] | None = None) -> dict[str, object]:
+        if self._provider_extra_fields:
+            raise ValueError(
+                f"Provider-executed MCP registrations do not accept {', '.join(self._provider_extra_fields)}"
+            )
+        return self.model_dump(include=fields_set, exclude_unset=True)
+
     @model_validator(mode="before")
     @classmethod
     def reject_execution_mode_changes(cls, values: object) -> object:

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -258,6 +258,7 @@ if MCP_AVAILABLE:
     )
     from litellm.proxy.mcp_server.server_resolution import (
         MCPServerTargetCatalog,
+        ResolvedMCPServer,
         authorize_mcp_server,
         resolve_mcp_server,
     )
@@ -1350,14 +1351,17 @@ if MCP_AVAILABLE:
             await global_mcp_server_manager.get_all_mcp_servers_unfiltered() if view_all or view_providers else []
         )
         visible_inventory: Final = [
-            server for server in inventory
+            server
-            if view_providers or getattr(server, "execution_mode", MCPExecutionMode.litellm) != MCPExecutionMode.provider
+            for server in inventory
+            if view_providers
+            or getattr(server, "execution_mode", MCPExecutionMode.litellm) != MCPExecutionMode.provider
         ]
         if view_all:
             return visible_inventory
 
         provider_servers: Final = [
-            server for server in visible_inventory
+            server
+            for server in visible_inventory
             if getattr(server, "execution_mode", MCPExecutionMode.litellm) == MCPExecutionMode.provider
         ]
         aggregated: Final[dict[str, LiteLLM_MCPServerTable]] = {server.server_id: server for server in provider_servers}
@@ -1964,20 +1968,40 @@ if MCP_AVAILABLE:
         --header 'Authorization: Bearer your_api_key_here'
         ```
         """
-        prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
-
         from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 
         client_ip: Final = IPAddressUtils.get_mcp_client_ip(request)
         is_admin_view: Final = user_api_key_has_admin_view(user_api_key_dict)
         is_restricted_virtual_key: Final = _is_restricted_virtual_key_request(user_api_key_dict)
-        resolved: Final = await resolve_mcp_server(
+        snapshot: Final = global_mcp_server_manager.catalog.current()
-            server_id,
+        config_match: Final = (
-            manager=global_mcp_server_manager,
+            next(
-            db_lookup=lambda sid: get_mcp_server(prisma_client, sid),
+                (
-            id_client_ip=client_ip,
+                    server
-            name_client_ip=client_ip,
+                    for server in snapshot.management_servers.values()
-            match_name=True,
+                    if server.is_config
+                    and server.execution_mode == MCPExecutionMode.provider
+                    and server_id in (server.server_id, server.server_name, server.alias)
+                ),
+                None,
+            )
+            if snapshot is not None
+            else None
+        )
+        resolved: Final = (
+            ResolvedMCPServer(table=config_match, runtime=None, source="registry")
+            if config_match is not None
+            else await resolve_mcp_server(
+                server_id,
+                manager=global_mcp_server_manager,
+                db_lookup=lambda sid: get_mcp_server(
+                    get_prisma_client_or_throw("Database not connected. Connect a database to your proxy"),
+                    sid,
+                ),
+                id_client_ip=client_ip,
+                name_client_ip=client_ip,
+                match_name=True,
+            )
         )
         authorized: Final = await authorize_mcp_server(
             resolved,
@@ -2109,6 +2133,8 @@ if MCP_AVAILABLE:
             raise HTTPException(
                 status_code=409, detail="An MCP registration already uses this identity or provider endpoint"
             ) from exc
+        except HTTPException:
+            raise
         except Exception as e:
             verbose_proxy_logger.exception("Error creating mcp server: %s", e)
             raise HTTPException(

litellm/proxy/mcp_server/catalog.py

@@ -26,6 +26,7 @@ from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.models.mcp_server import LiteLLM_MCPServerTable
 from litellm.proxy.mcp_server.result_conversion import age_freshness, aggregate_freshness
 from litellm.types.mcp import MCPExecutionMode
+from litellm.types.mcp_server.provider_registration import canonical_provider_mcp_url
 
 if TYPE_CHECKING:
     from mcp.types import ListToolsResult, PaginatedRequestParams, PaginatedResult
@@ -92,8 +93,39 @@ def _check_oauth_revision(selected: MCPServer, candidate: MCPServer | None) -> N
         raise HTTPException(status_code=503, detail="OAuth metadata discovery changed repeatedly; retry shortly")
 
 
+def _provider_registration_conflicts(first: LiteLLM_MCPServerTable, second: LiteLLM_MCPServerTable) -> bool:
+    if MCPExecutionMode.provider not in (first.execution_mode, second.execution_mode):
+        return False
+    identifiers: Final = frozenset(
+        value.lower() for value in (first.server_id, first.server_name, first.alias) if value
+    )
+    other_identifiers: Final = frozenset(
+        value.lower() for value in (second.server_id, second.server_name, second.alias) if value
+    )
+    same_provider_url: Final = (
+        first.execution_mode == second.execution_mode == MCPExecutionMode.provider
+        and first.url is not None
+        and second.url is not None
+        and canonical_provider_mcp_url(first.url) == canonical_provider_mcp_url(second.url)
+    )
+    return bool(identifiers & other_identifiers) or same_provider_url
+
+
+def validate_provider_registration_conflicts(
+    config_servers: Sequence[LiteLLM_MCPServerTable],
+    database_servers: Sequence[LiteLLM_MCPServerTable] = (),
+) -> None:
+    for index, config_server in enumerate(config_servers):
+        for other in (*config_servers[index + 1 :], *database_servers):
+            if _provider_registration_conflicts(config_server, other):
+                raise ValueError(
+                    f"MCP config registration '{config_server.server_name}' conflicts with registration '{other.server_name}'"
+                )
+
+
 def _snapshot(
-    manager: MCPServerManager, database_identity: str,
+    manager: MCPServerManager,
+    database_identity: str,
     registration_only_servers: Mapping[str, LiteLLM_MCPServerTable],
 ) -> CatalogSnapshot:
     from litellm.proxy.mcp_server.tool_registry import global_mcp_tool_registry
@@ -103,6 +135,7 @@ def _snapshot(
     serialized: Final = json.dumps(
         (
             database_identity,
+            tuple(server.model_dump(mode="json") for server in manager.catalog.config_registrations.values()),
             tuple(sorted(manager.registry)),
             tuple((key, _configuration_identity(value)) for key, value in sorted(manager.config_mcp_servers.items())),
         ),
@@ -113,10 +146,13 @@ def _snapshot(
         hashlib.sha256(serialized.encode()).hexdigest(),
         MappingProxyType(dict(global_mcp_tool_registry.published_tools)),
         dict(manager.published_tool_routes),
-        MappingProxyType({
+        MappingProxyType(
-            **{key: manager.build_mcp_server_table(value) for key, value in servers.items()},
+            {
-            **{key: value.model_copy(deep=True) for key, value in registration_only_servers.items()},
+                **{key: value.model_copy(deep=True) for key, value in manager.catalog.config_registrations.items()},
-        }),
+                **{key: manager.build_mcp_server_table(value) for key, value in servers.items()},
+                **{key: value.model_copy(deep=True) for key, value in registration_only_servers.items()},
+            }
+        ),
     )
 
 
@@ -126,6 +162,7 @@ class CatalogSnapshots:
         self._refresh_lock = asyncio.Lock()
         self._database_identity = ""
         self._registration_only_servers: Mapping[str, LiteLLM_MCPServerTable] = MappingProxyType({})
+        self.config_registrations: Mapping[str, LiteLLM_MCPServerTable] = MappingProxyType({})
         self._arrival_ticket = 0
         self._completed_ticket = 0
         self._shared_snapshot: CatalogSnapshot | None = None
@@ -139,6 +176,24 @@ class CatalogSnapshots:
             "mcp_catalog_routing", default=None
         )
 
+    def validate_config_inventory(self, servers: Sequence[LiteLLM_MCPServerTable]) -> None:
+        database_servers: Final = (
+            *(self.manager.build_mcp_server_table(server) for server in self.manager.registry.values()),
+            *self._registration_only_servers.values(),
+        )
+        validate_provider_registration_conflicts(servers, database_servers)
+
+    def publish_config_registrations(self, servers: Mapping[str, LiteLLM_MCPServerTable]) -> None:
+        self.config_registrations = MappingProxyType(dict(servers))
+        self._shared_snapshot = None
+
+    def validate_database_registration(self, server: LiteLLM_MCPServerTable) -> None:
+        config_servers: Final = (
+            *self.config_registrations.values(),
+            *(self.manager.build_mcp_server_table(value) for value in self.manager.config_mcp_servers.values()),
+        )
+        validate_provider_registration_conflicts(config_servers, (server,))
+
     def current(self) -> CatalogSnapshot | None:
         scoped: Final = self._operation.get()
         return scoped[0] if scoped is not None and not scoped[1].is_set() else None
@@ -222,9 +277,9 @@ class CatalogSnapshots:
             shared,
             servers=MappingProxyType({key: value.model_copy(deep=True) for key, value in shared.servers.items()}),
             routing=routing,
-            management_servers=MappingProxyType({
+            management_servers=MappingProxyType(
-                key: value.model_copy(deep=True) for key, value in shared.management_servers.items()
+                {key: value.model_copy(deep=True) for key, value in shared.management_servers.items()}
-            }),
+            ),
         )
         closed: Final = asyncio.Event()
         token: Final = self._operation.set((snapshot, closed))
@@ -507,13 +562,24 @@ class CatalogSnapshots:
         ).hexdigest()
         verbose_logger.info("Found %s MCP servers in database", len(raw_rows))
 
-        registration_only_servers: Final = MappingProxyType({
+        registration_only_servers: Final = MappingProxyType(
-            server.server_id: server
+            {
-            for server in (
+                server.server_id: server
-                LiteLLM_MCPServerTable.model_validate(row.model_dump())
+                for server in (
-                for row in raw_rows if row.model_dump().get("execution_mode") == MCPExecutionMode.provider
+                    LiteLLM_MCPServerTable.model_validate(row.model_dump())
+                    for row in raw_rows
+                    if row.model_dump().get("execution_mode") == MCPExecutionMode.provider
+                )
+            }
+        )
+        config_servers: Final = (
+            *self.config_registrations.values(),
+            *(self.manager.build_mcp_server_table(value) for value in self.manager.config_mcp_servers.values()),
+        )
+        if self.config_registrations or registration_only_servers:
+            validate_provider_registration_conflicts(
+                config_servers, tuple(LiteLLM_MCPServerTable.model_validate(row.model_dump()) for row in raw_rows)
             )
-        })
         previous_registry: Final = self.manager.registry
         new_registry: Final = await self._stage_servers(raw_rows, reuse_unchanged=reuse_unchanged)
 

litellm/proxy/mcp_server/db.py

@@ -1084,6 +1084,30 @@ async def delete_mcp_server(
     return deleted_server  # pyright: ignore[reportReturnType]  # prisma row, not domain LiteLLM_MCPServerTable
 
 
+def _validate_mcp_config_registration_write(values: Mapping[str, object]) -> None:
+    from litellm.proxy.mcp_server.mcp_server_manager import global_mcp_server_manager
+
+    catalog: Final = global_mcp_server_manager.catalog
+    if not catalog.config_registrations and values.get("execution_mode") != MCPExecutionMode.provider:
+        return
+    identity: Final = {
+        key: values[key]
+        for key in (
+            "server_id",
+            "server_name",
+            "alias",
+            "url",
+            "transport",
+            "execution_mode",
+        )
+        if key in values
+    }
+    try:
+        catalog.validate_database_registration(LiteLLM_MCPServerTable.model_validate(identity))
+    except ValueError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+
+
 async def create_mcp_server(
     prisma_client: PrismaClient, data: NewMCPServerRequest, touched_by: str
 ) -> LiteLLM_MCPServerTable:
@@ -1099,6 +1123,7 @@ async def create_mcp_server(
         data.server_id = str(uuid.uuid4())
 
     # Use helper to prepare data with proper JSON serialization
+    _validate_mcp_config_registration_write(data.model_dump())
     data_dict: Final = _prepare_mcp_server_data(data)
 
     # Add audit fields
@@ -1123,6 +1148,7 @@ async def create_mcp_server_if_identifier_free(
     if data.server_id is None:
         data.server_id = str(uuid.uuid4())
 
+    _validate_mcp_config_registration_write(data.model_dump())
     data_dict: Final = _prepare_mcp_server_data(data)
     data_dict["created_by"] = touched_by
     data_dict["updated_by"] = touched_by
@@ -1260,13 +1286,14 @@ def _validate_mcp_protocol_write(
 
 
 def _validate_provider_registration_update(
-    stored: "prisma_db_models.LiteLLM_MCPServerTable", data: UpdateMCPServerRequest,
+    stored: "prisma_db_models.LiteLLM_MCPServerTable",
+    data: UpdateMCPServerRequest,
     fields_set: set[str] | None = None,
 ) -> NewMCPServerRequest:
-    patch: Final = data.model_dump(include=fields_set, exclude_unset=True)
     try:
+        patch: Final = data.provider_registration_patch(fields_set)
         registration: Final = LiteLLM_MCPServerTable.model_validate(stored.model_dump())
-        merged: Final = {**registration.model_dump(), **patch}
+        merged: Final = {**registration.model_dump(include=set(NewMCPServerRequest.model_fields)), **patch}
         return NewMCPServerRequest.model_validate(merged)
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
@@ -1278,15 +1305,19 @@ def _prepare_provider_registration_update(
     touched_by: str,
     fields_set: set[str] | None,
 ) -> Mapping[str, object]:
-    from prisma import Json
+    from litellm.types.mcp_server.provider_registration import ProviderMCPRegistration
 
     validated: Final = _validate_provider_registration_update(stored, data, fields_set)
+    _validate_mcp_config_registration_write(validated.model_dump())
     supplied_fields: Final = fields_set if fields_set is not None else data.model_fields_set
-    normalized_patch: Final = data.model_copy(update=validated.model_dump(include=supplied_fields | {"mcp_info"}))
+    normalized_patch: Final = UpdateMCPServerRequest.model_validate(
+        validated.model_dump(
+            include=(supplied_fields & ProviderMCPRegistration.model_fields.keys()) | {"server_id", "mcp_info"}
+        )
+    )
     serialized: Final = _prepare_mcp_server_data(normalized_patch, exclude_unset=True, fields_set=supplied_fields)
     return {
         **serialized,
-        **({"credentials": Json(None)} if "credentials" in serialized and serialized["credentials"] is None else {}),
         "updated_by": touched_by,
     }
 
@@ -1378,6 +1409,9 @@ async def update_mcp_server(
             else LiteLLM_MCPServerTable.model_validate(updated_provider.model_dump())
         )
 
+    if registration is not None:
+        _validate_mcp_config_registration_write({**registration.model_dump(), **data.model_dump(exclude_unset=True)})
+
     # Use helper to prepare data with proper JSON serialization.
     # exclude_unset=True makes this a true partial update: fields the caller did
     # not provide are not written, so they keep their existing DB value instead
@@ -1385,15 +1419,17 @@ async def update_mcp_server(
     prepared_data: Final = _prepare_mcp_server_data(data, exclude_unset=True, fields_set=fields_set)
 
     # Pre-fetch existing record once if we need it for auth_type, url, or credential logic
-    existing = None
     has_credentials: Final = "credentials" in prepared_data and prepared_data["credentials"] is not None
     # 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: Final = bool(_TOKEN_EXCHANGE_COLUMN_FIELDS & prepared_data.keys())
     url_provided: Final = "url" in prepared_data and prepared_data["url"] is not None
     issuer_provided: Final = "issuer" in prepared_data
-    if data.auth_type or has_credentials or explicit_te_write or url_provided or issuer_provided:
+    existing: Final = (
-        existing = registration
+        registration
+        if data.auth_type or has_credentials or explicit_te_write or url_provided or issuer_provided
+        else None
+    )
 
     auth_type_changed: Final = bool(
         data.auth_type

litellm/proxy/mcp_server/mcp_server_manager.py

@@ -89,6 +89,7 @@ from litellm.proxy._types import (
     MCPEnvVar,
     MCPTransport,
     MCPTransportType,
+    NewMCPServerRequest,
     SpecialMCPServerNames,
     UserAPIKeyAuth,
     is_per_server_oauth_discovery_eligible,
@@ -408,6 +409,7 @@ class MCPServerConfig(TypedDict, total=False):
 
     server_id: ReadOnly[str]
     execution_mode: ReadOnly[MCPExecutionMode]
+    tool_name_to_description: ReadOnly[Mapping[str, str]]
     protocol_version: ReadOnly[MCPUpstreamProtocol]
     alias: str
     description: str
@@ -2558,13 +2560,83 @@ class MCPServerManager:
         store; a rowless server that is NOT config-declared is a throwaway temp/session server whose
         client must not be persisted. This never overrides the row-existence check: a server that has
         a LiteLLM_MCPServerTable row is always resolved to that row first."""
-        return server_id in self.config_mcp_servers
+        return server_id in self.config_mcp_servers or server_id in self.catalog.config_registrations
+
+    def _config_registration(
+        self,
+        name: str,
+        values: MCPServerConfig,
+        aliases: Mapping[str, str] | None,
+    ) -> LiteLLM_MCPServerTable:
+        alias: Final = _assigned_alias(name, values, aliases)
+        pinned_id: Final = _pinned_config_server_id(values.get("server_id"), name)
+        if values.get("execution_mode", MCPExecutionMode.litellm) == MCPExecutionMode.provider:
+            request: Final = NewMCPServerRequest.model_validate(
+                {
+                    **{field: value for field, value in values.items() if field != "access_groups"},
+                    "server_name": name,
+                    "alias": alias,
+                    "mcp_access_groups": values.get("access_groups", []),
+                }
+            )
+            stable_id: Final = hashlib.sha256(f"provider|{name}|{request.url}|{alias or ''}".encode()).hexdigest()[:32]
+            return LiteLLM_MCPServerTable.model_validate(
+                {
+                    **request.model_dump(exclude_none=True),
+                    "server_id": pinned_id or stable_id,
+                    "transport": request.transport,
+                    "is_config": True,
+                }
+            )
+        validate_mcp_gateway_execution(values.get("execution_mode", MCPExecutionMode.litellm))
+        return LiteLLM_MCPServerTable(
+            server_id=pinned_id
+            or self._generate_stable_server_id(
+                name,
+                values.get("url") or "",
+                values.get("transport", MCPTransport.http),
+                values.get("auth_type"),
+                alias,
+            ),
+            server_name=name,
+            alias=alias,
+            url=values.get("url"),
+            transport=values.get("transport", MCPTransport.http),
+            is_config=True,
+        )
 
     async def load_servers_from_config(
         self,
         mcp_servers_config: dict[str, MCPServerConfig],
         mcp_aliases: dict[str, str] | None = None,
-    ):
+    ) -> None:
+        _validate_config_server_names(mcp_servers_config)
+        registrations: Final = tuple(
+            self._config_registration(name, config, mcp_aliases) for name, config in mcp_servers_config.items()
+        )
+        self.catalog.validate_config_inventory(registrations)
+        providers: Final = {
+            server.server_id: server for server in registrations if server.execution_mode == MCPExecutionMode.provider
+        }
+        provider_names: Final = frozenset(server.server_name for server in providers.values())
+        for server in tuple(self.config_mcp_servers.values()):
+            if server.server_name in provider_names or server.server_id in providers:
+                self._cleanup_server_tool_routing_artifacts(server)
+                self.invalidate_oauth_discovery_state(server.server_id)
+                self.config_mcp_servers.pop(server.server_id, None)
+        gateway_config: Final = {
+            name: config
+            for name, config in mcp_servers_config.items()
+            if config.get("execution_mode", MCPExecutionMode.litellm) != MCPExecutionMode.provider
+        }
+        await self._load_gateway_servers_from_config(gateway_config, mcp_aliases)
+        self.catalog.publish_config_registrations(providers)
+
+    async def _load_gateway_servers_from_config(
+        self,
+        mcp_servers_config: dict[str, MCPServerConfig],
+        mcp_aliases: dict[str, str] | None = None,
+    ) -> None:
         """
         Load the MCP Servers from the config
 
@@ -3527,9 +3599,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 (
+        if mcp_server.execution_mode == MCPExecutionMode.provider or mcp_server.approval_status not in (
-            mcp_server.execution_mode == MCPExecutionMode.provider
+            None,
-            or mcp_server.approval_status not in (None, "active", "approved")
+            "active",
+            "approved",
         ):
             evicted = self.registry.pop(mcp_server.server_id, None)
             if evicted is None and mcp_server.server_name:

litellm/proxy/proxy_server.py

@@ -7453,17 +7453,18 @@ class ProxyConfig:
 
             global_agent_registry.load_agents_from_config(agent_config)
 
+        from litellm.proxy.mcp_server.utils import is_mcp_available
+
         mcp_servers_config: Final = config.get("mcp_servers", None)
-        if mcp_servers_config:
+        if mcp_servers_config or is_mcp_available():
-            from litellm.proxy.mcp_server.mcp_server_manager import (
+            from litellm.proxy.mcp_server.mcp_server_manager import global_mcp_server_manager
-                global_mcp_server_manager,
-            )
 
-            # Get mcp_aliases from litellm_settings if available
+            if mcp_servers_config:
-            litellm_settings: Final = config.get("litellm_settings", {})
+                litellm_settings: Final = config.get("litellm_settings", {})
-            mcp_aliases: Final = litellm_settings.get("mcp_aliases", None)
+                mcp_aliases: Final = litellm_settings.get("mcp_aliases", None)
-
+                await global_mcp_server_manager.load_servers_from_config(mcp_servers_config, mcp_aliases)
-            await global_mcp_server_manager.load_servers_from_config(mcp_servers_config, mcp_aliases)
+            elif global_mcp_server_manager.catalog.config_registrations:
+                global_mcp_server_manager.catalog.publish_config_registrations({})
 
         ## VECTOR STORES
         vector_store_registry_config: Final = config.get("vector_store_registry", None)

litellm/types/mcp_server/provider_registration.py

@@ -1,53 +1,100 @@
 import json
-from collections.abc import Mapping
+from collections.abc import Collection, Mapping
-from typing import Final
+from datetime import datetime
+from typing import Final, Literal
 from urllib.parse import urlsplit, urlunsplit
 
-from pydantic import AnyHttpUrl, TypeAdapter
+from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, TypeAdapter, field_validator, model_validator
+from typing_extensions import Self
+
+from litellm.types.mcp import MCPExecutionMode
 
 _STR: Final = TypeAdapter(str)
-_TOOLS: Final = TypeAdapter(list[str])
 _MAP: Final = TypeAdapter(dict[str, object])
 _URL: Final = TypeAdapter(AnyHttpUrl)
-_CONNECTION_FIELDS: Final = (
+
-    "transport",
+
-    "auth_type",
+def _is_neutral_registration_value(value: object) -> bool:
-    "credentials",
+    if isinstance(value, str):
-    "spec_path",
+        return value.strip() in ("", "null", "{}", "[]")
-    "extra_headers",
+    return value is None or value is False or value in ({}, [])
-    "static_headers",
+
-    "env_vars",
+
-    "command",
+def active_registration_extras(values: Mapping[str, object], known_fields: Collection[str]) -> tuple[str, ...]:
-    "args",
+    return tuple(
-    "env",
+        field
-    "issuer",
+        for field, value in values.items()
-    "authorization_url",
+        if field not in known_fields and not _is_neutral_registration_value(value)
-    "token_url",
+    )
-    "registration_url",
+
-    "oauth2_flow",
+
-    "token_exchange_endpoint",
+class ProviderMCPRegistration(BaseModel):
-    "audience",
+    model_config = ConfigDict(frozen=True, extra="forbid")
-    "subject_token_type",
+
-    "token_exchange_profile",
+    execution_mode: Literal[MCPExecutionMode.provider] = MCPExecutionMode.provider
-    "delegate_auth_to_upstream",
+    server_id: str | None = None
-    "oauth_passthrough",
+    server_name: str | None = None
-    "dcr_bridge",
+    alias: str | None = None
-    "per_server_oauth_discovery",
+    description: str | None = None
-    "is_byok",
+    url: str
-    "byok_description",
+    allowed_tools: list[str]
-    "byok_api_key_help_url",
+    mcp_info: dict[str, object] = Field(default_factory=dict, validate_default=True)
-    "timeout",
+    mcp_access_groups: list[str] = Field(default_factory=list)
-    "max_concurrent_requests",
+    tool_name_to_display_name: dict[str, str] | None = None
-    "rpm",
+    tool_name_to_description: dict[str, str] | None = None
-    "pinned_tools",
+    instructions: str | None = None
-)
+    source_url: str | None = None
-_FLAG_FIELDS: Final = frozenset(
+    allow_all_keys: bool = False
-    {"delegate_auth_to_upstream", "oauth_passthrough", "dcr_bridge", "per_server_oauth_discovery", "is_byok"}
+    available_on_public_internet: Literal[True] = True
-)
+    protocol_version: Literal["auto"] = "auto"
-_EMPTY_FIELDS: Final = frozenset(
+    approval_status: str | None = None
-    {"credentials", "extra_headers", "static_headers", "env_vars", "args", "env", "byok_description", "pinned_tools"}
+    submitted_by: str | None = None
-)
+    submitted_at: datetime | None = None
-_JSON_FIELDS: Final = frozenset({"credentials", "static_headers", "env_vars", "env", "pinned_tools"})
+
+    @model_validator(mode="before")
+    @classmethod
+    def discard_neutral_extras(cls, values: object) -> object:
+        if not isinstance(values, Mapping):
+            return values
+        fields: Final = _MAP.validate_python(values)
+        return {
+            field: value
+            for field, value in fields.items()
+            if field in cls.model_fields or not _is_neutral_registration_value(value)
+        }
+
+    @field_validator("url", mode="before")
+    @classmethod
+    def normalize_url(cls, value: object) -> str:
+        return canonical_provider_mcp_url(value)
+
+    @field_validator("allowed_tools")
+    @classmethod
+    def normalize_allowed_tools(cls, tools: list[str]) -> list[str]:
+        if any(not tool.strip() or any(char in tool for char in "*?[]") for tool in tools):
+            raise ValueError("allowed_tools must contain nonblank exact tool names without wildcards")
+        return list(dict.fromkeys(tools))
+
+    @field_validator("mcp_info", mode="before")
+    @classmethod
+    def normalize_metadata(cls, value: object) -> dict[str, object]:
+        info: Final = _MAP.validate_python(_decoded_json(value) or {})
+        if "tool_allowlist_enforced" in info and info["tool_allowlist_enforced"] is not True:
+            raise ValueError("Provider registrations require mcp_info.tool_allowlist_enforced=true")
+        if info.get("protocol_version", "auto") != "auto":
+            raise ValueError("Provider registrations do not configure the upstream MCP protocol")
+        return {**info, "tool_allowlist_enforced": True}
+
+    @field_validator("tool_name_to_description", "tool_name_to_display_name", mode="before")
+    @classmethod
+    def decode_stored_metadata(cls, value: object) -> object:
+        return _decoded_json(value)
+
+    @model_validator(mode="after")
+    def validate_tool_descriptions(self) -> Self:
+        if not (self.tool_name_to_description or {}).keys() <= set(self.allowed_tools):
+            raise ValueError("Tool descriptions must reference allowed_tools")
+        return self
 
 
 def canonical_provider_mcp_url(value: object) -> str:
@@ -71,39 +118,11 @@ def _decoded_json(value: object) -> object:
     return json.loads(value) if isinstance(value, str) else value
 
 
-def _active_connection_field(field: str, raw: object) -> bool:
-    value: Final = _decoded_json(raw) if field in _JSON_FIELDS else raw
-    return not (
-        value is None or (field in _FLAG_FIELDS and value is False) or (field in _EMPTY_FIELDS and value in ({}, []))
-    )
-
-
 def normalize_provider_registration(values: Mapping[str, object]) -> dict[str, object]:
-    unsupported: Final = next(
+    registration: Final = ProviderMCPRegistration.model_validate(values)
-        (field for field in _CONNECTION_FIELDS if _active_connection_field(field, values.get(field))), None
-    )
-    if unsupported is not None:
-        raise ValueError(f"Provider-executed MCP registrations do not accept {unsupported}")
-    if values.get("available_on_public_internet", True) is not True:
-        raise ValueError("Provider-executed MCP registrations do not support gateway network restrictions")
-    if values.get("allowed_tools") is None:
-        raise ValueError("Provider registrations require an explicit allowed_tools list")
-    tools: Final = _TOOLS.validate_python(values.get("allowed_tools"))
-    if any(not tool.strip() or any(char in tool for char in "*?[]") for tool in tools):
-        raise ValueError("allowed_tools must contain nonblank exact tool names without wildcards")
-    info: Final = _MAP.validate_python(_decoded_json(values.get("mcp_info")) or {})
-    if "tool_allowlist_enforced" in info and info["tool_allowlist_enforced"] is not True:
-        raise ValueError("Provider registrations require mcp_info.tool_allowlist_enforced=true")
-    if info.get("protocol_version", "auto") != "auto":
-        raise ValueError("Provider registrations do not configure the upstream MCP protocol")
-    descriptions: Final = _MAP.validate_python(_decoded_json(values.get("tool_name_to_description")) or {})
-    if not descriptions.keys() <= set(tools):
-        raise ValueError("Tool descriptions must reference allowed_tools")
     return {
-        **values,
+        **registration.model_dump(exclude_unset=True),
-        "url": canonical_provider_mcp_url(values.get("url")),
+        "mcp_info": registration.mcp_info,
-        "allowed_tools": list(dict.fromkeys(tools)),
-        "mcp_info": {**info, "tool_allowlist_enforced": True},
         "transport": None,
         "auth_type": None,
         "credentials": None,

tests/unit/proxy/mcp_server/test_catalog.py

@@ -891,3 +891,164 @@ async def test_management_and_runtime_inventory_share_refresh_and_request_snapsh
     assert await manager.get_all_mcp_servers_unfiltered() == []
     assert await manager.catalog.list() == {}
     assert read_rows.await_count == 4
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("pinned_id", [None, "configured-docs"])
+async def test_provider_config_inventory_reloads_without_entering_gateway(
+    monkeypatch: pytest.MonkeyPatch, pinned_id: str | None,
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    manager: Final = MCPServerManager()
+    config: Final = {
+        "docs": {
+            "execution_mode": "provider", "url": "https://DOCS.example:443/mcp", "allowed_tools": ["search"],
+            "tool_name_to_description": {"search": "Search documentation"},
+            **({"server_id": pinned_id} if pinned_id else {}),
+        },
+    }
+    await manager.load_servers_from_config(config)
+    first: Final = (await manager.get_all_mcp_servers_unfiltered())[0]
+    assert first.is_config and first.execution_mode == "provider"
+    assert first.url == "https://docs.example/mcp"
+    assert first.allowed_tools == ["search"]
+    assert first.tool_name_to_description == {"search": "Search documentation"}
+    assert first.transport is None and first.credentials is None
+    assert first.mcp_info["tool_allowlist_enforced"] is True
+    assert not manager.get_registry()
+    async with manager.catalog.operation():
+        await manager.load_servers_from_config({"docs": {
+            **config["docs"], "url": first.url, "allowed_tools": [], "tool_name_to_description": {},
+        }})
+        assert (await manager.get_all_mcp_servers_unfiltered())[0].allowed_tools == ["search"]
+    updated: Final = (await manager.get_all_mcp_servers_unfiltered())[0]
+    assert updated.server_id == first.server_id
+    assert updated.allowed_tools == []
+    await manager.load_servers_from_config({})
+    assert await manager.get_all_mcp_servers_unfiltered() == []
+    assert not manager.get_registry()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("field,value", [
+    ("authentication_token", "synthetic"), ("client_secret", "synthetic"),
+    ("aws_access_key_id", "synthetic"), ("client_private_key", "synthetic"),
+    ("allow_sampling", True), ("protocol_version", "2025-03-26"),
+    ("gateway_custom_option", {"enabled": True}),
+])
+async def test_provider_config_rejects_active_config_only_connection_fields(
+    monkeypatch: pytest.MonkeyPatch, field: str, value: object,
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    manager: Final = MCPServerManager()
+    config: Final = {"execution_mode": "provider", "url": "https://docs.example/mcp", "allowed_tools": ["search"]}
+    await manager.load_servers_from_config({"docs": config})
+    previous: Final = await manager.get_all_mcp_servers_unfiltered()
+    with pytest.raises(ValueError, match=field):
+        await manager.load_servers_from_config({"docs": {**config, field: value}})
+    assert await manager.get_all_mcp_servers_unfiltered() == previous
+    assert not manager.get_registry()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("other", [
+    {"execution_mode": "provider", "url": "https://DOCS.example:443/mcp", "allowed_tools": []},
+    {"execution_mode": "provider", "url": "https://other.example/mcp", "allowed_tools": [], "alias": "docs"},
+    {"url": "https://gateway.example/mcp", "transport": "http", "server_id": "docs"},
+])
+async def test_provider_config_rejects_ambiguous_config_identities(
+    monkeypatch: pytest.MonkeyPatch, other: dict[str, object],
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    manager: Final = MCPServerManager()
+    with pytest.raises(ValueError, match="conflicts"):
+        await manager.load_servers_from_config({
+            "docs": {"execution_mode": "provider", "url": "https://docs.example/mcp", "allowed_tools": []},
+            "other": other,
+        })
+    assert not manager.get_registry()
+    assert await manager.get_all_mcp_servers_unfiltered() == []
+
+
+@pytest.mark.asyncio
+async def test_provider_config_replaces_gateway_entry_without_retaining_execution(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    manager: Final = MCPServerManager()
+    await manager.load_servers_from_config({"docs": {"transport": "http", "url": "https://docs.example/mcp"}})
+    assert manager.get_mcp_server_by_name("docs") is not None
+    await manager.load_servers_from_config({"docs": {
+        "execution_mode": "provider", "url": "https://docs.example/mcp", "allowed_tools": [],
+    }})
+    assert manager.get_mcp_server_by_name("docs") is None
+    assert not manager.get_registry()
+    assert (await manager.get_all_mcp_servers_unfiltered())[0].execution_mode == "provider"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("database_first", [False, True])
+async def test_provider_config_rejects_database_url_conflict_in_either_loading_order(
+    monkeypatch: pytest.MonkeyPatch, database_first: bool,
+) -> None:
+    from fastapi import HTTPException
+
+    from litellm.models.mcp_server import LiteLLM_MCPServerTable
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    row: Final = LiteLLM_MCPServerTable(
+        server_id="database-docs", server_name="database-docs", execution_mode="provider", transport=None,
+        url="https://docs.example/mcp", allowed_tools=["search"],
+    )
+    tables: Final = SimpleNamespace(
+        litellm_mcpservertable=SimpleNamespace(find_many=AsyncMock(return_value=[row])),
+        litellm_config=SimpleNamespace(find_unique=AsyncMock(return_value=SimpleNamespace(reload_revision=1))),
+    )
+    monkeypatch.setattr(proxy_server, "prisma_client", SimpleNamespace(db=tables, writer_db=tables))
+    monkeypatch.setattr(proxy_server, "general_settings", {})
+    manager: Final = MCPServerManager()
+    config: Final = {"docs": {"execution_mode": "provider", "url": row.url, "allowed_tools": []}}
+    if database_first:
+        await manager.get_all_mcp_servers_unfiltered()
+        with pytest.raises(ValueError, match="conflicts"):
+            await manager.load_servers_from_config(config)
+    else:
+        await manager.load_servers_from_config(config)
+        with pytest.raises(HTTPException) as unavailable:
+            await manager.get_all_mcp_servers_unfiltered()
+        assert unavailable.value.status_code == 503
+    assert not manager.get_registry()
+
+
+@pytest.mark.asyncio
+async def test_provider_config_changes_invalidate_cached_inventory_without_database_revision_change(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from litellm.proxy import proxy_server
+    from litellm.proxy.mcp_server.mcp_server_manager import MCPServerManager
+
+    tables: Final = SimpleNamespace(
+        litellm_mcpservertable=SimpleNamespace(find_many=AsyncMock(return_value=[])),
+        litellm_config=SimpleNamespace(find_unique=AsyncMock(return_value=SimpleNamespace(reload_revision=1))),
+    )
+    monkeypatch.setattr(proxy_server, "prisma_client", SimpleNamespace(db=tables, writer_db=tables))
+    monkeypatch.setattr(proxy_server, "general_settings", {})
+    manager: Final = MCPServerManager()
+    initial: Final = {"execution_mode": "provider", "url": "https://docs.example/mcp", "allowed_tools": ["search"]}
+    await manager.load_servers_from_config({"docs": initial})
+    assert (await manager.get_all_mcp_servers_unfiltered())[0].allowed_tools == ["search"]
+    await manager.load_servers_from_config({"docs": {**initial, "allowed_tools": []}})
+    assert (await manager.get_all_mcp_servers_unfiltered())[0].allowed_tools == []
+    await manager.load_servers_from_config({})
+    assert await manager.get_all_mcp_servers_unfiltered() == []

tests/unit/proxy/mcp_server/test_mcp_partial_update.py

@@ -1375,6 +1375,7 @@ async def test_issuer_edit_preserves_replacement_with_same_client_id(replacement
         {"mcp_info": {"tool_allowlist_enforced": False}},
         {"tool_name_to_description": {"write": "Not approved"}},
         {"credentials": {"auth_value": "synthetic"}},
+        {"gateway_custom_option": {"enabled": True}},
     ],
 )
 def test_provider_update_validates_the_merged_stored_registration(patch: dict[str, object]) -> None:
@@ -1444,7 +1445,6 @@ async def test_provider_accepts_neutral_connection_settings_without_credentials_
     updated = await update_mcp_server(prisma, UpdateMCPServerRequest(server_id="provider", **neutral), "admin")
     assert updated.allowed_tools == ["search"]
     written = prisma.db.litellm_mcpservertable.update.call_args.kwargs["data"]
-    assert written["transport"] is None
+    assert not neutral.keys() & written.keys()
-    assert _credentials_cleared(written["credentials"])
     assert "allowed_tools" not in written
     assert json.loads(written["mcp_info"])["tool_allowlist_enforced"] is True

tests/unit/proxy/proxy_server/test_proxy_config.py

@@ -6264,3 +6264,34 @@ async def test_proxy_config_validates_advertised_mcp_versions_at_load(tmp_path,
         return
     with pytest.raises(ValidationError):
         await ProxyConfig().load_config(router=None, config_file_path=str(config))
+
+
+@pytest.mark.asyncio
+async def test_provider_config_management_and_removal_without_database(monkeypatch):
+    from starlette.requests import Request
+
+    from litellm.proxy import proxy_server
+    from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+    from litellm.proxy.management_endpoints.mcp_management_endpoints import fetch_all_mcp_servers, fetch_mcp_server
+    from litellm.proxy.mcp_server import mcp_server_manager
+    from litellm.proxy.management_endpoints import mcp_management_endpoints
+
+    manager = mcp_server_manager.MCPServerManager()
+    monkeypatch.setattr(mcp_server_manager, "global_mcp_server_manager", manager)
+    monkeypatch.setattr(mcp_management_endpoints, "global_mcp_server_manager", manager)
+    monkeypatch.setattr(proxy_server, "prisma_client", None)
+    config = proxy_server.ProxyConfig()
+    await config._init_non_llm_configs({"mcp_servers": {"config_docs": {
+        "server_id": "config-docs", "execution_mode": "provider", "url": "https://docs.example/mcp",
+        "allowed_tools": ["search"],
+    }}})
+    admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
+    inventory = await fetch_all_mcp_servers(admin, team_id=None, user_id=None)
+    assert len(inventory) == 1 and inventory[0].is_config
+    request = Request({"type": "http", "headers": [], "client": ("127.0.0.1", 12345)})
+    detail = await fetch_mcp_server(request, "config_docs", admin, include_reachability=True)
+    assert detail.server_id == "config-docs" and detail.allowed_tools == ["search"]
+    assert detail.last_health_check is None
+    await config._init_non_llm_configs({})
+    assert await fetch_all_mcp_servers(admin, team_id=None, user_id=None) == []
+    assert not manager.get_registry()

tests/unit/types/mcp_server/test_provider_registration.py

@@ -44,6 +44,7 @@ def test_unsafe_endpoint_identity_is_rejected(url: str) -> None:
         {"credentials": {"auth_value": "synthetic"}},
         {"static_headers": {"Authorization": "synthetic"}},
         {"command": "python"},
+        {"gateway_custom_option": {"enabled": True}},
         {"timeout": 0},
         {"mcp_info": {"tool_allowlist_enforced": False}},
         {"mcp_info": {"tool_allowlist_enforced": None}},
@@ -65,3 +66,31 @@ def test_empty_tool_list_is_enforced_without_transport_or_stored_credentials() -
     assert prepared["transport"] is None
     assert not prepared.get("credentials")
     assert prepared.get("auth_type") is None
+
+
+@pytest.mark.parametrize("neutral", [None, False, "", {}, []])
+def test_provider_registration_ignores_neutral_unknown_settings(neutral: object) -> None:
+    request: Final = NewMCPServerRequest(
+        execution_mode="provider", url="https://docs.example/mcp", allowed_tools=["search"],
+        gateway_custom_option=neutral,
+    )
+    assert request.allowed_tools == ["search"]
+    assert request.transport is None
+    assert "gateway_custom_option" not in request.model_dump()
+
+
+def test_provider_registration_preserves_supported_management_fields() -> None:
+    request: Final = NewMCPServerRequest(
+        execution_mode="provider", server_name="docs", alias="documentation",
+        url="https://docs.example/mcp", allowed_tools=["search"], description="Documentation",
+        instructions="Search docs", source_url="https://docs.example", mcp_access_groups=["research"],
+        allow_all_keys=True, tool_name_to_description={"search": "Search docs"},
+        tool_name_to_display_name={"search": "Search"}, mcp_info={"custom_metadata": "retained"},
+    )
+    assert request.server_name == "docs" and request.alias == "documentation"
+    assert request.description == "Documentation" and request.instructions == "Search docs"
+    assert request.source_url == "https://docs.example" and request.mcp_access_groups == ["research"]
+    assert request.allow_all_keys
+    assert request.tool_name_to_description == {"search": "Search docs"}
+    assert request.tool_name_to_display_name == {"search": "Search"}
+    assert request.mcp_info == {"custom_metadata": "retained", "tool_allowlist_enforced": True}