diff --git a/mistral/tests/unit/engine/test_environment.py b/mistral/tests/unit/engine/test_environment.py
index e056f33df..a88a750f0 100644
--- a/mistral/tests/unit/engine/test_environment.py
+++ b/mistral/tests/unit/engine/test_environment.py
@@ -19,6 +19,7 @@ from mistral.db.v2 import api as db_api
 from mistral.executors import default_executor as d_exe
 from mistral.executors import remote_executor as r_exe
 from mistral.services import workbooks as wb_service
+from mistral.services import workflows as wf_service
 from mistral.tests.unit.engine import base
 
 
@@ -199,3 +200,71 @@ class EnvironmentTest(base.EngineTestCase):
         }
 
         self._test_subworkflow(env)
+
+    def test_evaluate_env_parameter(self):
+        wf_text = """---
+        version: '2.0'
+
+        wf:
+          tasks:
+            task1:
+              action: std.noop
+              publish:
+                var1: <% env().var1 %>
+                var2: <% env().var2 %>
+        """
+
+        wf_service.create_workflows(wf_text)
+
+        env = {
+            "var1": "val1",
+            "var2": "<% env().var1 %>"
+        }
+
+        # Run with 'evaluate_env' set to True.
+
+        wf_ex = self.engine.start_workflow(
+            'wf',
+            {},
+            env=env,
+            evaluate_env=True
+        )
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            t = self._assert_single_item(wf_ex.task_executions, name='task1')
+
+            self.assertDictEqual(
+                {
+                    "var1": "val1",
+                    "var2": "val1"
+                },
+                t.published
+            )
+
+        # Run with 'evaluate_env' set to False.
+
+        wf_ex = self.engine.start_workflow(
+            'wf',
+            {},
+            env=env,
+            evaluate_env=False
+        )
+
+        self.await_workflow_success(wf_ex.id)
+
+        with db_api.transaction():
+            wf_ex = db_api.get_workflow_execution(wf_ex.id)
+
+            t = self._assert_single_item(wf_ex.task_executions, name='task1')
+
+            self.assertDictEqual(
+                {
+                    "var1": "val1",
+                    "var2": "<% env().var1 %>"
+                },
+                t.published
+            )
diff --git a/mistral/workflow/data_flow.py b/mistral/workflow/data_flow.py
index 11f501025..3175a389f 100644
--- a/mistral/workflow/data_flow.py
+++ b/mistral/workflow/data_flow.py
@@ -297,8 +297,14 @@ def add_environment_to_context(wf_ex):
     if 'env' in wf_ex.params:
         env = copy.deepcopy(wf_ex.params['env'])
 
-        # An env variable can be an expression of other env variables.
-        wf_ex.context['__env'] = expr.evaluate_recursively(env, {'__env': env})
+        if ('evaluate_env' in wf_ex.params and
+                not wf_ex.params['evaluate_env']):
+            wf_ex.context['__env'] = env
+        else:
+            wf_ex.context['__env'] = expr.evaluate_recursively(
+                env,
+                {'__env': env}
+            )
 
 
 def add_workflow_variables_to_context(wf_ex, wf_spec):