diff --git a/ecs_deploy/cli.py b/ecs_deploy/cli.py
index 6538ca9..f44b1b3 100644
--- a/ecs_deploy/cli.py
+++ b/ecs_deploy/cli.py
@@ -308,7 +308,9 @@ def scale(cluster, service, desired_count, access_key_id, secret_access_key, reg
 @click.option('--secret-access-key', help='AWS secret access key')
 @click.option('--profile', help='AWS configuration profile name')
 @click.option('--diff/--no-diff', default=True, help='Print what values were changed in the task definition')
-def run(cluster, task, count, command, env, secret, launchtype, subnet, securitygroup, public_ip, region, access_key_id, secret_access_key, profile, diff):
+@click.option('--exclusive-env', is_flag=True, default=False, help='Set the given environment variables exclusively and remove all other pre-existing env variables from all containers')
+@click.option('--exclusive-secrets', is_flag=True, default=False, help='Set the given secrets exclusively and remove all other pre-existing secrets from all containers')
+def run(cluster, task, count, command, env, secret, launchtype, subnet, securitygroup, public_ip, region, access_key_id, secret_access_key, profile, diff, exclusive_env, exclusive_secrets):
     """
     Run a one-off task.
 
@@ -323,8 +325,8 @@ def run(cluster, task, count, command, env, secret, launchtype, subnet, security
 
         td = action.get_task_definition(task)
         td.set_commands(**{key: value for (key, value) in command})
-        td.set_environment(env)
-        td.set_secrets(secret)
+        td.set_environment(env, exclusive_env)
+        td.set_secrets(secret, exclusive_secrets)
 
         if diff:
             print_diff(td, 'Using task definition: %s' % task)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 534cae1..939d6ab 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -643,12 +643,12 @@ def test_run_task_with_command(get_client, runner):
 
 
 @patch('ecs_deploy.cli.get_client')
-def test_run_task_with_environment_var(get_client, runner):
+def test_run_one_new_environment_variable(get_client, runner):
     get_client.return_value = EcsTestClient('acces_key', 'secret_key')
     result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'application', 'foo', 'bar'))
 
-    assert not result.exception
     assert result.exit_code == 0
+    assert not result.exception
 
     assert u"Using task definition: test-task" in result.output
     assert u'Changed environment "foo" of container "application" to: "bar"' in result.output
@@ -657,6 +657,157 @@ def test_run_task_with_environment_var(get_client, runner):
     assert u"- arn:lorem:ipsum" in result.output
 
 
+@patch('ecs_deploy.cli.get_client')
+def test_run_change_environment_variable_empty_string(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'application', 'foo', ''))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed environment "foo" of container "application" to: ""' in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_new_empty_environment_variable(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'application', 'new', ''))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed environment "new" of container "application" to: ""' in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_empty_environment_variable_again(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'webserver', 'empty', ''))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" not in result.output
+    assert u'Changed environment' not in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_previously_empty_environment_variable_with_value(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'webserver', 'empty', 'not-empty'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed environment "empty" of container "webserver" to: "not-empty"' in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_exclusive_environment(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'webserver', 'new-env', 'new-value', '--exclusive-env'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed environment "new-env" of container "webserver" to: "new-value"' in result.output
+
+    assert u'Removed environment "foo" of container "webserver"' in result.output
+    assert u'Removed environment "lorem" of container "webserver"' in result.output
+
+    assert u'Removed secret' not in result.output
+
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_exclusive_secret(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-s', 'webserver', 'new-secret', 'new-place', '--exclusive-secrets'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed secret "new-secret" of container "webserver" to: "new-place"' in result.output
+
+    assert u'Removed secret "baz" of container "webserver"' in result.output
+    assert u'Removed secret "dolor" of container "webserver"' in result.output
+
+    assert u'Removed environment' not in result.output
+
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_one_new_secret_variable(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2',
+                                        '-s', 'application', 'baz', 'qux',
+                                        '-s', 'webserver', 'baz', 'quux'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" in result.output
+    assert u'Changed secret "baz" of container "application" to: "qux"' in result.output
+    assert u'Changed secret "baz" of container "webserver" to: "quux"' in result.output
+    assert u'Changed secret "dolor" of container "webserver" to: "sit"' not in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_without_changing_environment_value(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-e', 'webserver', 'foo', 'bar'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" not in result.output
+    assert u'Changed environment' not in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
+@patch('ecs_deploy.cli.get_client')
+def test_run_without_changing_secrets_value(get_client, runner):
+    get_client.return_value = EcsTestClient('acces_key', 'secret_key')
+    result = runner.invoke(cli.run, (CLUSTER_NAME, 'test-task', '2', '-s', 'webserver', 'baz', 'qux'))
+
+    assert result.exit_code == 0
+    assert not result.exception
+
+    assert u"Using task definition: test-task" not in result.output
+    assert u'Changed secrets' not in result.output
+    assert u"Successfully started 2 instances of task: test-task:2" in result.output
+    assert u"- arn:foo:bar" in result.output
+    assert u"- arn:lorem:ipsum" in result.output
+
+
 @patch('ecs_deploy.cli.get_client')
 def test_run_task_without_diff(get_client, runner):
     get_client.return_value = EcsTestClient('acces_key', 'secret_key')