diff --git a/pyinfra/api/deploy.py b/pyinfra/api/deploy.py
index 85cdca64d..d5db44440 100644
--- a/pyinfra/api/deploy.py
+++ b/pyinfra/api/deploy.py
@@ -110,7 +110,7 @@ def decorated_func(*args, **kwargs):
                 'Deploy called without state/host: {0} ({1})'
             ).format(func, get_call_location()))
 
-        deploy_kwargs, _ = pop_global_op_kwargs(state, kwargs)
+        deploy_kwargs, _ = pop_global_op_kwargs(state, host, kwargs)
 
         # Name the deploy
         deploy_name = getattr(func, 'deploy_name', func.__name__)
diff --git a/pyinfra/api/operation.py b/pyinfra/api/operation.py
index 8429ae6ee..2c5c1b691 100644
--- a/pyinfra/api/operation.py
+++ b/pyinfra/api/operation.py
@@ -180,7 +180,7 @@ def decorated_func(*args, **kwargs):
         #
 
         # Get the meta kwargs (globals that apply to all hosts)
-        global_kwargs, global_kwarg_keys = pop_global_op_kwargs(state, kwargs)
+        global_kwargs, global_kwarg_keys = pop_global_op_kwargs(state, host, kwargs)
 
         # If this op is being called inside another, just return here
         # (any unwanted/op-related kwargs removed above).
diff --git a/pyinfra/api/operation_kwargs.py b/pyinfra/api/operation_kwargs.py
index 4c7cc5626..dace96560 100644
--- a/pyinfra/api/operation_kwargs.py
+++ b/pyinfra/api/operation_kwargs.py
@@ -123,9 +123,14 @@ def get_executor_kwarg_keys():
     return list(keys)
 
 
-def pop_global_op_kwargs(state, kwargs):
+def pop_global_op_kwargs(state, host, kwargs):
     '''
-    Pop and return operation global keyword arguments.
+    Pop and return operation global keyword arguments, in preferred order:
+
+    + From the current context (operation kwargs)
+    + From any current @deploy context (deploy kwargs)
+    + From the host data variables
+    + From the config variables
     '''
 
     meta_kwargs = state.deploy_kwargs or {}
@@ -149,6 +154,10 @@ def get_kwarg(key, default=None):
                 if default:
                     default = default(state.config)
 
+                host_default = getattr(host.data, key, None)
+                if host_default is not None:
+                    default = host_default
+
             value, has_key = get_kwarg(key, default=default)
             if handler:
                 value = handler(state.config, value)
diff --git a/tests/test_api/test_api_operation_kwargs.py b/tests/test_api/test_api_operation_kwargs.py
new file mode 100644
index 000000000..3a80c84f9
--- /dev/null
+++ b/tests/test_api/test_api_operation_kwargs.py
@@ -0,0 +1,49 @@
+from unittest import TestCase
+
+from pyinfra.api import Config, Inventory, State
+from pyinfra.api.operation_kwargs import pop_global_op_kwargs
+
+
+class TestOperationKwargs(TestCase):
+    def test_get_from_config(self):
+        config = Config(SUDO='config-value')
+        inventory = Inventory((('somehost',), {}))
+
+        state = State(config=config, inventory=inventory)
+
+        kwargs, keys = pop_global_op_kwargs(state, inventory.get_host('somehost'), {})
+        assert kwargs['sudo'] == 'config-value'
+
+    def test_get_from_host(self):
+        config = Config(SUDO='config-value')
+        inventory = Inventory(([('somehost', {'sudo': 'host-value'})], {}))
+
+        state = State(config=config, inventory=inventory)
+
+        kwargs, keys = pop_global_op_kwargs(state, inventory.get_host('somehost'), {})
+        assert kwargs['sudo'] == 'host-value'
+
+    def test_get_from_state_deploy_kwargs(self):
+        config = Config(SUDO='config-value')
+        inventory = Inventory(([('somehost', {'sudo': 'host-value'})], {}))
+
+        state = State(config=config, inventory=inventory)
+        state.deploy_kwargs = {'sudo': 'deploy-kwarg-value'}
+
+        kwargs, keys = pop_global_op_kwargs(state, inventory.get_host('somehost'), {})
+        assert kwargs['sudo'] == 'deploy-kwarg-value'
+
+    def test_get_from_kwargs(self):
+        config = Config(SUDO='config-value')
+        inventory = Inventory(([('somehost', {'sudo': 'host-value'})], {}))
+
+        state = State(config=config, inventory=inventory)
+        state.deploy_kwargs = {'sudo': 'deploy-kwarg-value'}
+
+        kwargs, keys = pop_global_op_kwargs(
+            state,
+            inventory.get_host('somehost'),
+            {'sudo': 'kwarg-value'},
+        )
+        assert kwargs['sudo'] == 'kwarg-value'
+        assert 'sudo' in keys