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

.circleci/scripts/run_integration.sh

@@ -147,6 +147,7 @@ start_proxy() {
   local port="$1"
   local log_name="$2"
   local -a cost_map_env
+  local -a stdio_env=()
   if [ "$suite" = cost ]; then
     cost_map_env=(
       "LITELLM_MODEL_COST_MAP_URL=$INTEGRATION_UPSTREAM_URL/_cost_map"
@@ -160,12 +161,15 @@ start_proxy() {
   else
     cost_map_env=("LITELLM_LOCAL_MODEL_COST_MAP=True")
   fi
+  if [ "$suite" = mcp ]; then
+    stdio_env=("LITELLM_ENABLE_STDIO_MCP=true")
+  fi
   local -a database_env=("DATABASE_URL=${INTEGRATION_PROXY_DATABASE_URL:-$DATABASE_URL}")
   if [ -n "$INTEGRATION_PROXY_READ_REPLICA_URL" ]; then
     database_env+=("DATABASE_URL_READ_REPLICA=$INTEGRATION_PROXY_READ_REPLICA_URL")
   fi
   setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \
-    "${database_env[@]}" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
+    "${database_env[@]}" "${stdio_env[@]}" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
     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:-}" \

.circleci/tests.yml

@@ -186,6 +186,9 @@ jobs:
       legacy_mcp_peer:
         type: boolean
         default: false
+      stdio_mcp:
+        type: boolean
+        default: false
       reruns:
         type: integer
         default: 0
@@ -227,6 +230,7 @@ jobs:
             rerun_args=(-p no:rerunfailures)
             if [ "<< parameters.reruns >>" -gt 0 ]; then rerun_args=(--reruns << parameters.reruns >> --reruns-delay 1 --rerun-except "from pytest-timeout"); fi
             test_env=(PATH="$PATH" HOME="$HOME" CI=true COVERAGE_CORE="$COVERAGE_CORE" LITELLM_LOCAL_MODEL_COST_MAP="$LITELLM_LOCAL_MODEL_COST_MAP")
+            if [ "<< parameters.stdio_mcp >>" = "true" ]; then test_env+=(LITELLM_ENABLE_STDIO_MCP=true); fi
             if [ -n "${MCP_TEST_PEER_PYTHON:-}" ]; then test_env+=(MCP_TEST_PEER_PYTHON="$MCP_TEST_PEER_PYTHON"); fi
             set +e
             env -i "${test_env[@]}" \
@@ -330,6 +334,7 @@ workflows:
           shards: 1
           workers: 2
           legacy_mcp_peer: true
+          stdio_mcp: true
           base_ref: << pipeline.event.name == "pull_request" and pipeline.event.github.pull_request.base.ref or "" >>
           pull_request_url: << pipeline.event.name == "pull_request" and pipeline.event.github.pull_request.url or "" >>
       - unit:

.github/workflows/_test-unit-base.yml

@@ -84,6 +84,11 @@ on:
         required: false
         type: boolean
         default: false
+      stdio_mcp:
+        description: "Enable stdio MCP for the proxy MCP test shard"
+        required: false
+        type: boolean
+        default: false
 
 permissions:
   contents: read
@@ -97,6 +102,8 @@ jobs:
     name: Run tests
     runs-on: ubuntu-latest
     timeout-minutes: ${{ inputs.job-timeout-minutes }}
+    env:
+      LITELLM_ENABLE_STDIO_MCP: ${{ inputs.stdio_mcp && 'true' || 'false' }}
     permissions:
       contents: read
       pull-requests: read

.github/workflows/test-unit.yml

@@ -295,6 +295,7 @@ jobs:
     with:
       test-path: ${{ matrix.test-path }}
       unit-flag: ${{ matrix.unit-flag || '' }}
+      stdio_mcp: ${{ matrix.shard == 'mcp-integration' }}
       workers: ${{ matrix.workers }}
       reruns: ${{ matrix.reruns }}
       timeout-minutes: ${{ matrix.timeout-minutes }}

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 @@
         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 @@
 
         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 @@
 
         # 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 @@
         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 @@
         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/integration/_support/process.py

@@ -194,6 +194,9 @@ def owned_proxy_process(
         "LITELLM_MASTER_KEY": gateway.key,
         "LITELLM_SALT_KEY": os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt"),
         "STORE_MODEL_IN_DB": "True",
+        **(
+            {} if "LITELLM_ENABLE_STDIO_MCP" in remove_environment else {"LITELLM_ENABLE_STDIO_MCP": "true"}
+        ),  # The suite intentionally exercises stdio transport
         **overrides,
     }
     output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(directory)))

tests/integration/mcp/test_mcp_management.py

@@ -253,13 +253,35 @@ def test_invalid_registrations_are_rejected(gateway: Gateway) -> None:
             {"server_name": alias + "c", "transport": "stdio", "command": "/bin/sh", "args": ["-c", "true"]},
         )
         assert bad_command.status_code in (400, 422), bad_command.text
+        assert "not in the allowed commands list" in bad_command.text, bad_command.text
         hyphenless: Final = gateway.request(
             "POST", "/v1/mcp/server", {"server_name": "bad name!", **peer.registration()}
         )
         assert hyphenless.status_code in (400, 422), hyphenless.text
         assert len([s for s in _servers(gateway).values() if str(s["server_name"]).startswith(alias)]) == 1
 
 
+def test_stdio_registration_requires_process_opt_in(gateway: Gateway, tmp_path: Path) -> None:
+    with owned_proxy(
+        gateway,
+        tmp_path,
+        {},
+        remove_environment=("LITELLM_ENABLE_STDIO_MCP",),
+    ) as isolated:
+        response: Final = isolated.request(
+            "POST",
+            "/v1/mcp/server",
+            {
+                "server_name": "stdio_default_off",
+                "transport": "stdio",
+                "command": "/bin/sh",
+                "args": ["-c", "true"],
+            },
+        )
+        assert response.status_code in (400, 422), response.text
+        assert "LITELLM_ENABLE_STDIO_MCP" in response.text, response.text
+
+
 def test_access_group_membership_follows_edits(gateway: Gateway) -> None:
     with mcp_peer() as peer, gateway.scenario() as scenario:
         group: Final = "grp" + uuid.uuid4().hex[:8]