pyAS4

Бібліотека Python для роботи з AS4 (OASIS ebXML AS4) протоколом обміну повідомленнями.

pyAS4 — це зручна обгортка навколо Zeep та MTOM/XOP-транспорту, яка спрощує створення й керування AS4-сумісними повідомленнями для систем електронної доставки та B2B-комунікацій.

Опис

pyAS4 надає удосконалений клієнт для роботи з AS4 веб-сервісами, включаючи:

• AS4Send - клієнт для відправлення AS4-повідомлень
• AS4Receive - клієнт для отримання AS4-повідомлень
• Header - конструктор заголовків AS4 для керування метаданими повідомлень
• Підтримка MTOM/XOP для передачі великих файлів
• Керування ідентифікаторами сторін і параметрами service/action
• Обробка вхідних та вихідних повідомлень

Вимоги

• Python >= 3.12
• lxml >= 6.1.1
• zeep >= 4.3.3
• pymtom-xop-fork >= 0.0.3

Встановлення

За допомогою uv:

uv add pyAS4

Або за допомогою pip:

python -m pip install pyAS4

Для розробки з вихідного коду:

git clone https://github.com/AndreyShapovalovVN/pyAS4.git
cd pyAS4
uv sync --extra dev

Використання

Спільні параметри клієнтів

AS4Send і AS4Receive використовують однакові параметри інфраструктури:

• wsdl — URL або локальний шлях до WSDL AS4/Domibus-сервісу.

• transport — транспорт Zeep. Для AS4Send обов'язково потрібен MtomTransport, оскільки вихідні файли передаються як MTOM/XOP-вкладення.

• plugins — список плагінів Zeep, наприклад [HistoryPlugin()]. Якщо плагіни не потрібні, слід передати порожній список [].

Різниця між клієнтами стосується лише даних AS4: для відправлення потрібен повний Header, а для отримання достатньо c4_party_id (finalRecipient).

Відправлення повідомлень

Контракт конструктора:

AS4Send(
    wsdl: str,
transport: MtomTransport,
plugins: list[HistoryPlugin],
header: Header,
)

Повний Header потрібен для формування ebMS-заголовка, включно з інформацією про сторони C1–C4, сервіс, дію, conversation ID і вкладення.

from pymtom_xop import MtomTransport
from zeep.exceptions import Fault
from zeep.plugins import HistoryPlugin

from pyAS4.AS4Client import AS4Send
from pyAS4.header import Header

transport = MtomTransport()
plugins = [HistoryPlugin()]

header = Header(
    c1_party_id="original-sender",
    c1_party_id_type="urn:example:identifier:sender",
    c2_party_id="sender-gateway",
    c2_party_id_type="urn:example:identifier:gateway",
    c3_party_id="receiver-gateway",
    c3_party_id_type="urn:example:identifier:gateway",
    c4_party_id="final-recipient",
    c4_party_id_type="urn:example:identifier:recipient",
    service="http://example.eu/services/evidence",
    service_type="urn:oasis:names:tc:ebcore:ebrs:ebms:binding:1.0",
    action="http://example.eu/actions/submit-evidence",
)

# Додаткові Property у MessageProperties (type є необов'язковим).
header.add_message_property("requestId", "request-123")
header.add_message_property("trackingId", "tracking-456", property_type="urn:example:tracking")

sender = AS4Send(
    wsdl="https://domibus.example.eu/services/backend?wsdl",
    transport=transport,
    plugins=plugins,
    header=header,
)

payloads = [
    {
        "content": b"<EvidenceRequest>...</EvidenceRequest>",
        "content_type": "application/xml",
        "cid": "evidence-request.xml",
    },
    {
        "content": b"%PDF-1.7 ...",
        "content_type": "application/pdf",
        "cid": "supporting-document.pdf",
    },
]

try:
    response = sender.send_message(payloads)
except Fault as exc:
    print(f"Domibus відхилив повідомлення: {exc}")

Контракт одного елемента payload:

from typing import NotRequired, TypedDict


class Payload(TypedDict):
    content: bytes | str
    content_type: str
    cid: NotRequired[str]

Сигнатура методу:

def send_message(self, payload: Sequence[Payload]) -> Any:
    ...

• content (bytes | str) — обов'язковий вміст вкладення. Рядок кодується як UTF-8.

• content_type (str) — обов'язковий MIME-тип, наприклад application/xml або application/pdf.

• cid (str, необов'язковий) — Content-ID вкладення. Якщо значення не передано, бібліотека генерує UUID.

send_message() додає вкладення до MTOM-транспорту, формує PayloadInfo, викликає SOAP-операцію submitMessage і повертає відповідь Zeep без додаткового перетворення. SOAP Fault та інші транспортні помилки логуються і передаються виклику користувача.

Отримання очікуючих повідомлень

Контракт конструктора:

AS4Receive(
    wsdl: str,
transport: Transport,
plugins: list[HistoryPlugin],
header: Header | None = None,
*,
c4_party_id: str | None = None,
)

Рекомендований спосіб — передати лише c4_party_id. Це значення використовується як finalRecipient в операції listPendingMessages. Тип ідентифікатора c4_party_id_type для цього SOAP-виклику не потрібен.

from pyAS4.AS4Client import AS4Receive

receiver = AS4Receive(
    wsdl="https://domibus.example.eu/services/backend?wsdl",
    transport=transport,
    plugins=plugins,
    c4_party_id="final-recipient",
)

for message in receiver.receive_message():
    print(message["messageId"])
    print(message["header"])
    for payload in message["payload"]:
        print(payload["content"])

Якщо повний Header уже існує, його також можна використати. У цьому випадку recipient береться з header.c4_party_id:

receiver = AS4Receive(
    wsdl="https://domibus.example.eu/services/backend?wsdl",
    transport=transport,
    plugins=plugins,
    header=header,
)

Якщо одночасно передані header і c4_party_id, явний c4_party_id має пріоритет. Якщо не передано жодного з них, конструктор піднімає ValueError.

receive_message() є генератором. Він викликає listPendingMessages, а потім retrieveMessage для кожного знайденого message ID. Кожне повідомлення має контракт:

{
    "messageId": str,
    "header": dict,
    "payload": list[dict[str, str]],
}

Приклад реального отриманого повідомлення

Нижче наведено анонімізований приклад повідомлення, отриманого від Domibus у сценарії OOTS ExecuteQueryRequest. Операційні ідентифікатори та персональні дані замінені демонстраційними значеннями.

import datetime

messages = [
    {
        "messageId": "00000000-0000-0000-0000-000000000001@domibus.eu",
        "header": {
            "messageId": "00000000-0000-0000-0000-000000000001@domibus.eu",
            "timestamp": datetime.datetime(2026, 9, 4, 10, 16, 33, 679505),
            "Party": {
                "From": {
                    "name": "domibus-red",
                    "type": "urn:oasis:names:tc:ebcore:partyid-type:unregistered:oots",
                    "role": "http://sdg.europa.eu/edelivery/gateway",
                },
                "To": {
                    "name": "domibus-blue",
                    "type": "urn:oasis:names:tc:ebcore:partyid-type:unregistered:oots",
                    "role": "http://sdg.europa.eu/edelivery/gateway",
                },
            },
            "CollaborationInfo": {
                "service": "QueryManager",
                "serviceType": "urn:oasis:names:tc:ebcore:ebrs:ebms:binding:1.0",
                "action": "ExecuteQueryRequest",
                "conversationId": "00000000-0000-0000-0000-000000000002",
            },
            "MessageProperties": [
                {
                    "finalRecipient": "EXAMPLE_PROVIDER",
                    "type": "urn:oasis:names:tc:ebcore:partyid-type:unregistered:XX",
                },
                {
                    "originalSender": "EXAMPLE_REQUESTER",
                    "type": "urn:oasis:names:tc:ebcore:partyid-type:unregistered:XX",
                },
            ],
            "PayloadInfo": [
                {
                    "href": "cid:ootsPayload@er.oots.eu",
                    "MimeType": "application/x-ebrs+xml",
                }
            ],
        },
        "payload": [
            {
                "href": "cid:ootsPayload@er.oots.eu",
                "content": """\
<query:QueryRequest
    xmlns:query="urn:oasis:names:tc:ebxml-regrep:xsd:query:4.0"
    xmlns:rim="urn:oasis:names:tc:ebxml-regrep:xsd:rim:4.0"
    xmlns:sdg="http://data.europa.eu/p4s"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    id="urn:uuid:00000000-0000-0000-0000-000000000003"
    xml:lang="EN">
  <!-- Скорочено: SpecificationIdentifier, Procedure, Requirements тощо -->
  <query:Query queryDefinition="DocumentQuery">
    <rim:Slot name="NaturalPerson">
      <rim:SlotValue xsi:type="rim:AnyValueType">
        <sdg:Person>
          <sdg:LevelOfAssurance>Substantial</sdg:LevelOfAssurance>
          <sdg:Identifier schemeID="eidas">XX/YY/REDACTED</sdg:Identifier>
          <sdg:FamilyName>REDACTED</sdg:FamilyName>
          <sdg:GivenName>REDACTED</sdg:GivenName>
          <sdg:DateOfBirth>REDACTED</sdg:DateOfBirth>
        </sdg:Person>
      </rim:SlotValue>
    </rim:Slot>
  </query:Query>
</query:QueryRequest>""",
            }
        ],
    }
]

Це Python-структура, а не готовий JSON: поле timestamp містить datetime.datetime. Перед серіалізацією в JSON його потрібно перетворити, наприклад, через timestamp.isoformat(). Повний XML payload доступний у message["payload"][0]["content"]; MIME-тип відповідного вкладення знаходиться в message["header"]["PayloadInfo"] за тим самим href.

SOAP Fault під час отримання списку pending messages передається користувачу. Помилки завантаження окремих повідомлень логуються, після чого генератор переходить до наступного повідомлення.

Робота з заголовками AS4

from pyAS4.header import Header

# Створіть заголовок AS4
header = Header(
    c1_party_id="sender-id",
    c1_party_id_type="urn:fdc:peppol.eu:2017:identifiers:C1",
    c2_party_id="recipient-id",
    c2_party_id_type="urn:fdc:peppol.eu:2017:identifiers:C2",
    c3_party_id="intermediary-id",
    c3_party_id_type="urn:fdc:peppol.eu:2017:identifiers:C3",
    c4_party_id="carrier-id",
    c4_party_id_type="urn:fdc:peppol.eu:2017:identifiers:C4",
    conversationid="unique-conversation-id",
    service="http://docs.oasis-open.org/ebxml-msg/as4/200902/service",
    service_type="urn:oasis:names:tc:ebcore:ebrs:ebms:binding:1.0",
    action="http://docs.oasis-open.org/ebxml-msg/as4/200902/action",
)

Структура проєкту

pyAS4/
├── pyAS4/
│   ├── __init__.py           # Точка входу бібліотеки
│   ├── AS4Client.py          # AS4Client, AS4Send та AS4Receive
│   └── header.py             # Клас Header для керування AS4-заголовками
├── README.md                 # Цей файл
├── pyproject.toml            # Конфігурація проєкту (PEP 517/518)
└── uv.lock                   # Зафіксовані залежності для uv

Архітектура

• AS4Send / AS4Receive — базові клієнти для відправлення й отримання AS4-повідомлень через WSDL, транспорт та плагіни Zeep. AS4Send використовує повний AS4-заголовок, тоді як AS4Receive може працювати лише з finalRecipient.

• Header — клас для конструювання та керування ebXML AS4-заголовками повідомлень. Генерує XML на основі стандартів OASIS ebXML.

• MtomTransport — транспорт із підтримкою MTOM/XOP для обробки великих бінарних вкладень.

Стандарти та протоколи

Проєкт реалізує наступні стандарти:

• OASIS ebXML AS4 v3.0 - Асинхронна обробка SOAP повідомлень
• MTOM/XOP - SOAP з вкладеннями (передача великих файлів)

Автор

Andrey Shapovalov (mt.andrey@gmail.com)

Ліцензія

EUPL v1.2

Поточна версія

0.1.23

Зміни у 0.1.23

• Виправлено параметр submitMessage: вкладення передаються через payload замість body.
• Додано Header.add_message_property(name, value, property_type=None) для додавання властивостей у MessageProperties з необов'язковим атрибутом type.

Внески

Розробка ведеться у репозиторії pyAS4 на GitHub.

Контакти

Для питань та пропозицій звертайтеся:

• Email: mt.andrey@gmail.com

---

Примітка: Бібліотека розроблена для роботи з системами електронної доставки, що відповідають стандартам OASIS ebXML AS4, та особливо для проєктів, що використовують OOTS мережу.