Data Validation

[Denny-g6labs]

Thank you for your library!

I'm using it to generate an invoice, and I'm wondering what is the best approach to do a data validation.

Example:

        pmt_id = pacs_008_001_08.PaymentIdentification7(
            # Is our transaction ID
            instr_id=transaction["id"],
            # Unique identification, as assigned by the initiating party, to unambiguously identify the transaction. This identification is passed on, unchanged, throughout the entire end-to-end chain.
            end_to_end_id=transaction["reference"],
            # Unique End-to-end Transaction Reference
            uetr=str(uuid.uuid4()),
        )

end_to_end_id has a max_length of 35, but no error is raised, nor is the value cut off when the XML is generated.

How to use these Dataclass field metadata info to do a validation?

Thanks!

EDIT:
Quick self made implementation:

        pmt_id = pacs_008_001_08.PaymentIdentification7(
            # Is our transaction ID
            instr_id=transaction["id"],
            # Unique identification, as assigned by the initiating party, to unambiguously identify the transaction. This identification is passed on, unchanged, throughout the entire end-to-end chain.
            end_to_end_id=self.__validate(
                pacs_008_001_08.PaymentIdentification7, "end_to_end_id", transaction["reference"]
            ),
            # Unique End-to-end Transaction Reference
            uetr=str(uuid.uuid4()),
        )

# or

        # Payment identification
        pmt_id = pacs_008_001_08.PaymentIdentification7(
            # Is our transaction ID
            instr_id=transaction["id"],
            # Unique identification, as assigned by the initiating party, to unambiguously identify the transaction. This identification is passed on, unchanged, throughout the entire end-to-end chain.
            end_to_end_id=transaction["reference"],
            # Unique End-to-end Transaction Reference
            uetr=str(uuid.uuid4()),
        )
        pmt_id.end_to_end_id = self.__validate(
            pacs_008_001_08.PaymentIdentification7, "end_to_end_id", pmt_id.end_to_end_id
        )

# [...]

    def __validate(self, dataclass, field_name, value):
        """
        Validate Metaclass fields (Basic implementation)
        """

        for field_info in fields(dataclass):
            if field_info.name == field_name:
                # Required
                if field_info.metadata.get("required"):
                    if not value:
                        raise BusinessError(
                            f"""Unable to generate ISO 20022: Field {field_name!r} ({dataclass.__name__}) is required but the value is empty"""
                        )

                # Pattern
                if pattern := field_info.metadata.get("pattern"):
                    if not re.findall(pattern, value or ""):
                        raise BusinessError(
                            f"""Unable to generate ISO 20022: Field {field_name!r} ({dataclass.__name__}) has a pattern of {pattern!r} but the value is {value!r}"""
                        )

                # Min Length
                if min_length := field_info.metadata.get("min_length"):
                    if value and len(value) < min_length:  # If value is empty, it is not required (tested above)
                        raise BusinessError(
                            f"""Unable to generate ISO 20022: Field {field_name!r} ({dataclass.__name__}) has a min length of {min_length} but the value is {value!r}"""
                        )

                # Max Length
                if max_length := field_info.metadata.get("max_length"):
                    if len(value or "") > max_length:
                        # raise BusinessError(
                        #     f"""Unable to generate ISO 20022: Field {field_name!r} ({dataclass.__name__}) has a max length of {max_length} but the value is {value!r}"""
                        # )
                        return value[:max_length]

        return value

[phoughton]

Hello,

Thanks, yes its a bit annoying when it comes to checking things like max_length etc. xsdata does not seem to support this out of the box.

There might be a need for a dataclass metadata validator, thanks for your example.

I've tended to revalidate against the XSD file, e.g.:

...
xml_content = serializer.render(doc, ns_map=ns_map_doc)

import xmlschema

schema = xmlschema.XMLSchema("iso20022/xsd/pain/pain_001/pain.001.001.08.xsd")

try:
    schema.validate(xml_content)
    print("XML is valid.")
except xmlschema.XMLSchemaValidationError as e:
    print("XML is invalid.")
    print(e)

Its usually give readable output:

$ python validcheck.py 
XML is invalid.
failed validating '1234567890123456789012345678901234567' with XsdMaxLengthFacet(value=35, fixed=False):

Reason: value length cannot be greater than 35

Schema component:

  <xs:maxLength xmlns:xs="http://www.w3.org/2001/XMLSchema" value="35" />

Instance type: <class 'xml.etree.ElementTree.Element'>

Instance:

  <MsgId xmlns="urn:iso:std:iso:20022:tech:xsd:pain.001.001.08">1234567890123456789012345678901234567</MsgId>

Path: /Document/CstmrCdtTrfInitn/GrpHdr/MsgId

Thanks
Pete

[Denny-g6labs]

Thanks!
Would you create a helper method for that? The xsd could be part of your lib then.

[phoughton]

Hey,
That's one option yes, although the package would get pretty large.
Maybe another option is to place them into another package.
It's something I'm looking into.

Thanks,
Pete.

[DennyWeinberg]

Another idea, use the install extras feature pip supports, like abc[der]==1.2.3.
I guess the issue should be reopened.

[phoughton]

Hello
The latest version has a validator (beta version) that can give basic info about of the document violates the dataclass metadata rules.

There's an example in the README.
Thanks
Pete

[Denny-g6labs]

Hey quick question @phoughton

We had some issues because the XML tags (for the address for example, or the bic) didn't have the correct order. It seems like the XSD somehow specifies the order of the tags (sequence). Do you know how that works? Is that also validated with the new feature?

Update: I just gave it a try, and it seems like that is not tested. I tried two different files where one has the wrong tag order and the result is PASSED:

#### 1
status: PASSED
errors: 0
warnings: 0


## Detailed Error Report
validation_status: PASSED
error_count: 0
warning_count: 0

#### 2
status: PASSED
errors: 0
warnings: 0


## Detailed Error Report
validation_status: PASSED
error_count: 0
warning_count: 0

[Denny-g6labs]

Oh, I'm realizing it doesn't make sense what I'm saying. The python object is of course correct and a serialization would wrote the file with all tags in the correct order.
In my case I could load the file, parse it, serialize it, and in the end it actually has the correct order.

[Denny-g6labs]

FYI, I'm going to use your (Beta) validation method in our system.

[Denny-g6labs]

Hi,

I had the issue that a processing failed because IntrBkSttlmDt was missing. Can we add more validation checks to the library?

Missing IntrBkSttlmDt (root cause of the error)

<IntrBkSttlmAmt Ccy="USD">1.00</IntrBkSttlmAmt>
<IntrBkSttlmDt>2026-01-23</IntrBkSttlmDt>   ← missing
<InstdAmt Ccy="USD">1.00</InstdAmt>
<ChrgBr>SLEV</ChrgBr>

The schema validator encounters ChrgBr where it expects IntrBkSttlmDt, which is exactly what the error says.
Add the settlement date:

<IntrBkSttlmAmt Ccy="USD">3670.00</IntrBkSttlmAmt>
<IntrBkSttlmDt>2026-05-19</IntrBkSttlmDt>
<ChrgBr>SLEV</ChrgBr>

Yes, the IntrBkSttlmDt (Interbank Settlement Date) is a required field for ISO 20022 wire transfers (such as pacs.008 and pacs.009 messages), but where it is required depends on the specific message structure and the network's implementation rules.
Under the core ISO 20022 base specification, there is a strict Cross-Element Complex Rule that dictates its placement:

1. Group Header Level (GrpHdr): If the Interbank Settlement Date is placed in the Group Header, it applies to all transactions within that message. If it is present here, it cannot be repeated at the transaction level.
2. Transaction Level (CdtTrfTxInf): If the Interbank Settlement Date is not present in the Group Header, it must be present within each individual Credit Transfer Transaction Information block.
   In other words, the date must exist exactly once for every transaction in the message.

How Wire Networks Implement This

Because high-value wire networks (like SWIFT CBPR+, Fedwire Funds Service, CHAPS, and TARGET2) typically process messages as single, individual transactions rather than batches, they enforce strict guidelines:

• SWIFT CBPR+ / Target2 / CHAPS: These guidelines generally restrict the message to a single transaction (NumberOfTransactions = 1) and mandate that the IntrBkSttlmDt must be placed in the Group Header (GrpHdr) as a required [1..1] element.
• Fedwire Funds Service: Similarly, Fedwire ISO 20022 guidelines require the IntrBkSttlmDt in the Group Header.
  If you are formatting a pacs.008 or pacs.009 for a real-time gross settlement (RTGS) or cross-border wire, you should almost always populate in the block using the standard YYYY-MM-DD ISO date format. Failing to include it will cause the message to be rejected by the clearing system or correspondent bank.

[phoughton]

Thanks, I'll take a look. I've also experienced poor XSD sourced error messages (generally).

Out of interest, which scheme (e.g. CBPR+, CHAPS, T2, FedWire etc) are you validating? and which message and version ? (e.g.: pacs.008.001.09 etc)

This library does not handle usage rules currently and is only setup for the base ISO20022 spec.

[Denny-g6labs]

We support pacs 001, 004, 008, 009. We mostly generate 008, for which I'd love to see support for such more extended validation mechanisms.

I wrote two helper modules for the Fedwire envelope logic:
helper_iso20022.py

from typing import Any, Dict

from shared_library.tools.exceptions import BusinessError
from shared_library.transaction.transaction_types.iso20022.iso20022_interface import (
    IS2O0022Document,
    IS2O0022PacsCustomerCreditTransferDocument,
    IS2O0022PacsFICreditTransferDocument,
    IS2O0022PacsPaymentReturnDocument,
)
from xsdata.formats.dataclass.parsers import XmlParser
from xsdata.formats.dataclass.parsers.config import ParserConfig
from xsdata.formats.dataclass.serializers import XmlSerializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig


class Iso20022Helper:
    # To parse
    config = ParserConfig(fail_on_unknown_properties=False)
    parser = XmlParser(config)

    # To serialize
    ns_map_doc_008: Dict[str | None, Any] = {
        None: "urn:iso:std:iso:20022:tech:xsd:pacs.008.001.08",
        "xsi": "http://www.w3.org/2001/XMLSchema-instance",
    }
    ns_map_doc_009: Dict[str | None, Any] = {
        None: "urn:iso:std:iso:20022:tech:xsd:pacs.009.001.08",
        "xsi": "http://www.w3.org/2001/XMLSchema-instance",
    }
    ns_map_doc_004: Dict[str | None, Any] = {
        None: "urn:iso:std:iso:20022:tech:xsd:pacs.004.001.10",
        "xsi": "http://www.w3.org/2001/XMLSchema-instance",
    }

    # @classmethod

    # def parse_head(cls, document_bytes: bytes) -> head_001_001_03.AppHdr:
    #     return cls.parser.from_bytes(document_bytes, head_001_001_03.AppHdr)

    @staticmethod
    def serialize_document(
        document: Any,
        ns_map_doc: Dict[str | None, Any],
        include_xml_declaration: bool = True,
    ) -> str:
        """
        Serializes a dataclass object into an XML string.

        Args:
            document: The dataclass object to serialize.
            schema_location_doc: The schema location string.
            ns_map_doc: The namespace map dictionary.
            include_xml_declaration: If True, includes the <?xml ... ?> header.
        """

        config = SerializerConfig(
            indent="  ",
            xml_declaration=include_xml_declaration,
            encoding="UTF-8",
        )

        serializer = XmlSerializer(config=config)

        xml_content = serializer.render(document, ns_map=ns_map_doc)

        return xml_content

    @classmethod
    def parse_pacs_document(cls, document_bytes: bytes) -> IS2O0022Document:
        """
        Parses bytes into an IS2O0022Document object.
        """

        content = document_bytes.decode("utf8")

        if """urn:iso:std:iso:20022:tech:xsd:pacs.004""" in content:
            # Return
            return cls.parser.from_bytes(document_bytes, IS2O0022PacsPaymentReturnDocument)

        elif """urn:iso:std:iso:20022:tech:xsd:pacs.008""" in content:
            # Customer Credit Transfer
            return cls.parser.from_bytes(document_bytes, IS2O0022PacsCustomerCreditTransferDocument)

        elif """urn:iso:std:iso:20022:tech:xsd:pacs.009""" in content:
            # Financial Institution Credit Transfer
            return cls.parser.from_bytes(document_bytes, IS2O0022PacsFICreditTransferDocument)

        else:
            # Should include the xmlns (for example `<Document xmlns="urn:iso:std:iso:20022:tech:xsd:pacs.002.001.10">`)
            raise BusinessError(
                f"""Unsupported ISO 20022 document: {content[1:65]}"""
            )  # Do not include the "<" character

    @classmethod
    def serialize_pacs_document(cls, document: IS2O0022Document) -> str:
        """
        Serializes a single IS2O0022Document object into an XML string.
        """

        match document:
            case IS2O0022PacsCustomerCreditTransferDocument():
                ns = cls.ns_map_doc_008
            case IS2O0022PacsFICreditTransferDocument():
                ns = cls.ns_map_doc_009
            case IS2O0022PacsPaymentReturnDocument():
                ns = cls.ns_map_doc_004

        return cls.serialize_document(document, ns)

helper_iso20022_fedwire.py

import re
import xml.etree.ElementTree as ET
from typing import List, Literal

from shared_library.tools.exceptions import BusinessError
from shared_library.transaction.transaction_types.iso20022.helper.iso20022.helper_iso20022 import (
    IS2O0022PacsCustomerCreditTransferDocument,
    IS2O0022PacsFICreditTransferDocument,
    Iso20022Helper,
)
from shared_library.transaction.transaction_types.iso20022.iso20022_interface import (
    IS2O0022Document,
    IS2O0022PacsPaymentReturnDocument,
)

BIZSvcType = Literal["TEST", "PROD"]


class Iso20022FedwireHelper:
    """
    A parser for fedwire wrapped ISO 20022 messages.
    """

    @classmethod
    def parse_fedwire_pacs(cls, xml_binary: bytes) -> List[IS2O0022Document]:
        """
        Parses an XML binary that contains one or more ISO 20022 messages
        wrapped in a custom <FedwireFundsIncoming> tag.

        Args:
            xml_binary: The XML content as bytes.

        Returns:
            A list of pacs.008.001.08 (Customer Credit Transfer) Documents.
            For now we ignore the custom headers
        """

        results: List[IS2O0022Document] = []

        # Decode the XML from bytes to a string
        xml_content = xml_binary.decode("utf-8").strip()

        # This prevents a ParseError when wrapping the content in a single root element.
        xml_content_no_decl = re.sub(r"<\?xml[^>]*\?>", "", xml_content).strip()

        # To robustly handle one or more root-level XML elements in the input,
        # we wrap the entire content in a single dummy root tag. This makes the
        # input a single, well-formed XML document for the initial parsing stage.
        xml_string = f"<root>{xml_content_no_decl}</root>"

        # Parse the wrapped XML structure
        tree = ET.fromstring(xml_string)

        # Iterate over each <FedwireFundsIncoming> element found under our dummy root
        for fedwire_msg_element in tree:
            document_element = None
            for child in fedwire_msg_element.iter():
                # The tag includes the namespace in braces, e.g., {urn...}Document.
                # We check if the local name (the part after the '}') is 'Document'.
                if child.tag.endswith("}Document"):
                    document_element = child
                    break  # Found the first one, stop searching

            if document_element is not None:
                # Convert these specific elements back into byte strings
                # app_hdr_bytes = ET.tostring(app_hdr_element)
                document_bytes = ET.tostring(document_element)

                # Parse
                # parsed_head = self.__parse_head(app_hdr_bytes)
                parsed_doc = Iso20022Helper.parse_pacs_document(document_bytes)
                results.append(parsed_doc)

        return results

    @classmethod
    def envelop_pacs_for_fedwire(cls, docs: List[IS2O0022Document], biz_svc: BIZSvcType) -> str:
        types = set([type(doc) for doc in docs])

        if all(doc_type == IS2O0022PacsCustomerCreditTransferDocument for doc_type in types):
            return Iso20022FedwireHelper.envelop_pacs_cct_for_fedwire(docs, biz_svc)  # type: ignore
        if all(doc_type == IS2O0022PacsFICreditTransferDocument for doc_type in types):
            return Iso20022FedwireHelper.envelop_pacs_fi_cdt_for_fedwire(docs, biz_svc)  # type: ignore
        if all(doc_type == IS2O0022PacsPaymentReturnDocument for doc_type in types):
            return Iso20022FedwireHelper.envelop_pacs_r_for_fedwire(docs, biz_svc)  # type: ignore
        else:
            raise BusinessError(f"""Can't envelope different document types ({types})""")

    @classmethod
    def envelop_pacs_cct_for_fedwire(
        cls, docs: List[IS2O0022PacsCustomerCreditTransferDocument], biz_svc: BIZSvcType
    ) -> str:
        """
        Envelops all ISO 20022 messages
        into a custom <FedwireFundsIncoming> tag to output an XML binary.

        Args:
            docs: A list of pacs.008.001.08 (Customer Credit Transfer) Documents.

        Returns:
            The XML content as bytes.
        """

        final_xml_parts = []

        for doc in docs:
            # --- 1. Extract data needed for the AppHdr from the pacs.008 document ---
            credit_transfer_info = doc.fito_ficstmr_cdt_trf.cdt_trf_tx_inf[  # type:ignore
                0
            ]  # Filled in ISO20022PacsGeneratorFromWire
            group_header = doc.fito_ficstmr_cdt_trf.grp_hdr  # Filled in ISO20022PacsGeneratorFromWire  # type:ignore

            # The Fedwire envelope helper REQUIRES 'clr_sys_mmb_id' (ABA).
            try:
                from_id = (
                    credit_transfer_info.instg_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore  # Should be filled in ISO20022PacsGeneratorFromWire
                )
            except AttributeError:
                raise BusinessError(
                    "Fedwire Envelope Error: The Instructing Agent (Sender) is missing or does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId). This field is mandatory for Fedwire."
                )

            try:
                to_id = (
                    credit_transfer_info.instd_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore  # Should be filled in ISO20022PacsGeneratorFromWire
                )
            except AttributeError:
                # Attempt to provide a helpful hint if a BIC was provided instead
                hint = ""
                try:
                    if credit_transfer_info.instd_agt.fin_instn_id.bicfi:  # type:ignore
                        hint = " A BIC was detected, but Fedwire cannot route directly to a BIC. For international wires, you must map the Instructed Agent to your US Correspondent Bank (ABA)."
                except AttributeError:
                    pass

                raise BusinessError(
                    f"Fedwire Envelope Error: The Instructed Agent (Receiver) is missing or does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId).{hint}"
                )

            biz_msg_idr = group_header.msg_id  # Should be filled in ISO20022PacsGeneratorFromWire  # type:ignore
            cre_dt = str(group_header.cre_dt_tm)  # Should be filled in ISO20022PacsGeneratorFromWire  # type:ignore

            # --- 2. Construct the AppHdr XML as a string ---
            app_hdr_xml = f"""<AppHdr xmlns="urn:iso:std:iso:20022:tech:xsd:head.001.001.03">
            <Fr>
                <FIId>
                    <FinInstnId>
                        <ClrSysMmbId>
                            <MmbId>{from_id}</MmbId>
                        </ClrSysMmbId>
                    </FinInstnId>
                </FIId>
            </Fr>
            <To>
                <FIId>
                    <FinInstnId>
                        <ClrSysMmbId>
                            <MmbId>{to_id}</MmbId>
                        </ClrSysMmbId>
                    </FinInstnId>
                </FIId>
            </To>
            <BizMsgIdr>{biz_msg_idr}</BizMsgIdr>
            <MsgDefIdr>pacs.008.001.08</MsgDefIdr>
            <BizSvc>{biz_svc}</BizSvc>
            <MktPrctc>
                <Regy>www2.swift.com/mystandards/#/group/Federal_Reserve_Financial_Services/Fedwire_Funds_Service</Regy>
                <Id>frb.fedwire.01</Id>
            </MktPrctc>
            <CreDt>{cre_dt}</CreDt>
            <PssblDplct>false</PssblDplct>
        </AppHdr>"""

            # --- 3. Serialize the main pacs.008 Document object to an XML string ---
            # Use the internal helper to render the XML without the `<?xml...?>` declaration
            document_xml = Iso20022Helper.serialize_document(
                document=doc,
                ns_map_doc=Iso20022Helper.ns_map_doc_008,
                include_xml_declaration=False,
            ).strip()

            # --- 4. Assemble the final, fully-wrapped Fedwire message ---
            fedwire_message_xml = f"""<FedwireFundsIncoming xmlns="urn:fedwirefunds:incoming:v001">
    <FedwireFundsIncomingMessage>
        <FedwireFundsCustomerCreditTransfer>
            {app_hdr_xml}
            {document_xml}
        </FedwireFundsCustomerCreditTransfer>
    </FedwireFundsIncomingMessage>
</FedwireFundsIncoming>"""
            final_xml_parts.append(fedwire_message_xml)

        # --- 5. Join all the generated message blocks and encode to bytes ---
        # The final output is a concatenation of one or more <FedwireFundsIncoming> blocks
        full_output = "\n".join(final_xml_parts)
        return full_output

    @classmethod
    def envelop_pacs_fi_cdt_for_fedwire(
        cls, docs: List[IS2O0022PacsFICreditTransferDocument], biz_svc: BIZSvcType
    ) -> str:
        """
        Envelops pacs.009 messages
        into a custom <FedwireFundsOutgoing> tag for FI credit transfers.

        Args:
            docs: A list of pacs.009.001.08 (Financial Institution Credit Transfer) Documents.
            biz_svc: The business service ("TEST" or "PROD").

        Returns:
            The XML content as a string.
        """

        final_xml_parts = []

        for doc in docs:
            # --- 1. Extract data needed for the AppHdr from the pacs.009 document ---
            credit_transfer_info = doc.ficdt_trf.cdt_trf_tx_inf[0]  # type:ignore
            group_header = doc.ficdt_trf.grp_hdr  # type:ignore

            try:
                from_id = credit_transfer_info.instg_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore
            except AttributeError:
                raise BusinessError(
                    "Fedwire Envelope Error (FI Credit Transfer): The Instructing Agent (Sender) is missing or does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId). This field is mandatory for Fedwire."
                )

            try:
                to_id = credit_transfer_info.instd_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore
            except AttributeError:
                hint = ""
                try:
                    if credit_transfer_info.instd_agt.fin_instn_id.bicfi:  # type:ignore
                        hint = " A BIC was detected, but Fedwire cannot route directly to a BIC."
                except AttributeError:
                    pass

                raise BusinessError(
                    f"Fedwire Envelope Error (FI Credit Transfer): The Instructed Agent (Receiver) is missing or does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId).{hint}"
                )

            biz_msg_idr = group_header.msg_id  # type:ignore
            cre_dt = str(group_header.cre_dt_tm)  # type:ignore

            # --- 2. Construct the AppHdr XML ---
            app_hdr_xml = f"""<AppHdr xmlns="urn:iso:std:iso:20022:tech:xsd:head.001.001.03">
            <Fr>
                <FIId>
                    <FinInstnId>
                        <ClrSysMmbId>
                            <MmbId>{from_id}</MmbId>
                        </ClrSysMmbId>
                    </FinInstnId>
                </FIId>
            </Fr>
            <To>
                <FIId>
                    <FinInstnId>
                        <ClrSysMmbId>
                            <MmbId>{to_id}</MmbId>
                        </ClrSysMmbId>
                    </FinInstnId>
                </FIId>
            </To>
            <BizMsgIdr>{biz_msg_idr}</BizMsgIdr>
            <MsgDefIdr>pacs.009.001.08</MsgDefIdr>
            <BizSvc>{biz_svc}</BizSvc>
            <MktPrctc>
                <Regy>www2.swift.com/mystandards/#/group/Federal_Reserve_Financial_Services/Fedwire_Funds_Service</Regy>
                <Id>frb.fedwire.01</Id>
            </MktPrctc>
            <CreDt>{cre_dt}</CreDt>
        </AppHdr>"""

            # --- 3. Serialize the main pacs.009 Document object to an XML string ---
            document_xml = Iso20022Helper.serialize_document(
                document=doc,
                ns_map_doc=Iso20022Helper.ns_map_doc_009,
                include_xml_declaration=False,
            ).strip()

            # --- 4. Assemble the final Fedwire message ---
            fedwire_message_xml = f"""<FedwireFundsOutgoing xmlns="urn:fedwirefunds:outgoing:v001">
    <FedwireFundsOutgoingMessage>
        <FedwireFundsFinancialInstitutionCreditTransfer>
            {app_hdr_xml}
            {document_xml}
        </FedwireFundsFinancialInstitutionCreditTransfer>
    </FedwireFundsOutgoingMessage>
</FedwireFundsOutgoing>"""
            final_xml_parts.append(fedwire_message_xml)

        return "\n".join(final_xml_parts)

    @classmethod
    def envelop_pacs_r_for_fedwire(
        cls, docs: List[IS2O0022PacsPaymentReturnDocument], biz_svc: Literal["TEST", "PROD"]
    ) -> str:
        """
        Envelops pacs.004 messages
        into a custom <FedwireFundsOutgoing> tag for returns.

        Args:
            docs: A list of pacs.004.001.10 (Payment Return) Documents.
            biz_svc: The business service ("TEST" or "PROD").

        Returns:
            The XML content as a string.
        """

        final_xml_parts = []

        for doc in docs:
            # --- 1. Extract data from the pacs.004 document ---
            # Note: For returns, the direction is reversed. The "Instructing Agent" of the return
            # is the recipient of the original payment.
            return_tx_info = doc.pmt_rtr.tx_inf[0]  # type:ignore
            group_header = doc.pmt_rtr.grp_hdr  # type:ignore

            # The 'from' and 'to' in the AppHdr refer to the parties in the return message itself.
            # The Fedwire envelope helper REQUIRES 'clr_sys_mmb_id' (ABA).
            try:
                from_id = return_tx_info.instg_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore
            except AttributeError:
                raise BusinessError(
                    "Fedwire Envelope Error (Return): The Instructing Agent (Sender of the Return) is missing or "
                    "does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId). "
                    "Fedwire requires an ABA for returns."
                )

            try:
                to_id = return_tx_info.instd_agt.fin_instn_id.clr_sys_mmb_id.mmb_id  # type:ignore
            except AttributeError:
                # Attempt to provide a helpful hint if a BIC was provided instead
                hint = ""
                try:
                    if return_tx_info.instd_agt.fin_instn_id.bicfi:  # type:ignore
                        hint = " A BIC was detected, but Fedwire cannot route directly to a BIC. Ensure the return is routed to the original sender's US Correspondent Bank."
                except AttributeError:
                    pass

                raise BusinessError(
                    f"Fedwire Envelope Error (Return): The Instructed Agent (Receiver of the Return) is missing or does not have a valid US ABA Routing Number (ClrSysMmbId/MmbId).{hint}"
                )
            biz_msg_idr = group_header.msg_id  # type:ignore
            cre_dt = str(group_header.cre_dt_tm)  # type:ignore

            # --- 2. Construct the AppHdr for the return ---
            app_hdr_xml = f"""<AppHdr xmlns="urn:iso:std:iso:20022:tech:xsd:head.001.001.03">
                <Fr>
                    <FIId>
                        <FinInstnId>
                            <ClrSysMmbId>
                            <MmbId>{from_id}</MmbId>
                            </ClrSysMmbId>
                        </FinInstnId>
                    </FIId>
                </Fr>
                <To>
                    <FIId>
                        <FinInstnId>
                            <ClrSysMmbId>
                                <MmbId>{to_id}</MmbId>
                            </ClrSysMmbId>
                        </FinInstnId>
                    </FIId>
                </To>
                <BizMsgIdr>{biz_msg_idr}</BizMsgIdr>
                <MsgDefIdr>pacs.004.001.10</MsgDefIdr>
                <BizSvc>{biz_svc}</BizSvc>
                <MktPrctc>
                    <Regy>www2.swift.com/mystandards/#/group/Federal_Reserve_Financial_Services/Fedwire_Funds_Service</Regy>
                    <Id>frb.fedwire.01</Id>
                </MktPrctc>
                <CreDt>{cre_dt}</CreDt>
            </AppHdr>"""

            # --- 3. Serialize the main pacs.004 Document object ---
            document_xml = Iso20022Helper.serialize_document(
                document=doc,
                ns_map_doc=Iso20022Helper.ns_map_doc_004,
                include_xml_declaration=False,
            ).strip()

            # --- 4. Assemble the final Fedwire return message ---
            fedwire_message_xml = f"""<FedwireFundsOutgoing xmlns="urn:fedwirefunds:outgoing:v001">
    <FedwireFundsOutgoingMessage>
        <FedwireFundsPaymentReturn>
            {app_hdr_xml}
            {document_xml}
        </FedwireFundsPaymentReturn>
    </FedwireFundsOutgoingMessage>
</FedwireFundsOutgoing>"""
            final_xml_parts.append(fedwire_message_xml)

        return "\n".join(final_xml_parts)

It might not be perfect but it works.

I'm using your validator to check line length and so on.

def validate_wire20022_file(
    ctx: DBContext,
    data_direction: DataDirectionType | None,
    document_list: Sequence[IS2O0022Document],
):
    """
    Validate a previously created WIRE20022 file,
    to get potential errors
    """

    logger.debug(f"""validate_wire20022_file""")

    # Custom validation
    error: str = do_custom_wire20022_validation(ctx, data_direction, document_list) or ""

    # pyiso20022 validation
    errors = []
    for doc in document_list:
        result = validate_message(doc, strict=False)  # If False, collects all errors

        if not result.is_valid:
            if result.errors:
                errors.append(f"Errors:\n{'\n'.join([str(error) for error in result.errors])}")

            if result.warnings:
                errors.append(f"Warnings:\n{'\n'.join([str(warning) for warning in result.warnings])}")

    error += "\n".join(errors)

    return error