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)

Envelope `dataclass` ¶

Envelope(
    sender: str = "", recipients: tuple[str, ...] = ()
)

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.

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(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

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,
    )

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

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

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

MxRavenError ¶

Bases: Exception

Base class for every error raised by the mxRaven SDK.

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 = "",
    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 = "",
    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 = "",
    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.

StorageStatus `dataclass` ¶

StorageStatus(
    event_type: str = "",
    task_id: str = "",
    tenant_id: str = "",
    listener_id: 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(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(), "")

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.

API reference¶

Root package¶

mxraven ¶

Attachment dataclass ¶

Client ¶

__enter__ ¶

__exit__ ¶

close ¶

send ¶

send_raw ¶

Envelope dataclass ¶

FeedbackError ¶

detail property ¶

retryable property ¶

Header dataclass ¶

InvalidSignatureError ¶

Message ¶

attach ¶

attach_file ¶

attach_inline ¶

bcc ¶

build ¶

cc ¶

date ¶

from_ ¶

header ¶

html ¶

in_reply_to ¶

message_id ¶

null_sender ¶

references ¶

reply_to ¶

sender ¶

subject ¶

text ¶

to ¶

MxRavenError ¶

RecipientResult dataclass ¶

Result dataclass ¶

SMTPCapabilityError ¶

SMTPConnectionError ¶

SMTPError ¶

permanent property ¶

transient property ¶

SMTPProtocolError ¶

SMTPTimeoutError ¶

SMTPTransactionError ¶

WebhookError ¶

Asynchronous client¶

aio ¶

Client ¶

__aenter__ async ¶

__aexit__ async ¶

close async ¶

send async ¶

send_raw async ¶

Webhooks¶

webhook ¶

DeliveryStatus dataclass ¶

Envelope dataclass ¶

Event dataclass ¶

EventType ¶

HeaderField dataclass ¶

InboundEmail dataclass ¶

InvalidSignatureError ¶

MessageSummary dataclass ¶

RawEmail dataclass ¶

fetch ¶

fetch_async async ¶

RoutingDecision dataclass ¶

StatusOutcome ¶

StorageStatus dataclass ¶

TerminalAction ¶

Verdicts dataclass ¶

Verifier ¶

verify ¶

verify_and_decode ¶

WebhookError ¶

WebhookRequest dataclass ¶

create classmethod ¶

Attachment `dataclass` ¶

enter ¶

exit ¶

Envelope `dataclass` ¶

detail `property` ¶

retryable `property` ¶

Header `dataclass` ¶

RecipientResult `dataclass` ¶

Result `dataclass` ¶

permanent `property` ¶

transient `property` ¶

aenter `async` ¶

aexit `async` ¶

close `async` ¶

send `async` ¶

send_raw `async` ¶

DeliveryStatus `dataclass` ¶

Envelope `dataclass` ¶

Event `dataclass` ¶

HeaderField `dataclass` ¶

InboundEmail `dataclass` ¶

MessageSummary `dataclass` ¶

RawEmail `dataclass` ¶

fetch_async `async` ¶

RoutingDecision `dataclass` ¶

StorageStatus `dataclass` ¶

Verdicts `dataclass` ¶

WebhookRequest `dataclass` ¶

create `classmethod` ¶

from_asgi `classmethod` ¶

from_wsgi `classmethod` ¶

learn `async` ¶

learn_ham `async` ¶

learn_spam `async` ¶

unsubscribe `async` ¶

detail `property` ¶

retryable `property` ¶

LearningResult `dataclass` ¶