diff --git a/changelogs/fragments/12120-consul-kv-empty-value.yml b/changelogs/fragments/12120-consul-kv-empty-value.yml
new file mode 100644
index 0000000000..f812364b54
--- /dev/null
+++ b/changelogs/fragments/12120-consul-kv-empty-value.yml
@@ -0,0 +1,2 @@
+minor_changes:
+  - consul_kv lookup plugin - add ``empty_value`` option to control what is returned for null Consul values (https://github.com/ansible-collections/community.general/issues/11039, https://github.com/ansible-collections/community.general/pull/12120).
diff --git a/plugins/lookup/consul_kv.py b/plugins/lookup/consul_kv.py
index 3ef88027ac..ede8fdb370 100644
--- a/plugins/lookup/consul_kv.py
+++ b/plugins/lookup/consul_kv.py
@@ -94,6 +94,16 @@ options:
     ini:
       - section: lookup_consul
         key: url
+  empty_value:
+    description:
+      - Controls what is returned when a Consul value is null.
+    type: str
+    default: 'textual_none'
+    choices:
+      textual_none: Return the string V(None). This is the legacy behavior.
+      python_none: Return a Python V(null)/V(None) value.
+      empty_string: Return an empty string.
+    version_added: 13.1.0
 """
 
 EXAMPLES = r"""
@@ -136,6 +146,13 @@ except ImportError:
     HAS_CONSUL = False
 
 
+_EMPTY_VALUE_MAP = {
+    "textual_none": "None",
+    "python_none": None,
+    "empty_string": "",
+}
+
+
 class LookupModule(LookupBase):
     def run(self, terms, variables=None, **kwargs):
         if not HAS_CONSUL:
@@ -165,6 +182,7 @@ class LookupModule(LookupBase):
 
         verify = (ca_path or validate_certs) if validate_certs else False
 
+        empty_value = _EMPTY_VALUE_MAP[self.get_option("empty_value")]
         values = []
         try:
             for term in terms:
@@ -182,9 +200,11 @@ class LookupModule(LookupBase):
                     # responds with a single or list of result maps
                     if isinstance(results[1], list):
                         for r in results[1]:
-                            values.append(to_text(r["Value"]))
+                            v = r["Value"]
+                            values.append(to_text(v) if v is not None else empty_value)
                     else:
-                        values.append(to_text(results[1]["Value"]))
+                        v = results[1]["Value"]
+                        values.append(to_text(v) if v is not None else empty_value)
         except Exception as e:
             raise AnsibleError(f"Error locating '{term}' in kv store. Error was {e}") from e