fix(mcp): refuse stdio MCP transport unless LITELLM_ENABLE_STDIO_MCP is set

litellm/constants.py

@@ -207,10 +207,16 @@
 MCP_BYOK_CREDENTIAL_CACHE_TTL_SECONDS: Final = 60
 MCP_BYOK_CREDENTIAL_CACHE_MAX_SIZE: Final = 4096
 
-# Allowlist of commands permitted for MCP stdio transport.
-# Prevents arbitrary command execution via /mcp-rest/test/* endpoints or server creation.
-# Note: allowlisted runtimes can still execute code via args (e.g. python -c "...").
-# This is an accepted residual risk since these endpoints require PROXY_ADMIN.
+# Config and DB environment_variables are applied after import, so they cannot enable stdio.
+MCP_STDIO_TRANSPORT_ENABLED: Final[bool] = os.getenv("LITELLM_ENABLE_STDIO_MCP", "false").lower() == "true"
+MCP_STDIO_DISABLED_MESSAGE: Final[str] = (
+    "stdio MCP transport is disabled on this proxy. stdio servers run a command on the proxy host, so they are off "
+    "by default. To allow them, the operator must set LITELLM_ENABLE_STDIO_MCP=true in the proxy's process "
+    "environment. Alternatively, run the server behind a stdio-to-HTTP bridge and register it as an HTTP MCP server"
+)
+
+# Defense-in-depth command allowlist for MCP stdio transport, which only applies after operator opt-in.
+# The opt-in is the real control; allowlisted runtimes can still execute arbitrary code through their arguments.
 # Extend via LITELLM_MCP_STDIO_EXTRA_COMMANDS env var (comma-separated).
 _MCP_STDIO_EXTRA_COMMANDS: Final = os.getenv("LITELLM_MCP_STDIO_EXTRA_COMMANDS", "")
 MCP_STDIO_ALLOWED_COMMANDS: Final[frozenset] = frozenset(

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -52,6 +52,7 @@
 from typing_extensions import ReadOnly
 
 import litellm
+import litellm.constants as litellm_constants
 from litellm._logging import verbose_logger
 from litellm.caching.in_memory_cache import InMemoryCache
 from litellm.constants import (
@@ -2382,6 +2383,19 @@ def is_config_declared_server(self, server_id: str) -> bool:
         a LiteLLM_MCPServerTable row is always resolved to that row first."""
         return server_id in self.config_mcp_servers
 
+    @staticmethod
+    def _warn_if_stdio_transport_disabled(server_name: str, server_config: MCPServerConfig) -> None:
+        if (
+            litellm_constants.MCP_STDIO_TRANSPORT_ENABLED
+            or server_config.get("transport", MCPTransport.http) != MCPTransport.stdio
+        ):
+            return
+        verbose_logger.warning(
+            "MCP server '%s' uses stdio transport, which is disabled by default. Set "
+            "LITELLM_ENABLE_STDIO_MCP=true in the proxy's process environment to enable it",
+            server_name,
+        )
+
     async def load_servers_from_config(
         self,
         mcp_servers_config: dict[str, MCPServerConfig],
@@ -2408,6 +2422,7 @@ async def load_servers_from_config(
 
         for server_name, raw_server_config in mcp_servers_config.items():
             server_config: MCPServerConfig = raw_server_config
+            self._warn_if_stdio_transport_disabled(server_name, server_config)
             _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()
@@ -4280,6 +4295,9 @@ async def _create_mcp_client(
 
         # Handle stdio transport
         if transport == MCPTransport.stdio:
+            if not litellm_constants.MCP_STDIO_TRANSPORT_ENABLED:
+                raise HTTPException(status_code=403, detail=litellm_constants.MCP_STDIO_DISABLED_MESSAGE)
+
             resolved_env: Final = (
                 stdio_env
                 if stdio_env is not None

litellm/proxy/_types.py

@@ -21,6 +21,7 @@
 )
 from typing_extensions import NotRequired, ReadOnly, Required, TypedDict
 
+import litellm.constants as litellm_constants
 from litellm._uuid import uuid
 from litellm.constants import DEFAULT_STAGGER_WINDOW_SECONDS, MCP_STDIO_ALLOWED_COMMANDS
 from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
@@ -1652,6 +1653,8 @@ def validate_transport_fields(cls, values):
         if isinstance(values, dict):
             transport: Final = values.get("transport")
             if transport == MCPTransport.stdio:
+                if not litellm_constants.MCP_STDIO_TRANSPORT_ENABLED:
+                    raise ValueError(litellm_constants.MCP_STDIO_DISABLED_MESSAGE)
                 if not values.get("command"):
                     raise ValueError("command is required for stdio transport")
                 if not values.get("args"):
@@ -1751,6 +1754,8 @@ def validate_transport_fields(cls, values):
         if isinstance(values, dict):
             transport: Final = values.get("transport")
             if transport == MCPTransport.stdio:
+                if not litellm_constants.MCP_STDIO_TRANSPORT_ENABLED:
+                    raise ValueError(litellm_constants.MCP_STDIO_DISABLED_MESSAGE)
                 if not values.get("command"):
                     raise ValueError("command is required for stdio transport")
                 if not values.get("args"):

litellm/proxy/proxy_server.py

@@ -6174,6 +6174,8 @@ def parse_search_tools(self, config: dict) -> list[SearchToolTypedDict] | None:
         "LOGNAME",
         "NO_PROXY",
         "no_proxy",
+        "LITELLM_ENABLE_STDIO_MCP",
+        "LITELLM_MCP_STDIO_EXTRA_COMMANDS",
     }
 
     def _load_environment_variables(self, config: dict):
@@ -6182,8 +6184,11 @@ def _load_environment_variables(self, config: dict):
         environment_variables: Final = config.get("environment_variables", None)
         if environment_variables:
             for key, value in environment_variables.items():
-                if key in self._BLOCKED_ENV_KEYS:
-                    verbose_proxy_logger.warning("Skipping blocked environment variable key: %s", key)
+                if key.upper() in self._BLOCKED_ENV_KEYS:
+                    verbose_proxy_logger.warning(
+                        "Skipping blocked environment variable key: %s",
+                        cast(str, key),  # cast-ok: config environment-variable keys are strings
+                    )
                     continue
                 #########################################################
                 # handles this scenario:
@@ -7586,6 +7591,12 @@ def _decrypt_and_set_db_env_variables(
         """
         decrypted_env_vars: Final = {}
         for k, v in environment_variables.items():
+            if k.upper() in self._BLOCKED_ENV_KEYS:
+                verbose_proxy_logger.warning(
+                    "Skipping blocked environment variable key: %s",
+                    cast(str, k),  # cast-ok: stored environment-variable keys are strings
+                )
+                continue
             try:
                 decrypted_value = decrypt_value_helper(value=v, key=k, return_original_value=return_original_value)
                 if decrypted_value is not None:
@@ -8033,6 +8044,12 @@ def _prepared_db_settings_values(self, section: Section, value: object) -> Mappi
                 **{key.upper(): decrypted_value for key, decrypted_value in decrypted.items()},
             }
             for key, decrypted_value in normalized.items():
+                if key.upper() in self._BLOCKED_ENV_KEYS:
+                    verbose_proxy_logger.warning(
+                        "Skipping blocked environment variable key: %s",
+                        cast(str, key),  # cast-ok: normalized environment-variable keys are strings
+                    )
+                    continue
                 os.environ[key] = decrypted_value
             return _as_settings_mapping(normalized)
 

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

@@ -5,7 +5,7 @@
 import logging
 import os
 import sys
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, Final, Literal, Optional
@@ -29,6 +29,7 @@
 import httpx
 import httpx2
 from mcp import ReadResourceResult, Resource
+from mcp.server.context import ServerRequestContext
 from mcp.types import (
     CallToolResult,
     GetPromptResult,
@@ -39,6 +40,7 @@
 from mcp.types import Tool as MCPTool
 from pydantic import AnyUrl, TypeAdapter
 
+import litellm.constants as litellm_constants
 from litellm.constants import MCP_METADATA_TIMEOUT
 from litellm.proxy._experimental.mcp_server.tool_outcome import TextResult
 from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
@@ -314,8 +316,46 @@ def test_get_mcp_server_by_id_rejects_private_server_for_public_ip(self):
         with patch.object(manager, "_get_general_settings", return_value={}):
             assert manager.get_mcp_server_by_id(server.server_id, client_ip="8.8.8.8") is None
 
-    async def test_create_mcp_client_stdio(self):
+    async def test_create_mcp_client_stdio_disabled_rejects_before_client_construction(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", False)
+        server = MCPServer(
+            server_id="stdio-disabled",
+            name="stdio_disabled",
+            transport=MCPTransport.stdio,
+            command="python3",
+            args=["-m", "test_server"],
+        )
+
+        with (
+            patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.MCPClient") as mcp_client,
+            pytest.raises(HTTPException) as exc_info,
+        ):
+            await MCPServerManager()._create_mcp_client(server)
+
+        assert exc_info.value.status_code == 403
+        assert exc_info.value.detail == litellm_constants.MCP_STDIO_DISABLED_MESSAGE
+        mcp_client.assert_not_called()
+
+    async def test_create_mcp_client_http_is_unaffected_when_stdio_is_disabled(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", False)
+        server = MCPServer(
+            server_id="http-enabled",
+            name="http_enabled",
+            url="https://example.com/mcp",
+            transport=MCPTransport.http,
+        )
+
+        client = await MCPServerManager()._create_mcp_client(server)
+
+        assert client.transport_type == MCPTransport.http
+
+    async def test_create_mcp_client_stdio(self, monkeypatch: pytest.MonkeyPatch) -> None:
         """Test creating MCP client for stdio transport"""
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         manager = MCPServerManager()
 
         stdio_server = MCPServer(
@@ -458,11 +498,14 @@ async def test_half_configured_id_jag_fails_closed_instead_of_deferring_to_v1(se
         assert exc_info.value.status_code == 500
         assert "oauth2_id_jag" in str(exc_info.value.detail)
 
-    async def test_create_mcp_client_stdio_injects_npm_config_cache(self):
+    async def test_create_mcp_client_stdio_injects_npm_config_cache(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
         """Test that _create_mcp_client injects NPM_CONFIG_CACHE when not already set,
         and preserves user-provided NPM_CONFIG_CACHE when present."""
         from litellm.constants import MCP_NPM_CACHE_DIR
 
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         manager = MCPServerManager()
 
         # Case 1: NPM_CONFIG_CACHE not set -> should be injected
@@ -585,6 +628,32 @@ async def test_load_servers_from_config_warns_on_invalid_alias(self, config_only
 
         assert any("invalid alias 'bad/name'" in message for message in caplog.messages)
 
+    @pytest.mark.asyncio
+    async def test_load_servers_from_config_keeps_disabled_stdio_server_and_warns(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        config_only_mcp_manager_factory: type[MCPServerManager],
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", False)
+        manager = config_only_mcp_manager_factory()
+
+        with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+            await manager.load_servers_from_config(
+                {
+                    "local_server": {
+                        "transport": "stdio",
+                        "command": "python3",
+                        "args": ["-m", "test_server"],
+                    }
+                }
+            )
+
+        assert any(server.name == "local_server" for server in manager.config_mcp_servers.values())
+        assert any(
+            "local_server" in message and "LITELLM_ENABLE_STDIO_MCP" in message for message in caplog.messages
+        )
+
     @pytest.mark.asyncio
     async def test_load_servers_from_config_accepts_valid_alias(self, config_only_mcp_manager_factory, caplog):
         """Valid aliases should be accepted and populate the registry."""
@@ -9876,7 +9945,8 @@ async def test_static_token_missing_rejects_before_connecting(self):
         assert exc.value.status_code == 500
         assert "credential" in str(exc.value.detail)
 
-    async def test_stdio_migrated_auth_type_still_defers_to_v1(self):
+    async def test_stdio_migrated_auth_type_still_defers_to_v1(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         client = await MCPServerManager()._create_mcp_client(
             MCPServer(
                 server_id="stdio-graft",
@@ -13563,7 +13633,13 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("transport", ["http", "stdio"])
-async def test_debug_reports_legacy_signing_and_non_http_transport(_mcp_request_ctx, transport: Literal["http", "stdio"]) -> None:
+async def test_debug_reports_legacy_signing_and_non_http_transport(
+    _mcp_request_ctx: Callable[..., ServerRequestContext],
+    transport: Literal["http", "stdio"],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    if transport == "stdio":
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
     from litellm.proxy._experimental.mcp_server.mcp_context import active_mcp_request_ctx_var
     from starlette.requests import Request
 

tests/unit/proxy/_experimental/mcp_server/test_rest_endpoints.py

@@ -17,6 +17,7 @@
 from mcp.types import CallToolResult, TextContent
 from starlette.requests import Request
 
+import litellm.constants as litellm_constants
 from litellm.constants import MCP_TOOL_LISTING_TIMEOUT
 from litellm.proxy._experimental.mcp_server import rest_endpoints
 from litellm.proxy._experimental.mcp_server.auth import (
@@ -3730,6 +3731,10 @@ async def fake_get_tools_from_server(**kwargs):
 class TestStdioCommandAllowlist:
     """Tests for MCP stdio command allowlist validation."""
 
+    @pytest.fixture(autouse=True)
+    def enable_stdio_for_allowlist_tests(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
+
     def test_allowed_command_passes_validation(self):
         """npx, uvx, python, etc. should be accepted."""
         req = NewMCPServerRequest(

tests/unit/proxy/management_endpoints/test_mcp_connector_import.py

@@ -1,5 +1,6 @@
 import pytest
 
+import litellm.constants as litellm_constants
 from litellm.proxy.management_endpoints.mcp_connector_import import (
     ConnectorConversionError,
     ConvertedConnector,
@@ -124,7 +125,8 @@ def test_sse_type_maps_to_sse_transport(self):
         assert isinstance(result, ConvertedConnector)
         assert result.request.transport == MCPTransport.sse
 
-    def test_stdio_connector(self):
+    def test_stdio_connector(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         result = _single(
             {
                 "mcpServers": {
@@ -142,11 +144,18 @@ def test_stdio_connector(self):
         assert result.request.args == ["-y", "@example/mcp-server"]
         assert result.request.env == {"API_KEY": "value"}
 
-    def test_disallowed_stdio_command_returns_error(self):
+    def test_disallowed_stdio_command_returns_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         result = _single({"mcpServers": {"evil": {"command": "rm", "args": ["-rf", "/"]}}})
         assert isinstance(result, ConnectorConversionError)
         assert "not in the allowed commands list" in result.error
 
+    def test_stdio_connector_is_rejected_when_operator_gate_is_off(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", False)
+        result = _single({"mcpServers": {"local": {"command": "python3", "args": ["-m", "test_server"]}}})
+        assert isinstance(result, ConnectorConversionError)
+        assert litellm_constants.MCP_STDIO_DISABLED_MESSAGE in result.error
+
     def test_unsupported_type_returns_error(self):
         result = _single({"mcpServers": {"ws": {"type": "websocket", "url": "wss://x.example"}}})
         assert isinstance(result, ConnectorConversionError)

tests/unit/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -20,6 +20,7 @@
 from fastapi.testclient import TestClient
 
 import litellm
+import litellm.constants as litellm_constants
 from litellm._uuid import uuid
 from litellm.caching.caching import DualCache
 from litellm.integrations.custom_guardrail import CustomGuardrail
@@ -151,6 +152,25 @@ def create_mcp_router_test_client() -> TestClient:
     return TestClient(app)
 
 
+def test_post_mcp_server_returns_422_when_stdio_is_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", False)
+    client = create_mcp_router_test_client()
+    client.app.dependency_overrides[mgmt_endpoints.user_api_key_auth] = lambda: generate_mock_user_api_key_auth()
+
+    response = client.post(
+        "/v1/mcp/server",
+        json={
+            "server_name": "stdio-gate-test",
+            "transport": "stdio",
+            "command": "python3",
+            "args": ["-m", "test_server"],
+        },
+    )
+
+    assert response.status_code == 422
+    assert litellm_constants.MCP_STDIO_DISABLED_MESSAGE in response.text
+
+
 def patch_proxy_general_settings(settings: dict):
     fake_proxy_server_module = types.SimpleNamespace(general_settings=settings)
     return patch.dict(
@@ -4881,7 +4901,8 @@ async def test_register_mcp_server_requires_team_key(self):
         assert "team" in str(exc_info.value.detail).lower()
 
     @pytest.mark.asyncio
-    async def test_register_mcp_server_rejects_stdio_transport(self):
+    async def test_register_mcp_server_rejects_stdio_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm_constants, "MCP_STDIO_TRANSPORT_ENABLED", True)
         # stdio servers spawn a local subprocess on the proxy host. Accepting
         # them from the non-admin submission endpoint would let a team member
         # propose a config that an admin could rubber-stamp into local code