API reference

The reference is generated from the source with mkdocstrings.

Root package

mxraven

The official Python SDK for the mxRaven mail runtime.

The root package exposes SMTP submission. The webhook and feedback runtime surfaces live in :mod:mxraven.webhook and :mod:mxraven.feedback.

Control-plane administration (tenants, domains, suppressions, analytics) is out of scope and is served by the separate mxRaven control-plane SDK.

Attachment dataclass

Attachment(
    filename: str = "",
    content_type: str = "",
    data: bytes = b"",
    inline: bool = False,
    content_id: str = "",
)

A message attachment.

Attributes:

Name	Type	Description
filename	str	The attachment filename. It may be empty.
content_type	str	The media type; defaults to application/octet-stream when empty.
data	bytes	The raw attachment content. The caller retains ownership; it is not mutated.
inline	bool	Whether the attachment is meant for inline display, for example an image referenced by content_id from an HTML body.
content_id	str	The inline content identifier, ignored unless inline is true.

Client

Client(
    host: str,
    *,
    port: int = DEFAULT_ADDRESS_PORT,
    username: str,
    secret: str,
    tls: SSLContext | None = None,
    pool_size: int = _DEFAULT_POOL_SIZE,
    connect_timeout: float = _DEFAULT_CONNECT_TIMEOUT,
    read_timeout: float = _DEFAULT_READ_TIMEOUT,
    write_timeout: float = _DEFAULT_WRITE_TIMEOUT,
    local_name: str | None = None,
    starttls: bool = True,
    require_tls: bool = True,
    implicit_tls: bool = False,
    local_address: str | None = None,
    auth_mechanisms: Sequence[str] = (),
)

Submits mail to the mxRaven SMTP submission service.

A client maintains a bounded pool of authenticated connections. Use it as a context manager or call :meth:close to release them.

Example

client = Client(
    host="smtp.mxraven.com",
    username="mxr_tx_ab12cd34ef56",
    secret=secret,
)
with client:
    result = client.send(message)

Parameters:

Name	Type	Description	Default
host	str	The submission server host; a port must not be included.	required
port	int	The submission server port. Defaults to 587.	DEFAULT_ADDRESS_PORT
username	str	The submission API key username.	required
secret	str	The submission API key secret.	required
tls	SSLContext | None	A custom STARTTLS ssl.SSLContext. The default verifies the server certificate against the system roots.	None
pool_size	int	The maximum number of pooled connections. Defaults to 5.	_DEFAULT_POOL_SIZE
connect_timeout	float	The TCP connect timeout in seconds.	_DEFAULT_CONNECT_TIMEOUT
read_timeout	float	The per-reply read timeout in seconds.	_DEFAULT_READ_TIMEOUT
write_timeout	float	The per-write timeout in seconds.	_DEFAULT_WRITE_TIMEOUT
local_name	str | None	The name sent in EHLO. Defaults to localhost.	None
starttls	bool	Whether to issue STARTTLS when advertised.	True
require_tls	bool	Whether to fail when STARTTLS is not advertised.	True
implicit_tls	bool	Whether the connection uses TLS from the first byte (SMTPS, typically port 465). When true, STARTTLS is not issued.	False
local_address	str | None	An optional local bind address, host or host:port.	None
auth_mechanisms	Sequence[str]	Preferred SASL mechanisms in order. When empty, the client prefers PLAIN then LOGIN.	()

Raises:

Type	Description
ValueError	A required argument is missing or invalid.

Initialize the client and its connection pool.

Source code in src/mxraven/client.py

def __init__(
    self,
    host: str,
    *,
    port: int = DEFAULT_ADDRESS_PORT,
    username: str,
    secret: str,
    tls: ssl.SSLContext | None = None,
    pool_size: int = _DEFAULT_POOL_SIZE,
    connect_timeout: float = _DEFAULT_CONNECT_TIMEOUT,
    read_timeout: float = _DEFAULT_READ_TIMEOUT,
    write_timeout: float = _DEFAULT_WRITE_TIMEOUT,
    local_name: str | None = None,
    starttls: bool = True,
    require_tls: bool = True,
    implicit_tls: bool = False,
    local_address: str | None = None,
    auth_mechanisms: Sequence[str] = (),
) -> None:
    """Initialize the client and its connection pool."""
    validate_options(
        host=host,
        port=port,
        username=username,
        secret=secret,
        pool_size=pool_size,
        connect_timeout=connect_timeout,
        read_timeout=read_timeout,
        write_timeout=write_timeout,
    )
    config = ConnectionConfig(
        host=host,
        port=port,
        local_name=local_name if local_name else "localhost",
        username=username,
        password=secret,
        tls_context=tls,
        starttls=starttls,
        require_tls=require_tls,
        implicit_tls=implicit_tls,
        local_address=local_address,
        auth_mechanisms=tuple(auth_mechanisms),
        connect_timeout=connect_timeout,
        read_timeout=read_timeout,
        write_timeout=write_timeout,
    )
    self._pool = Pool(config, pool_size)

__enter__

__enter__() -> Self

Return the client.

Source code in src/mxraven/client.py

def __enter__(self) -> Self:
    """Return the client."""
    return self

__exit__

__exit__(*exc_info: object) -> None

Release the pooled connections.

Source code in src/mxraven/client.py

def __exit__(self, *exc_info: object) -> None:
    """Release the pooled connections."""
    self.close()

close

close() -> None

Release the pooled connections. Safe to call more than once.

Source code in src/mxraven/client.py

def close(self) -> None:
    """Release the pooled connections. Safe to call more than once."""
    self._pool.close()

send

send(message: Message) -> Result

Submit a composed message and return the server's result.

Parameters:

Name	Type	Description	Default
message	Message	The message to submit. It may be sent more than once.	required

Returns:

Type	Description
Result	The server's result for the submission, including the mxRaven
Result	message_ref and per-recipient acceptance.

Raises:

Type	Description
TypeError	message is not a :class:Message.
ValueError	The message cannot be built.
SMTPError	The server rejected a command.
SMTPTransactionError	Every recipient was rejected; the per-recipient detail is on the exception's result.

Source code in src/mxraven/client.py

def send(self, message: Message) -> Result:
    """Submit a composed message and return the server's result.

    Args:
        message: The message to submit. It may be sent more than once.

    Returns:
        The server's result for the submission, including the mxRaven
        `message_ref` and per-recipient acceptance.

    Raises:
        TypeError: `message` is not a :class:`Message`.
        ValueError: The message cannot be built.
        SMTPError: The server rejected a command.
        SMTPTransactionError: Every recipient was rejected; the
            per-recipient detail is on the exception's `result`.
    """
    candidate: object = message
    if not isinstance(candidate, Message):
        error = "mail: message is required"
        raise TypeError(error)
    built = message.build()
    return self._pool.send(transaction_from_built(built), built.data)

send_raw

send_raw(
    envelope: Envelope,
    data: bytes | bytearray | memoryview | Iterable[bytes],
    *,
    prefer_bdat: bool = False,
) -> Result

Stream an already serialized RFC 5322 message with an explicit envelope.

The message is not parsed, so the caller is responsible for RFC 5322 correctness. Prefer this for large or pre-rendered messages.

Parameters:

Name	Type	Description	Default
envelope	Envelope	The SMTP envelope, independent of the message headers.	required
data	bytes | bytearray | memoryview | Iterable[bytes]	The complete message bytes, or an iterable of chunks that is streamed through DATA.	required
prefer_bdat	bool	Whether to use BDAT when CHUNKING is advertised. Streaming sources always use DATA.	False

Returns:

Type	Description
Result	The server's result for the submission.

Raises:

Type	Description
TypeError	envelope is not an :class:Envelope.
ValueError	The envelope has no recipients or an address is invalid.
SMTPError	The server rejected a command.
SMTPTransactionError	Every recipient was rejected.

Source code in src/mxraven/client.py

def send_raw(
    self,
    envelope: Envelope,
    data: bytes | bytearray | memoryview | Iterable[bytes],
    *,
    prefer_bdat: bool = False,
) -> Result:
    """Stream an already serialized RFC 5322 message with an explicit envelope.

    The message is not parsed, so the caller is responsible for RFC 5322
    correctness. Prefer this for large or pre-rendered messages.

    Args:
        envelope: The SMTP envelope, independent of the message headers.
        data: The complete message bytes, or an iterable of chunks that is
            streamed through ``DATA``.
        prefer_bdat: Whether to use ``BDAT`` when ``CHUNKING`` is
            advertised. Streaming sources always use ``DATA``.

    Returns:
        The server's result for the submission.

    Raises:
        TypeError: `envelope` is not an :class:`Envelope`.
        ValueError: The envelope has no recipients or an address is
            invalid.
        SMTPError: The server rejected a command.
        SMTPTransactionError: Every recipient was rejected.
    """
    candidate: object = envelope
    if not isinstance(candidate, Envelope):
        error = "mail: envelope is required"
        raise TypeError(error)
    if isinstance(data, bytes | bytearray | memoryview):
        raw = bytes(data)
        return self._pool.send(
            transaction_from_envelope(envelope, raw), raw, prefer_bdat=prefer_bdat
        )
    return self._pool.send(transaction_from_envelope(envelope, None), data)

DecodeError

DecodeError(subject: str, message: str = '')

Bases: NotificationError

A notification body that does not match the SDK's payload model.

This normally means the SDK is older than the server that produced the event. The event is terminated rather than redelivered, because redelivery can never succeed. When an error handler is configured, it receives the error and consumption continues; otherwise consumption stops and this error is raised.

Initialize the error.

Parameters:

Name	Type	Description	Default
subject	str	The notification subject the undecodable body was published to.	required
message	str	A short description of the decode failure.	''

Source code in src/mxraven/errors.py

def __init__(self, subject: str, message: str = "") -> None:
    """Initialize the error.

    Args:
        subject: The notification subject the undecodable body was
            published to.
        message: A short description of the decode failure.
    """
    self.subject = subject
    self.message = message
    detail = message.strip()
    text = f"notify: decode {subject!r}"
    if detail:
        text = f"{text}: {detail}"
    super().__init__(text)

DeliveryBy dataclass

DeliveryBy(
    seconds: int,
    mode: DeliveryByMode = NOTIFY,
    trace: bool = False,
)

An RFC 2852 DELIVERBY request.

Attributes:

Name	Type	Description
seconds	int	The delivery interval in seconds. It must be positive when mode is DeliveryByMode.RETURN; 0 is allowed with DeliveryByMode.NOTIFY.
mode	DeliveryByMode	How the server handles a message that misses the deadline.
trace	bool	Whether to request a trace of the delivery.

DeliveryByMode

Bases: StrEnum

How a server handles a message that misses its DELIVERBY deadline.

The values are the RFC 2852 wire values.

NOTIFY class-attribute instance-attribute

NOTIFY = 'N'

Request a delayed delivery notification (wire value N).

RETURN class-attribute instance-attribute

RETURN = 'R'

Request that the message be returned to the sender (wire value R).

Envelope dataclass

Envelope(
    sender: str = "",
    recipients: tuple[str, ...] = (),
    require_tls: bool = False,
    delivery_by: DeliveryBy | None = None,
)

The SMTP envelope for a raw message.

Attributes:

Name	Type	Description
sender	str	The envelope sender. An empty value requests a null reverse-path, which is appropriate for bounce messages.
recipients	tuple[str, ...]	At least one envelope recipient.
require_tls	bool	Whether to request that TLS be used on every delivery hop (RFC 8689). The session must already be encrypted and the service must advertise REQUIRETLS.
delivery_by	DeliveryBy | None	An RFC 2852 DELIVERBY request, when the service advertises DELIVERBY.

FeedbackError

FeedbackError(status_code: int, message: str = '')

Bases: MxRavenError

The feedback service returned a non-success response.

Initialize the error.

Parameters:

Name	Type	Description	Default
status_code	int	The HTTP response status.	required
message	str	The service's error field, or the HTTP status text.	''

Source code in src/mxraven/errors.py

def __init__(self, status_code: int, message: str = "") -> None:
    """Initialize the error.

    Args:
        status_code: The HTTP response status.
        message: The service's ``error`` field, or the HTTP status text.
    """
    self.status_code = status_code
    self.message = message
    detail = message.strip()
    text = f"feedback: request failed with status {status_code}"
    if detail:
        text = f"{text}: {detail}"
    super().__init__(text)

detail property

detail: str

The service's error detail, if any. Alias of :attr:message.

retryable property

retryable: bool

Whether the request may succeed if retried later (429 or 5xx).

Header dataclass

Header(name: str, value: str)

A custom message header.

Attributes:

Name	Type	Description
name	str	The header field name.
value	str	The header field value. It must not contain a line break.

InvalidSignatureError

Bases: WebhookError

A webhook request's HMAC signature did not match.

Message

Message()

A mutable, chainable builder for an email message.

Chained methods return the receiver. Set both :meth:text and :meth:html to produce a multipart/alternative body; attachments produce a multipart/mixed body. Bcc recipients are added to the envelope but not to the headers.

Example

message = (
    Message()
    .from_("Acme <noreply@acme.example>")
    .to("customer@example.com")
    .subject("Your receipt")
    .text("Thanks for your order.")
)

Create an empty message builder.

Source code in src/mxraven/message.py

def __init__(self) -> None:
    """Create an empty message builder."""
    self._from = ""
    self._null_sender = False
    self._sender = ""
    self._reply_to = ""
    self._to: list[str] = []
    self._cc: list[str] = []
    self._bcc: list[str] = []
    self._subject = ""
    self._text = ""
    self._html = ""
    self._headers: list[Header] = []
    self._attachments: list[Attachment] = []
    self._message_id = ""
    self._in_reply_to = ""
    self._references: list[str] = []
    self._date: datetime | None = None
    self._require_tls = False
    self._delivery_by: DeliveryBy | None = None

attach

attach(attachment: Attachment) -> Self

Append an attachment.

Note

The attachment data is retained by reference until the message is sent. Callers must not mutate it in the meantime.

Source code in src/mxraven/message.py

def attach(self, attachment: Attachment) -> Self:
    """Append an attachment.

    Note:
        The attachment data is retained by reference until the message is
        sent. Callers must not mutate it in the meantime.
    """
    self._attachments.append(attachment)
    return self

attach_file

attach_file(
    filename: str, data: bytes, *, content_type: str = ""
) -> Self

Append a file attachment with an optional explicit content type.

Source code in src/mxraven/message.py

def attach_file(self, filename: str, data: bytes, *, content_type: str = "") -> Self:
    """Append a file attachment with an optional explicit content type."""
    return self.attach(Attachment(filename=filename, data=data, content_type=content_type))

attach_inline

attach_inline(
    filename: str,
    content_id: str,
    data: bytes,
    *,
    content_type: str = "",
) -> Self

Append an inline attachment referenced by content_id from HTML.

Source code in src/mxraven/message.py

def attach_inline(
    self, filename: str, content_id: str, data: bytes, *, content_type: str = ""
) -> Self:
    """Append an inline attachment referenced by `content_id` from HTML."""
    return self.attach(
        Attachment(
            filename=filename,
            data=data,
            content_type=content_type,
            inline=True,
            content_id=content_id,
        )
    )

bcc

bcc(*addresses: str) -> Self

Add envelope recipients without a visible Bcc header.

Source code in src/mxraven/message.py

def bcc(self, *addresses: str) -> Self:
    """Add envelope recipients without a visible ``Bcc`` header."""
    self._bcc.extend(addresses)
    return self

build

build() -> BuiltMessage

Serialize the message into bytes and its submission envelope.

Returns:

Type	Description
BuiltMessage	The serialized message and envelope.

Raises:

Type	Description
ValueError	The message has no From address or header, has no recipients, an address is invalid, or a header is invalid.

Source code in src/mxraven/message.py

def build(self) -> BuiltMessage:
    """Serialize the message into bytes and its submission envelope.

    Returns:
        The serialized message and envelope.

    Raises:
        ValueError: The message has no ``From`` address or header, has no
            recipients, an address is invalid, or a header is invalid.
    """
    addresses = self._resolve_addresses()
    headers = self._assemble_headers(addresses)
    body_text, content_type, encoding = self._render_body()
    headers.append(("MIME-Version", "1.0"))
    headers.append(("Content-Type", content_type))
    headers.append(("Content-Transfer-Encoding", encoding))

    body = body_text.encode("utf-8")
    data = serialize_headers(headers) + b"\r\n" + body
    body_type = "8BITMIME" if any(byte > _ASCII_LIMIT for byte in body) else ""
    return BuiltMessage(
        data=data,
        sender=addresses.envelope_sender,
        recipients=tuple(mailbox_to_string(address) for address in addresses.all),
        smtp_utf8=_needs_smtp_utf8(headers, addresses),
        size=len(data),
        body_type=body_type,
        require_tls=self._require_tls,
        delivery_by=self._delivery_by,
    )

cc

cc(*addresses: str) -> Self

Add envelope and Cc header recipients.

Source code in src/mxraven/message.py

def cc(self, *addresses: str) -> Self:
    """Add envelope and ``Cc`` header recipients."""
    self._cc.extend(addresses)
    return self

date

date(when: datetime) -> Self

Set the Date header. When unset, the time of building is used.

A naive when is interpreted as UTC.

Source code in src/mxraven/message.py

def date(self, when: datetime) -> Self:
    """Set the ``Date`` header. When unset, the time of building is used.

    A naive `when` is interpreted as UTC.
    """
    self._date = when
    return self

deliver_by

deliver_by(request: DeliveryBy) -> Self

Request delivery within a bounded window (RFC 2852).

The submission service must advertise DELIVERBY, otherwise sending raises :class:mxraven.SMTPCapabilityError.

Source code in src/mxraven/message.py

def deliver_by(self, request: DeliveryBy) -> Self:
    """Request delivery within a bounded window (RFC 2852).

    The submission service must advertise ``DELIVERBY``, otherwise sending
    raises :class:`mxraven.SMTPCapabilityError`.
    """
    self._delivery_by = request
    return self

from_

from_(address: str) -> Self

Set the envelope sender and the From header.

The name is from_ because from is a Python keyword. The address may include a display name, for example Acme <noreply@acme.example>.

Source code in src/mxraven/message.py

def from_(self, address: str) -> Self:
    """Set the envelope sender and the ``From`` header.

    The name is `from_` because `from` is a Python keyword. The address may
    include a display name, for example ``Acme <noreply@acme.example>``.
    """
    self._from = address
    return self

header

header(name: str, value: str) -> Self

Append a custom header.

Source code in src/mxraven/message.py

def header(self, name: str, value: str) -> Self:
    """Append a custom header."""
    self._headers.append(Header(name, value))
    return self

html

html(body: str) -> Self

Set the HTML body.

Source code in src/mxraven/message.py

def html(self, body: str) -> Self:
    """Set the HTML body."""
    self._html = body
    return self

in_reply_to

in_reply_to(message_id: str) -> Self

Set the In-Reply-To header for threading.

Source code in src/mxraven/message.py

def in_reply_to(self, message_id: str) -> Self:
    """Set the ``In-Reply-To`` header for threading."""
    self._in_reply_to = message_id
    return self

message_id

message_id(message_id: str) -> Self

Set the Message-ID header, wrapping the value in angle brackets.

Source code in src/mxraven/message.py

def message_id(self, message_id: str) -> Self:
    """Set the ``Message-ID`` header, wrapping the value in angle brackets."""
    self._message_id = message_id
    return self

null_sender

null_sender() -> Self

Use a null reverse-path while keeping the From header.

Intended for bounce and other auto-generated messages. A From header is still required for a valid message.

Source code in src/mxraven/message.py

def null_sender(self) -> Self:
    """Use a null reverse-path while keeping the ``From`` header.

    Intended for bounce and other auto-generated messages. A ``From`` header
    is still required for a valid message.
    """
    self._null_sender = True
    return self

references

references(*message_ids: str) -> Self

Set the References header for threading.

Source code in src/mxraven/message.py

def references(self, *message_ids: str) -> Self:
    """Set the ``References`` header for threading."""
    self._references.extend(message_ids)
    return self

reply_to

reply_to(address: str) -> Self

Set the Reply-To header.

Source code in src/mxraven/message.py

def reply_to(self, address: str) -> Self:
    """Set the ``Reply-To`` header."""
    self._reply_to = address
    return self

require_tls

require_tls() -> Self

Require TLS for every hop that handles the message (RFC 8689).

The session must already be encrypted and the submission service must advertise REQUIRETLS; otherwise sending raises :class:mxraven.SMTPError before the message is transmitted.

Source code in src/mxraven/message.py

def require_tls(self) -> Self:
    """Require TLS for every hop that handles the message (RFC 8689).

    The session must already be encrypted and the submission service must
    advertise ``REQUIRETLS``; otherwise sending raises
    :class:`mxraven.SMTPError` before the message is transmitted.
    """
    self._require_tls = True
    return self

sender

sender(address: str) -> Self

Set the Sender header, required when From has several mailboxes.

Source code in src/mxraven/message.py

def sender(self, address: str) -> Self:
    """Set the ``Sender`` header, required when ``From`` has several mailboxes."""
    self._sender = address
    return self

subject

subject(subject: str) -> Self

Set the Subject header. Non-ASCII subjects are encoded.

Source code in src/mxraven/message.py

def subject(self, subject: str) -> Self:
    """Set the ``Subject`` header. Non-ASCII subjects are encoded."""
    self._subject = subject
    return self

text

text(body: str) -> Self

Set the plain-text body.

Source code in src/mxraven/message.py

def text(self, body: str) -> Self:
    """Set the plain-text body."""
    self._text = body
    return self

to

to(*addresses: str) -> Self

Add envelope and To header recipients.

Source code in src/mxraven/message.py

def to(self, *addresses: str) -> Self:
    """Add envelope and ``To`` header recipients."""
    self._to.extend(addresses)
    return self

MimeError

Bases: MxRavenError

A MIME message cannot be parsed within the parser's limits.

Malformed message content never raises; the parser returns the best interpretation it can. Only structurally unbounded input raises, so an untrusted message cannot exhaust the stack or memory.

MimeNestingError

Bases: MimeError

MIME nesting exceeded the parser's maximum depth.

MimePartsError

Bases: MimeError

A message contained more MIME entities than the parser allows.

MxRavenError

Bases: Exception

Base class for every error raised by the mxRaven SDK.

NotificationError

Bases: MxRavenError

A failure connecting to or consuming the client notification stream.

The underlying transport or JetStream failure is preserved as :attr:__cause__. Most connection failures are transient and are retried automatically; this error is only raised when an operation cannot continue.

RecipientResult dataclass

RecipientResult(
    address: str,
    accepted: bool,
    error: MxRavenError | None = None,
)

Whether a single envelope recipient was accepted.

Attributes:

Name	Type	Description
address	str	The recipient address as submitted, without angle brackets.
accepted	bool	Whether the server accepted the recipient.
error	MxRavenError | None	The rejection reason when accepted is false, otherwise None.

Result dataclass

Result(
    message_ref: str = "",
    code: int = 0,
    message: str = "",
    recipients: tuple[RecipientResult, ...] = (),
)

The outcome of a submission.

Attributes:

Name	Type	Description
message_ref	str	The mxRaven message reference parsed from the final DATA reply, or an empty string when the server did not report one. This is the durable reference to use for correlation.
code	int	The final SMTP reply code.
message	str	The final SMTP reply text. It is intended for humans and is not stable.
recipients	tuple[RecipientResult, ...]	Per-recipient acceptance, in envelope order.

SMTPCapabilityError

Bases: MxRavenError

The server does not support a feature the message requires.

SMTPConnectionError

Bases: MxRavenError

A connection could not be established or was lost.

The underlying socket or TLS failure is preserved as :attr:__cause__.

SMTPError

SMTPError(code: int, message: str, enhanced_code: str = '')

Bases: MxRavenError

An SMTP reply that rejected a command.

Inspect :attr:code or :attr:enhanced_code to make a delivery decision. :attr:message is the server's human-readable text and is not stable.

Initialize the error.

Parameters:

Name	Type	Description	Default
code	int	The three-digit SMTP reply code.	required
message	str	The server's reply text.	required
enhanced_code	str	The RFC 3463 X.Y.Z enhanced status code, when the server supplied one.	''

Source code in src/mxraven/errors.py

def __init__(self, code: int, message: str, enhanced_code: str = "") -> None:
    """Initialize the error.

    Args:
        code: The three-digit SMTP reply code.
        message: The server's reply text.
        enhanced_code: The RFC 3463 ``X.Y.Z`` enhanced status code, when the
            server supplied one.
    """
    self.code = code
    self.enhanced_code = enhanced_code
    self.message = message
    super().__init__(_format_smtp_error(code, enhanced_code, message))

permanent property

permanent: bool

Whether the failure is permanent (5xx); retrying is unlikely to help.

transient property

transient: bool

Whether the failure is transient (4xx); the message may be retried.

SMTPProtocolError

Bases: MxRavenError

The server sent a reply the client could not parse.

SMTPTimeoutError

Bases: SMTPConnectionError

An SMTP operation exceeded its deadline.

SMTPTransactionError

SMTPTransactionError(message: str, result: Result)

Bases: MxRavenError

Every envelope recipient was rejected.

The per-recipient detail is available on :attr:result.

Initialize the error.

Parameters:

Name	Type	Description	Default
message	str	A human-readable summary.	required
result	Result	The transaction result, including each recipient's outcome.	required

Source code in src/mxraven/errors.py

def __init__(self, message: str, result: Result) -> None:
    """Initialize the error.

    Args:
        message: A human-readable summary.
        result: The transaction result, including each recipient's outcome.
    """
    self.result = result
    super().__init__(message)

WebhookError

Bases: MxRavenError

Base class for webhook verification and decoding failures.

Asynchronous client

aio

Asynchronous mxRaven mail runtime.

Exposes :class:mxraven.aio.Client, the asyncio counterpart of :class:mxraven.Client, with the same semantics and error surface.

Client

Client(
    host: str,
    *,
    port: int = DEFAULT_ADDRESS_PORT,
    username: str,
    secret: str,
    tls: SSLContext | None = None,
    pool_size: int = _DEFAULT_POOL_SIZE,
    connect_timeout: float = _DEFAULT_CONNECT_TIMEOUT,
    read_timeout: float = _DEFAULT_READ_TIMEOUT,
    write_timeout: float = _DEFAULT_WRITE_TIMEOUT,
    local_name: str | None = None,
    starttls: bool = True,
    require_tls: bool = True,
    implicit_tls: bool = False,
    local_address: str | None = None,
    auth_mechanisms: Sequence[str] = (),
)

Submits mail to the mxRaven SMTP submission service using asyncio.

A client maintains a bounded pool of authenticated connections. Use it as an asynchronous context manager or call :meth:close.

Example

async with Client(host="smtp.mxraven.com", username=user, secret=secret) as client:
    result = await client.send(message)

Parameters:

Name	Type	Description	Default
host	str	The submission server host; a port must not be included.	required
port	int	The submission server port. Defaults to 587.	DEFAULT_ADDRESS_PORT
username	str	The submission API key username.	required
secret	str	The submission API key secret.	required
tls	SSLContext | None	A custom STARTTLS ssl.SSLContext.	None
pool_size	int	The maximum number of pooled connections. Defaults to 5.	_DEFAULT_POOL_SIZE
connect_timeout	float	The TCP connect timeout in seconds.	_DEFAULT_CONNECT_TIMEOUT
read_timeout	float	The per-reply read timeout in seconds.	_DEFAULT_READ_TIMEOUT
write_timeout	float	The per-write timeout in seconds.	_DEFAULT_WRITE_TIMEOUT
local_name	str | None	The name sent in EHLO. Defaults to localhost.	None
starttls	bool	Whether to issue STARTTLS when advertised.	True
require_tls	bool	Whether to fail when STARTTLS is not advertised.	True
implicit_tls	bool	Whether the connection uses TLS from the first byte (SMTPS, typically port 465). When true, STARTTLS is not issued.	False
local_address	str | None	An optional local bind address, host or host:port.	None
auth_mechanisms	Sequence[str]	Preferred SASL mechanisms in order.	()

Raises:

Type	Description
ValueError	A required argument is missing or invalid.

Initialize the client and its connection pool.

Source code in src/mxraven/aio/client.py

def __init__(
    self,
    host: str,
    *,
    port: int = DEFAULT_ADDRESS_PORT,
    username: str,
    secret: str,
    tls: ssl.SSLContext | None = None,
    pool_size: int = _DEFAULT_POOL_SIZE,
    connect_timeout: float = _DEFAULT_CONNECT_TIMEOUT,
    read_timeout: float = _DEFAULT_READ_TIMEOUT,
    write_timeout: float = _DEFAULT_WRITE_TIMEOUT,
    local_name: str | None = None,
    starttls: bool = True,
    require_tls: bool = True,
    implicit_tls: bool = False,
    local_address: str | None = None,
    auth_mechanisms: Sequence[str] = (),
) -> None:
    """Initialize the client and its connection pool."""
    validate_options(
        host=host,
        port=port,
        username=username,
        secret=secret,
        pool_size=pool_size,
        connect_timeout=connect_timeout,
        read_timeout=read_timeout,
        write_timeout=write_timeout,
    )
    config = ConnectionConfig(
        host=host,
        port=port,
        local_name=local_name if local_name else "localhost",
        username=username,
        password=secret,
        tls_context=tls,
        starttls=starttls,
        require_tls=require_tls,
        implicit_tls=implicit_tls,
        local_address=local_address,
        auth_mechanisms=tuple(auth_mechanisms),
        connect_timeout=connect_timeout,
        read_timeout=read_timeout,
        write_timeout=write_timeout,
    )
    self._pool = AsyncPool(config, pool_size)

__aenter__ async

__aenter__() -> Self

Return the client.

Source code in src/mxraven/aio/client.py

async def __aenter__(self) -> Self:
    """Return the client."""
    return self

__aexit__ async

__aexit__(*exc_info: object) -> None

Release the pooled connections.

Source code in src/mxraven/aio/client.py

async def __aexit__(self, *exc_info: object) -> None:
    """Release the pooled connections."""
    await self.close()

close async

close() -> None

Release the pooled connections. Safe to call more than once.

Source code in src/mxraven/aio/client.py

async def close(self) -> None:
    """Release the pooled connections. Safe to call more than once."""
    await self._pool.close()

send async

send(message: Message) -> Result

Submit a composed message and return the server's result.

Raises:

Type	Description
TypeError	message is not a :class:Message.
ValueError	The message cannot be built.
SMTPError	The server rejected a command.
SMTPTransactionError	Every recipient was rejected.

Source code in src/mxraven/aio/client.py

async def send(self, message: Message) -> Result:
    """Submit a composed message and return the server's result.

    Raises:
        TypeError: `message` is not a :class:`Message`.
        ValueError: The message cannot be built.
        SMTPError: The server rejected a command.
        SMTPTransactionError: Every recipient was rejected.
    """
    candidate: object = message
    if not isinstance(candidate, Message):
        error = "mail: message is required"
        raise TypeError(error)
    built = message.build()
    return await self._pool.send(transaction_from_built(built), built.data)

send_raw async

send_raw(
    envelope: Envelope,
    data: bytes
    | bytearray
    | memoryview
    | Iterable[bytes]
    | AsyncIterable[bytes],
    *,
    prefer_bdat: bool = False,
) -> Result

Stream an already serialized RFC 5322 message with an explicit envelope.

Raises:

Type	Description
TypeError	envelope is not an :class:Envelope.
ValueError	The envelope has no recipients or an address is invalid.
SMTPError	The server rejected a command.
SMTPTransactionError	Every recipient was rejected.

Source code in src/mxraven/aio/client.py

async def send_raw(
    self,
    envelope: Envelope,
    data: bytes | bytearray | memoryview | Iterable[bytes] | AsyncIterable[bytes],
    *,
    prefer_bdat: bool = False,
) -> Result:
    """Stream an already serialized RFC 5322 message with an explicit envelope.

    Raises:
        TypeError: `envelope` is not an :class:`Envelope`.
        ValueError: The envelope has no recipients or an address is
            invalid.
        SMTPError: The server rejected a command.
        SMTPTransactionError: Every recipient was rejected.
    """
    candidate: object = envelope
    if not isinstance(candidate, Envelope):
        error = "mail: envelope is required"
        raise TypeError(error)
    if isinstance(data, bytes | bytearray | memoryview):
        raw = bytes(data)
        return await self._pool.send(
            transaction_from_envelope(envelope, raw), raw, prefer_bdat=prefer_bdat
        )
    if hasattr(data, "__aiter__"):
        collected = await _collect(cast("AsyncIterable[bytes]", data))
        return await self._pool.send(
            transaction_from_envelope(envelope, collected), collected, prefer_bdat=prefer_bdat
        )
    return await self._pool.send(transaction_from_envelope(envelope, None), data)

Webhooks

webhook

mxRaven webhook verification and decoding.

Verifies the X-MxRaven-* HMAC-SHA256 signature over the canonical request string and decodes DELIVER_WEBHOOK (inbound email) and NOTIFY_WEBHOOK (SMTP or object-storage delivery status) payloads.

Example

from mxraven.webhook import Verifier, WebhookRequest

verifier = Verifier(secret=signing_secret)
request = WebhookRequest.create(method, url, headers, body)
event = verifier.verify_and_decode(request)

DeliveryStatus dataclass

DeliveryStatus(
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    status: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    source_ip: str = "",
    destination_domain: str = "",
    remote_host: str = "",
    smtp_code: int = 0,
    enhanced_status_code: str = "",
    reason: str = "",
    remote_response: str = "",
    next_retry_at_utc: int = 0,
    correlation_task_id: str = "",
)

The SMTP NOTIFY_WEBHOOK payload.

Envelope dataclass

Envelope(
    mail_from: str = "", rcpt_to: tuple[str, ...] = ()
)

The SMTP envelope of a message.

Event dataclass

Event(
    type: EventType,
    inbound_email: InboundEmail | None = None,
    delivery_status: DeliveryStatus | None = None,
    storage_status: StorageStatus | None = None,
)

A decoded webhook delivery.

Exactly one of inbound_email, delivery_status, or storage_status is set, matching type.

EventType

Bases: StrEnum

The shape of a decoded webhook payload.

HeaderField dataclass

HeaderField(name: str, value: str)

One message header occurrence.

InboundEmail dataclass

InboundEmail(
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    routing_decision: RoutingDecision | None = None,
    verdicts: Verdicts | None = None,
    envelope: Envelope = Envelope(),
    message: MessageSummary = MessageSummary(),
    headers: tuple[HeaderField, ...] = (),
    raw_email: RawEmail = RawEmail(),
)

The DELIVER_WEBHOOK payload: a complete inbound message.

InvalidSignatureError

Bases: WebhookError

A webhook request's HMAC signature did not match.

MessageSummary dataclass

MessageSummary(
    subject: str = "",
    from_: tuple[str, ...] = (),
    to: tuple[str, ...] = (),
    cc: tuple[str, ...] = (),
    message_id: str = "",
    date: str = "",
)

A summary of the parsed message headers.

RawEmail dataclass

RawEmail(
    url: str = "",
    token_type: str = "",
    access_token: str = "",
    expires_at_utc: int = 0,
    size_bytes: int = 0,
    sha256_hex: str = "",
    content_type: str = "",
)

Time-limited access to the raw RFC 822 message.

fetch

fetch(
    *,
    timeout: float = 60.0,
    transport: SyncTransport | None = None,
) -> bytes

Download and verify the raw message.

Parameters:

Name	Type	Description	Default
timeout	float	The request timeout in seconds.	60.0
transport	SyncTransport | None	An optional transport to use instead of the default.	None

Returns:

Type	Description
bytes	The verified raw message bytes.

Raises:

Type	Description
WebhookError	The URL or token is missing, the response is not 200, the size does not match, or the SHA-256 digest does not match.

Note

The raw-email access token is a secret; do not log it.

Source code in src/mxraven/webhook/payload.py

def fetch(self, *, timeout: float = 60.0, transport: SyncTransport | None = None) -> bytes:
    """Download and verify the raw message.

    Args:
        timeout: The request timeout in seconds.
        transport: An optional transport to use instead of the default.

    Returns:
        The verified raw message bytes.

    Raises:
        WebhookError: The URL or token is missing, the response is not 200,
            the size does not match, or the SHA-256 digest does not match.

    Note:
        The raw-email access token is a secret; do not log it.
    """
    chosen = transport or DEFAULT_SYNC_TRANSPORT
    _require_raw_fields(self)
    response = chosen.request(
        "GET",
        self.url.strip(),
        headers={"Authorization": _authorization(self)},
        body=None,
        timeout=timeout,
    )
    return _verify_raw(self, response.status, response.reason, response.body)

fetch_async async

fetch_async(
    *,
    timeout: float = 60.0,
    transport: AsyncTransport | None = None,
) -> bytes

Asynchronously download and verify the raw message.

Parameters:

Name	Type	Description	Default
timeout	float	The request timeout in seconds.	60.0
transport	AsyncTransport | None	An optional transport to use instead of the default.	None

Returns:

Type	Description
bytes	The verified raw message bytes.

Raises:

Type	Description
WebhookError	The URL or token is missing, the response is not 200, the size does not match, or the SHA-256 digest does not match.

Note

The raw-email access token is a secret; do not log it.

Source code in src/mxraven/webhook/payload.py

async def fetch_async(
    self, *, timeout: float = 60.0, transport: AsyncTransport | None = None
) -> bytes:
    """Asynchronously download and verify the raw message.

    Args:
        timeout: The request timeout in seconds.
        transport: An optional transport to use instead of the default.

    Returns:
        The verified raw message bytes.

    Raises:
        WebhookError: The URL or token is missing, the response is not 200,
            the size does not match, or the SHA-256 digest does not match.

    Note:
        The raw-email access token is a secret; do not log it.
    """
    chosen = transport or DEFAULT_ASYNC_TRANSPORT
    _require_raw_fields(self)
    response = await chosen.request(
        "GET",
        self.url.strip(),
        headers={"Authorization": _authorization(self)},
        body=None,
        timeout=timeout,
    )
    return _verify_raw(self, response.status, response.reason, response.body)

RoutingDecision dataclass

RoutingDecision(
    terminal_action: str = "",
    matched_rule_id: str = "",
    used_listener_default: bool = False,
)

The final routing outcome for an inbound message.

Attributes:

Name	Type	Description
terminal_action	str	The final terminal action. Compare it against the :class:TerminalAction members; unknown future values are preserved as their raw string.
matched_rule_id	str	The rule that selected the action, when one matched.
used_listener_default	bool	Whether the listener default was used.

StatusOutcome

Bases: StrEnum

The lifecycle state reported by a delivery status webhook.

The set is open on the wire: a status this SDK does not know is still decoded and preserved as its raw string.

StorageStatus dataclass

StorageStatus(
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    status: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    storage_ref: str = "",
    bucket_name: str = "",
    object_key: str = "",
    endpoint_host: str = "",
    status_code: int = 0,
    error_code: str = "",
    message: str = "",
    next_retry_at_utc: int = 0,
)

The object-storage NOTIFY_WEBHOOK payload.

TerminalAction

Bases: StrEnum

The final routing action recorded for an inbound message.

Verdicts dataclass

Verdicts(
    action: str = "",
    score: float = 0.0,
    required_score: float = 0.0,
    is_spam: bool = False,
    has_malware: bool = False,
    malware_names: tuple[str, ...] = (),
    is_skipped: bool = False,
    error: str = "",
)

Spam and malware scan results.

Verifier

Verifier(
    *,
    secret: str = "",
    keys: Mapping[str, str] | None = None,
    tolerance: float = _DEFAULT_TOLERANCE,
    max_body_bytes: int = _DEFAULT_MAX_BODY,
    clock: Callable[[], float] | None = None,
)

Verifies the signature of an mxRaven webhook request.

A verifier is safe for concurrent use once constructed. Configure it with a signing secret, with per-key secrets for rotation, or both.

Example

verifier = Verifier(secret=signing_secret)
event = verifier.verify_and_decode(request)

Parameters:

Name	Type	Description	Default
secret	str	The signing secret, used as literal bytes and never decoded.	''
keys	Mapping[str, str] | None	Per-key secrets for rotation, keyed by signing key ID.	None
tolerance	float	The maximum accepted clock skew in seconds; 0 disables the timestamp check. Defaults to five minutes.	_DEFAULT_TOLERANCE
max_body_bytes	int	The maximum body size in bytes. Defaults to 1 MiB.	_DEFAULT_MAX_BODY
clock	Callable[[], float] | None	A callable returning the current Unix time; used for testing.	None

Raises:

Type	Description
ValueError	No secret was supplied or an option is invalid.

Initialize the verifier.

Source code in src/mxraven/webhook/verifier.py

def __init__(
    self,
    *,
    secret: str = "",
    keys: Mapping[str, str] | None = None,
    tolerance: float = _DEFAULT_TOLERANCE,
    max_body_bytes: int = _DEFAULT_MAX_BODY,
    clock: Callable[[], float] | None = None,
) -> None:
    """Initialize the verifier."""
    self._secret = secret
    self._keys: dict[str, str] = {}
    for kid, value in (keys or {}).items():
        if not kid.strip():
            message = "webhook: signing key ID must not be empty"
            raise ValueError(message)
        if value == "":
            message = "webhook: signing secret must not be empty"
            raise ValueError(message)
        self._keys[kid] = value
    if secret == "" and not self._keys:
        message = "webhook: a signing secret is required"
        raise ValueError(message)
    if tolerance < 0:
        message = "webhook: tolerance must not be negative"
        raise ValueError(message)
    if max_body_bytes <= 0:
        message = "webhook: maximum body size must be positive"
        raise ValueError(message)
    self._tolerance = tolerance
    self._max_body_bytes = max_body_bytes
    self._clock = clock if clock is not None else time.time

verify

verify(request: WebhookRequest) -> None

Verify a request's signature.

Parameters:

Name	Type	Description	Default
request	WebhookRequest	The inbound request.	required

Raises:

Type	Description
InvalidSignatureError	The signature does not match.
WebhookError	A required header is missing, the timestamp is stale, or the body exceeds the configured limit.

Source code in src/mxraven/webhook/verifier.py

def verify(self, request: WebhookRequest) -> None:
    """Verify a request's signature.

    Args:
        request: The inbound request.

    Raises:
        InvalidSignatureError: The signature does not match.
        WebhookError: A required header is missing, the timestamp is stale,
            or the body exceeds the configured limit.
    """
    self._check(request)

verify_and_decode

verify_and_decode(request: WebhookRequest) -> Event

Verify a request and decode its payload.

Parameters:

Name	Type	Description	Default
request	WebhookRequest	The inbound request.	required

Returns:

Type	Description
Event	The decoded event.

Raises:

Type	Description
InvalidSignatureError	The signature does not match.
WebhookError	Verification or decoding failed.

Source code in src/mxraven/webhook/verifier.py

def verify_and_decode(self, request: WebhookRequest) -> Event:
    """Verify a request and decode its payload.

    Args:
        request: The inbound request.

    Returns:
        The decoded event.

    Raises:
        InvalidSignatureError: The signature does not match.
        WebhookError: Verification or decoding failed.
    """
    self._check(request)
    return decode(request.body)

WebhookError

Bases: MxRavenError

Base class for webhook verification and decoding failures.

WebhookRequest dataclass

WebhookRequest(
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes = b"",
)

The parts of an inbound HTTP request the signature covers.

Header names are stored lowercased so lookups are case-insensitive.

Attributes:

Name	Type	Description
method	str	The HTTP method.
url	str	The public URL the caller used, including scheme and host.
headers	Mapping[str, str]	The request headers keyed by lowercase name.
body	bytes	The exact raw request body bytes.

create classmethod

create(
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes | bytearray | memoryview = b"",
) -> WebhookRequest

Build a request, normalizing header names and body type.

Source code in src/mxraven/webhook/request.py

@classmethod
def create(
    cls,
    method: str,
    url: str,
    headers: Mapping[str, str],
    body: bytes | bytearray | memoryview = b"",
) -> WebhookRequest:
    """Build a request, normalizing header names and body type."""
    normalized = {str(name).lower(): str(value) for name, value in headers.items()}
    return cls(method=method.upper(), url=url, headers=normalized, body=bytes(body))

from_asgi classmethod

from_asgi(
    scope: Mapping[str, object], body: bytes
) -> WebhookRequest

Build a request from an ASGI scope and pre-read body.

Source code in src/mxraven/webhook/request.py

@classmethod
def from_asgi(cls, scope: Mapping[str, object], body: bytes) -> WebhookRequest:
    """Build a request from an ASGI ``scope`` and pre-read body."""
    method = str(scope.get("method", ""))
    scheme = str(scope.get("scheme", "https"))
    raw_headers = scope.get("headers")
    headers: dict[str, str] = {}
    if isinstance(raw_headers, list):
        for name, value in raw_headers:
            headers[bytes(name).decode("latin-1").lower()] = bytes(value).decode("latin-1")
    host = headers.get("host", "")
    path = str(scope.get("path", "/"))
    query_bytes = scope.get("query_string", b"")
    query = bytes(query_bytes).decode("latin-1") if isinstance(query_bytes, bytes) else ""
    url = f"{scheme}://{host}{path}" + (f"?{query}" if query else "")
    return cls.create(method, url, headers, body)

from_wsgi classmethod

from_wsgi(environ: Mapping[str, object]) -> WebhookRequest

Build a request from a WSGI environ.

Note

The signature covers the exact raw body and the public URL, so read the body here rather than letting a framework parse it first.

Source code in src/mxraven/webhook/request.py

@classmethod
def from_wsgi(cls, environ: Mapping[str, object]) -> WebhookRequest:
    """Build a request from a WSGI ``environ``.

    Note:
        The signature covers the exact raw body and the public URL, so read
        the body here rather than letting a framework parse it first.
    """
    method = str(environ.get("REQUEST_METHOD", ""))
    scheme = str(environ.get("wsgi.url_scheme", "https"))
    host = str(environ.get("HTTP_HOST") or environ.get("SERVER_NAME") or "")
    path = str(environ.get("PATH_INFO", ""))
    query = str(environ.get("QUERY_STRING", ""))
    url = f"{scheme}://{host}{path}" + (f"?{query}" if query else "")
    headers = {
        str(key)[5:].replace("_", "-").lower(): str(value)
        for key, value in environ.items()
        if str(key).startswith("HTTP_")
    }
    content_type = environ.get("CONTENT_TYPE")
    if isinstance(content_type, str) and content_type:
        headers["content-type"] = content_type
    length_value = environ.get("CONTENT_LENGTH")
    length = int(str(length_value)) if length_value else 0
    stream = environ.get("wsgi.input")
    body = stream.read(length) if length and hasattr(stream, "read") else b""
    return cls.create(method, url, headers, body)

header

header(name: str) -> str

Return a header value, or an empty string when absent.

Source code in src/mxraven/webhook/request.py

def header(self, name: str) -> str:
    """Return a header value, or an empty string when absent."""
    return self.headers.get(name.lower(), "")

MIME parsing

mime

Inbound email parsing for mxRaven deliveries.

mxRaven webhook payloads describe inbound mail as JSON, and the full message remains available as raw RFC 5322 bytes behind the payload's raw_email reference. :func:parse_email turns those bytes into headers, addresses, bodies, and decoded attachments; :func:fetch_and_parse_raw_email downloads, verifies, and parses them in one step.

Example

from mxraven.mime import fetch_and_parse_raw_email
from mxraven.webhook import Verifier, EventType

verifier = Verifier(secret=signing_secret)
event = verifier.verify_and_decode(request)

if event.type is EventType.INBOUND_EMAIL and event.inbound_email is not None:
    email = fetch_and_parse_raw_email(event.inbound_email.raw_email)
    print(email.subject, len(email.attachments))

The parser is defensive: malformed messages yield the best available interpretation instead of an error, and unknown charsets fall back to UTF-8 and then Windows-1252. It understands multipart/mixed, multipart/alternative, and multipart/related, decodes base64 and quoted-printable transfer encodings, decodes RFC 2047 encoded words, and reassembles RFC 2231 parameters. message/rfc822 parts are surfaced as attachments rather than parsed recursively. Only unbounded input -- very deep nesting or thousands of parts -- is rejected.

MimeError

Bases: MxRavenError

A MIME message cannot be parsed within the parser's limits.

Malformed message content never raises; the parser returns the best interpretation it can. Only structurally unbounded input raises, so an untrusted message cannot exhaust the stack or memory.

MimeNestingError

Bases: MimeError

MIME nesting exceeded the parser's maximum depth.

MimePartsError

Bases: MimeError

A message contained more MIME entities than the parser allows.

ParsedAddress dataclass

ParsedAddress(address: str, name: str = '')

One mailbox from a parsed address header.

Attributes:

Name	Type	Description
address	str	The bare local@domain form.
name	str	The display name, with RFC 2047 encoded words decoded, when one was present.

ParsedAttachment dataclass

ParsedAttachment(
    content_type: str,
    content: bytes,
    filename: str = "",
    content_id: str = "",
    inline: bool = False,
)

One non-body MIME part.

Attributes:

Name	Type	Description
content_type	str	The media type, for example application/pdf.
content	bytes	The decoded attachment bytes. It is a fresh copy that the caller owns and may mutate.
filename	str	The decoded filename, when one was present.
content_id	str	The Content-ID value without angle brackets, for inline parts.
inline	bool	Whether the part is marked inline or carries a Content-ID.

ParsedEmail dataclass

ParsedEmail(
    headers: tuple[ParsedHeader, ...] = (),
    subject: str = "",
    from_: tuple[ParsedAddress, ...] = (),
    to: tuple[ParsedAddress, ...] = (),
    cc: tuple[ParsedAddress, ...] = (),
    bcc: tuple[ParsedAddress, ...] = (),
    reply_to: tuple[ParsedAddress, ...] = (),
    message_id: str = "",
    date: datetime | None = None,
    text: str = "",
    html: str = "",
    attachments: tuple[ParsedAttachment, ...] = (),
)

A parsed inbound RFC 5322 / MIME message.

An empty value means the corresponding header or body was absent.

Attributes:

Name	Type	Description
headers	tuple[ParsedHeader, ...]	Every header field, in order, with continuation lines unfolded. Values are not decoded.
subject	str	The decoded Subject, when the header was present.
from_	tuple[ParsedAddress, ...]	The From addresses. It is from_ because from is a Python keyword.
to	tuple[ParsedAddress, ...]	The To addresses.
cc	tuple[ParsedAddress, ...]	The Cc addresses.
bcc	tuple[ParsedAddress, ...]	The Bcc addresses.
reply_to	tuple[ParsedAddress, ...]	The Reply-To addresses.
message_id	str	The Message-ID without angle brackets, when present.
date	datetime | None	The parsed Date, when present and parseable.
text	str	The decoded first text/plain body, when the message has one.
html	str	The decoded first text/html body, when the message has one.
attachments	tuple[ParsedAttachment, ...]	Every non-body part, with transfer encoding and filename decoding applied. message/rfc822 parts appear here rather than being parsed recursively.

ParsedHeader dataclass

ParsedHeader(name: str, value: str)

One header field from a parsed message.

Attributes:

Name	Type	Description
name	str	The field name as it appeared in the message.
value	str	The unfolded field value. Encoded words are not decoded; use :attr:ParsedEmail.subject or an address field for decoded text.

fetch_and_parse_raw_email

fetch_and_parse_raw_email(
    raw: RawEmail,
    *,
    timeout: float = _DEFAULT_FETCH_TIMEOUT,
    transport: SyncTransport | None = None,
) -> ParsedEmail

Download the raw message an inbound-email payload references and parse it.

The download verifies the declared size and SHA-256 digest and is bounded exactly like :meth:mxraven.webhook.RawEmail.fetch; the payload's short-lived bearer token is sent in the Authorization header and must not be logged.

Parameters:

Name	Type	Description	Default
raw	RawEmail	The raw_email object from an inbound-email payload.	required
timeout	float	The request timeout in seconds.	_DEFAULT_FETCH_TIMEOUT
transport	SyncTransport | None	An optional transport to use instead of the default.	None

Returns:

Type	Description
ParsedEmail	The parsed message. Attachment content is a fresh copy.

Raises:

Type	Description
WebhookError	The download failed or the size or digest did not match.
MimeNestingError	The message nests deeper than the parser allows.
MimePartsError	The message has more MIME entities than the parser allows.

Source code in src/mxraven/mime/fetch.py

def fetch_and_parse_raw_email(
    raw: RawEmail,
    *,
    timeout: float = _DEFAULT_FETCH_TIMEOUT,
    transport: SyncTransport | None = None,
) -> ParsedEmail:
    """Download the raw message an inbound-email payload references and parse it.

    The download verifies the declared size and SHA-256 digest and is bounded
    exactly like :meth:`mxraven.webhook.RawEmail.fetch`; the payload's
    short-lived bearer token is sent in the ``Authorization`` header and must
    not be logged.

    Args:
        raw: The ``raw_email`` object from an inbound-email payload.
        timeout: The request timeout in seconds.
        transport: An optional transport to use instead of the default.

    Returns:
        The parsed message. Attachment content is a fresh copy.

    Raises:
        WebhookError: The download failed or the size or digest did not match.
        MimeNestingError: The message nests deeper than the parser allows.
        MimePartsError: The message has more MIME entities than the parser
            allows.
    """
    return parse_email(raw.fetch(timeout=timeout, transport=transport))

parse_email

parse_email(
    data: bytes | bytearray | memoryview,
) -> ParsedEmail

Parse an inbound RFC 5322 / MIME message.

The parser never validates the message against the grammar and never raises on malformed content; it extracts what it can. Multipart mixed, alternative, and related structures are traversed, base64 and quoted-printable transfer encodings are decoded, RFC 2047 encoded words in structured headers are decoded, and RFC 2231 parameters are reassembled. message/rfc822 parts are surfaced as attachments rather than parsed recursively.

Parameters:

Name	Type	Description	Default
data	bytes | bytearray | memoryview	The raw message bytes.	required

Returns:

Type	Description
ParsedEmail	The parsed message. Attachment content is a fresh copy.

Raises:

Type	Description
MimeNestingError	The message nests deeper than the parser allows.
MimePartsError	The message has more MIME entities than the parser allows.

Example

from mxraven.mime import parse_email

email = parse_email(raw_bytes)
print(email.subject, len(email.attachments))

Source code in src/mxraven/mime/parse.py

def parse_email(data: bytes | bytearray | memoryview) -> ParsedEmail:
    """Parse an inbound RFC 5322 / MIME message.

    The parser never validates the message against the grammar and never raises
    on malformed content; it extracts what it can. Multipart ``mixed``,
    ``alternative``, and ``related`` structures are traversed, ``base64`` and
    ``quoted-printable`` transfer encodings are decoded, RFC 2047 encoded words
    in structured headers are decoded, and RFC 2231 parameters are reassembled.
    ``message/rfc822`` parts are surfaced as attachments rather than parsed
    recursively.

    Args:
        data: The raw message bytes.

    Returns:
        The parsed message. Attachment content is a fresh copy.

    Raises:
        MimeNestingError: The message nests deeper than the parser allows.
        MimePartsError: The message has more MIME entities than the parser
            allows.

    Example:
        ```python
        from mxraven.mime import parse_email

        email = parse_email(raw_bytes)
        print(email.subject, len(email.attachments))
        ```
    """
    payload = bytes(data)
    state = _WalkState()
    root = _parse_entity(payload) if payload else None
    if root is not None:
        _walk_entity(root, 0, state)
    headers = root.headers if root is not None else ()

    return ParsedEmail(
        headers=headers,
        subject=_decode_optional_header(headers, "subject"),
        from_=_header_addresses(headers, "from"),
        to=_header_addresses(headers, "to"),
        cc=_header_addresses(headers, "cc"),
        bcc=_header_addresses(headers, "bcc"),
        reply_to=_header_addresses(headers, "reply-to"),
        message_id=_message_id(headers),
        date=_date_header(headers),
        text=state.text if state.text is not None else "",
        html=state.html if state.html is not None else "",
        attachments=tuple(state.attachments),
    )

Feedback

feedback

mxRaven recipient feedback.

Teaches the spam filter with tenant-supplied raw messages and performs RFC 8058 one-click unsubscribes.

AsyncClient

AsyncClient(
    base_url: str,
    *,
    username: str = "",
    secret: str = "",
    timeout: float = _DEFAULT_TIMEOUT,
    transport: AsyncTransport | None = None,
)

Calls the mxRaven feedback service using asyncio.

Parameters:

Name	Type	Description	Default
base_url	str	The feedback service base URL.	required
username	str	The submission API key username, used for learning.	''
secret	str	The submission API key secret, used for learning.	''
timeout	float	The request timeout in seconds. Defaults to 30.	_DEFAULT_TIMEOUT
transport	AsyncTransport | None	An optional asynchronous HTTP transport.	None

Raises:

Type	Description
ValueError	base_url is empty.

Initialize the client.

Source code in src/mxraven/feedback/client.py

def __init__(
    self,
    base_url: str,
    *,
    username: str = "",
    secret: str = "",
    timeout: float = _DEFAULT_TIMEOUT,
    transport: AsyncTransport | None = None,
) -> None:
    """Initialize the client."""
    _validate_options(base_url, username, secret, timeout)
    self._base_url = base_url.rstrip("/")
    self._username = username
    self._secret = secret
    self._timeout = timeout
    self._transport: AsyncTransport = transport or DEFAULT_ASYNC_TRANSPORT

learn async

learn(
    disposition: Disposition,
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Submit one training example asynchronously.

Source code in src/mxraven/feedback/client.py

async def learn(
    self, disposition: Disposition, raw_mime: bytes | bytearray | memoryview
) -> LearningResult:
    """Submit one training example asynchronously."""
    if not self._username or not self._secret:
        message = "feedback: credentials are required for learning"
        raise ValueError(message)
    resolved = _require_disposition(disposition)
    endpoint = f"{self._base_url}/v1/feedback/learn/{resolved}"
    headers = {
        "Content-Type": _LEARN_CONTENT_TYPE,
        "Authorization": _basic_auth(self._username, self._secret),
    }
    try:
        response = await self._transport.request(
            "POST",
            endpoint,
            headers=headers,
            body=bytes(raw_mime),
            timeout=self._timeout,
        )
    except Exception as error:
        message = "feedback: submit learning request"
        raise FeedbackError(0, message) from error
    return _parse_learning(response)

learn_ham async

learn_ham(
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Teach the spam filter that raw_mime is not spam.

Source code in src/mxraven/feedback/client.py

async def learn_ham(self, raw_mime: bytes | bytearray | memoryview) -> LearningResult:
    """Teach the spam filter that `raw_mime` is not spam."""
    return await self.learn(Disposition.HAM, raw_mime)

learn_spam async

learn_spam(
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Teach the spam filter that raw_mime is spam.

Source code in src/mxraven/feedback/client.py

async def learn_spam(self, raw_mime: bytes | bytearray | memoryview) -> LearningResult:
    """Teach the spam filter that `raw_mime` is spam."""
    return await self.learn(Disposition.SPAM, raw_mime)

unsubscribe async

unsubscribe(token: str) -> None

Perform an RFC 8058 one-click unsubscribe for token.

Source code in src/mxraven/feedback/client.py

async def unsubscribe(self, token: str) -> None:
    """Perform an RFC 8058 one-click unsubscribe for `token`."""
    value = token.strip()
    if value == "":
        message = "feedback: unsubscribe token is empty"
        raise ValueError(message)
    endpoint = f"{self._base_url}/v1/feedback/unsubscribe/{quote(value, safe='')}"
    try:
        response = await self._transport.request(
            "POST",
            endpoint,
            headers={"Content-Type": _UNSUBSCRIBE_CONTENT_TYPE},
            body=_UNSUBSCRIBE_BODY,
            timeout=self._timeout,
        )
    except Exception as error:
        message = "feedback: submit unsubscribe request"
        raise FeedbackError(0, message) from error
    _require_success(response)

Client

Client(
    base_url: str,
    *,
    username: str = "",
    secret: str = "",
    timeout: float = _DEFAULT_TIMEOUT,
    transport: SyncTransport | None = None,
)

Calls the mxRaven feedback service synchronously.

A client is safe for concurrent use.

Parameters:

Name	Type	Description	Default
base_url	str	The feedback service base URL, for example https://feedback.mxraven.com.	required
username	str	The submission API key username, used for learning.	''
secret	str	The submission API key secret, used for learning.	''
timeout	float	The request timeout in seconds. Defaults to 30.	_DEFAULT_TIMEOUT
transport	SyncTransport | None	An optional HTTP transport to use instead of the default.	None

Raises:

Type	Description
ValueError	base_url is empty.

Initialize the client.

Source code in src/mxraven/feedback/client.py

def __init__(
    self,
    base_url: str,
    *,
    username: str = "",
    secret: str = "",
    timeout: float = _DEFAULT_TIMEOUT,
    transport: SyncTransport | None = None,
) -> None:
    """Initialize the client."""
    _validate_options(base_url, username, secret, timeout)
    self._base_url = base_url.rstrip("/")
    self._username = username
    self._secret = secret
    self._timeout = timeout
    self._transport: SyncTransport = transport or DEFAULT_SYNC_TRANSPORT

learn

learn(
    disposition: Disposition,
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Submit one training example.

Parameters:

Name	Type	Description	Default
disposition	Disposition	The training label.	required
raw_mime	bytes | bytearray | memoryview	The exact raw RFC 822 bytes mxRaven processed; the service matches them against stored evidence by SHA-256.	required

Returns:

Type	Description
LearningResult	The service's learning result.

Raises:

Type	Description
ValueError	Credentials are missing or the disposition is invalid.
FeedbackError	The service returned a non-success response.

Source code in src/mxraven/feedback/client.py

def learn(
    self, disposition: Disposition, raw_mime: bytes | bytearray | memoryview
) -> LearningResult:
    """Submit one training example.

    Args:
        disposition: The training label.
        raw_mime: The exact raw RFC 822 bytes mxRaven processed; the
            service matches them against stored evidence by SHA-256.

    Returns:
        The service's learning result.

    Raises:
        ValueError: Credentials are missing or the disposition is invalid.
        FeedbackError: The service returned a non-success response.
    """
    if not self._username or not self._secret:
        message = "feedback: credentials are required for learning"
        raise ValueError(message)
    resolved = _require_disposition(disposition)
    endpoint = f"{self._base_url}/v1/feedback/learn/{resolved}"
    headers = {
        "Content-Type": _LEARN_CONTENT_TYPE,
        "Authorization": _basic_auth(self._username, self._secret),
    }
    try:
        response = self._transport.request(
            "POST",
            endpoint,
            headers=headers,
            body=bytes(raw_mime),
            timeout=self._timeout,
        )
    except Exception as error:
        message = "feedback: submit learning request"
        raise FeedbackError(0, message) from error
    return _parse_learning(response)

learn_ham

learn_ham(
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Teach the spam filter that raw_mime is not spam.

Source code in src/mxraven/feedback/client.py

def learn_ham(self, raw_mime: bytes | bytearray | memoryview) -> LearningResult:
    """Teach the spam filter that `raw_mime` is not spam."""
    return self.learn(Disposition.HAM, raw_mime)

learn_spam

learn_spam(
    raw_mime: bytes | bytearray | memoryview,
) -> LearningResult

Teach the spam filter that raw_mime is spam.

Source code in src/mxraven/feedback/client.py

def learn_spam(self, raw_mime: bytes | bytearray | memoryview) -> LearningResult:
    """Teach the spam filter that `raw_mime` is spam."""
    return self.learn(Disposition.SPAM, raw_mime)

unsubscribe

unsubscribe(token: str) -> None

Perform an RFC 8058 one-click unsubscribe for token.

Parameters:

Name	Type	Description	Default
token	str	The unsubscribe token from the List-Unsubscribe URL.	required

Raises:

Type	Description
ValueError	The token is empty.
FeedbackError	The service returned a non-success response.

Source code in src/mxraven/feedback/client.py

def unsubscribe(self, token: str) -> None:
    """Perform an RFC 8058 one-click unsubscribe for `token`.

    Args:
        token: The unsubscribe token from the ``List-Unsubscribe`` URL.

    Raises:
        ValueError: The token is empty.
        FeedbackError: The service returned a non-success response.
    """
    value = token.strip()
    if value == "":
        message = "feedback: unsubscribe token is empty"
        raise ValueError(message)
    endpoint = f"{self._base_url}/v1/feedback/unsubscribe/{quote(value, safe='')}"
    try:
        response = self._transport.request(
            "POST",
            endpoint,
            headers={"Content-Type": _UNSUBSCRIBE_CONTENT_TYPE},
            body=_UNSUBSCRIBE_BODY,
            timeout=self._timeout,
        )
    except Exception as error:
        message = "feedback: submit unsubscribe request"
        raise FeedbackError(0, message) from error
    _require_success(response)

Disposition

Bases: StrEnum

The training label for a message.

FeedbackError

FeedbackError(status_code: int, message: str = '')

Bases: MxRavenError

The feedback service returned a non-success response.

Initialize the error.

Parameters:

Name	Type	Description	Default
status_code	int	The HTTP response status.	required
message	str	The service's error field, or the HTTP status text.	''

Source code in src/mxraven/errors.py

def __init__(self, status_code: int, message: str = "") -> None:
    """Initialize the error.

    Args:
        status_code: The HTTP response status.
        message: The service's ``error`` field, or the HTTP status text.
    """
    self.status_code = status_code
    self.message = message
    detail = message.strip()
    text = f"feedback: request failed with status {status_code}"
    if detail:
        text = f"{text}: {detail}"
    super().__init__(text)

detail property

detail: str

The service's error detail, if any. Alias of :attr:message.

retryable property

retryable: bool

Whether the request may succeed if retried later (429 or 5xx).

LearningResult dataclass

LearningResult(
    status: str = "",
    disposition: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    matched_hash_kind: str = "",
)

The outcome of a successful learning request.

Attributes:

Name	Type	Description
status	str	The service status, normally learned.
disposition	str	The training label that was applied.
tenant_id	str	The tenant that owns the matched message.
listener_id	str	The listener that processed the matched message.
matched_hash_kind	str	Which stored hash matched the submitted bytes.

Notifications

notify

mxRaven client notification events.

Consumes the durable mail events a listener publishes over NATS JetStream, using the same API key as SMTP submission. The transport requires the optional notify extra; install mxraven[notify].

Example

from mxraven.notify import NotificationClient

with NotificationClient.connect(
    url="wss://nats.mxraven.com",
    username="mxr_tx_ab12cd34ef56",
    secret=api_key_secret,
    listener=listener_id,
) as client:
    for delivery in client.events():
        if delivery.event.kind == "accepted":
            handle(delivery.event.accepted)
        delivery.ack()
        break

NotificationHandler module-attribute

NotificationHandler: TypeAlias = (
    "Callable[[NotificationEvent], NotificationResult]"
)

Processes one notification event synchronously.

Returning None acknowledges the event. Raising leaves it unacknowledged so the server redelivers it with exponential backoff. Returning :data:STOP_CONSUMING acknowledges the event and stops consumption.

NotificationResult module-attribute

NotificationResult: TypeAlias = 'StopConsuming | None'

What a notification handler returns: None to acknowledge, or :data:STOP_CONSUMING to acknowledge and stop.

STOP_CONSUMING module-attribute

STOP_CONSUMING: Final[StopConsuming] = StopConsuming()

Return this from a handler to acknowledge the current event and stop.

AttemptChannel

Bases: StrEnum

The egress worker that produced an :class:AttemptEvent.

The set is open on the wire: a channel this SDK does not know is preserved as a raw string on :attr:AttemptEvent.channel, which still compares equal to the matching member.

AttemptEvent dataclass

AttemptEvent(
    channel: str = "",
    status: str = "",
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    correlation_id: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    source_ip: str = "",
    destination_domain: str = "",
    remote_host: str = "",
    smtp_code: int = 0,
    enhanced_status_code: str = "",
    reason: str = "",
    remote_response: str = "",
    correlation_task_id: str = "",
    storage_ref: str = "",
    bucket_name: str = "",
    object_key: str = "",
    endpoint_host: str = "",
    error_code: str = "",
    status_code: int = 0,
    message: str = "",
    next_retry_at_utc: int = 0,
    retry_after_ms: int = 0,
    target_url_host: str = "",
    webhook_ref: str = "",
    signing_key_kid: str = "",
    raw_token_ttl_seconds: int = 0,
)

One delivery attempt for an accepted message.

:attr:channel names the egress worker that produced the attempt; the remaining fields describe that worker's protocol and default when they do not apply. A rule that chains side-effect actions reports each action it triggers, so one message can produce more than one attempt event. Every attempt carries a :attr:correlation_id that points at the task_id of the accepted event for the message it stems from.

Attributes:

Name	Type	Description
event_type	str	The producer's event-type marker when one is set. SMTP attempt events omit it.
channel	str	The egress worker that produced the attempt. Compare it against :class:AttemptChannel; unknown future values are preserved as their raw string.
task_id	str	The delivery task. For a submission listener it equals the accepted event's task_id.
tenant_id	str	The owning tenant.
listener_id	str	The source listener.
correlation_id	str	The task_id of the accepted event for the message, when one was recorded.
status	str	The attempt outcome, one of :class:mxraven.webhook.StatusOutcome.
attempt	int	The delivery attempt number, starting at 1.
accepted_at_utc	int	The Unix time, in seconds, when mxRaven accepted the message. Zero when the producer does not report it.
occurred_at_utc	int	The Unix time, in seconds, when the attempt was built.
source_ip	str	The SMTP egress source address, when known.
destination_domain	str	The recipient domain for an SMTP attempt.
remote_host	str	The remote MTA hostname, when known.
smtp_code	int	The remote SMTP reply code.
enhanced_status_code	str	The RFC 3463 enhanced status code.
reason	str	The bounded, machine-readable delivery classification.
remote_response	str	The remote reply text, truncated by the server.
correlation_task_id	str	Links a generated DSN back to its original task.
storage_ref	str	The storage integration identifier for an s3 attempt.
bucket_name	str	The destination bucket for an s3 attempt.
object_key	str	The stored object key for an s3 attempt.
endpoint_host	str	The storage endpoint host for an s3 attempt.
error_code	str	The storage provider error code, when one was returned.
status_code	int	The storage or webhook HTTP response status.
message	str	The failure description, truncated by the server.
next_retry_at_utc	int	The Unix time, in seconds, of the next attempt when the attempt will be retried.
retry_after_ms	int	The webhook retry-after hint, in milliseconds.
target_url_host	str	The webhook target host.
webhook_ref	str	The webhook endpoint identifier.
signing_key_kid	str	The key ID used to sign the webhook attempt.
raw_token_ttl_seconds	int	The raw-message token lifetime the webhook attempt used.

DecodeError

DecodeError(subject: str, message: str = '')

Bases: NotificationError

A notification body that does not match the SDK's payload model.

This normally means the SDK is older than the server that produced the event. The event is terminated rather than redelivered, because redelivery can never succeed. When an error handler is configured, it receives the error and consumption continues; otherwise consumption stops and this error is raised.

Initialize the error.

Parameters:

Name	Type	Description	Default
subject	str	The notification subject the undecodable body was published to.	required
message	str	A short description of the decode failure.	''

Source code in src/mxraven/errors.py

def __init__(self, subject: str, message: str = "") -> None:
    """Initialize the error.

    Args:
        subject: The notification subject the undecodable body was
            published to.
        message: A short description of the decode failure.
    """
    self.subject = subject
    self.message = message
    detail = message.strip()
    text = f"notify: decode {subject!r}"
    if detail:
        text = f"{text}: {detail}"
    super().__init__(text)

InboundEmail dataclass

InboundEmail(
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    routing_decision: RoutingDecision | None = None,
    verdicts: Verdicts | None = None,
    envelope: Envelope = Envelope(),
    message: MessageSummary = MessageSummary(),
    headers: tuple[HeaderField, ...] = (),
    raw_email: RawEmail = RawEmail(),
)

The DELIVER_WEBHOOK payload: a complete inbound message.

NotificationClient

NotificationClient(
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: SSLContext | None = None,
    on_error: ErrorHandler | None = None,
)

Consumes one listener's client notification events synchronously.

Notifications are off by default; enable them for the listener with the mxRaven administrative API first, which creates the durable consumer this client reads from. The connection authenticates with the listener's ordinary API key and is granted access only to that listener's consumer.

A client is safe for concurrent use. A consume call and an events iterator share the durable consumer with any other client using the same API key: events are distributed between them, not duplicated to each.

This client owns a private asyncio event loop on a worker thread. Handlers run on a worker thread, not the calling thread, so a slow handler does not block close from another thread. Do not call close from the thread that is blocked in consume.

Use :meth:connect to build a connected client; do not construct one directly. A client is a context manager.

Parameters:

Name	Type	Description	Default
url	str	The client notification tier WebSocket URL, for example wss://nats.mxraven.com. Only the ws and wss schemes are accepted.	required
username	str	The listener API key prefix, for example mxr_tx_ab12cd34ef56.	required
secret	str	The listener API key secret. Treat it as a secret and do not log it.	required
listener	str	The listener whose events are consumed. It must not contain ., *, >, /, a backslash, or whitespace.	required
consumer	str | None	Overrides the durable consumer name. Defaults to notify-<listener>.	None
connect_timeout	float	The connection deadline in seconds. Defaults to 10.0.	DEFAULT_CONNECT_TIMEOUT
tls	SSLContext | None	An optional TLS context for wss:// connections, for example to trust a private certificate authority.	None
on_error	ErrorHandler | None	Receives non-fatal connection and decode errors. It runs on the worker thread and must return promptly. When omitted, an undecodable event stops consumption with a DecodeError.	None

Raises:

Type	Description
ValueError	An option is missing or invalid.

Validate the options. The client is not connected until connect.

Source code in src/mxraven/notify/client.py

def __init__(
    self,
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: ssl.SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> None:
    """Validate the options. The client is not connected until `connect`."""
    self._options = validate_options(
        url=url,
        username=username,
        secret=secret,
        listener=listener,
        consumer=consumer,
        connect_timeout=connect_timeout,
    )
    self._tls = tls
    self._on_error = on_error
    self._runner = LoopRunner("mxraven-notify")
    self._client: _AsyncNotificationClient | None = None
    self._active: Future[None] | None = None
    self._closed = False
    self._lock = threading.Lock()

__enter__

__enter__() -> Self

Return this client.

Source code in src/mxraven/notify/client.py

def __enter__(self) -> Self:
    """Return this client."""
    return self

__exit__

__exit__(*exc_info: object) -> None

Close the client.

Source code in src/mxraven/notify/client.py

def __exit__(self, *exc_info: object) -> None:
    """Close the client."""
    self.close()

close

close() -> None

Release the client's connection.

Active consume calls and iterators stop, and events they did not acknowledge are redelivered later. Safe to call more than once.

Source code in src/mxraven/notify/client.py

def close(self) -> None:
    """Release the client's connection.

    Active `consume` calls and iterators stop, and events they did not
    acknowledge are redelivered later. Safe to call more than once.
    """
    with self._lock:
        if self._closed:
            return
        self._closed = True
        active = self._active
    client = self._client
    runner = self._runner
    try:
        if client is not None:
            runner.run(client.close())
        if active is not None:
            # The consuming call's own outcome is not close()'s concern;
            # closing already asked it to stop.
            with contextlib.suppress(Exception):
                active.result(5.0)
    finally:
        runner.stop()
        runner.join(5.0)
        runner.close()

connect classmethod

connect(
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> NotificationClient

Connect to the notification tier and return a client.

The initial connection fails here rather than on the first event. After a successful connection the transport reconnects indefinitely when the tier is unreachable, and the consumer resumes from its acknowledgement floor.

Parameters:

Name	Type	Description	Default
url	str	The notification tier WebSocket URL (ws:// or wss://).	required
username	str	The listener API key prefix.	required
secret	str	The listener API key secret.	required
listener	str	The listener whose events are consumed.	required
consumer	str | None	An optional durable consumer name override.	None
connect_timeout	float	The connection deadline in seconds.	DEFAULT_CONNECT_TIMEOUT
tls	SSLContext | None	An optional TLS context for wss:// connections.	None
on_error	ErrorHandler | None	An optional non-fatal error handler.	None

Returns:

Type	Description
NotificationClient	A connected client.

Raises:

Type	Description
ValueError	An option is missing or invalid.
NotificationError	The connection failed, the API key was rejected, or the optional notify extra is not installed.

Source code in src/mxraven/notify/client.py

@classmethod
def connect(
    cls,
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: ssl.SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> NotificationClient:
    """Connect to the notification tier and return a client.

    The initial connection fails here rather than on the first event. After
    a successful connection the transport reconnects indefinitely when the
    tier is unreachable, and the consumer resumes from its acknowledgement
    floor.

    Args:
        url: The notification tier WebSocket URL (``ws://`` or ``wss://``).
        username: The listener API key prefix.
        secret: The listener API key secret.
        listener: The listener whose events are consumed.
        consumer: An optional durable consumer name override.
        connect_timeout: The connection deadline in seconds.
        tls: An optional TLS context for ``wss://`` connections.
        on_error: An optional non-fatal error handler.

    Returns:
        A connected client.

    Raises:
        ValueError: An option is missing or invalid.
        NotificationError: The connection failed, the API key was rejected,
            or the optional ``notify`` extra is not installed.
    """
    client = cls(
        url=url,
        username=username,
        secret=secret,
        listener=listener,
        consumer=consumer,
        connect_timeout=connect_timeout,
        tls=tls,
        on_error=on_error,
    )
    client._start()
    return client

consume

consume(handler: NotificationHandler) -> None

Consume events until stopped, closed, or the handler stops.

A handler that returns None acknowledges the event. A handler that raises leaves it unacknowledged so the server redelivers it after a delay that doubles with each attempt up to 30 seconds. A handler that returns STOP_CONSUMING acknowledges the event and stops. An undecodable event is terminated so it cannot poison the stream, and either stops consumption with a DecodeError or is reported to on_error and skipped.

This call blocks until consumption stops. Call :meth:close from another thread, or return STOP_CONSUMING from the handler, to stop it.

Parameters:

Name	Type	Description	Default
handler	NotificationHandler	Processes one event at a time on a worker thread.	required

Raises:

Type	Description
NotificationError	The client is closed, consumption is already running, or consumption failed.
DecodeError	An event could not be decoded and no error handler is configured.

Source code in src/mxraven/notify/client.py

def consume(self, handler: NotificationHandler) -> None:
    """Consume events until stopped, closed, or the handler stops.

    A handler that returns ``None`` acknowledges the event. A handler that
    raises leaves it unacknowledged so the server redelivers it after a
    delay that doubles with each attempt up to 30 seconds. A handler that
    returns ``STOP_CONSUMING`` acknowledges the event and stops. An
    undecodable event is terminated so it cannot poison the stream, and
    either stops consumption with a `DecodeError` or is reported to
    `on_error` and skipped.

    This call blocks until consumption stops. Call :meth:`close` from
    another thread, or return `STOP_CONSUMING` from the handler, to stop it.

    Args:
        handler: Processes one event at a time on a worker thread.

    Raises:
        NotificationError: The client is closed, consumption is already
            running, or consumption failed.
        DecodeError: An event could not be decoded and no error handler is
            configured.
    """
    client = self._require_client()

    async def adapted(event: NotificationEvent) -> NotificationResult:
        return await asyncio.to_thread(handler, event)

    with self._lock:
        if self._active is not None:
            message = "notify: consumption is already running"
            raise NotificationError(message)
        active = self._runner.submit(client.consume(adapted))
        self._active = active
    try:
        active.result()
    finally:
        with self._lock:
            self._active = None

events

events() -> NotificationIterator

Return an iterator for manual acknowledgement.

The caller acknowledges each event with :meth:~NotificationDelivery.ack, nak, or term. Stopping iteration leaves unacknowledged events for redelivery after the consumer's acknowledgement timeout.

Source code in src/mxraven/notify/client.py

def events(self) -> NotificationIterator:
    """Return an iterator for manual acknowledgement.

    The caller acknowledges each event with
    :meth:`~NotificationDelivery.ack`, `nak`, or `term`. Stopping iteration
    leaves unacknowledged events for redelivery after the consumer's
    acknowledgement timeout.
    """
    client = self._require_client()
    return NotificationIterator(self._runner, client.events())

NotificationDelivery

NotificationDelivery(
    runner: LoopRunner, delivery: NotificationDelivery
)

One event yielded by a :class:NotificationIterator.

Exactly one of ack, nak, or term must be called for the event; calling more than one raises NotificationError.

Wrap an asynchronous delivery for blocking settlement.

Source code in src/mxraven/notify/client.py

def __init__(self, runner: LoopRunner, delivery: _AsyncDelivery) -> None:
    """Wrap an asynchronous delivery for blocking settlement."""
    self._runner = runner
    self._delivery = delivery

event property

event: NotificationEvent

The decoded notification.

ack

ack() -> None

Acknowledge the event so it is not redelivered.

Source code in src/mxraven/notify/client.py

def ack(self) -> None:
    """Acknowledge the event so it is not redelivered."""
    self._runner.run(self._delivery.ack())

nak

nak(delay: float | None = None) -> None

Negatively acknowledge the event so the server redelivers it.

Parameters:

Name	Type	Description	Default
delay	float | None	An optional redelivery delay in seconds.	None

Source code in src/mxraven/notify/client.py

def nak(self, delay: float | None = None) -> None:
    """Negatively acknowledge the event so the server redelivers it.

    Args:
        delay: An optional redelivery delay in seconds.
    """
    self._runner.run(self._delivery.nak(delay))

term

term() -> None

Tell the server never to redeliver the event.

Source code in src/mxraven/notify/client.py

def term(self) -> None:
    """Tell the server never to redeliver the event."""
    self._runner.run(self._delivery.term())

NotificationError

Bases: MxRavenError

A failure connecting to or consuming the client notification stream.

The underlying transport or JetStream failure is preserved as :attr:__cause__. Most connection failures are transient and are retried automatically; this error is only raised when an operation cannot continue.

NotificationEvent dataclass

NotificationEvent(
    kind: NotificationKind,
    accepted: InboundEmail | None = None,
    attempt: AttemptEvent | None = None,
)

A decoded client notification.

Exactly one of :attr:accepted or :attr:attempt is set, matching :attr:kind.

NotificationIterator

NotificationIterator(
    runner: LoopRunner, iterator: NotificationIterator
)

An iterator over a listener's notification events.

Use for and break when finished; the iterator stops pulling and leaves unacknowledged events for redelivery. Iteration also stops when the client's connection closes.

Wrap an asynchronous iterator for blocking iteration.

Source code in src/mxraven/notify/client.py

def __init__(self, runner: LoopRunner, iterator: _AsyncIterator) -> None:
    """Wrap an asynchronous iterator for blocking iteration."""
    self._runner = runner
    self._iterator = iterator

__iter__

__iter__() -> Self

Return this iterator.

Source code in src/mxraven/notify/client.py

def __iter__(self) -> Self:
    """Return this iterator."""
    return self

__next__

__next__() -> NotificationDelivery

Yield the next delivery, or stop iteration.

Source code in src/mxraven/notify/client.py

def __next__(self) -> NotificationDelivery:
    """Yield the next delivery, or stop iteration."""
    try:
        delivery = self._runner.run(self._iterator.__anext__())
    except StopAsyncIteration:
        raise StopIteration from None
    return NotificationDelivery(self._runner, delivery)

close

close() -> None

Stop iterating. Safe to call more than once.

Source code in src/mxraven/notify/client.py

def close(self) -> None:
    """Stop iterating. Safe to call more than once."""
    self._runner.run(self._iterator.aclose())

NotificationKind

Bases: StrEnum

The kind of notification a :class:NotificationEvent carries.

StopConsuming

The type of the :data:STOP_CONSUMING sentinel.

Do not construct it directly; return the exported :data:STOP_CONSUMING singleton instead.

__repr__

__repr__() -> str

Return a readable representation.

Source code in src/mxraven/notify/handler.py

def __repr__(self) -> str:
    """Return a readable representation."""
    return "STOP_CONSUMING"

decode_notification

decode_notification(
    subject: str, body: bytes
) -> NotificationEvent

Decode one notification body.

The event kind is the final token of subject. When the subject is unavailable or unrecognized, the body shape is probed instead.

Parameters:

Name	Type	Description	Default
subject	str	The notification subject the body was published to.	required
body	bytes	The raw JSON body bytes.	required

Returns:

Type	Description
NotificationEvent	The decoded notification.

Raises:

Type	Description
DecodeError	The body is not valid JSON, has a field of the wrong type, or cannot be identified.

Source code in src/mxraven/notify/event.py

def decode_notification(subject: str, body: bytes) -> NotificationEvent:
    """Decode one notification body.

    The event kind is the final token of `subject`. When the subject is
    unavailable or unrecognized, the body shape is probed instead.

    Args:
        subject: The notification subject the body was published to.
        body: The raw JSON body bytes.

    Returns:
        The decoded notification.

    Raises:
        DecodeError: The body is not valid JSON, has a field of the wrong type,
            or cannot be identified.
    """
    source = _parse(subject, body)
    kind = _subject_kind(subject)
    if kind == _ACCEPTED:
        return NotificationEvent(NotificationKind.ACCEPTED, accepted=_accepted(subject, source))
    if kind == _ATTEMPT:
        return NotificationEvent(NotificationKind.ATTEMPT, attempt=_attempt(subject, source))

    if source.get("event_type") == _INBOUND:
        return NotificationEvent(NotificationKind.ACCEPTED, accepted=_accepted(subject, source))
    channel = source.get("channel")
    if isinstance(channel, str) and channel != "":
        return NotificationEvent(NotificationKind.ATTEMPT, attempt=_attempt(subject, source))
    message = "unrecognized notification body"
    raise DecodeError(subject, message)

Asynchronous notifications

notify

Asynchronous mxRaven client notification events.

The asyncio counterpart of :mod:mxraven.notify, with the same payloads and error surface. The transport requires the optional notify extra; install mxraven[notify].

NotificationHandler module-attribute

NotificationHandler: TypeAlias = (
    "Callable[[NotificationEvent], NotificationResult]"
)

Processes one notification event synchronously.

Returning None acknowledges the event. Raising leaves it unacknowledged so the server redelivers it with exponential backoff. Returning :data:STOP_CONSUMING acknowledges the event and stops consumption.

NotificationResult module-attribute

NotificationResult: TypeAlias = 'StopConsuming | None'

What a notification handler returns: None to acknowledge, or :data:STOP_CONSUMING to acknowledge and stop.

STOP_CONSUMING module-attribute

STOP_CONSUMING: Final[StopConsuming] = StopConsuming()

Return this from a handler to acknowledge the current event and stop.

AttemptChannel

Bases: StrEnum

The egress worker that produced an :class:AttemptEvent.

The set is open on the wire: a channel this SDK does not know is preserved as a raw string on :attr:AttemptEvent.channel, which still compares equal to the matching member.

AttemptEvent dataclass

AttemptEvent(
    channel: str = "",
    status: str = "",
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    correlation_id: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    source_ip: str = "",
    destination_domain: str = "",
    remote_host: str = "",
    smtp_code: int = 0,
    enhanced_status_code: str = "",
    reason: str = "",
    remote_response: str = "",
    correlation_task_id: str = "",
    storage_ref: str = "",
    bucket_name: str = "",
    object_key: str = "",
    endpoint_host: str = "",
    error_code: str = "",
    status_code: int = 0,
    message: str = "",
    next_retry_at_utc: int = 0,
    retry_after_ms: int = 0,
    target_url_host: str = "",
    webhook_ref: str = "",
    signing_key_kid: str = "",
    raw_token_ttl_seconds: int = 0,
)

One delivery attempt for an accepted message.

:attr:channel names the egress worker that produced the attempt; the remaining fields describe that worker's protocol and default when they do not apply. A rule that chains side-effect actions reports each action it triggers, so one message can produce more than one attempt event. Every attempt carries a :attr:correlation_id that points at the task_id of the accepted event for the message it stems from.

Attributes:

Name	Type	Description
event_type	str	The producer's event-type marker when one is set. SMTP attempt events omit it.
channel	str	The egress worker that produced the attempt. Compare it against :class:AttemptChannel; unknown future values are preserved as their raw string.
task_id	str	The delivery task. For a submission listener it equals the accepted event's task_id.
tenant_id	str	The owning tenant.
listener_id	str	The source listener.
correlation_id	str	The task_id of the accepted event for the message, when one was recorded.
status	str	The attempt outcome, one of :class:mxraven.webhook.StatusOutcome.
attempt	int	The delivery attempt number, starting at 1.
accepted_at_utc	int	The Unix time, in seconds, when mxRaven accepted the message. Zero when the producer does not report it.
occurred_at_utc	int	The Unix time, in seconds, when the attempt was built.
source_ip	str	The SMTP egress source address, when known.
destination_domain	str	The recipient domain for an SMTP attempt.
remote_host	str	The remote MTA hostname, when known.
smtp_code	int	The remote SMTP reply code.
enhanced_status_code	str	The RFC 3463 enhanced status code.
reason	str	The bounded, machine-readable delivery classification.
remote_response	str	The remote reply text, truncated by the server.
correlation_task_id	str	Links a generated DSN back to its original task.
storage_ref	str	The storage integration identifier for an s3 attempt.
bucket_name	str	The destination bucket for an s3 attempt.
object_key	str	The stored object key for an s3 attempt.
endpoint_host	str	The storage endpoint host for an s3 attempt.
error_code	str	The storage provider error code, when one was returned.
status_code	int	The storage or webhook HTTP response status.
message	str	The failure description, truncated by the server.
next_retry_at_utc	int	The Unix time, in seconds, of the next attempt when the attempt will be retried.
retry_after_ms	int	The webhook retry-after hint, in milliseconds.
target_url_host	str	The webhook target host.
webhook_ref	str	The webhook endpoint identifier.
signing_key_kid	str	The key ID used to sign the webhook attempt.
raw_token_ttl_seconds	int	The raw-message token lifetime the webhook attempt used.

DecodeError

DecodeError(subject: str, message: str = '')

Bases: NotificationError

A notification body that does not match the SDK's payload model.

This normally means the SDK is older than the server that produced the event. The event is terminated rather than redelivered, because redelivery can never succeed. When an error handler is configured, it receives the error and consumption continues; otherwise consumption stops and this error is raised.

Initialize the error.

Parameters:

Name	Type	Description	Default
subject	str	The notification subject the undecodable body was published to.	required
message	str	A short description of the decode failure.	''

Source code in src/mxraven/errors.py

def __init__(self, subject: str, message: str = "") -> None:
    """Initialize the error.

    Args:
        subject: The notification subject the undecodable body was
            published to.
        message: A short description of the decode failure.
    """
    self.subject = subject
    self.message = message
    detail = message.strip()
    text = f"notify: decode {subject!r}"
    if detail:
        text = f"{text}: {detail}"
    super().__init__(text)

InboundEmail dataclass

InboundEmail(
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: str = "",
    message_ref: str = "",
    attempt: int = 0,
    accepted_at_utc: int = 0,
    occurred_at_utc: int = 0,
    routing_decision: RoutingDecision | None = None,
    verdicts: Verdicts | None = None,
    envelope: Envelope = Envelope(),
    message: MessageSummary = MessageSummary(),
    headers: tuple[HeaderField, ...] = (),
    raw_email: RawEmail = RawEmail(),
)

The DELIVER_WEBHOOK payload: a complete inbound message.

NotificationClient

NotificationClient(
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: SSLContext | None = None,
    on_error: ErrorHandler | None = None,
)

Consumes one listener's client notification events asynchronously.

Notifications are off by default; enable them for the listener with the mxRaven administrative API first, which creates the durable consumer this client reads from. The connection authenticates with the listener's ordinary API key and is granted access only to that listener's consumer.

A client is safe for concurrent use. A consume call and an events iterator share the durable consumer with any other client using the same API key: events are distributed between them, not duplicated to each.

Use :meth:connect to build a connected client; do not construct one directly. A client is an async context manager.

Parameters:

Name	Type	Description	Default
url	str	The client notification tier WebSocket URL, for example wss://nats.mxraven.com. Only the ws and wss schemes are accepted.	required
username	str	The listener API key prefix, for example mxr_tx_ab12cd34ef56.	required
secret	str	The listener API key secret. Treat it as a secret and do not log it.	required
listener	str	The listener whose events are consumed. It must not contain ., *, >, /, a backslash, or whitespace.	required
consumer	str | None	Overrides the durable consumer name. Defaults to notify-<listener>.	None
connect_timeout	float	The connection deadline in seconds. Defaults to 10.0.	DEFAULT_CONNECT_TIMEOUT
tls	SSLContext | None	An optional TLS context for wss:// connections, for example to trust a private certificate authority.	None
on_error	ErrorHandler | None	Receives non-fatal connection and decode errors. When omitted, an undecodable event stops consumption with a DecodeError.	None

Raises:

Type	Description
ValueError	An option is missing or invalid.

Validate the options. The client is not connected until connect.

Source code in src/mxraven/aio/notify/client.py

def __init__(
    self,
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: ssl.SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> None:
    """Validate the options. The client is not connected until `connect`."""
    self._options = validate_options(
        url=url,
        username=username,
        secret=secret,
        listener=listener,
        consumer=consumer,
        connect_timeout=connect_timeout,
    )
    self._tls = tls
    self._on_error = on_error
    self._session: Session | None = None
    self._closed = False

__aenter__ async

__aenter__() -> Self

Return this client.

Source code in src/mxraven/aio/notify/client.py

async def __aenter__(self) -> Self:
    """Return this client."""
    return self

__aexit__ async

__aexit__(*exc_info: object) -> None

Close the client.

Source code in src/mxraven/aio/notify/client.py

async def __aexit__(self, *exc_info: object) -> None:
    """Close the client."""
    await self.close()

close async

close() -> None

Release the client's connection.

Active consume calls and iterators stop, and events they did not acknowledge are redelivered later. Safe to call more than once.

Source code in src/mxraven/aio/notify/client.py

async def close(self) -> None:
    """Release the client's connection.

    Active `consume` calls and iterators stop, and events they did not
    acknowledge are redelivered later. Safe to call more than once.
    """
    if self._closed:
        return
    self._closed = True
    session = self._session
    if session is not None:
        await session.close()

connect async classmethod

connect(
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> NotificationClient

Connect to the notification tier and return a client.

The initial connection fails here rather than on the first event. After a successful connection the transport reconnects indefinitely when the tier is unreachable, and the consumer resumes from its acknowledgement floor.

Parameters:

Name	Type	Description	Default
url	str	The notification tier WebSocket URL (ws:// or wss://).	required
username	str	The listener API key prefix.	required
secret	str	The listener API key secret.	required
listener	str	The listener whose events are consumed.	required
consumer	str | None	An optional durable consumer name override.	None
connect_timeout	float	The connection deadline in seconds.	DEFAULT_CONNECT_TIMEOUT
tls	SSLContext | None	An optional TLS context for wss:// connections.	None
on_error	ErrorHandler | None	An optional non-fatal error handler.	None

Returns:

Type	Description
NotificationClient	A connected client.

Raises:

Type	Description
ValueError	An option is missing or invalid.
NotificationError	The connection failed, the API key was rejected, or the optional notify extra is not installed.

Source code in src/mxraven/aio/notify/client.py

@classmethod
async def connect(
    cls,
    *,
    url: str,
    username: str,
    secret: str,
    listener: str,
    consumer: str | None = None,
    connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
    tls: ssl.SSLContext | None = None,
    on_error: ErrorHandler | None = None,
) -> NotificationClient:
    """Connect to the notification tier and return a client.

    The initial connection fails here rather than on the first event. After
    a successful connection the transport reconnects indefinitely when the
    tier is unreachable, and the consumer resumes from its acknowledgement
    floor.

    Args:
        url: The notification tier WebSocket URL (``ws://`` or ``wss://``).
        username: The listener API key prefix.
        secret: The listener API key secret.
        listener: The listener whose events are consumed.
        consumer: An optional durable consumer name override.
        connect_timeout: The connection deadline in seconds.
        tls: An optional TLS context for ``wss://`` connections.
        on_error: An optional non-fatal error handler.

    Returns:
        A connected client.

    Raises:
        ValueError: An option is missing or invalid.
        NotificationError: The connection failed, the API key was rejected,
            or the optional ``notify`` extra is not installed.
    """
    client = cls(
        url=url,
        username=username,
        secret=secret,
        listener=listener,
        consumer=consumer,
        connect_timeout=connect_timeout,
        tls=tls,
        on_error=on_error,
    )
    await client._connect()
    return client

consume async

consume(handler: NotificationHandler) -> None

Consume events until stopped, closed, or the handler stops.

A handler that returns None acknowledges the event. A handler that raises leaves it unacknowledged so the server redelivers it after a delay that doubles with each attempt up to 30 seconds. A handler that returns STOP_CONSUMING acknowledges the event and stops. An undecodable event is terminated so it cannot poison the stream, and either stops consumption with a DecodeError or is reported to on_error and skipped.

Cancellation is a normal stop. A permanently lost consumer or connection raises NotificationError.

Parameters:

Name	Type	Description	Default
handler	NotificationHandler	Processes one event at a time. It may be a coroutine function or a plain callable.	required

Raises:

Type	Description
NotificationError	The client is closed, or consumption failed.
DecodeError	An event could not be decoded and no error handler is configured.

Source code in src/mxraven/aio/notify/client.py

async def consume(self, handler: NotificationHandler) -> None:
    """Consume events until stopped, closed, or the handler stops.

    A handler that returns ``None`` acknowledges the event. A handler that
    raises leaves it unacknowledged so the server redelivers it after a
    delay that doubles with each attempt up to 30 seconds. A handler that
    returns ``STOP_CONSUMING`` acknowledges the event and stops. An
    undecodable event is terminated so it cannot poison the stream, and
    either stops consumption with a `DecodeError` or is reported to
    `on_error` and skipped.

    Cancellation is a normal stop. A permanently lost consumer or
    connection raises `NotificationError`.

    Args:
        handler: Processes one event at a time. It may be a coroutine
            function or a plain callable.

    Raises:
        NotificationError: The client is closed, or consumption failed.
        DecodeError: An event could not be decoded and no error handler is
            configured.
    """
    session = self._require_session()
    subscription = await session.pull()
    try:
        while True:
            try:
                messages = await subscription.fetch(PULL_TIMEOUT)
            except TimeoutError:
                if session.is_closed:
                    return
                continue
            except Exception as error:
                if session.is_closed:
                    return
                operation = "notify: consume"
                raise _wrap(operation, error) from error
            for message in messages:
                if await self._dispatch(message, handler):
                    return
    finally:
        await subscription.close()

events

events() -> NotificationIterator

Return an async iterator for manual acknowledgement.

The caller acknowledges each event with :meth:~NotificationDelivery.ack, nak, or term. Stopping iteration leaves unacknowledged events for redelivery after the consumer's acknowledgement timeout.

Source code in src/mxraven/aio/notify/client.py

def events(self) -> NotificationIterator:
    """Return an async iterator for manual acknowledgement.

    The caller acknowledges each event with
    :meth:`~NotificationDelivery.ack`, `nak`, or `term`. Stopping iteration
    leaves unacknowledged events for redelivery after the consumer's
    acknowledgement timeout.
    """
    return NotificationIterator(self)

NotificationDelivery

NotificationDelivery(
    client: NotificationClient,
    message: Message,
    event: NotificationEvent,
)

One event yielded by a :class:NotificationIterator.

Exactly one of ack, nak, or term must be called for the event; calling more than one raises NotificationError. An event that is never settled is redelivered after the consumer's acknowledgement timeout.

Wrap a decoded event for manual settlement.

Source code in src/mxraven/aio/notify/client.py

def __init__(
    self, client: NotificationClient, message: Message, event: NotificationEvent
) -> None:
    """Wrap a decoded event for manual settlement."""
    self._client = client
    self._message = message
    self._event = event
    self._settled = False

event property

event: NotificationEvent

The decoded notification.

ack async

ack() -> None

Acknowledge the event so it is not redelivered.

Raises:

Type	Description
NotificationError	The event was already settled.

Source code in src/mxraven/aio/notify/client.py

async def ack(self) -> None:
    """Acknowledge the event so it is not redelivered.

    Raises:
        NotificationError: The event was already settled.
    """
    self._claim()
    await self._client._settle("notify: acknowledge event", self._message.ack())

nak async

nak(delay: float | None = None) -> None

Negatively acknowledge the event so the server redelivers it.

Parameters:

Name	Type	Description	Default
delay	float | None	An optional redelivery delay in seconds. When omitted or zero, the server redelivers immediately.	None

Raises:

Type	Description
ValueError	delay is negative.
NotificationError	The event was already settled.

Source code in src/mxraven/aio/notify/client.py

async def nak(self, delay: float | None = None) -> None:
    """Negatively acknowledge the event so the server redelivers it.

    Args:
        delay: An optional redelivery delay in seconds. When omitted or
            zero, the server redelivers immediately.

    Raises:
        ValueError: `delay` is negative.
        NotificationError: The event was already settled.
    """
    if delay is not None and delay < 0:
        message = f"notify: invalid nak delay {delay!r}"
        raise ValueError(message)
    self._claim()
    resolved = delay if delay else None
    await self._client._settle("notify: negate event", self._message.nak(resolved))

term async

term() -> None

Tell the server never to redeliver the event.

Raises:

Type	Description
NotificationError	The event was already settled.

Source code in src/mxraven/aio/notify/client.py

async def term(self) -> None:
    """Tell the server never to redeliver the event.

    Raises:
        NotificationError: The event was already settled.
    """
    self._claim()
    await self._client._settle("notify: terminate event", self._message.term())

NotificationError

Bases: MxRavenError

A failure connecting to or consuming the client notification stream.

The underlying transport or JetStream failure is preserved as :attr:__cause__. Most connection failures are transient and are retried automatically; this error is only raised when an operation cannot continue.

NotificationEvent dataclass

NotificationEvent(
    kind: NotificationKind,
    accepted: InboundEmail | None = None,
    attempt: AttemptEvent | None = None,
)

A decoded client notification.

Exactly one of :attr:accepted or :attr:attempt is set, matching :attr:kind.

NotificationIterator

NotificationIterator(client: NotificationClient)

An async iterator over a listener's notification events.

Use async for and break when finished; the iterator stops pulling and leaves unacknowledged events for redelivery. Iteration also stops when the client's connection closes.

Build an iterator over client.

Source code in src/mxraven/aio/notify/client.py

def __init__(self, client: NotificationClient) -> None:
    """Build an iterator over `client`."""
    self._client = client
    self._subscription: Subscription | None = None
    self._closed = False

__aiter__

__aiter__() -> Self

Return this iterator.

Source code in src/mxraven/aio/notify/client.py

def __aiter__(self) -> Self:
    """Return this iterator."""
    return self

__anext__ async

__anext__() -> NotificationDelivery

Yield the next delivery, or stop iteration.

Source code in src/mxraven/aio/notify/client.py

async def __anext__(self) -> NotificationDelivery:
    """Yield the next delivery, or stop iteration."""
    if self._closed:
        raise StopAsyncIteration
    session = self._client._require_session()
    subscription = await self._subscription_for(session)
    while True:
        try:
            messages = await subscription.fetch(PULL_TIMEOUT)
        except TimeoutError:
            if session.is_closed:
                raise StopAsyncIteration from None
            continue
        except Exception as error:
            if session.is_closed:
                raise StopAsyncIteration from None
            operation = "notify: iterate events"
            raise _wrap(operation, error) from error
        if not messages:
            continue
        message = messages[0]
        try:
            event = decode_notification(message.subject, message.data)
        except DecodeError as error:
            await self._client._settle("notify: terminate event", message.term())
            if self._client._on_error is None:
                await self.aclose()
                raise
            await self._client._report(error)
            continue
        return NotificationDelivery(self._client, message, event)

aclose async

aclose() -> None

Stop iterating. Safe to call more than once.

Source code in src/mxraven/aio/notify/client.py

async def aclose(self) -> None:
    """Stop iterating. Safe to call more than once."""
    if self._closed:
        return
    self._closed = True
    subscription = self._subscription
    if subscription is not None:
        await subscription.close()

NotificationKind

Bases: StrEnum

The kind of notification a :class:NotificationEvent carries.

StopConsuming

The type of the :data:STOP_CONSUMING sentinel.

Do not construct it directly; return the exported :data:STOP_CONSUMING singleton instead.

__repr__

__repr__() -> str

Return a readable representation.

Source code in src/mxraven/notify/handler.py

def __repr__(self) -> str:
    """Return a readable representation."""
    return "STOP_CONSUMING"

decode_notification

decode_notification(
    subject: str, body: bytes
) -> NotificationEvent

Decode one notification body.

The event kind is the final token of subject. When the subject is unavailable or unrecognized, the body shape is probed instead.

Parameters:

Name	Type	Description	Default
subject	str	The notification subject the body was published to.	required
body	bytes	The raw JSON body bytes.	required

Returns:

Type	Description
NotificationEvent	The decoded notification.

Raises:

Type	Description
DecodeError	The body is not valid JSON, has a field of the wrong type, or cannot be identified.

Source code in src/mxraven/notify/event.py

def decode_notification(subject: str, body: bytes) -> NotificationEvent:
    """Decode one notification body.

    The event kind is the final token of `subject`. When the subject is
    unavailable or unrecognized, the body shape is probed instead.

    Args:
        subject: The notification subject the body was published to.
        body: The raw JSON body bytes.

    Returns:
        The decoded notification.

    Raises:
        DecodeError: The body is not valid JSON, has a field of the wrong type,
            or cannot be identified.
    """
    source = _parse(subject, body)
    kind = _subject_kind(subject)
    if kind == _ACCEPTED:
        return NotificationEvent(NotificationKind.ACCEPTED, accepted=_accepted(subject, source))
    if kind == _ATTEMPT:
        return NotificationEvent(NotificationKind.ATTEMPT, attempt=_attempt(subject, source))

    if source.get("event_type") == _INBOUND:
        return NotificationEvent(NotificationKind.ACCEPTED, accepted=_accepted(subject, source))
    channel = source.get("channel")
    if isinstance(channel, str) and channel != "":
        return NotificationEvent(NotificationKind.ATTEMPT, attempt=_attempt(subject, source))
    message = "unrecognized notification body"
    raise DecodeError(subject, message)

Asynchronous MIME parsing

mime

Asynchronous inbound email parsing for mxRaven deliveries.

The asyncio counterpart of :mod:mxraven.mime. :func:parse_email is pure and shared with the synchronous surface; :func:fetch_and_parse_raw_email downloads asynchronously.

MimeError

Bases: MxRavenError

A MIME message cannot be parsed within the parser's limits.

Malformed message content never raises; the parser returns the best interpretation it can. Only structurally unbounded input raises, so an untrusted message cannot exhaust the stack or memory.

MimeNestingError

Bases: MimeError

MIME nesting exceeded the parser's maximum depth.

MimePartsError

Bases: MimeError

A message contained more MIME entities than the parser allows.

ParsedAddress dataclass

ParsedAddress(address: str, name: str = '')

One mailbox from a parsed address header.

Attributes:

Name	Type	Description
address	str	The bare local@domain form.
name	str	The display name, with RFC 2047 encoded words decoded, when one was present.

ParsedAttachment dataclass

ParsedAttachment(
    content_type: str,
    content: bytes,
    filename: str = "",
    content_id: str = "",
    inline: bool = False,
)

One non-body MIME part.

Attributes:

Name	Type	Description
content_type	str	The media type, for example application/pdf.
content	bytes	The decoded attachment bytes. It is a fresh copy that the caller owns and may mutate.
filename	str	The decoded filename, when one was present.
content_id	str	The Content-ID value without angle brackets, for inline parts.
inline	bool	Whether the part is marked inline or carries a Content-ID.

ParsedEmail dataclass

ParsedEmail(
    headers: tuple[ParsedHeader, ...] = (),
    subject: str = "",
    from_: tuple[ParsedAddress, ...] = (),
    to: tuple[ParsedAddress, ...] = (),
    cc: tuple[ParsedAddress, ...] = (),
    bcc: tuple[ParsedAddress, ...] = (),
    reply_to: tuple[ParsedAddress, ...] = (),
    message_id: str = "",
    date: datetime | None = None,
    text: str = "",
    html: str = "",
    attachments: tuple[ParsedAttachment, ...] = (),
)

A parsed inbound RFC 5322 / MIME message.

An empty value means the corresponding header or body was absent.

Attributes:

Name	Type	Description
headers	tuple[ParsedHeader, ...]	Every header field, in order, with continuation lines unfolded. Values are not decoded.
subject	str	The decoded Subject, when the header was present.
from_	tuple[ParsedAddress, ...]	The From addresses. It is from_ because from is a Python keyword.
to	tuple[ParsedAddress, ...]	The To addresses.
cc	tuple[ParsedAddress, ...]	The Cc addresses.
bcc	tuple[ParsedAddress, ...]	The Bcc addresses.
reply_to	tuple[ParsedAddress, ...]	The Reply-To addresses.
message_id	str	The Message-ID without angle brackets, when present.
date	datetime | None	The parsed Date, when present and parseable.
text	str	The decoded first text/plain body, when the message has one.
html	str	The decoded first text/html body, when the message has one.
attachments	tuple[ParsedAttachment, ...]	Every non-body part, with transfer encoding and filename decoding applied. message/rfc822 parts appear here rather than being parsed recursively.

ParsedHeader dataclass

ParsedHeader(name: str, value: str)

One header field from a parsed message.

Attributes:

Name	Type	Description
name	str	The field name as it appeared in the message.
value	str	The unfolded field value. Encoded words are not decoded; use :attr:ParsedEmail.subject or an address field for decoded text.

fetch_and_parse_raw_email async

fetch_and_parse_raw_email(
    raw: RawEmail,
    *,
    timeout: float = _DEFAULT_FETCH_TIMEOUT,
    transport: AsyncTransport | None = None,
) -> ParsedEmail

Download and parse an inbound-email payload's raw message.

The asynchronous counterpart of :func:mxraven.mime.fetch_and_parse_raw_email; the download verifies the declared size and SHA-256 digest.

Parameters:

Name	Type	Description	Default
raw	RawEmail	The raw_email object from an inbound-email payload.	required
timeout	float	The request timeout in seconds.	_DEFAULT_FETCH_TIMEOUT
transport	AsyncTransport | None	An optional asynchronous transport to use instead of the default.	None

Returns:

Type	Description
ParsedEmail	The parsed message. Attachment content is a fresh copy.

Raises:

Type	Description
WebhookError	The download failed or the size or digest did not match.
MimeNestingError	The message nests deeper than the parser allows.
MimePartsError	The message has more MIME entities than the parser allows.

Source code in src/mxraven/aio/mime.py

async def fetch_and_parse_raw_email(
    raw: RawEmail,
    *,
    timeout: float = _DEFAULT_FETCH_TIMEOUT,
    transport: AsyncTransport | None = None,
) -> ParsedEmail:
    """Download and parse an inbound-email payload's raw message.

    The asynchronous counterpart of
    :func:`mxraven.mime.fetch_and_parse_raw_email`; the download verifies the
    declared size and SHA-256 digest.

    Args:
        raw: The ``raw_email`` object from an inbound-email payload.
        timeout: The request timeout in seconds.
        transport: An optional asynchronous transport to use instead of the
            default.

    Returns:
        The parsed message. Attachment content is a fresh copy.

    Raises:
        WebhookError: The download failed or the size or digest did not match.
        MimeNestingError: The message nests deeper than the parser allows.
        MimePartsError: The message has more MIME entities than the parser
            allows.
    """
    return parse_email(await raw.fetch_async(timeout=timeout, transport=transport))

parse_email

parse_email(
    data: bytes | bytearray | memoryview,
) -> ParsedEmail

Parse an inbound RFC 5322 / MIME message.

The parser never validates the message against the grammar and never raises on malformed content; it extracts what it can. Multipart mixed, alternative, and related structures are traversed, base64 and quoted-printable transfer encodings are decoded, RFC 2047 encoded words in structured headers are decoded, and RFC 2231 parameters are reassembled. message/rfc822 parts are surfaced as attachments rather than parsed recursively.

Parameters:

Name	Type	Description	Default
data	bytes | bytearray | memoryview	The raw message bytes.	required

Returns:

Type	Description
ParsedEmail	The parsed message. Attachment content is a fresh copy.

Raises:

Type	Description
MimeNestingError	The message nests deeper than the parser allows.
MimePartsError	The message has more MIME entities than the parser allows.

Example

from mxraven.mime import parse_email

email = parse_email(raw_bytes)
print(email.subject, len(email.attachments))

Source code in src/mxraven/mime/parse.py

def parse_email(data: bytes | bytearray | memoryview) -> ParsedEmail:
    """Parse an inbound RFC 5322 / MIME message.

    The parser never validates the message against the grammar and never raises
    on malformed content; it extracts what it can. Multipart ``mixed``,
    ``alternative``, and ``related`` structures are traversed, ``base64`` and
    ``quoted-printable`` transfer encodings are decoded, RFC 2047 encoded words
    in structured headers are decoded, and RFC 2231 parameters are reassembled.
    ``message/rfc822`` parts are surfaced as attachments rather than parsed
    recursively.

    Args:
        data: The raw message bytes.

    Returns:
        The parsed message. Attachment content is a fresh copy.

    Raises:
        MimeNestingError: The message nests deeper than the parser allows.
        MimePartsError: The message has more MIME entities than the parser
            allows.

    Example:
        ```python
        from mxraven.mime import parse_email

        email = parse_email(raw_bytes)
        print(email.subject, len(email.attachments))
        ```
    """
    payload = bytes(data)
    state = _WalkState()
    root = _parse_entity(payload) if payload else None
    if root is not None:
        _walk_entity(root, 0, state)
    headers = root.headers if root is not None else ()

    return ParsedEmail(
        headers=headers,
        subject=_decode_optional_header(headers, "subject"),
        from_=_header_addresses(headers, "from"),
        to=_header_addresses(headers, "to"),
        cc=_header_addresses(headers, "cc"),
        bcc=_header_addresses(headers, "bcc"),
        reply_to=_header_addresses(headers, "reply-to"),
        message_id=_message_id(headers),
        date=_date_header(headers),
        text=state.text if state.text is not None else "",
        html=state.html if state.html is not None else "",
        attachments=tuple(state.attachments),
    )