feat(mcp)!: disable stdio MCP servers by default

.circleci/scripts/run_integration.sh

@@ -169,7 +169,7 @@ start_proxy() {
     INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \
     LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" LITELLM_SALT_KEY="$LITELLM_SALT_KEY" LITELLM_UI_PATH="$LITELLM_UI_PATH" PROXY_BASE_URL="http://127.0.0.1:$port" \
     LITELLM_LICENSE="${LITELLM_LICENSE:-}" \
-    LITELLM_MODE=PRODUCTION STORE_MODEL_IN_DB=True "${cost_map_env[@]}" \
+    LITELLM_MODE=PRODUCTION STORE_MODEL_IN_DB=True LITELLM_ENABLE_MCP_STDIO=true "${cost_map_env[@]}" \
     AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 COVERAGE_FILE="$coverage_data" \
     "${proxy_command[@]}" --config tests/integration/proxy_config.yaml \
     --host 127.0.0.1 --port "$port" --num_workers 1 --telemetry False \

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -142,6 +142,12 @@
 from litellm.proxy._experimental.mcp_server.sampling_handler import (
     MCP_SAMPLING_AVAILABLE,
 )
+from litellm.proxy._experimental.mcp_server.stdio_gate import (
+    MCP_STDIO_DISABLED_MESSAGE,
+    is_mcp_stdio_blocked,
+    is_mcp_stdio_enabled,
+    warn_if_mcp_stdio_blocked,
+)
 from litellm.proxy._experimental.mcp_server.tool_catalog_guard import (
     CatalogAlert,
     apply_description_overrides,
@@ -2464,6 +2470,7 @@ async def load_servers_from_config(
                 alias=alias,
                 server_name=server_name,
             )
+            warn_if_mcp_stdio_blocked(server_name, server_config.get("transport"))
 
             auth_type = server_config.get("auth_type", None)
             manual_issuer = _blank_to_none(server_config.get("issuer"))
@@ -3282,6 +3289,7 @@ async def add_server(self, mcp_server: LiteLLM_MCPServerTable):
                 # `credentials` field is the only one still encrypted here).
                 # Re-decrypting plaintext would zero the values, so build with
                 # env_vars_are_encrypted=False.
+                self._warn_if_newly_blocked_stdio(mcp_server, None)
                 new_server: Final = await self.build_mcp_server_from_table(mcp_server, env_vars_are_encrypted=False)
                 self._assign_unique_short_prefix(new_server)
                 self._invalidate_server_definition_caches(mcp_server.server_id)
@@ -4280,6 +4288,8 @@ async def _create_mcp_client(
 
         # Handle stdio transport
         if transport == MCPTransport.stdio:
+            if not is_mcp_stdio_enabled():
+                raise HTTPException(status_code=403, detail=MCP_STDIO_DISABLED_MESSAGE)
             resolved_env: Final = (
                 stdio_env
                 if stdio_env is not None
@@ -4439,6 +4449,9 @@ async def _get_tools_from_server(
             global_mcp_tool_registry,
         )
 
+        if self._skip_blocked_stdio_listing(server, "tool"):
+            return []
+
         verbose_logger.debug("Connecting to url: %s", server.url)
         verbose_logger.info("_get_tools_from_server for %s...", server.name)
 
@@ -4638,6 +4651,19 @@ def _discovery_key(
         )
         return server.server_id, hashlib.sha256(material.encode()).hexdigest()
 
+    @staticmethod
+    def _warn_if_newly_blocked_stdio(row: LiteLLM_MCPServerTable, previous: MCPServer | None) -> None:
+        if previous is None or previous.transport != row.transport:
+            warn_if_mcp_stdio_blocked(row.alias or row.server_name, row.transport)
+
+    def _skip_blocked_stdio_listing(self, server: MCPServer, listing: str) -> bool:
+        if not is_mcp_stdio_blocked(server.transport):
+            return False
+        verbose_logger.debug(
+            "Skipping %s listing for MCP server %s: %s", listing, server.name, MCP_STDIO_DISABLED_MESSAGE
+        )
+        return True
+
     async def get_prompts_from_server(
         self,
         server: MCPServer,
@@ -4648,6 +4674,8 @@ async def get_prompts_from_server(
         raw_headers: dict[str, str] | None = None,
         client_ip: str | None = None,
     ) -> list[Prompt]:
+        if self._skip_blocked_stdio_listing(server, "prompt"):
+            return []
         try:
             headers: Final = (
                 dict(
@@ -4694,6 +4722,8 @@ async def get_resources_from_server(
         raw_headers: dict[str, str] | None = None,
         client_ip: str | None = None,
     ) -> list[Resource]:
+        if self._skip_blocked_stdio_listing(server, "resource"):
+            return []
         try:
             headers: Final = (
                 dict(
@@ -4740,6 +4770,8 @@ async def get_resource_templates_from_server(
         raw_headers: dict[str, str] | None = None,
         client_ip: str | None = None,
     ) -> list[ResourceTemplate]:
+        if self._skip_blocked_stdio_listing(server, "resource template"):
+            return []
         try:
             headers: Final = (
                 dict(
@@ -6324,6 +6356,8 @@ def _candidate_matches_server_name(candidate: MCPServer) -> bool:
                 mcp_server = fallback
         if mcp_server is None:
             raise ValueError(f"Tool {name} not found")
+        if is_mcp_stdio_blocked(mcp_server.transport):
+            raise HTTPException(status_code=403, detail=MCP_STDIO_DISABLED_MESSAGE)
 
         if resolved_by_server_name_only and not self.server_exposes_tool(mcp_server, name):
             raise ValueError(f"Tool {name} not found")
@@ -6708,7 +6742,10 @@ def _get_mcp_server_from_tool_name(self, tool_name: str) -> MCPServer | None:
         if matched is not None:
             matched_prefix, original_tool_name = matched
             matched_server: Final = prefix_to_server.get(matched_prefix)
-            if matched_server is not None and self.server_exposes_tool(matched_server, original_tool_name):
+            if matched_server is not None and (
+                self.server_exposes_tool(matched_server, original_tool_name)
+                or is_mcp_stdio_blocked(matched_server.transport)
+            ):
                 return matched_server
 
         return None
@@ -6770,6 +6807,7 @@ async def reload_servers_from_database(self):
                     alias=getattr(server, "alias", None),
                     server_name=getattr(server, "server_name", None),
                 )
+                self._warn_if_newly_blocked_stdio(server, existing_server)
                 verbose_logger.debug("Building server from DB: %s (%s)", server.server_id, server.server_name)
                 # raw_rows come straight from the DB, so their global env var
                 # values (like credentials) are still encrypted here, unlike the

litellm/proxy/_experimental/mcp_server/stdio_gate.py

@@ -0,0 +1,28 @@
+import os
+from typing import Final
+
+from litellm._logging import verbose_logger
+from litellm.types.mcp import MCPTransport
+
+MCP_STDIO_ENABLED_ENV_VAR: Final = "LITELLM_ENABLE_MCP_STDIO"
+MCP_STDIO_DISABLED_MESSAGE: Final = (
+    f"stdio MCP servers are disabled on this proxy. "
+    f"Set {MCP_STDIO_ENABLED_ENV_VAR}=true on the proxy and restart to enable them"
+)
+
+
+def is_mcp_stdio_enabled() -> bool:
+    return os.getenv(MCP_STDIO_ENABLED_ENV_VAR, "").strip().lower() == "true"
+
+
+def is_mcp_stdio_flag_key(env_var_name: str) -> bool:
+    return env_var_name.upper() == MCP_STDIO_ENABLED_ENV_VAR
+
+
+def is_mcp_stdio_blocked(transport: str | None) -> bool:
+    return transport == MCPTransport.stdio and not is_mcp_stdio_enabled()
+
+
+def warn_if_mcp_stdio_blocked(server_name: str | None, transport: str | None) -> None:
+    if is_mcp_stdio_blocked(transport):
+        verbose_logger.warning("MCP server '%s' will not start: %s", server_name, MCP_STDIO_DISABLED_MESSAGE)

litellm/proxy/_types.py

@@ -28,6 +28,7 @@
     validate_langfuse_span_scope_value,
     validate_no_callback_env_reference,
 )
+from litellm.proxy._experimental.mcp_server.stdio_gate import MCP_STDIO_DISABLED_MESSAGE, is_mcp_stdio_enabled
 from litellm.types.agents import AgentCaller, AgentResponse
 from litellm.types.integrations.compression_interception import (
     CompressionSavingsMetadata,
@@ -1529,6 +1530,30 @@
     raise _per_server_oauth_discovery_error()
 
 
+def _validate_mcp_transport_fields(values: object) -> None:
+    if not isinstance(values, dict):
+        return
+    transport: Final = values.get("transport")
+    if transport in (MCPTransport.http, MCPTransport.sse):
+        if not values.get("url") and not values.get("spec_path"):
+            raise ValueError("url or spec_path is required for HTTP/SSE transport")
+        return
+    if transport != MCPTransport.stdio:
+        return
+    if not is_mcp_stdio_enabled():
+        raise ValueError(MCP_STDIO_DISABLED_MESSAGE)
+    command: Final = values.get("command")
+    if not command:
+        raise ValueError("command is required for stdio transport")
+    if not values.get("args"):
+        raise ValueError("args is required for stdio transport")
+    if os.path.basename(str(command)) not in MCP_STDIO_ALLOWED_COMMANDS:
+        raise ValueError(
+            f"Command '{command}' is not in the allowed commands list "
+            f"for stdio transport. Allowed commands: {sorted(MCP_STDIO_ALLOWED_COMMANDS)}"
+        )
+
+
 class NewMCPServerRequest(LiteLLMPydanticObjectBase):
     server_id: str | None = None
     server_name: str | None = None
@@ -1595,23 +1620,7 @@
     @model_validator(mode="before")
     @classmethod
     def validate_transport_fields(cls, values):
-        if isinstance(values, dict):
-            transport: Final = values.get("transport")
-            if transport == MCPTransport.stdio:
-                if not values.get("command"):
-                    raise ValueError("command is required for stdio transport")
-                if not values.get("args"):
-                    raise ValueError("args is required for stdio transport")
-                # Validate command against allowlist to prevent arbitrary execution
-                base_command: Final = os.path.basename(values["command"])
-                if base_command not in MCP_STDIO_ALLOWED_COMMANDS:
-                    raise ValueError(
-                        f"Command '{values['command']}' is not in the allowed commands list "
-                        f"for stdio transport. Allowed commands: {sorted(MCP_STDIO_ALLOWED_COMMANDS)}"
-                    )
-            elif transport in [MCPTransport.http, MCPTransport.sse]:
-                if not values.get("url") and not values.get("spec_path"):
-                    raise ValueError("url or spec_path is required for HTTP/SSE transport")
+        _validate_mcp_transport_fields(values)
         return values
 
     @model_validator(mode="before")
@@ -1694,23 +1703,7 @@
     @model_validator(mode="before")
     @classmethod
     def validate_transport_fields(cls, values):
-        if isinstance(values, dict):
-            transport: Final = values.get("transport")
-            if transport == MCPTransport.stdio:
-                if not values.get("command"):
-                    raise ValueError("command is required for stdio transport")
-                if not values.get("args"):
-                    raise ValueError("args is required for stdio transport")
-                # Validate command against allowlist to prevent arbitrary execution
-                base_command: Final = os.path.basename(values["command"])
-                if base_command not in MCP_STDIO_ALLOWED_COMMANDS:
-                    raise ValueError(
-                        f"Command '{values['command']}' is not in the allowed commands list "
-                        f"for stdio transport. Allowed commands: {sorted(MCP_STDIO_ALLOWED_COMMANDS)}"
-                    )
-            elif transport in [MCPTransport.http, MCPTransport.sse]:
-                if not values.get("url") and not values.get("spec_path"):
-                    raise ValueError("url or spec_path is required for HTTP/SSE transport")
+        _validate_mcp_transport_fields(values)
         return values
 
     @model_validator(mode="before")

litellm/proxy/discovery_endpoints/ui_discovery_endpoints.py

@@ -4,6 +4,7 @@
 
 from fastapi import APIRouter
 
+from litellm.proxy._experimental.mcp_server.stdio_gate import is_mcp_stdio_enabled
 from litellm.proxy.common_utils.html_forms.default_credentials_hint import should_hide_default_credentials_hint
 from litellm.types.proxy.discovery_endpoints.ui_discovery_endpoints import (
     UiDiscoveryEndpoints,
@@ -41,4 +42,5 @@ async def get_ui_config():
         hide_default_credentials_hint=hide_default_credentials_hint,
         is_control_plane=is_control_plane,
         workers=proxy_config.worker_registry if is_control_plane else [],
+        mcp_stdio_enabled=is_mcp_stdio_enabled(),
     )

litellm/proxy/proxy_server.py

@@ -335,6 +335,7 @@ def generate_feedback_box():
 from litellm.llms.openai_like.model_info import MODEL_INFO_REFRESH_SECONDS
 from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
 from litellm.proxy._experimental.mcp_server.byok_credential_cache import byok_credential_cache
+from litellm.proxy._experimental.mcp_server.stdio_gate import MCP_STDIO_ENABLED_ENV_VAR, is_mcp_stdio_flag_key
 from litellm.proxy._lazy_features import attach_lazy_features, reserve_lazy_slot
 from litellm.proxy._types import *
 from litellm.proxy.analytics_endpoints.analytics_endpoints import (
@@ -5401,6 +5402,7 @@ def __init__(self) -> None:
         self._last_cyberark_config: dict[str, object] | None = None  # mutable-ok: change-detection cache
         self._last_cleanup_schedule_attempt: tuple[object, ...] | None = None
         self._cleanup_reschedule_failed: bool = False
+        self._warned_db_mcp_stdio_flag_ignored: bool = False
         self._cyberark_boot_env: dict[str, str | None] | None = None  # mutable-ok: deployment env snapshot, set once
         self.worker_registry: list[WorkerRegistryEntry] = []
         self.config_sync_subscriber: ConfigSyncSubscriber | None = None
@@ -6184,6 +6186,12 @@ def _load_environment_variables(self, config: dict):
                 if key in self._BLOCKED_ENV_KEYS:
                     verbose_proxy_logger.warning("Skipping blocked environment variable key: %s", key)
                     continue
+                if isinstance(key, str) and is_mcp_stdio_flag_key(key):
+                    verbose_proxy_logger.warning(
+                        "Ignoring %s set in the config file. Set it in the proxy's environment instead",
+                        MCP_STDIO_ENABLED_ENV_VAR,
+                    )
+                    continue
                 #########################################################
                 # handles this scenario:
                 # ```yaml
@@ -7585,6 +7593,14 @@ def _decrypt_and_set_db_env_variables(
         """
         decrypted_env_vars: Final = {}
         for k, v in environment_variables.items():
+            if isinstance(k, str) and is_mcp_stdio_flag_key(k):
+                if not self._warned_db_mcp_stdio_flag_ignored:
+                    verbose_proxy_logger.warning(
+                        "Ignoring %s stored in the database. Set it in the proxy's environment instead",
+                        MCP_STDIO_ENABLED_ENV_VAR,
+                    )
+                    self._warned_db_mcp_stdio_flag_ignored = True
+                continue
             try:
                 decrypted_value = decrypt_value_helper(value=v, key=k, return_original_value=return_original_value)
                 if decrypted_value is not None:

litellm/types/proxy/discovery_endpoints/ui_discovery_endpoints.py

@@ -11,4 +11,5 @@ class UiDiscoveryEndpoints(BaseModel):
     sso_configured: bool
     hide_default_credentials_hint: bool = False
     is_control_plane: bool = False
+    mcp_stdio_enabled: bool = False
     workers: list[WorkerRegistryEntry] = []

tests/mcp_tests/test_proxy_mcp_e2e.py

@@ -55,6 +55,7 @@ def _clear_proxy_database_env() -> typing.Iterator[None]:
     # the config file. We must set it here so the lifespan doesn't reset it to None.
     mp.setenv("LITELLM_MASTER_KEY", "sk-1234")
     mp.setenv("LITELLM_DANGEROUSLY_PERMIT_WEAK_OR_UNSET_MASTER_KEY", "true")
+    mp.setenv("LITELLM_ENABLE_MCP_STDIO", "true")
     try:
         yield
     finally: