diff --git a/pyproject.toml b/pyproject.toml
index a8ef954c2..047d571a0 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -88,6 +88,7 @@ devel-test = [
   "pytest-bdd        ~=7.0",
   "pytest-cov        ~=5.0",
   "testcontainers    ~=3.0",
+  "jsonpath-ng",
 ]
 devel = ["pact-python[devel-types,devel-docs,devel-test]", "ruff==0.4.2"]
 
diff --git a/src/pact/v3/ffi.py b/src/pact/v3/ffi.py
index e23d94b86..858ddd1be 100644
--- a/src/pact/v3/ffi.py
+++ b/src/pact/v3/ffi.py
@@ -178,25 +178,216 @@ class MatchingRuleKeyValuePair: ...
 class MatchingRuleResult: ...
 
 
-class Message: ...
+class Message:
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new Message
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct Message *":
+            msg = (
+                "ptr must be a struct Message, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "Message"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"Message({self._ptr!r})"
+
+    @property
+    def description(self) -> str:
+        return ffi.string(lib.pactffi_message_get_description(self._ptr)).decode('utf-8')
+
+    @property
+    def contents(self) -> str:
+        _contents = lib.pactffi_message_get_contents(self._ptr)
+        if _contents != ffi.NULL:
+            return ffi.string(lib.pactffi_message_get_contents(self._ptr)).decode('utf-8')
+        return None
+
+    @property
+    def metadata(self) -> list[MessageMetadataPair]:
+        return [{m.key: m.value} for m in message_get_metadata_iter(self)]
 
 
 class MessageContents: ...
 
 
-class MessageHandle: ...
+class MessageHandle:
+    """
+    Handle to a Message.
+
+    [Rust
+    `MessageHandle`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/mock_server/handles/struct.MessageHandle.html)
+    """
+
+    def __init__(self, ref: int) -> None:
+        """
+        Initialise a new Message Handle.
+
+        Args:
+            ref:
+                Reference to the Message Handle.
+        """
+        self._ref: int = ref
+
+    def __str__(self) -> str:
+        """
+        String representation of the Message Handle.
+        """
+        return f"MessageHandle({self._ref})"
+
+    def __repr__(self) -> str:
+        """
+        String representation of the Message Handle.
+        """
+        return f"MessageHandle({self._ref!r})"
+
+
+class MessageMetadataIterator:
+    """
+    Iterator over a Message's metadata
+    """
+
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new Message Metadata Iterator.
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct MessageMetadataIterator *":
+            msg = (
+                "ptr must be a struct MessageMetadataIterator, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "MessageMetadataIterator"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"MessageMetadataIterator({self._ptr!r})"
+
+    def __del__(self) -> None:
+        """
+        Destructor for the Message Metadata Iterator.
+        """
+        message_metadata_iter_delete(self)
+
+    def __iter__(self) -> MessageMetadataIterator:
+        return self
+
+    def __next__(self) -> MessageMetadataPair:
+        """
+        Get the next interaction from the iterator.
+        """
+        msg = message_metadata_iter_next(self)
+        if msg == ffi.NULL:
+            raise StopIteration
+        return MessageMetadataPair(msg)
+
+
+
+class MessageMetadataPair:
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new MessageMetadataPair.
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct MessageMetadataPair *":
+            msg = (
+                "ptr must be a struct MessageMetadataPair, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
 
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "MessageMetadataPair"
 
-class MessageMetadataIterator: ...
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"MessageMetadataPair({self._ptr!r})"
 
+    @property
+    def key(self) -> str:
+        return ffi.string(self._ptr.key).decode('utf-8')
 
-class MessageMetadataPair: ...
+    @property
+    def value(self) -> str:
+        return ffi.string(self._ptr.value).decode('utf-8')
 
 
 class MessagePact: ...
 
 
-class MessagePactHandle: ...
+class MessagePactHandle:
+    """
+    Handle to a Pact.
+
+    [Rust
+    `PactHandle`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/mock_server/handles/struct.PactHandle.html)
+    """
+
+    def __init__(self, ref: int) -> None:
+        """
+        Initialise a new Message Pact Handle.
+
+        Args:
+            ref:
+                Rust library reference to the Pact Handle.
+        """
+        self._ref: int = ref
+
+    def __del__(self) -> None:
+        """
+        Destructor for the Message Pact Handle.
+        """
+        cleanup_plugins(self)
+        free_message_pact_handle(self)
+
+    def __str__(self) -> str:
+        """
+        String representation of the Message Pact Handle.
+        """
+        return f"MessagePactHandle({self._ref})"
+
+    def __repr__(self) -> str:
+        """
+        String representation of the Message Pact Handle.
+        """
+        return f"MessagePactHandle({self._ref!r})"
+
 
 
 class MessagePactMessageIterator: ...
@@ -307,7 +498,35 @@ def port(self) -> int:
         return self._ref
 
 
-class PactInteraction: ...
+class PactInteraction:
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new Pact Interaction.
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct PactInteraction *":
+            msg = (
+                "ptr must be a struct PactInteraction, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "PactInteraction"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"PactInteraction({self._ptr!r})"
+
 
 
 class PactInteractionIterator:
@@ -352,6 +571,9 @@ def __del__(self) -> None:
         """
         pact_interaction_iter_delete(self)
 
+    def __iter__(self) -> PactInteractionIterator:
+        return self
+
     def __next__(self) -> PactInteraction:
         """
         Get the next interaction from the iterator.
@@ -517,19 +739,228 @@ def __next__(self) -> SynchronousMessage:
 class Provider: ...
 
 
-class ProviderState: ...
+class ProviderState:
+
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new ProviderState
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct ProviderState *":
+            msg = (
+                "ptr must be a struct ProviderState, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "ProviderState"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"ProviderState({self._ptr!r})"
+
+    @property
+    def name(self) -> str:
+        return ffi.string(lib.pactffi_provider_state_get_name(self._ptr)).decode('utf-8')
+
+    @property
+    def parameters(self) -> list[ProviderStateParamPair]:
+        return [{p.key: p.value} for p in provider_state_get_param_iter(self)]
+
+
+class ProviderStateIterator:
+    """
+    Iterator over an interactions ProviderStates
+    """
+
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new Provider State Iterator
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct ProviderStateIterator *":
+            msg = (
+                "ptr must be a struct ProviderStateIterator, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "ProviderStateIterator"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"ProviderStateIterator({self._ptr!r})"
+
+    def __del__(self) -> None:
+        """
+        Destructor for the Provider State Iterator.
+        """
+        provider_state_iter_delete(self)
+
+    def __iter__(self) -> ProviderStateIterator:
+        """
+        Return the iterator itself.
+        """
+        return self
+
+    def __next__(self) -> ProviderStateParamPair:
+        """
+        Get the next message from the iterator.
+        """
+        return provider_state_iter_next(self)
+
+
+class ProviderStateParamIterator:
+    """
+    Iterator over a Provider States Parameters
+    """
+
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new Provider State Param Iterator
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct ProviderStateParamIterator *":
+            msg = (
+                "ptr must be a struct ProviderStateParamIterator, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "ProviderStateParamIterator"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"ProviderStateParamIterator({self._ptr!r})"
+
+    def __del__(self) -> None:
+        """
+        Destructor for the Provider State Param Iterator.
+        """
+        provider_state_param_iter_delete(self)
+
+    def __iter__(self) -> ProviderStateParamIterator:
+        """
+        Return the iterator itself.
+        """
+        return self
+
+    def __next__(self) -> ProviderStateParam:
+        """
+        Get the next message from the iterator.
+        """
+        return provider_state_param_iter_next(self)
+
+
+class ProviderStateParamPair:
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new ProviderStateParamPair.
+
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct ProviderStateParamPair *":
+            msg = (
+                "ptr must be a struct ProviderStateParamPair, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
+
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "ProviderStateParamPair"
+
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"ProviderStateParamPair({self._ptr!r})"
+
+    @property
+    def key(self) -> str:
+        return ffi.string(self._ptr.key).decode('utf-8')
+
+    @property
+    def value(self) -> str:
+        return ffi.string(self._ptr.value).decode('utf-8')
 
 
-class ProviderStateIterator: ...
+class SynchronousHttp:
+    def __init__(self, ptr: cffi.FFI.CData) -> None:
+        """
+        Initialise a new SynchronousHttp.
 
+        Args:
+            ptr:
+                CFFI data structure.
+        """
+        if ffi.typeof(ptr).cname != "struct SynchronousHttp *":
+            msg = (
+                "ptr must be a struct SynchronousHttp, got"
+                f" {ffi.typeof(ptr).cname}"
+            )
+            raise TypeError(msg)
+        self._ptr = ptr
 
-class ProviderStateParamIterator: ...
+    def __str__(self) -> str:
+        """
+        Nice string representation.
+        """
+        return "SynchronousHttp"
 
+    def __repr__(self) -> str:
+        """
+        Debugging representation.
+        """
+        return f"SynchronousHttp({self._ptr!r})"
 
-class ProviderStateParamPair: ...
+    @property
+    def request_contents(self) -> str:
+        contents = lib.pactffi_sync_http_get_request_contents(self._ptr)
+        if contents != ffi.NULL:
+            return ffi.string(contents).decode('utf-8')
 
+    @property
+    def response_contents(self) -> str:
+        contents = lib.pactffi_sync_http_get_response_contents(self._ptr)
+        if contents != ffi.NULL:
+            return ffi.string(contents).decode('utf-8')
 
-class SynchronousHttp: ...
 
 
 class SynchronousMessage: ...
@@ -646,7 +1077,7 @@ def __repr__(self) -> str:
         """
         Information-rich string representation of the Interaction Part.
         """
-        return f"InteractionPath.{self.name}"
+        return f"InteractionPart.{self.name}"
 
 
 class LevelFilter(Enum):
@@ -952,7 +1383,7 @@ def init_with_log_level(level: str = "INFO") -> None:
 
     Exported functions are inherently unsafe.
     """
-    raise NotImplementedError
+    lib.pactffi_init_with_log_level(level.encode("utf-8"))
 
 
 def enable_ansi_support() -> None:
@@ -1373,7 +1804,7 @@ def pact_model_interaction_iterator(pact: Pact) -> PactInteractionIterator:
 
     # Errors On any error, this function will return a NULL pointer.
     """
-    raise NotImplementedError
+    return PactInteractionIterator(lib.pactffi_pact_model_interaction_iterator(pact))
 
 
 def pact_spec_version(pact: Pact) -> PactSpecification:
@@ -2549,7 +2980,7 @@ def sync_http_get_request(interaction: SynchronousHttp) -> HttpRequest:
 
     If the interaction is NULL, returns NULL.
     """
-    raise NotImplementedError
+    return lib.pactffi_sync_http_get_request(interaction)
 
 
 def sync_http_get_request_contents(interaction: SynchronousHttp) -> str:
@@ -3033,7 +3464,6 @@ def pact_message_iter_next(iter: PactMessageIterator) -> Message:
     ptr = lib.pactffi_pact_message_iter_next(iter._ptr)
     if ptr == ffi.NULL:
         raise StopIteration
-    raise NotImplementedError
     return Message(ptr)
 
 
@@ -3071,7 +3501,6 @@ def pact_sync_http_iter_next(iter: PactSyncHttpIterator) -> SynchronousHttp:
     ptr = lib.pactffi_pact_sync_http_iter_next(iter._ptr)
     if ptr == ffi.NULL:
         raise StopIteration
-    raise NotImplementedError
     return SynchronousHttp(ptr)
 
 
@@ -3095,7 +3524,6 @@ def pact_interaction_iter_next(iter: PactInteractionIterator) -> PactInteraction
     ptr = lib.pactffi_pact_interaction_iter_next(iter._ptr)
     if ptr == ffi.NULL:
         raise StopIteration
-    raise NotImplementedError
     return PactInteraction(ptr)
 
 
@@ -3207,7 +3635,11 @@ def message_new_from_json(
 
     If the JSON string is invalid or not UTF-8 encoded, returns a NULL.
     """
-    raise NotImplementedError
+    return lib.pactffi_message_new_from_json(
+        ffi.new('unsigned int *', index)[0],
+        ffi.new('char[]', json_str.encode("utf-8")),
+        spec_version,
+    )
 
 
 def message_new_from_body(body: str, content_type: str) -> Message:
@@ -3436,7 +3868,7 @@ def message_get_provider_state_iter(message: Message) -> ProviderStateIterator:
 
     Returns NULL if an error occurs.
     """
-    raise NotImplementedError
+    return ProviderStateIterator(lib.pactffi_message_get_provider_state_iter(message._ptr))
 
 
 def provider_state_iter_next(iter: ProviderStateIterator) -> ProviderState:
@@ -3457,7 +3889,10 @@ def provider_state_iter_next(iter: ProviderStateIterator) -> ProviderState:
 
     Returns NULL if an error occurs.
     """
-    raise NotImplementedError
+    provider_state = lib.pactffi_provider_state_iter_next(iter._ptr)
+    if provider_state == ffi.NULL:
+        raise StopIteration
+    return ProviderState(provider_state)
 
 
 def provider_state_iter_delete(iter: ProviderStateIterator) -> None:
@@ -3493,7 +3928,7 @@ def message_find_metadata(message: Message, key: str) -> str:
     This function may fail if the provided `key` string contains invalid UTF-8,
     or if the Rust string contains embedded null ('\0') bytes.
     """
-    raise NotImplementedError
+    return ffi.string(lib.pactffi_message_find_metadata(message._ptr, key.encode('utf-8')))
 
 
 def message_insert_metadata(message: Message, key: str, value: str) -> int:
@@ -3536,7 +3971,10 @@ def message_metadata_iter_next(iter: MessageMetadataIterator) -> MessageMetadata
 
     If no further data is present, returns NULL.
     """
-    raise NotImplementedError
+    message_metadata = lib.pactffi_message_metadata_iter_next(iter._ptr)
+    if message_metadata == ffi.NULL:
+        raise StopIteration
+    return message_metadata
 
 
 def message_get_metadata_iter(message: Message) -> MessageMetadataIterator:
@@ -3561,7 +3999,7 @@ def message_get_metadata_iter(message: Message) -> MessageMetadataIterator:
     This function may fail if any of the Rust strings contain embedded null
     ('\0') bytes.
     """
-    raise NotImplementedError
+    return MessageMetadataIterator(lib.pactffi_message_get_metadata_iter(message._ptr))
 
 
 def message_metadata_iter_delete(iter: MessageMetadataIterator) -> None:
@@ -3571,7 +4009,7 @@ def message_metadata_iter_delete(iter: MessageMetadataIterator) -> None:
     [Rust
     `pactffi_message_metadata_iter_delete`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_message_metadata_iter_delete)
     """
-    raise NotImplementedError
+    lib.pactffi_message_metadata_iter_delete(iter._ptr)
 
 
 def message_metadata_pair_delete(pair: MessageMetadataPair) -> None:
@@ -3922,7 +4360,7 @@ def provider_state_get_param_iter(
     This function may fail if any of the Rust strings contain embedded null
     ('\0') bytes.
     """
-    raise NotImplementedError
+    return ProviderStateParamIterator(lib.pactffi_provider_state_get_param_iter(provider_state._ptr))
 
 
 def provider_state_param_iter_next(
@@ -3947,7 +4385,10 @@ def provider_state_param_iter_next(
 
     Returns NULL if there's no further elements or the iterator is NULL.
     """
-    raise NotImplementedError
+    provider_state_param = lib.pactffi_provider_state_param_iter_next(iter._ptr)
+    if provider_state_param == ffi.NULL:
+        raise StopIteration
+    return ProviderStateParamPair(provider_state_param)
 
 
 def provider_state_delete(provider_state: ProviderState) -> None:
@@ -3967,7 +4408,7 @@ def provider_state_param_iter_delete(iter: ProviderStateParamIterator) -> None:
     [Rust
     `pactffi_provider_state_param_iter_delete`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_provider_state_param_iter_delete)
     """
-    raise NotImplementedError
+    lib.pactffi_provider_state_param_iter_delete(iter._ptr)
 
 
 def provider_state_param_pair_delete(pair: ProviderStateParamPair) -> None:
@@ -4710,16 +5151,11 @@ def write_pact_file(
     if ret == 0:
         return
     if ret == 1:
-        msg = (
-            f"The function panicked while writing the Pact for {mock_server_handle} in"
-            f" {directory}."
-        )
-    elif ret == 2:  # noqa: PLR2004
         msg = (
             f"The Pact file for {mock_server_handle} could not be written in"
             f" {directory}."
         )
-    elif ret == 3:  # noqa: PLR2004
+    elif ret == 2:  # noqa: PLR2004
         msg = f"The Pact for the {mock_server_handle} was not found."
     else:
         msg = (
@@ -4863,7 +5299,7 @@ def pact_handle_to_pointer(pact: PactHandle) -> Pact:
     The returned Pact model must be freed with the `pactffi_pact_model_delete`
     function when no longer needed.
     """
-    raise NotImplementedError
+    return lib.pactffi_pact_handle_to_pointer(pact._ref)
 
 
 def new_interaction(pact: PactHandle, description: str) -> InteractionHandle:
@@ -6052,7 +6488,12 @@ def new_message_pact(consumer_name: str, provider_name: str) -> MessagePactHandl
     Returns a new `MessagePactHandle`. The handle will need to be freed with the
     `pactffi_free_message_pact_handle` function to release its resources.
     """
-    raise NotImplementedError
+    return MessagePactHandle(
+        lib.pactffi_new_message_pact(
+            consumer_name.encode("utf-8"),
+            provider_name.encode("utf-8"),
+        ),
+    )
 
 
 def new_message(pact: MessagePactHandle, description: str) -> MessageHandle:
@@ -6067,7 +6508,12 @@ def new_message(pact: MessagePactHandle, description: str) -> MessageHandle:
 
     Returns a new `MessageHandle`.
     """
-    raise NotImplementedError
+    return MessageHandle(
+        lib.pactffi_new_message(
+            pact._ref,
+            description.encode("utf-8"),
+        ),
+    )
 
 
 def message_expects_to_receive(message: MessageHandle, description: str) -> None:
@@ -6085,15 +6531,24 @@ def message_expects_to_receive(message: MessageHandle, description: str) -> None
 
 def message_given(message: MessageHandle, description: str) -> None:
     """
-    Adds a provider state to the Interaction.
+    Adds a provider state to the Message
 
     [Rust
-    `pactffi_message_given`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_message_given)
+    `pactffi_given`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_given)
+
+    Args:
+        message:
+            Handle to the Message.
+
+        description:
+            The provider state description. It needs to be unique.
 
-    * `description` - The provider state description. It needs to be unique for
-      each message
+    Raises:
+        RuntimeError: If the provider state could not be specified.
     """
-    raise NotImplementedError
+    # message_given does not return anything,
+    # so we can't check for errors
+    lib.pactffi_message_given(message._ref, description.encode("utf-8"))
 
 
 def message_given_with_param(
@@ -6103,22 +6558,57 @@ def message_given_with_param(
     value: str,
 ) -> None:
     """
-    Adds a provider state to the Message with a parameter key and value.
+    Adds a parameter key and value to a provider state to the Message.
+
+    If the provider state does not exist, a new one will be created, otherwise
+    the parameter will be merged into the existing one. The parameter value will
+    be parsed as JSON.
 
     [Rust
-    `pactffi_message_given_with_param`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_message_given_with_param)
+    `pactffi_given_with_param`](https://docs.rs/pact_ffi/0.4.19/pact_ffi/?search=pactffi_message_given_with_param)
+
+    Args:
+        message:
+            Handle to the Message.
+
+        description:
+            The provider state description.
+
+        name:
+            Parameter name.
+
+        value:
+            Parameter value as JSON.
+
+    Raises:
+        RuntimeError: If the interaction state could not be updated.
+
+    # Errors
+
+    Returns EXIT_FAILURE (1) if the interaction or Pact can't be modified (i.e.
+    the mock server for it has already started).
+
+    Returns 2 and sets the error message (which can be retrieved with
+    `pactffi_get_error_message`) if the parameter values con't be parsed as
+    JSON.
+
+    Returns 3 if any of the C strings are not valid.
 
-    * `description` - The provider state description. It needs to be unique.
-    * `name` - Parameter name.
-    * `value` - Parameter value.
     """
-    raise NotImplementedError
+    # message_given_with_param does not return anything,
+    # so we can't check for errors
+    lib.pactffi_message_given_with_param(
+        message._ref,
+        description.encode("utf-8"),
+        name.encode("utf-8"),
+        value.encode("utf-8"),
+    )
 
 
 def message_with_contents(
     message_handle: MessageHandle,
     content_type: str,
-    body: List[int],
+    body: str | bytes,
     size: int,
 ) -> None:
     """
@@ -6144,7 +6634,14 @@ def message_with_contents(
     * `size` - number of bytes in the message body to read. This is not required
       for text bodies (JSON, XML, etc.).
     """
-    raise NotImplementedError
+    if isinstance(body, str):
+        body = body.encode("utf-8")
+    lib.pactffi_message_with_contents(
+        message_handle._ref,
+        content_type.encode("utf-8"),
+        body,
+        size
+    )
 
 
 def message_with_metadata(message_handle: MessageHandle, key: str, value: str) -> None:
@@ -6201,7 +6698,11 @@ def message_with_metadata_v2(
 
     See [IntegrationJson.md](https://github.com/pact-foundation/pact-reference/blob/libpact_ffi-v0.4.19/rust/pact_ffi/IntegrationJson.md).
     """
-    raise NotImplementedError
+    lib.pactffi_message_with_metadata_v2(
+        message_handle._ref,
+        key.encode("utf-8"),
+        value.encode("utf-8"),
+    )
 
 
 def message_reify(message_handle: MessageHandle) -> OwnedString:
@@ -6221,7 +6722,7 @@ def message_reify(message_handle: MessageHandle) -> OwnedString:
     from a Rust function that has a Tokio runtime in its call stack can result
     in a deadlock.
     """
-    raise NotImplementedError
+    return OwnedString(lib.pactffi_message_reify(message_handle._ref))
 
 
 def write_message_pact_file(
@@ -6256,6 +6757,31 @@ def write_message_pact_file(
     | 1 | The pact file was not able to be written |
     | 2 | The message pact for the given handle was not found |
     """
+    ret: int = lib.pactffi_write_message_pact_file(
+        pact._ref,
+        str(directory).encode("utf-8"),
+        overwrite,
+    )
+    if ret == 0:
+        return
+    if ret == 1:
+        msg = (
+            f"The function panicked while writing the Pact for {mock_server_handle} in"
+            f" {directory}."
+        )
+    elif ret == 2:  # noqa: PLR2004
+        msg = (
+            f"The Pact file for {mock_server_handle} could not be written in"
+            f" {directory}."
+        )
+    elif ret == 3:  # noqa: PLR2004
+        msg = f"The Pact for the {mock_server_handle} was not found."
+    else:
+        msg = (
+            "An unknown error occurred while writing the Pact for"
+            f" {mock_server_handle} in {directory}."
+        )
+    raise RuntimeError(msg)
     raise NotImplementedError
 
 
@@ -6362,7 +6888,14 @@ def free_message_pact_handle(pact: MessagePactHandle) -> int:
       that it was previously deleted.
 
     """
-    raise NotImplementedError
+    ret: int = lib.pactffi_free_message_pact_handle(pact._ref)
+    if ret == 0:
+        return
+    if ret == 1:
+        msg = f"{pact} is not valid or does not refer to a valid Pact."
+    else:
+        msg = f"There was an unknown error freeing {pact}."
+    raise RuntimeError(msg)
 
 
 def verify(args: str) -> int:
diff --git a/src/pact/v3/interaction/_async_message_interaction.py b/src/pact/v3/interaction/_async_message_interaction.py
index 011853af3..fc7702381 100644
--- a/src/pact/v3/interaction/_async_message_interaction.py
+++ b/src/pact/v3/interaction/_async_message_interaction.py
@@ -3,6 +3,8 @@
 """
 
 from __future__ import annotations
+from typing import Callable
+import json
 
 import pact.v3.ffi
 from pact.v3.interaction._base import Interaction
@@ -22,7 +24,11 @@ class AsyncMessageInteraction(Interaction):
         This class is not yet fully implemented and is not yet usable.
     """
 
-    def __init__(self, pact_handle: pact.v3.ffi.PactHandle, description: str) -> None:
+    def __init__(
+        self,
+        pact_handle: pact.v3.ffi.PactHandle,
+        description: str
+    ) -> None:
         """
         Initialise a new Asynchronous Message Interaction.
 
@@ -55,3 +61,43 @@ def _handle(self) -> pact.v3.ffi.InteractionHandle:
     @property
     def _interaction_part(self) -> pact.v3.ffi.InteractionPart:
         return pact.v3.ffi.InteractionPart.REQUEST
+
+    def with_content(self, content: dict[str, Any] | str, content_type='application/json') -> Self:
+        """
+        Set the content of the message.
+
+        Args:
+            content:
+                The content of the message, as a dictionary.
+
+        Returns:
+            The current instance of the interaction.
+        """
+        if isinstance(content, dict):
+            content = json.dumps(content)
+
+        pact.v3.ffi.message_with_contents(
+            self._handle,
+            content_type,
+            content,
+            len(content)
+        )
+        return self
+
+    def with_metadata(self, metadata: dict[str, Any]) -> Self:
+        """
+        Set the metadata of the message.
+
+        Args:
+            metadata:
+                The metadata of the message, as a dictionary.
+
+        Returns:
+            The current instance of the interaction.
+        """
+        [
+            pact.v3.ffi.message_with_metadata_v2(self._handle, k, v)
+            for k, v in metadata.items()
+        ]
+        return self
+
diff --git a/src/pact/v3/pact.py b/src/pact/v3/pact.py
index 231c01dd7..451a64143 100644
--- a/src/pact/v3/pact.py
+++ b/src/pact/v3/pact.py
@@ -87,7 +87,7 @@
 logger = logging.getLogger(__name__)
 
 
-class Pact:
+class BasePact:
     """
     A Pact between a consumer and a provider.
 
@@ -279,73 +279,14 @@ def upon_receiving(
         """
         if interaction == "HTTP":
             return HttpInteraction(self._handle, description)
-        if interaction == "Async":
+        elif interaction == "Async":
             return AsyncMessageInteraction(self._handle, description)
-        if interaction == "Sync":
+        elif interaction == "Sync":
             return SyncMessageInteraction(self._handle, description)
 
         msg = f"Invalid interaction type: {interaction}"
         raise ValueError(msg)
 
-    def serve(  # noqa: PLR0913
-        self,
-        addr: str = "localhost",
-        port: int = 0,
-        transport: str = "http",
-        transport_config: str | None = None,
-        *,
-        raises: bool = True,
-        verbose: bool = True,
-    ) -> PactServer:
-        """
-        Return a mock server for the Pact.
-
-        This function configures a mock server for the Pact. The mock server
-        is then started when the Pact is entered into a `with` block:
-
-        ```python
-        pact = Pact("consumer", "provider")
-        with pact.serve() as srv:
-            ...
-        ```
-
-        Args:
-            addr:
-                Address to bind the mock server to. Defaults to `localhost`.
-
-            port:
-                Port to bind the mock server to. Defaults to `0`, which will
-                select a random port.
-
-            transport:
-                Transport to use for the mock server. Defaults to `HTTP`.
-
-            transport_config:
-                Configuration for the transport. This is specific to the
-                transport being used and should be a JSON string.
-
-            raises:
-                Whether to raise an exception if there are mismatches between
-                the Pact and the server. If set to `False`, then the mismatches
-                must be handled manually.
-
-            verbose:
-                Whether or not to print the mismatches to the logger. This works
-                independently of `raises`.
-
-        Returns:
-            A [`PactServer`][pact.v3.pact.PactServer] instance.
-        """
-        return PactServer(
-            self._handle,
-            addr,
-            port,
-            transport,
-            transport_config,
-            raises=raises,
-            verbose=verbose,
-        )
-
     def messages(self) -> pact.v3.ffi.PactMessageIterator:
         """
         Iterate over the messages in the Pact.
@@ -443,6 +384,109 @@ def write_file(
         )
 
 
+class Pact(BasePact):
+
+    def serve(  # noqa: PLR0913
+        self,
+        addr: str = "localhost",
+        port: int = 0,
+        transport: str = "http",
+        transport_config: str | None = None,
+        *,
+        raises: bool = True,
+        verbose: bool = True,
+    ) -> PactServer:
+        """
+        Return a mock server for the Pact.
+
+        This function configures a mock server for the Pact. The mock server
+        is then started when the Pact is entered into a `with` block:
+
+        ```python
+        pact = Pact("consumer", "provider")
+        with pact.serve() as srv:
+            ...
+        ```
+
+        Args:
+            addr:
+                Address to bind the mock server to. Defaults to `localhost`.
+
+            port:
+                Port to bind the mock server to. Defaults to `0`, which will
+                select a random port.
+
+            transport:
+                Transport to use for the mock server. Defaults to `HTTP`.
+
+            transport_config:
+                Configuration for the transport. This is specific to the
+                transport being used and should be a JSON string.
+
+            raises:
+                Whether to raise an exception if there are mismatches between
+                the Pact and the server. If set to `False`, then the mismatches
+                must be handled manually.
+
+            verbose:
+                Whether or not to print the mismatches to the logger. This works
+                independently of `raises`.
+
+        Returns:
+            A [`PactServer`][pact.v3.pact.PactServer] instance.
+        """
+        return PactServer(
+            self._handle,
+            addr,
+            port,
+            transport,
+            transport_config,
+            raises=raises,
+            verbose=verbose,
+        )
+
+
+
+class MessagePact(BasePact):
+    def get_provider_states(self):
+        """
+        Get the provider states for the interaction.
+
+        Returns:
+            A list of provider states for the interaction.
+        """
+        provider_state_data = []
+        for message in pact.v3.ffi.pact_handle_get_message_iter(self._handle):
+            for provider_state in pact.v3.ffi.message_get_provider_state_iter(message):
+                provider_state_data.append({
+                    'name': provider_state.name,
+                    'params': provider_state.parameters
+                })
+        return provider_state_data
+
+    def verify(self, handler) -> list[dict[str, Any]]:
+        processed_messages = []
+        _mutable_pact = pact.v3.ffi.pact_handle_to_pointer(self._handle)
+        for interaction in pact.v3.ffi.pact_model_interaction_iterator(_mutable_pact):
+            msg_iter = pact.v3.ffi.pact_handle_get_message_iter(self._handle)
+            for msg in msg_iter:
+                processed_messages.append({
+                    "description": msg.description,
+                    "contents": msg.contents,
+                    "metadata": msg.metadata,
+                })
+                try:
+                    async_message = context = {}
+                    if msg.contents is not None:
+                        async_message = json.loads(msg.contents)
+                    if msg.metadata is not None:
+                        context = msg.metadata
+                    handler(async_message, context)
+                except Exception as e:
+                    raise e
+        return processed_messages
+
+
 class MismatchesError(Exception):
     """
     Exception raised when there are mismatches between the Pact and the server.
@@ -754,3 +798,4 @@ def write_file(
             str(directory),
             overwrite=overwrite,
         )
+
diff --git a/tests/v3/compatibility_suite/test_v3_message_consumer.py b/tests/v3/compatibility_suite/test_v3_message_consumer.py
new file mode 100644
index 000000000..08faa4ca6
--- /dev/null
+++ b/tests/v3/compatibility_suite/test_v3_message_consumer.py
@@ -0,0 +1,620 @@
+"""V3 Message consumer feature tests."""
+from __future__ import annotations
+
+import ast
+import json
+import logging
+import re
+from pathlib import Path
+from typing import Any, Generator, NamedTuple
+
+from jsonpath_ng import parse
+from pytest_bdd import (
+    given,
+    parsers,
+    scenario,
+    then,
+    when,
+)
+
+from pact.v3.pact import AsyncMessageInteraction
+from pact.v3.pact import MessagePact as Pact
+from tests.v3.compatibility_suite.util import FIXTURES_ROOT, parse_markdown_table
+
+logger = logging.getLogger(__name__)
+
+class PactInteraction(NamedTuple):
+    """Holder class for Pact and Interaction."""
+    pact: Pact
+    interaction: AsyncMessageInteraction
+
+
+class PactResult(NamedTuple):
+    """Holder class for Pact Result objects."""
+    received_payload: dict[str, Any]
+    pact_data: dict[str, Any] | None
+    error: str | None
+
+class ReceivedPayload(NamedTuple):
+    """Holder class for Message Received Payload."""
+    message: any
+    context: any
+
+class UnknownTypeError(Exception):
+    """Unknown type error."""
+    def __init__(self, expected_type: str) -> None:
+        """Initialize the UnknownTypeError."""
+        super().__init__(f"Unknown type: {expected_type}")
+
+class UnknownGeneratorCategoryError(Exception):
+    """Unknown type error."""
+    def __init__(self, generator_category: str) -> None:
+        """Initialize the UnknownGeneratorCategoryError."""
+        super().__init__(f"Unknown generator category: {generator_category}")
+
+class TestFailedError(Exception):
+    """Test failed error."""
+    def __init__(self) -> None:
+        """Initialize the TestFailedError."""
+        super().__init__("Test failed")
+
+NUM_RE = re.compile(r"^-?[.0-9]+$")
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "Supports arbitrary message metadata"
+)
+def test_supports_arbitrary_message_metadata() -> None:
+    """Supports arbitrary message metadata."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "Supports data for provider states"
+)
+def test_supports_data_for_provider_states() -> None:
+    """Supports data for provider states."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "Supports specifying provider states"
+)
+def test_supports_specifying_provider_states() -> None:
+    """Supports specifying provider states."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "Supports the use of generators with message metadata"
+)
+def test_supports_the_use_of_generators_with_message_metadata() -> None:
+    """Supports the use of generators with message metadata."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "Supports the use of generators with the message body"
+)
+def test_supports_the_use_of_generators_with_the_message_body() -> None:
+    """Supports the use of generators with the message body."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "When all messages are successfully processed"
+)
+def test_when_all_messages_are_successfully_processed() -> None:
+    """When all messages are successfully processed."""
+
+
+@scenario(
+    "definition/features/V3/message_consumer.feature",
+    "When not all messages are successfully processed"
+)
+def test_when_not_all_messages_are_successfully_processed() -> None:
+    """When not all messages are successfully processed."""
+
+
+################################################################################
+## Given
+################################################################################
+
+
+@given(
+    "a message integration is being defined for a consumer test",
+    target_fixture="pact_interaction"
+)
+def a_message_integration_is_being_defined_for_a_consumer_test() -> (
+    Generator[tuple[Pact, AsyncMessageInteraction], Any, None]
+):
+    """A message integration is being defined for a consumer test."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V3")
+    yield PactInteraction(pact, pact.upon_receiving("a request", "Async"))
+
+
+@given("a message is defined")
+def _a_message_is_defined() -> None:
+    """A message is defined."""
+
+
+@given(
+    parsers.re(
+        r'a provider state "(?P<state>[^"]+)" for the message '
+        r'is specified with the following data:\n(?P<table>.+)',
+        re.DOTALL,
+    ),
+    converters={"table": parse_markdown_table},
+)
+def a_provider_state_for_the_message_is_specified_with_the_following_data(
+    pact_interaction: PactInteraction,
+    state: str,
+    table: list[dict[str, Any]]
+) -> None:
+    """A provider state for the message is specified with the following data."""
+    for parameters in table:
+        state_params = { k: ast.literal_eval(v) for k, v in parameters.items() }
+        pact_interaction.interaction.given(state, parameters=state_params)
+
+
+@given(
+    parsers.re(r'a provider state "(?P<state>[^"]+)" for the message is specified')
+)
+def a_provider_state_for_the_message_is_specified(
+    pact_interaction: PactInteraction,
+    state: str,
+) -> None:
+    """A provider state for the message is specified."""
+    pact_interaction.interaction.given(state)
+
+
+@given(
+    parsers.re(
+        "the message contains the following "
+        "metadata:\n(?P<table>.+)", re.DOTALL
+    ),
+    converters={"table": parse_markdown_table},
+)
+def the_message_contains_the_following_metadata(
+    pact_interaction: PactInteraction,
+    table: list[dict[str, Any]]
+) -> None:
+    """The message contains the following metadata."""
+    for metadata in table:
+        if metadata.get("value","").startswith("JSON: "):
+            metadata["value"] = metadata["value"].replace("JSON:", "")
+        pact_interaction.interaction.with_metadata({metadata["key"]: metadata["value"]})
+
+
+@given(
+    parsers.re(
+        "the message is configured with the following:\n"
+        "(?P<table>.+)", re.DOTALL
+    ),
+    converters={"table": parse_markdown_table},
+)
+def the_message_is_configured_with_the_following(
+    pact_interaction: PactInteraction,
+    table: list[dict[str, Any]],
+) -> None:
+    """The message is configured with the following."""
+    body_json, generator_json, metadata_json = _build_message_data(table)
+    if generator_json:
+        category = next(iter(generator_json.keys()))
+        if category == "body":
+            _build_body_generator(generator_json, body_json)
+        elif category == "metadata":
+            _build_metadata_generator(generator_json, metadata_json)
+        else:
+            raise UnknownGeneratorCategoryError(category)
+    pact_interaction.interaction.with_content(body_json)
+    for k, v in metadata_json.items():
+        v_str = v
+        if isinstance(v, dict):
+            v_str = json.dumps(v)
+        pact_interaction.interaction.with_metadata({k: str(v_str)})
+
+
+@given(
+    parsers.re('the message payload contains the "(?P<json_doc>[^"]+)" JSON document')
+)
+def the_message_payload_contains_the_basic_json_document(
+    pact_interaction: PactInteraction,
+    json_doc: str
+) -> None:
+    """The message payload contains the "basic" JSON document."""
+    pact_interaction.interaction.with_content(read_json(f"{json_doc}.json"))
+
+
+################################################################################
+## When
+################################################################################
+
+
+@when(
+    'the message is NOT successfully processed with a "Test failed" exception',
+    target_fixture="pact_result"
+)
+def the_message_is_not_successfully_processed_with_an_exception(
+    pact_interaction: PactInteraction
+) -> None:
+    """The message is NOT successfully processed with a "Test failed" exception."""
+    # using a dict here because it's mutable
+    received_payload = {"data": None}
+    def fail(
+        async_message: str | dict[any: any], context: dict[any: any]
+    ) -> None:
+        received_payload["data"] = ReceivedPayload(async_message, context)
+        raise TestFailedError
+    try:
+        pact_interaction.pact.verify(fail)
+    except Exception as e: # noqa: BLE001
+        return PactResult(received_payload["data"], None, e)
+
+
+@when(
+    "the message is successfully processed",
+    target_fixture="pact_result"
+)
+def the_message_is_successfully_processed(
+    pact_interaction: PactInteraction,
+    temp_dir: Path
+) -> None:
+    """The message is successfully processed."""
+    received_payload = {"data": None}
+    def handler(
+        async_message: str | dict[Any, Any],
+        context: dict[Any, Any],
+    ) -> None:
+        received_payload["data"] = ReceivedPayload(async_message, context)
+    pact_interaction.pact.verify(handler)
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact_interaction.pact.write_file(temp_dir / "pacts")
+    with (
+        temp_dir / "pacts" / "consumer-provider.json"
+    ).open() as file:
+        yield PactResult(received_payload["data"], json.load(file), None)
+
+
+################################################################################
+## Then
+################################################################################
+
+
+@then("a Pact file for the message interaction will NOT have been written")
+def a_pact_file_for_the_message_interaction_will_not_have_been_written(
+    temp_dir: Path
+) -> None:
+    """A Pact file for the message interaction will NOT have been written."""
+    assert not Path(
+        temp_dir / "pacts" / "consumer-provider.json"
+    ).exists()
+
+
+@then("a Pact file for the message interaction will have been written")
+def a_pact_file_for_the_message_interaction_will_have_been_written(
+    temp_dir: Path
+) -> None:
+    """A Pact file for the message interaction will have been written."""
+    assert Path(
+        temp_dir / "pacts" / "consumer-provider.json"
+    ).exists()
+
+
+@then(
+    parsers.re(r'the consumer test error will be "(?P<error>[^"]+)"')
+)
+def the_consumer_test_error_will_be_test_failed(
+    pact_result: PactResult,
+    error: str,
+) -> None:
+    """The consumer test error will be "Test failed"."""
+    assert str(pact_result.error) == error
+
+
+@then("the consumer test will have failed")
+def the_consumer_test_will_have_failed(
+    pact_result: PactResult
+) -> None:
+    """The consumer test will have failed."""
+    assert type(pact_result.error) == TestFailedError
+    assert pact_result.pact_data is None
+
+
+@then("the consumer test will have passed")
+def the_consumer_test_will_have_passed(
+    pact_result: PactResult
+) -> None:
+    """The consumer test will have passed."""
+    assert pact_result.error is None
+    assert pact_result.pact_data is not None
+
+
+@then(
+    parsers.re(
+        r'the first message in the Pact file will contain '
+        'provider state "(?P<state>[^"]+)"'
+    )
+)
+def the_first_message_in_the_pact_file_will_contain_provider_state(
+    pact_result: PactResult,
+    state: str,
+) -> None:
+    """The first message in the Pact file will contain provider state."""
+    assert state in [
+        provider_state["name"]
+        for provider_state in pact_result.pact_data["messages"][0]["providerStates"]
+    ]
+
+
+@then(
+    parsers.re(
+        r'the first message in the pact file content type '
+        'will be "(?P<content_type>[^"]+)"'
+    )
+)
+def the_first_message_in_the_pact_file_content_type_will_be(
+    pact_result: PactResult,
+    content_type: str,
+) -> None:
+    """The first message in the pact file content type will be "application/json"."""
+    messages: list[dict[str, Any]] = pact_result.pact_data["messages"]
+    assert messages[0]["metadata"]["contentType"] == content_type
+
+
+@then(
+    parsers.re(
+        r"the first message in the pact file will contain "
+        "(?P<state_count>[0-9]+) provider states?"
+    ),
+    converters={"state_count": int},
+)
+def the_first_message_in_the_pact_file_will_contain(
+    pact_result: PactResult,
+    state_count: int,
+) -> None:
+    """The first message in the pact file will contain 1 provider state."""
+    assert len(pact_result.pact_data["messages"][0]["providerStates"]) == state_count
+
+
+@then(
+    parsers.re(
+        'the first message in the pact file will contain '
+        'the "(?P<json_doc>[^"]+)" document'
+    )
+)
+def the_first_message_in_the_pact_file_will_contain_the_basic_json_document(
+    pact_result: PactResult,
+    json_doc: str
+) -> None:
+    """The first message in the pact file will contain the "basic.json" document."""
+    assert pact_result.pact_data["messages"][0]["contents"] == read_json(json_doc)
+
+
+@then(
+    parsers.re(
+        r'the first message in the pact file will contain '
+        r'the message metadata "(?P<key>[^"]+)" == "(?P<value>[^"\\]*(?:\\.[^"\\]*)*)"'
+    )
+)
+def the_first_message_in_the_pact_file_will_contain_the_message_metadata(
+    pact_result: PactResult,
+    key: str,
+    value: any,
+) -> None:
+    """The first message in the pact file will contain the message metadata."""
+    if value.startswith("JSON: "):
+        value = value.replace("JSON: ", "")
+        value = value.replace('\\"', '"')
+        value = json.loads(value)
+    assert pact_result.pact_data["messages"][0]["metadata"][key] == value
+
+
+@then(
+    parsers.re(
+        r'the message contents for "(?P<replace_token>[^"]+)" '
+        'will have been replaced with an? "(?P<expected_type>[^"]+)"'
+    )
+)
+def the_message_contents_will_have_been_replaced_with(
+    pact_result: PactResult,
+    replace_token: str,
+    expected_type: str,
+) -> None:
+    """The message contents for "$.one" will have been replaced with an "integer"."""
+    path = parse(replace_token)
+    values = [match.value for match in path.find(pact_result.received_payload.message)]
+    for v in values:
+        assert compare_type(expected_type, v)
+
+
+@then(
+    parsers.parse(
+        "the pact file will contain {interaction_count:d} message interaction"
+    )
+)
+def the_pact_file_will_contain_message_interaction(
+    pact_result: PactResult,
+    interaction_count: int,
+) -> None:
+    """The pact file will contain 1 message interaction."""
+    assert len(pact_result.pact_data["messages"]) == interaction_count
+
+
+@then(
+    parsers.re(
+        r'the provider state "(?P<state>[^"]+)" for the message '
+        r'will contain the following parameters:\n(?P<parameters>.+)',
+        re.DOTALL,
+    ),
+    converters={"parameters": parse_markdown_table},
+)
+def the_provider_state_for_the_message_will_contain_the_following_parameters(
+    pact_interaction: PactInteraction,
+    state: str,
+    parameters: list[dict[str, Any]],
+) -> None:
+    """The provider state for the message will contain the following parameters."""
+    provider_state_params = None
+    expected_params = json.loads(parameters[0]["parameters"])
+    for provider_state in pact_interaction.pact.get_provider_states():
+        if provider_state["name"] == state:
+            provider_state_params = provider_state["params"]
+            break
+    # if we have provider_state_params, we found the expected provider state name
+    assert provider_state_params is not None
+    found = { k: False for k in expected_params }
+    for k, v in expected_params.items():
+        for provider_state_param in provider_state_params:
+            if provider_state_param.get(k):
+                assert ast.literal_eval(provider_state_param[k]) == v
+                found[k] = True
+                break
+    assert all(found.values())
+
+@then(
+    parsers.re(r'the received message content type will be "(?P<content_type>[^"]+)"')
+)
+def the_received_message_content_type_will_be(
+    pact_result: PactResult,
+    content_type: str,
+) -> None:
+    """The received message content type will be "application/json"."""
+    assert any(
+        context.get("contentType") == content_type
+        for context in pact_result.received_payload.context
+    )
+
+@then(
+    parsers.re(
+        r'the received message metadata will contain "(?P<key>[^"]+)" '
+        'replaced with an? "(?P<expected_type>[^"]+)"'
+    )
+)
+def the_received_message_metadata_will_contain_replaced_with(
+    pact_result: PactResult,
+    key: str,
+    expected_type: str,
+) -> None:
+    """The received message metadata will contain "ID" replaced with an "integer"."""
+    found = False
+    for metadata in pact_result.received_payload.context:
+        if metadata.get(key):
+            assert compare_type(expected_type, metadata[key])
+            found = True
+    assert found
+
+
+@then(
+    parsers.re(
+        r'the received message metadata will contain '
+        r'"(?P<key>[^"]+)" == "(?P<value>[^"\\]*(?:\\.[^"\\]*)*)"'
+    )
+)
+def the_received_message_metadata_will_contain(
+    pact_result: PactResult,
+    key: str,
+    value: any,
+) -> None:
+    """The received message metadata will contain "Origin" == "Some Text"."""
+    found = False
+    if value.startswith("JSON: "):
+        value = value.replace("JSON: ", "")
+        value = value.replace('\\"', '"')
+        value = json.loads(value)
+    for metadata in pact_result.received_payload.context:
+        if metadata.get(key):
+            if isinstance(value, dict):
+                assert json.loads(metadata[key]) == value
+            elif NUM_RE.match(metadata[key]):
+                assert ast.literal_eval(metadata[key]) == value
+            else:
+                assert metadata[key] == value
+            found = True
+    assert found
+
+
+@then(
+    parsers.re(
+        r'the received message payload will contain '
+        'the "(?P<json_doc>[^"]+)" JSON document'
+    )
+)
+def the_received_message_payload_will_contain_the_basic_json_document(
+    pact_result: PactResult,
+    json_doc: str
+) -> None:
+    """The received message payload will contain the "basic" JSON document."""
+    assert pact_result.received_payload.message == read_json(f"{json_doc}.json")
+
+
+def read_json(file: str) -> dict[str, Any]:
+    with Path(FIXTURES_ROOT / file).open() as f:
+        return json.loads(f.read())
+
+
+def compare_type(expected_type: str, t: str | int | None) -> bool:
+    if expected_type == "integer":
+        try:
+            int(t)
+        except ValueError:
+            return False
+        return True
+    raise UnknownTypeError(expected_type)
+
+
+def _build_message_data(
+    table: list[dict[str, Any]]
+) -> (dict[str, Any], dict[str, Any], dict[str, Any]):
+    body_json = generator_json = metadata_json = {}
+    for entry in table:
+        for k, v in entry.items():
+            if k == "generators":
+                if v.startswith("JSON: "):
+                    generator_json = json.loads(v.replace("JSON:", ""))
+                else:
+                    generator_json = read_json(v)
+            elif k == "body":
+                if v.startswith("file: "):
+                    file = v.replace("file: ", "")
+                    body_json = read_json(file)
+            elif k == "metadata":
+                metadata_json = json.loads(v)
+    return body_json, generator_json, metadata_json
+
+def _build_body_generator(
+    generator_json: dict[str, Any],
+    body_json: dict[str, Any]
+) -> None:
+    for k, v in generator_json["body"].items():
+        path = parse(k)
+        current_values = [match.value for match in path.find(body_json)]
+        matches = path.find(body_json)
+        for i, _ in enumerate(matches):
+            generator_type = v["type"]
+            del v["type"]
+            replacement_value = {
+                "value": current_values[i],
+                "pact:matcher:type": "notEmpty",
+                "pact:generator:type": generator_type,
+            }
+            replacement_value.update(v)
+            matches[i].full_path.update(body_json, replacement_value)
+
+def _build_metadata_generator(
+    generator_json: dict[str, Any],
+    metadata_json: dict[str, Any]
+) -> None:
+    for k in generator_json["metadata"]:
+        metadata = metadata_json[k]
+        if not isinstance(metadata, dict):
+            metadata = { "value": metadata }
+            metadata_json[k] = metadata
+        generator_data = generator_json["metadata"][k]
+        metadata.update({
+            "pact:generator:type": generator_data["type"],
+            "pact:matcher:type": "notEmpty",
+        })
+        del generator_data["type"]
+        metadata.update(generator_data)
diff --git a/tests/v3/compatibility_suite/test_v3_message_producer.py b/tests/v3/compatibility_suite/test_v3_message_producer.py
new file mode 100644
index 000000000..598752e61
--- /dev/null
+++ b/tests/v3/compatibility_suite/test_v3_message_producer.py
@@ -0,0 +1,377 @@
+"""V3 Message provider feature tests."""
+from __future__ import annotations
+
+import json
+import logging
+import pickle
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+from pytest_bdd import (
+    given,
+    parsers,
+    scenario,
+)
+
+from pact.v3.pact import Pact
+from tests.v3.compatibility_suite.util import (
+    InteractionDefinition,
+    parse_horizontal_markdown_table,
+    parse_markdown_table,
+)
+from tests.v3.compatibility_suite.util.provider import (
+    a_provider_is_started_that_can_generate_the_message,
+    the_provider_state_callback_will_be_called_after_the_verification_is_run,
+    the_provider_state_callback_will_be_called_before_the_verification_is_run,
+    the_provider_state_callback_will_receive_a_setup_call,
+    the_verification_is_run_with_start_context,
+    the_verification_results_will_contain_a_error,
+    the_verification_will_be_successful,
+)
+
+if TYPE_CHECKING:
+    from pact.v3.verifier import Verifier
+
+TEST_PACT_FILE_DIRECTORY = Path(Path(__file__).parent / "pacts")
+
+logger = logging.getLogger(__name__)
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Incorrect message is generated by the provider"
+)
+def test_incorrect_message_is_generated_by_the_provider() -> None:
+    """Incorrect message is generated by the provider."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with JSON body (negative case)"
+)
+def test_message_with_json_body_negative_case() -> None:
+    """Message with JSON body (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with JSON body (positive case)"
+)
+def test_message_with_json_body_positive_case() -> None:
+    """Message with JSON body (positive case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with XML body (negative case)"
+)
+def test_message_with_xml_body_negative_case() -> None:
+    """Message with XML body (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with XML body (positive case)"
+)
+def test_message_with_xml_body_positive_case() -> None:
+    """Message with XML body (positive case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with binary body (negative case)"
+)
+def test_message_with_binary_body_negative_case() -> None:
+    """Message with binary body (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with binary body (positive case)"
+)
+def test_message_with_binary_body_positive_case() -> None:
+    """Message with binary body (positive case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with plain text body (negative case)"
+)
+def test_message_with_plain_text_body_negative_case() -> None:
+    """Message with plain text body (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Message with plain text body (positive case)"
+)
+def test_message_with_plain_text_body_positive_case() -> None:
+    """Message with plain text body (positive case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Supports matching rules for the message body (negative case)"
+)
+def test_supports_matching_rules_for_the_message_body_negative_case() -> None:
+    """Supports matching rules for the message body (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Supports matching rules for the message body (positive case)"
+)
+def test_supports_matching_rules_for_the_message_body_positive_case() -> None:
+    """Supports matching rules for the message body (positive case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Supports matching rules for the message metadata (negative case)"
+)
+def test_supports_matching_rules_for_the_message_metadata_negative_case() -> None:
+    """Supports matching rules for the message metadata (negative case)."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Supports matching rules for the message metadata (positive case)"
+)
+def test_supports_matching_rules_for_the_message_metadata_positive_case() -> None:
+    """Supports matching rules for the message metadata (positive case)."""
+
+@pytest.mark.skip("Currently unable to implement")
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Supports messages with body formatted for the Kafka schema registry"
+)
+def test_supports_messages_with_body_formatted_for_the_kafka_schema_registry() -> None:
+    """Supports messages with body formatted for the Kafka schema registry."""
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Verifies the message metadata"
+)
+def test_verifies_the_message_metadata() -> None:
+    """Verifies the message metadata."""
+
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Verifying a simple message"
+)
+def test_verifying_a_simple_message() -> None:
+    """Verifying a simple message."""
+
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Verifying an interaction with a defined provider state"
+)
+def test_verifying_an_interaction_with_a_defined_provider_state() -> None:
+    """Verifying an interaction with a defined provider state."""
+
+
+@scenario(
+    "definition/features/V3/message_provider.feature",
+    "Verifying multiple Pact files"
+)
+def test_verifying_multiple_pact_files() -> None:
+    """Verifying multiple Pact files."""
+
+
+################################################################################
+## Given
+################################################################################
+
+
+@given(
+    parsers.re(
+        r'a Pact file for "(?P<name>[^"]+)" is to be verified with the following:\n'
+        r'(?P<table>.+)',
+        re.DOTALL,
+    ),
+    converters={"table": parse_horizontal_markdown_table},
+)
+def a_pact_file_for_is_to_be_verified_with_the_following(
+    verifier: Verifier,
+    temp_dir: Path,
+    name: str,
+    table: dict[str, str],
+) -> None:
+    """A Pact file for "basic" is to be verified with the following."""
+    metadata = {}
+    if table.get("metadata"):
+        def _repl(x: str) -> tuple[str, str]:
+            return (z.replace("JSON: ", "") for z in x.split("="))
+        metadata = dict(
+            _repl(x) for x in table["metadata"].split("; ")
+        )
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V3")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        metadata=metadata,
+        response_body=table["body"],
+        matching_rules=table.get("matching rules"),
+    )
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+
+
+@given(
+    parsers.parse('a Pact file for "{name}":"{fixture}" is to be verified')
+)
+def a_pact_file_for_is_to_be_verified(
+    verifier: Verifier,
+    temp_dir: Path,
+    name: str,
+    fixture: str
+) -> None:
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V3")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        response_body=fixture,
+    )
+    # for plain text message, the mime type needs to be set
+    if not re.match(r"^(file:|JSON:)", fixture):
+        interaction_definition.response_body.mime_type = "text/html;charset=utf-8"
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+
+
+@given(
+    parsers.parse(
+        'a Pact file for "{name}":"{fixture}" is to be '
+        'verified with provider state "{provider_state}"'
+    )
+)
+def a_pact_file_for_is_to_be_verified_with_provider_state(
+    temp_dir: Path,
+    verifier: Verifier,
+    name: str,
+    fixture: str,
+    provider_state: str,
+) -> None:
+    """A Pact file is to be verified with provider state."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V3")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        response_body=fixture,
+    )
+    states = [InteractionDefinition.State(provider_state)]
+    interaction_definition.states = states
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+    with (temp_dir / "provider_states").open("w") as f:
+        logger.debug("Writing provider state to %s", temp_dir / "provider_states")
+        json.dump([
+            s.as_dict()
+            for s in [InteractionDefinition.State(provider_state)]
+        ], f)
+
+
+@given(
+    parsers.re(
+        r'a Pact file for "(?P<name>[^"]+)":"(?P<fixture>[^"]+)" is '
+        'to be verified with the following metadata:\n'
+        r'(?P<metadata>.+)',
+        re.DOTALL,
+    ),
+    converters={"metadata": parse_markdown_table},
+)
+def a_pact_file_for_is_to_be_verified_with_the_following_metadata(
+    temp_dir: Path,
+    verifier: Verifier,
+    name: str,
+    fixture: str,
+    metadata: dict[str, str],
+) -> None:
+    """A Pact file is to be verified with the following metadata."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V3")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        metadata={ h["key"]: h["value"].replace("JSON: ", "") for h in metadata },
+        response_body=fixture,
+    )
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+
+a_provider_is_started_that_can_generate_the_message()
+
+@given(
+    parsers.re(
+        r'a provider is started that can generate the "(?P<name>[^"]+)" '
+        r'message with "(?P<fixture>[^"]+)" and the following metadata:\n'
+        r'(?P<metadata>.+)',
+        re.DOTALL,
+    ),
+    converters={"metadata": parse_markdown_table},
+)
+def a_provider_is_started_that_can_generate_the_message_with_the_following_metadata(
+    temp_dir: Path,
+    name: str,
+    fixture: str,
+    metadata: dict[str, str],
+) -> None:
+    """A provider is started that can generate the message with the following metadata.""" # noqa: E501
+    interaction_definitions = []
+    if ( temp_dir / "interactions.pkl").exists():
+        with (temp_dir / "interactions.pkl").open("rb") as pkl_file:
+            interaction_definitions = pickle.load(pkl_file) # noqa: S301
+
+    def parse_metadata_value(value: str) -> str:
+        return json.loads(
+            value.replace("JSON: ", "")
+        ) if value.startswith("JSON: ") else value
+
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        metadata={ m["key"]: parse_metadata_value(m["value"]) for m in metadata },
+        response_body=fixture
+    )
+    interaction_definitions.append(interaction_definition)
+    with (temp_dir / "interactions.pkl").open("wb") as pkl_file:
+        pickle.dump(interaction_definitions, pkl_file)
+
+
+@given(
+    "a provider state callback is configured",
+    target_fixture="callback"
+)
+def a_provider_state_callback_is_configured() -> None:
+    """A provider state callback is configured."""
+
+
+################################################################################
+## When
+################################################################################
+
+
+the_verification_is_run_with_start_context()
+
+
+################################################################################
+## Then
+################################################################################
+
+
+the_provider_state_callback_will_be_called_before_the_verification_is_run()
+
+the_provider_state_callback_will_be_called_after_the_verification_is_run()
+
+the_provider_state_callback_will_receive_a_setup_call()
+
+the_verification_will_be_successful()
+
+the_verification_results_will_contain_a_error()
diff --git a/tests/v3/compatibility_suite/test_v4_message_consumer.py b/tests/v3/compatibility_suite/test_v4_message_consumer.py
new file mode 100644
index 000000000..978e3d3c3
--- /dev/null
+++ b/tests/v3/compatibility_suite/test_v4_message_consumer.py
@@ -0,0 +1,175 @@
+"""Message consumer feature tests."""
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any, NamedTuple
+
+from pytest_bdd import (
+    given,
+    parsers,
+    scenario,
+    then,
+    when,
+)
+
+from pact.v3.pact import AsyncMessageInteraction
+from pact.v3.pact import MessagePact as Pact
+from tests.v3.compatibility_suite.util import string_to_int
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+class PactInteraction(NamedTuple):
+    """Holder class for Pact and Interaction."""
+    pact: Pact
+    interaction: AsyncMessageInteraction
+
+
+@scenario(
+    "definition/features/V4/message_consumer.feature",
+    "Sets the type for the interaction"
+)
+def test_sets_the_type_for_the_interaction() -> None:
+    """Sets the type for the interaction."""
+
+
+@scenario(
+    "definition/features/V4/message_consumer.feature",
+    "Supports adding comments"
+)
+def test_supports_adding_comments() -> None:
+    """Supports adding comments."""
+
+
+@scenario(
+    "definition/features/V4/message_consumer.feature",
+    "Supports specifying a key for the interaction"
+)
+def test_supports_specifying_a_key_for_the_interaction() -> None:
+    """Supports specifying a key for the interaction."""
+
+
+@scenario(
+    "definition/features/V4/message_consumer.feature",
+    "Supports specifying the interaction is pending"
+)
+def test_supports_specifying_the_interaction_is_pending() -> None:
+    """Supports specifying the interaction is pending."""
+
+
+################################################################################
+## Given
+################################################################################
+
+
+@given(
+    parsers.re(r'a comment "(?P<comment>[^"]+)" is added to the message interaction')
+)
+def a_comment_is_added_to_the_message_interaction(
+    pact_interaction: PactInteraction,
+    comment: str
+) -> None:
+    """A comment "{comment}" is added to the message interaction."""
+    pact_interaction.interaction.add_text_comment(comment)
+
+
+@given(parsers.re(
+    r'a key of "(?P<key>[^"]+)" is specified for the message interaction')
+)
+def a_key_is_specified_for_the_http_interaction(
+    pact_interaction: PactInteraction,
+    key: str,
+) -> None:
+    """A key is specified for the HTTP interaction."""
+    pact_interaction.interaction.set_key(key)
+
+
+@given(
+    "a message interaction is being defined for a consumer test",
+    target_fixture="pact_interaction"
+)
+def a_message_interaction_is_being_defined_for_a_consumer_test() -> None:
+    """A message integration is being defined for a consumer test."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V4")
+    yield PactInteraction(pact, pact.upon_receiving("a request", "Async"))
+
+
+@given("the message interaction is marked as pending")
+def the_message_interaction_is_marked_as_pending(
+    pact_interaction: PactInteraction
+) -> None:
+    """The message interaction is marked as pending."""
+    pact_interaction.interaction.set_pending(pending=True)
+
+
+################################################################################
+## When
+################################################################################
+
+
+@when(
+    "the Pact file for the test is generated",
+    target_fixture="pact_data"
+)
+def the_pact_file_for_the_test_is_generated(
+    pact_interaction: PactInteraction,
+    temp_dir: Path
+) -> None:
+    """The Pact file for the test is generated."""
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact_interaction.pact.write_file(temp_dir / "pacts")
+    with (
+        temp_dir / "pacts" / "consumer-provider.json"
+    ).open() as file:
+        yield json.load(file)
+
+
+################################################################################
+## Then
+################################################################################
+
+
+@then(
+    parsers.re(
+        r"the (?P<num>[^ ]+) interaction in the Pact file"
+        r" will have \"(?P<key>[^\"]+)\" = '(?P<value>[^']+)'"
+    ),
+    converters={"num": string_to_int},
+)
+def the_interaction_in_the_pact_file_will_have_a_key_of(
+    pact_data: dict[str, Any],
+    num: int,
+    key: str,
+    value: str,
+) -> None:
+    """The interaction in the Pact file will have a key of value."""
+    assert "interactions" in pact_data
+    assert len(pact_data["interactions"]) >= num
+    interaction = pact_data["interactions"][num - 1]
+    assert key in interaction
+    value = json.loads(value)
+    if isinstance(value, list):
+        assert interaction[key] in value
+    else:
+        assert interaction[key] == value
+
+@then(
+    parsers.re(
+        r"the (?P<num>[^ ]+) interaction in the Pact file"
+        r' will have a type of "(?P<interaction_type>[^"]+)"'
+    ),
+    converters={"num": string_to_int},
+)
+def the_interaction_in_the_pact_file_will_container_provider_states(
+    pact_data: dict[str, Any],
+    num: int,
+    interaction_type: str,
+) -> None:
+    """The interaction in the Pact file will container provider states."""
+    assert "interactions" in pact_data
+    assert len(pact_data["interactions"]) >= num
+    interaction = pact_data["interactions"][num - 1]
+    assert interaction["type"] == interaction_type
+
diff --git a/tests/v3/compatibility_suite/test_v4_message_producer.py b/tests/v3/compatibility_suite/test_v4_message_producer.py
new file mode 100644
index 000000000..dff10db1f
--- /dev/null
+++ b/tests/v3/compatibility_suite/test_v4_message_producer.py
@@ -0,0 +1,172 @@
+"""Message provider feature tests."""
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+from pytest_bdd import (
+    given,
+    parsers,
+    scenario,
+    then,
+)
+
+from pact.v3 import Pact
+from tests.v3.compatibility_suite.util import (
+    InteractionDefinition,
+    parse_markdown_table,
+)
+from tests.v3.compatibility_suite.util.provider import (
+    VERIFIER_ERROR_MAP,
+    a_provider_is_started_that_can_generate_the_message,
+    the_verification_is_run_with_start_context,
+    the_verification_will_be_successful,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from pact.v3.verifier import Verifier
+
+@scenario(
+    "definition/features/V4/message_provider.feature",
+    "Verifying a message interaction with comments"
+)
+def test_verifying_a_message_interaction_with_comments() -> None:
+    """Verifying a message interaction with comments."""
+
+
+@scenario(
+    "definition/features/V4/message_provider.feature",
+    "Verifying a pending message interaction"
+)
+def test_verifying_a_pending_message_interaction() -> None:
+    """Verifying a pending message interaction."""
+
+
+################################################################################
+## Given
+################################################################################
+
+
+@given(
+    parsers.re(
+        r'a Pact file for "(?P<name>[^"]+)":"(?P<fixture>[^"]+)" is to be verified '
+        r'with the following comments:\n(?P<comments>.+)',
+        re.DOTALL,
+    ),
+    converters={"comments": parse_markdown_table}
+)
+def a_pact_file_is_to_be_verified_with_comments(
+    verifier: Verifier,
+    temp_dir: Path,
+    name: str,
+    fixture: str,
+    comments: list[dict[str, str]],
+) -> None:
+    """A Pact file is to be verified with comments."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V4")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        response_body=fixture,
+    )
+    for comment in comments:
+        comment_type = comment.get("type", "text")
+        if comment_type == "text":
+            interaction_definition.text_comments.append(comment.get("comment"))
+        elif comment_type == "testname":
+            interaction_definition.test_name = comment.get("comment")
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+
+
+@given(
+    parsers.parse(
+        'a Pact file for "{name}":"{fixture}" is to be verified, but is marked pending'
+    )
+)
+def a_pact_file_is_to_be_verified_but_is_marked_pending(
+    verifier: Verifier,
+    temp_dir: Path,
+    name: str,
+    fixture: str
+) -> None:
+    """A Pact file is to be verified, but is marked pending."""
+    pact = Pact("consumer", "provider")
+    pact.with_specification("V4")
+    interaction_definition = InteractionDefinition(
+        method="POST",
+        path=f"/{name}",
+        is_async_message=True,
+        response_body=fixture,
+        is_pending=True,
+    )
+    interaction_definition.add_to_pact(pact, name, "Async")
+    (temp_dir / "pacts").mkdir(exist_ok=True, parents=True)
+    pact.write_file(temp_dir / "pacts")
+    verifier.add_source(temp_dir / "pacts")
+
+a_provider_is_started_that_can_generate_the_message()
+
+
+################################################################################
+## When
+################################################################################
+
+
+the_verification_is_run_with_start_context()
+
+
+################################################################################
+## Then
+################################################################################
+
+
+@then(
+    parsers.parse('the "{test_name}" will displayed as the original test name')
+)
+def the_test_name_will_displayed_as_the_original_test_name(
+    verifier_result: list[any, any],
+    test_name: str
+) -> None:
+    """The expected test name will displayed as the original test name."""
+    verifier_output: str = verifier_result[0].output
+    assert f"Test Name: {test_name}" in verifier_output()
+
+
+@then(
+    parsers.parse('the comment "{comment}" will have been printed to the console')
+)
+def the_comment_will_have_been_printed_to_the_console(
+    verifier_result: list[any, any],
+    comment: str
+) -> None:
+    """The expected comment will have been printed to the console."""
+    verifier_output: str = verifier_result[0].output
+    assert comment in verifier_output()
+
+
+the_verification_will_be_successful()
+
+
+@then(
+    parsers.parse('there will be a pending "{pending_error}" error')
+)
+def there_will_be_a_pending_error(
+    verifier_result: list[any, any],
+    pending_error: str
+) -> None:
+    """There will be a pending  error."""
+    expected_mismatch_type = VERIFIER_ERROR_MAP.get(pending_error)
+    mismatch_types = [
+        mismatch["type"]
+        for error in verifier_result[0].results["pendingErrors"]
+        for mismatch in error["mismatch"]["mismatches"]
+    ]
+    assert expected_mismatch_type in mismatch_types
+
diff --git a/tests/v3/compatibility_suite/util/__init__.py b/tests/v3/compatibility_suite/util/__init__.py
index 1351b1146..37b04f753 100644
--- a/tests/v3/compatibility_suite/util/__init__.py
+++ b/tests/v3/compatibility_suite/util/__init__.py
@@ -29,18 +29,23 @@ def _():
 import sys
 import typing
 from collections.abc import Collection, Mapping
+from contextlib import suppress
 from datetime import date, datetime, time
 from pathlib import Path
 from typing import Any
-from xml.etree import ElementTree
 
 import flask
+from defusedxml import ElementTree
 from flask import request
 from multidict import MultiDict
 from typing_extensions import Self
 from yarl import URL
 
 if typing.TYPE_CHECKING:
+    from pact.v3.interaction import (
+        AsyncMessageInteraction,
+        Interaction,
+    )
     from pact.v3.pact import Pact
 
 logger = logging.getLogger(__name__)
@@ -157,6 +162,32 @@ def parse_markdown_table(content: str) -> list[dict[str, str]]:
     return [dict(zip(rows[0], row)) for row in rows[1:]]
 
 
+def parse_horizontal_markdown_table(content: str) -> list[dict[str, str]]:
+    """
+    Parse a Markdown table into a list of dictionaries.
+
+    The table is expected to be in the following format:
+
+    ```markdown
+    | key1 | val1 |
+    | key2 | val2 |
+    | key3 | val3 |
+    ```
+
+    """
+    rows = [
+        list(map(str.strip, row.split("|")))[1:-1]
+        for row in content.split("\n")
+        if row.strip()
+    ]
+
+    if len(rows[0]) > 2:
+        msg = f"Expected at most two columns in the table, got {len(rows[0])}"
+        raise ValueError(msg)
+
+    return {row[0]: row[1] for row in rows}
+
+
 def serialize(obj: Any) -> Any:  # noqa: ANN401, PLR0911
     """
     Convert an object to a dictionary.
@@ -263,7 +294,7 @@ class Body:
         - An XML document
         """
 
-        def __init__(self, data: str) -> None:
+        def __init__(self, data: str | bytes) -> None:
             """
             Instantiate the interaction body.
             """
@@ -271,6 +302,10 @@ def __init__(self, data: str) -> None:
             self.bytes: bytes | None = None
             self.mime_type: str | None = None
 
+            if isinstance(data, bytes):
+                self.bytes = data
+                return
+
             if data.startswith("file: ") and data.endswith("-body.xml"):
                 self.parse_fixture(FIXTURES_ROOT / data[6:])
                 return
@@ -311,7 +346,7 @@ def parse_fixture(self, fixture: Path) -> None:
             This is used to parse the fixture files that contain additional
             metadata about the body (such as the content type).
             """
-            etree = ElementTree.parse(fixture)  # noqa: S314
+            etree = ElementTree.parse(fixture)
             root = etree.getroot()
             if not root or root.tag != "body":
                 msg = "Invalid XML fixture document"
@@ -416,6 +451,7 @@ def __init__(self, **kwargs: str) -> None:
         self.method: str = kwargs.pop("method")
         self.path: str = kwargs.pop("path")
         self.response: int = int(kwargs.pop("response", 200))
+        self.is_async_message: bool = kwargs.pop("is_async_message", False)
         self.query: str | None = None
         self.headers: MultiDict[str] = MultiDict()
         self.body: InteractionDefinition.Body | None = None
@@ -424,6 +460,8 @@ def __init__(self, **kwargs: str) -> None:
         self.response_body: InteractionDefinition.Body | None = None
         self.matching_rules: str | None = None
         self.response_matching_rules: str | None = None
+        self.metadata: dict[str, Any] | None = None
+        self.is_pending: bool = kwargs.pop("is_pending", False)
 
         self.update(**kwargs)
 
@@ -505,6 +543,9 @@ def update(self, **kwargs: str) -> None:  # noqa: C901, PLR0912
         ):
             self.response_matching_rules = parse_matching_rules(matching_rules)
 
+        if metadata := kwargs.pop("metadata", None):
+            self.metadata = metadata
+
         if len(kwargs) > 0:
             msg = f"Unexpected arguments: {kwargs.keys()}"
             raise TypeError(msg)
@@ -517,7 +558,88 @@ def __repr__(self) -> str:
             ", ".join(f"{k}={v!r}" for k, v in vars(self).items()),
         )
 
-    def add_to_pact(self, pact: Pact, name: str) -> None:  # noqa: C901, PLR0912, PLR0915
+    def _add_body(
+        self,
+        body: InteractionDefinition.Body,
+        interaction: Interaction
+    ) -> None:
+        if body.string:
+            logger.info(
+                "with_body(%s, %s)",
+                truncate(body.string),
+                body.mime_type,
+            )
+            interaction.with_body(
+                body.string,
+                body.mime_type,
+            )
+        elif body.bytes:
+            logger.info(
+                "with_binary_file(%s, %s)",
+                truncate(body.bytes),
+                body.mime_type,
+            )
+            interaction.with_binary_body(
+                body.bytes,
+                body.mime_type,
+            )
+        else:
+            msg = "Unexpected body definition"
+            raise RuntimeError(msg)
+
+    def _add_async_message_to_pact(
+        self,
+        interaction: AsyncMessageInteraction,
+    ) -> None:
+        if self.response_body.mime_type == "application/xml":
+            def _element_to_json(
+                element: ElementTree.Element
+            ) -> dict[str, Any]:
+                json_dict = {
+                    "name": element.tag,
+                }
+                if element.attrib:
+                    json_dict["attributes"] = element.attrib
+                if len(element):
+                    json_dict["children"] = [
+                        _element_to_json(child) for child in element
+                    ]
+                else:
+                    json_dict["children"] = [ { "content": element.text } ]
+                return json_dict
+            with suppress(ElementTree.ParseError):
+                # try to parse the content as XML
+                # it _may_ be JSON, so it's ok if this errors
+                self.response_body.string  = json.dumps(
+                    {
+                        "root": _element_to_json(
+                            ElementTree.fromstring(self.response_body.string)
+                        )
+                    }
+                )
+
+        logger.info(
+            "with_content(%s, %s)",
+            truncate(
+                self.response_body.string
+                if self.response_body.string
+                else self.response_body.bytes
+            ),
+            self.response_body.mime_type,
+        )
+        interaction.with_content(
+            self.response_body.string
+            if self.response_body.string
+            else self.response_body.bytes,
+            self.response_body.mime_type
+        )
+
+    def add_to_pact(
+        self,
+        pact: Pact,
+        name: str,
+        interaction_type: typing.Literal["HTTP", "Sync", "Async"] = "HTTP"
+    ) -> None:
         """
         Add the interaction to the pact.
 
@@ -532,9 +654,10 @@ def add_to_pact(self, pact: Pact, name: str) -> None:  # noqa: C901, PLR0912, PL
             name:
                 Name for this interaction. Must be unique for the pact.
         """
-        interaction = pact.upon_receiving(name)
-        logger.info("with_request(%s, %s)", self.method, self.path)
-        interaction.with_request(self.method, self.path)
+        interaction = pact.upon_receiving(name, interaction_type)
+        logger.info("with_request(%s, %s, %s)", self.method, self.path, interaction_type)
+        if not self.is_async_message:
+            interaction.with_request(self.method, self.path)
 
         for state in self.states or []:
             if state.parameters:
@@ -549,8 +672,9 @@ def add_to_pact(self, pact: Pact, name: str) -> None:  # noqa: C901, PLR0912, PL
             interaction.set_pending(pending=True)
 
         if self.text_comments:
-            logger.info("set_comment(text, %s)", self.text_comments)
-            interaction.set_comment("text", self.text_comments)
+            for comment in self.text_comments:
+                logger.info("add_text_comment(%s)", comment)
+                interaction.add_text_comment(comment)
 
         for key, value in self.comments.items():
             logger.info("set_comment(%s, %s)", key, value)
@@ -570,71 +694,39 @@ def add_to_pact(self, pact: Pact, name: str) -> None:  # noqa: C901, PLR0912, PL
             interaction.with_headers(self.headers.items())
 
         if self.body:
-            if self.body.string:
-                logger.info(
-                    "with_body(%s, %s)",
-                    truncate(self.body.string),
-                    self.body.mime_type,
-                )
-                interaction.with_body(
-                    self.body.string,
-                    self.body.mime_type,
-                )
-            elif self.body.bytes:
-                logger.info(
-                    "with_binary_file(%s, %s)",
-                    truncate(self.body.bytes),
-                    self.body.mime_type,
-                )
-                interaction.with_binary_body(
-                    self.body.bytes,
-                    self.body.mime_type,
-                )
-            else:
-                msg = "Unexpected body definition"
-                raise RuntimeError(msg)
+            self._add_body(self.body, interaction)
 
         if self.matching_rules:
             logger.info("with_matching_rules(%s)", self.matching_rules)
             interaction.with_matching_rules(self.matching_rules)
 
         if self.response:
-            logger.info("will_respond_with(%s)", self.response)
-            interaction.will_respond_with(self.response)
+            if not self.is_async_message:
+                logger.info("will_respond_with(%s)", self.response)
+                interaction.will_respond_with(self.response)
 
         if self.response_headers:
             logger.info("with_headers(%s)", self.response_headers)
             interaction.with_headers(self.response_headers.items())
 
         if self.response_body:
-            if self.response_body.string:
-                logger.info(
-                    "with_body(%s, %s)",
-                    truncate(self.response_body.string),
-                    self.response_body.mime_type,
-                )
-                interaction.with_body(
-                    self.response_body.string,
-                    self.response_body.mime_type,
-                )
-            elif self.response_body.bytes:
-                logger.info(
-                    "with_binary_file(%s, %s)",
-                    truncate(self.response_body.bytes),
-                    self.response_body.mime_type,
-                )
-                interaction.with_binary_body(
-                    self.response_body.bytes,
-                    self.response_body.mime_type,
-                )
+            if self.is_async_message:
+                self._add_async_message_to_pact(interaction)
             else:
-                msg = "Unexpected body definition"
-                raise RuntimeError(msg)
+                self._add_body(self.response_body, interaction)
 
         if self.response_matching_rules:
             logger.info("with_matching_rules(%s)", self.response_matching_rules)
             interaction.with_matching_rules(self.response_matching_rules)
 
+        if self.metadata:
+            for key, value in self.metadata.items():
+                interaction.with_metadata({key: value})
+
+        if self.is_pending:
+            logger.info("set_pending(True)")
+            interaction.set_pending(pending=True)
+
     def add_to_flask(self, app: flask.Flask) -> None:
         """
         Add an interaction to a Flask app.
@@ -697,3 +789,24 @@ def route_fn() -> flask.Response:
             view_func=route_fn,
             methods=[self.method],
         )
+
+    def create_message_response(self) -> flask.Response:
+        """Creates a flask response for an async message."""
+        if self.metadata:
+            self.response_headers.add(
+                "Pact-Message-Metadata",
+                base64.b64encode(
+                    json.dumps(self.metadata).encode("utf-8")
+                ).decode("utf-8")
+            )
+        return flask.Response(
+            response=self.response_body.bytes or self.response_body.string or None
+            if self.response_body
+            else None,
+            status=self.response,
+            headers=dict(**self.response_headers),
+            content_type=self.response_body.mime_type
+            if self.response_body
+            else None,
+            direct_passthrough=True,
+        )
diff --git a/tests/v3/compatibility_suite/util/provider.py b/tests/v3/compatibility_suite/util/provider.py
index 4c9d0d5dd..993cdcfd2 100644
--- a/tests/v3/compatibility_suite/util/provider.py
+++ b/tests/v3/compatibility_suite/util/provider.py
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import sys
+from contextlib import contextmanager
 from pathlib import Path
 
 import pytest
@@ -87,6 +88,13 @@
 start of the scenario.
 """
 
+VERIFIER_ERROR_MAP: dict[str,str] = {
+    "Response status did not match": "StatusMismatch",
+    "Headers had differences": "HeaderMismatch",
+    "Body had differences": "BodyMismatch",
+    "Metadata had differences": "MetadataMismatch",
+}
+
 
 def next_version() -> str:
     """
@@ -136,6 +144,7 @@ def __init__(self, provider_dir: Path | str) -> None:
                 called `interactions.pkl`. This file must contain a list of
                 [`InteractionDefinition`] objects.
         """
+        self._messages = {}
         self.provider_dir = Path(provider_dir)
         if not self.provider_dir.is_dir():
             msg = f"Directory {self.provider_dir} does not exist"
@@ -284,7 +293,26 @@ def _add_interactions(self, app: flask.Flask) -> None:
             interactions: list[InteractionDefinition] = pickle.load(f)  # noqa: S301
 
         for interaction in interactions:
-            interaction.add_to_flask(app)
+            if interaction.is_async_message:
+                self._messages.update({ interaction.path[1:]: interaction })
+            else:
+                interaction.add_to_flask(app)
+
+        @app.route("/message_handler", methods=["POST"])
+        def handle_messages() -> flask.Response:
+            body = json.loads(request.data.decode("utf-8"))
+            message = self._messages.get(body.get("description", ""))
+            if message:
+                return message.create_message_response()
+            return flask.Response(
+                response=json.dumps(
+                    {"error": f"Message {body.get('description')} not found"}
+                ),
+                status=404,
+                headers={"Content-Type": "application/json"},
+                content_type="application/json",
+                direct_passthrough=True,
+            )
 
     def run(self) -> None:
         """
@@ -578,6 +606,38 @@ def _(
         yield from start_provider(temp_dir)
 
 
+def a_provider_is_started_that_can_generate_the_message(
+    stacklevel: int = 1,
+) -> None:
+    @given(
+        parsers.parse(
+            'a provider is started that can generate the "{name}" message with "{body}"'
+        ),
+        stacklevel=stacklevel + 1,
+    )
+    def _(
+        temp_dir: Path,
+        name: str,
+        body: str,
+    ) -> None:
+        interaction_definitions = []
+        if ( temp_dir / "interactions.pkl").exists():
+            with (temp_dir / "interactions.pkl").open("rb") as pkl_file:
+                interaction_definitions = pickle.load(pkl_file) # noqa: S301
+
+        body = body.replace('\\"', '"')
+        interaction_definition = InteractionDefinition(
+            method="POST",
+            path=f"/{name}",
+            is_async_message=True,
+            response_body=body
+        )
+        interaction_definitions.append(interaction_definition)
+        with (temp_dir / "interactions.pkl").open("wb") as pkl_file:
+            pickle.dump(interaction_definitions, pkl_file)
+
+
+
 def start_provider(provider_dir: str | Path) -> Generator[URL, None, None]:  # noqa: C901
     """Start the provider app with the given interactions."""
     process = subprocess.Popen(
@@ -985,6 +1045,33 @@ def _(
         return verifier, None
 
 
+def the_verification_is_run_with_start_context(
+    stacklevel: int = 1,
+) -> tuple[Verifier, Exception | None]:
+    @when(
+        "the verification is run",
+        target_fixture="verifier_result",
+        stacklevel=stacklevel + 1,
+    )
+    def _(
+        verifier: Verifier,
+        temp_dir: Path,
+    ) -> tuple[Verifier, Exception | None]:
+        """Run the verification."""
+        start_provider_context_manager = contextmanager(start_provider)
+
+        with start_provider_context_manager(temp_dir) as provider_url:
+            verifier.set_state(
+                provider_url / "_test" / "callback",
+                teardown=True,
+            )
+            verifier.set_info("provider", url=f"{provider_url}/message_handler")
+            try:
+                verifier.verify()
+            except Exception as e:  # noqa: BLE001
+                return verifier, e
+            return verifier, None
+
 ################################################################################
 ## Then
 ################################################################################
@@ -1030,18 +1117,13 @@ def _(verifier_result: tuple[Verifier, Exception | None], error: str) -> None:
         verifier = verifier_result[0]
         logger.debug("Verification results: %s", json.dumps(verifier.results, indent=2))
 
-        if error == "Response status did not match":
-            mismatch_type = "StatusMismatch"
-        elif error == "Headers had differences":
-            mismatch_type = "HeaderMismatch"
-        elif error == "Body had differences":
-            mismatch_type = "BodyMismatch"
-        elif error == "State change request failed":
-            assert "One or more of the setup state change handlers has failed" in [
-                error["mismatch"]["message"] for error in verifier.results["errors"]
-            ]
-            return
-        else:
+        mismatch_type = VERIFIER_ERROR_MAP.get(error)
+        if not mismatch_type:
+            if error == "State change request failed":
+                assert "One or more of the setup state change handlers has failed" in [
+                    error["mismatch"]["message"] for error in verifier.results["errors"]
+                ]
+                return
             msg = f"Unknown error type: {error}"
             raise ValueError(msg)