diff --git a/changelogs/fragments/48306-ansible-connection-json b/changelogs/fragments/48306-ansible-connection-json
new file mode 100644
index 00000000000..130b12bb4fe
--- /dev/null
+++ b/changelogs/fragments/48306-ansible-connection-json
@@ -0,0 +1,4 @@
+---
+bugfixes:
+- Use custom JSON encoder in conneciton.py so that ansible objects (AnsibleVaultEncryptedUnicode, for example)
+  can be sent to the persistent connection process
diff --git a/lib/ansible/module_utils/connection.py b/lib/ansible/module_utils/connection.py
index 08017a52a1c..41aa8139d7a 100644
--- a/lib/ansible/module_utils/connection.py
+++ b/lib/ansible/module_utils/connection.py
@@ -33,9 +33,11 @@ import socket
 import struct
 import traceback
 import uuid
+from datetime import date, datetime
 
 from functools import partial
 from ansible.module_utils._text import to_bytes, to_text
+from ansible.module_utils.common._collections_compat import Mapping
 from ansible.module_utils.six import iteritems
 from ansible.module_utils.six.moves import cPickle
 
@@ -141,7 +143,14 @@ class Connection(object):
                                   '\nSee the socket_path issue catergory in Network Debug and Troubleshooting Guide')
 
         try:
-            data = json.dumps(req)
+            data = json.dumps(req, cls=AnsibleJSONEncoder)
+        except TypeError as exc:
+            raise ConnectionError(
+                "Failed to encode some variables as JSON for communication with ansible-connection. "
+                "The original exception was: %s" % to_text(exc)
+            )
+
+        try:
             out = self.send(data)
             response = json.loads(out)
 
@@ -189,3 +198,29 @@ class Connection(object):
         sf.close()
 
         return to_text(response, errors='surrogate_or_strict')
+
+
+# NOTE: This is a modified copy of the class in parsing.ajson to get around not
+#       being able to import that directly, nor some of the type classes
+class AnsibleJSONEncoder(json.JSONEncoder):
+    '''
+    Simple encoder class to deal with JSON encoding of Ansible internal types
+    '''
+
+    def default(self, o):
+        if type(o).__name__ == 'AnsibleVaultEncryptedUnicode':
+            # vault object
+            value = {'__ansible_vault': to_text(o._ciphertext, errors='surrogate_or_strict', nonstring='strict')}
+        elif type(o).__name__ == 'AnsibleUnsafe':
+            # unsafe object
+            value = {'__ansible_unsafe': to_text(o, errors='surrogate_or_strict', nonstring='strict')}
+        elif isinstance(o, Mapping):
+            # hostvars and other objects
+            value = dict(o)
+        elif isinstance(o, (date, datetime)):
+            # date object
+            value = o.isoformat()
+        else:
+            # use default encoder
+            value = super(AnsibleJSONEncoder, self).default(o)
+        return value