Code Examples

The examples implement the Payments API contract in Python. Together they form one module: the signing helper and the HTTP helper are used by all other examples.

Example	Classification
Signing helper	Helper example; used by the runnable examples
HTTP helper and error handling	Helper example; used by the runnable examples
Read a balance	Runnable example; requires the helpers above
Create a transfer and track its status	Runnable example with recovery; requires the helpers above

Setup

• Python 3.10 or later.
• Dependencies: pip install cryptography requests.
• Environment variables:

Variable	Value
EXN_HOST	Base URL of the API host provided during onboarding, for example https://<your API host>
EXN_API_KEY	Your API key
EXN_PRIVATE_KEY_PATH	Path to your Ed25519 signing key as a PEM (PKCS#8) file
EXN_SOURCE_ACCOUNT, EXN_DESTINATION_ACCOUNT, EXN_AMOUNT	Used only when you run the module as a script: the two of your own accounts and the amount of the transfer it creates
EXN_IDEMPOTENCY_KEY	Optional, script only: the key of a transfer whose outcome is unknown. Set it to the printed key and keep the same accounts and amount. The script starts this record as reserved, because an earlier attempt may have created the transfer

Live transfers

Running the module as a script creates a real transfer between the accounts you set. Use accounts and an amount that you intend to move.

Copy the Python blocks below into one file, in order. The script prints the idempotency key before sending the first request and the transfer_id as soon as creation succeeds. Keep both values. If the process stops before a transfer_id is printed, resume only with the printed key and the original accounts and amount. Once you have a transfer_id, use get_transfer() or wait_for_final_status() to check that transfer; do not run creation again.

The signing helper reads the signing key as a PEM (PKCS#8) file. If your signing key material is issued in another format, replace load_private_key() accordingly. The signing rules are described in Request signing.

Example values

We recommend checking the transfer status every 2 seconds, as shown below. The request timeout, total waiting time, and number of retries are example values you can adjust. Keep all requests within the limits in API Configuration and Limits, including requests from other applications or workers using the same API key.

Signing helper

import base64
import hashlib
import json
import os
import time

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


def b64url_nopad(raw: bytes) -> str:
    """base64url without '=' padding, as required for EXN-DATA, EXN-SIGN, and body_hash."""
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")


def load_private_key() -> Ed25519PrivateKey:
    """Load the Ed25519 signing key from a PEM (PKCS#8) file.
    Replace this function if your key material uses another format."""
    with open(os.environ["EXN_PRIVATE_KEY_PATH"], "rb") as key_file:
        key = serialization.load_pem_private_key(key_file.read(), password=None)
    if not isinstance(key, Ed25519PrivateKey):
        raise ValueError("EXN_PRIVATE_KEY_PATH must point to an Ed25519 private key")
    return key


def signed_headers(method: str, path: str, body: bytes, idempotency_key: str) -> dict:
    """Build the EXN-* headers for one request.

    - `path` is the request path with the query string exactly as it is sent.
    - `body` is the exact body bytes that are sent; b"" for GET requests.
    - `idempotency_key` is "" for GET requests.
    """
    api_key = os.environ["EXN_API_KEY"]
    timestamp = int(time.time() * 1000)  # Unix time in milliseconds
    payload = {
        "api_key": api_key,
        "idempotency_key": idempotency_key,
        "timestamp": timestamp,
        "sign_version": 1,
        "method": method.upper(),
        "path": path,
        "body_hash": b64url_nopad(hashlib.sha256(body).digest()),
    }
    payload_bytes = json.dumps(payload, separators=(",", ":")).encode()
    signature = load_private_key().sign(payload_bytes)  # signs the decoded payload bytes
    return {
        "EXN-API-KEY": api_key,
        "EXN-IDEMPOTENCY-KEY": idempotency_key,
        "EXN-TIMESTAMP": str(timestamp),
        "EXN-SIGN-VERSION": "1",
        "EXN-DATA": b64url_nopad(payload_bytes),
        "EXN-SIGN": b64url_nopad(signature),
    }

HTTP helper and error handling

Every error response contains one error with code, error_message, and description. The helper raises ApiError for every HTTP status of 400 or above, so your code branches on error_message or code, never on the description text.

For success, the helper requires the expected HTTP status and a JSON object. It does not follow redirects. An unexpected status or an unreadable success response raises UnexpectedResponse; after transfer creation, this leaves the outcome unknown.

from urllib.parse import quote

import requests

HOST = os.environ["EXN_HOST"]
TIMEOUT_SECONDS = 30  # example value: the client-side timeout of one request


PUBLIC_ERRORS = {  # code -> error_message: the public error codes of this API
    2: "SYSTEM_TEMPORARY_UNAVAILABLE",
    3: "SYSTEM_INTERNAL_ERROR",
    1000: "AUTH_INVALID_API_KEY",
    1001: "AUTH_INVALID_SIGNATURE",
    1002: "AUTH_PERMISSION_DENIED",
    2000: "ACCOUNT_NOT_FOUND",
    3000: "REQUEST_INVALID",
    3001: "REQUEST_INVALID_PARAMETERS",
    3015: "REQUEST_RATE_LIMIT",
    3016: "REQUEST_IDEMPOTENCY_KEY_REUSED",
    7000: "TRANSFER_NOT_SUPPORTED",
    7001: "TRANSFER_DESTINATION_NOT_SUPPORTED",
    7002: "TRANSFER_SOURCE_NOT_ALLOWED",
    7003: "OPERATION_NOT_ALLOWED",
    7004: "TRANSFER_INSUFFICIENT_FUNDS",
    7005: "TRANSFER_AMOUNT_TOO_SMALL",
    7006: "TRANSFER_AMOUNT_TOO_LARGE",
    7007: "TRANSFER_LIMIT_REACHED",
    7008: "TRANSFER_OUTCOME_UNKNOWN",
    7009: "TRANSFER_MT_ACCOUNT_NOT_SUPPORTED",
    7010: "TRANSFER_NOT_FOUND",
}


class ApiError(Exception):
    def __init__(self, http_status: int, body: dict):
        self.http_status = http_status
        self.code = body.get("code")
        self.error_message = body.get("error_message")
        self.description = body.get("description")
        super().__init__(f"HTTP {http_status}: {self.code} {self.error_message}")

    @property
    def is_api_error(self) -> bool:
        """False for a non-API response: an HTML
        page, or a JSON body from a proxy or load balancer, even one with a code."""
        return (type(self.code) is int and self.code in PUBLIC_ERRORS
                and PUBLIC_ERRORS[self.code] == self.error_message)


class UnexpectedResponse(requests.RequestException):
    """The response does not confirm the result of the request."""


def send(method: str, path: str, body: bytes = b"", idempotency_key: str = "",
         *, expected_status: int = 200) -> dict:
    headers = signed_headers(method, path, body, idempotency_key)
    if body:
        headers["Content-Type"] = "application/json"
    response = requests.request(
        method, HOST + path, headers=headers, data=body or None,
        timeout=TIMEOUT_SECONDS, allow_redirects=False
    )
    try:
        payload = response.json()
    except ValueError:
        payload = None
    if response.status_code >= 400:
        raise ApiError(response.status_code, payload if isinstance(payload, dict) else {})
    if response.status_code != expected_status:
        raise UnexpectedResponse(f"Expected HTTP {expected_status}, got {response.status_code}")
    if not isinstance(payload, dict):
        raise UnexpectedResponse(f"HTTP {response.status_code} without a readable JSON object")
    return payload

Read a balance

def get_balance(account_id: str) -> dict:
    """Returns {"account_id": ..., "balance": ..., "currency": ...}.
    `balance` is the amount available for transfers, as a decimal string."""
    return send("GET", f"/v1/accounts/{quote(account_id, safe='')}/balance")

Create a transfer and track its status

Build the request body once and keep it unchanged for every retry of the same transfer.

import uuid

FINAL_STATUSES = {"done", "rejected"}


def new_idempotency_key() -> str:
    """One key per transfer: a random UUID in its canonical text form."""
    return str(uuid.uuid4())


def transfer_body(destination_type: str, source_account: str,
                  destination_account: str, amount: str) -> bytes:
    """`amount` is a decimal string in the currency of the source account."""
    return json.dumps(
        {
            "destination_type": destination_type,
            "source_account": source_account,
            "destination_account": destination_account,
            "amount": amount,
        },
        separators=(",", ":"),
    ).encode()


def create_transfer(body: bytes, idempotency_key: str) -> dict:
    """Returns {"transfer_id": ..., "source": {...}, "destination": {...}} on HTTP 201."""
    created = send("POST", "/v1/transfers", body, idempotency_key, expected_status=201)
    transfer_id = created.get("transfer_id")
    if not (isinstance(transfer_id, str) and transfer_id.isascii() and transfer_id.isdigit()):
        raise UnexpectedResponse("HTTP 201 without a valid transfer_id")
    return created


def get_transfer(transfer_id: str) -> dict:
    """Returns {"transfer_id": ..., "status": "processing" | "done" | "rejected"}."""
    return send("GET", f"/v1/transfers/{quote(transfer_id, safe='')}")


def wait_for_final_status(transfer_id: str, check_interval_seconds: float,
                          max_wait_seconds: float) -> dict:
    """Check the transfer status until it is final or the waiting time is over.

    We recommend checking every 2 seconds. Keep all status requests within
    the limit of 3 requests per second for your API key.
    max_wait_seconds controls how long this function keeps checking.
    It does not cancel the transfer. If the status is still `processing`,
    return the latest response and check it again later.
    """
    deadline = time.monotonic() + max_wait_seconds
    while True:
        transfer = get_transfer(transfer_id)
        if transfer["status"] in FINAL_STATUSES:
            return transfer
        if time.monotonic() + check_interval_seconds > deadline:
            return transfer
        time.sleep(check_interval_seconds)

If the example returns processing, the transfer is still in progress. Contact Customer Support with the transfer_id if it stays processing beyond the waiting time chosen for your application.

Recover from an unknown outcome

A timeout does not mean the transfer failed. create_with_recovery() retries the same request with the same idempotency key and returns the response with transfer_id on success.

If it raises OutcomeUnknown, the transfer may already exist. If record_closed is True, do not send that record again. Follow Recovery and Idempotency for the next steps.

Save records in your application

This example keeps records in memory. Save the complete record before sending a request and restore it after a restart. Allow only one sender to process each record at a time.

import threading
from datetime import datetime, timezone

RETRY_WITH_SAME_KEY = {
    "TRANSFER_OUTCOME_UNKNOWN",
    "SYSTEM_TEMPORARY_UNAVAILABLE",
    "SYSTEM_INTERNAL_ERROR",
}
FIX_AND_RESEND = {
    "AUTH_INVALID_API_KEY",
    "AUTH_INVALID_SIGNATURE",
    "AUTH_PERMISSION_DENIED",
    "REQUEST_RATE_LIMIT",
}


class OutcomeUnknown(Exception):
    """No attempt returned 201, and an earlier attempt may have created the transfer.
    Keep the idempotency key and the body. Never send the transfer with a new key
    while the outcome is unknown."""

    def __init__(self, idempotency_key: str, last_error: Exception, record_closed: bool = False):
        self.idempotency_key = idempotency_key
        self.last_error = last_error
        self.record_closed = record_closed
        super().__init__(f"outcome unknown for key {idempotency_key}: {last_error}")


class RecordClosed(Exception):
    """The record is created or closed: its key is never sent again."""


class TransferRecord:
    """State for one transfer, kept in memory here. In your integration, save each
    change in one database transaction and process the record with one sender."""

    def __init__(self, body: bytes, idempotency_key: str, *, state: str = "new"):
        if state not in ("new", "reserved"):
            raise ValueError("Only new or reserved records can be sent")
        self.body = body
        self.idempotency_key = idempotency_key
        self.attempt_times = []
        self.state = state  # new | reserved | created | closed
        self.transfer_id = None
        self._lock = threading.Lock()

    def before_send(self) -> tuple[int, bool]:
        """Records an attempt before it is sent and reserves the key. Returns the attempt
        number and whether the key was already reserved. If it raises, nothing is sent."""
        with self._lock:
            if self.state in ("created", "closed"):
                raise RecordClosed(f"record {self.idempotency_key} is {self.state}")
            reserved_before = self.state == "reserved"
            self.attempt_times.append(datetime.now(timezone.utc).isoformat())
            self.state = "reserved"
            return len(self.attempt_times), reserved_before

    def no_transfer_created(self, attempt: int, close: bool) -> bool:
        """Records that this attempt created no transfer. Returns False and changes
        nothing if another attempt was recorded after it."""
        with self._lock:
            if self.state != "reserved" or attempt != len(self.attempt_times):
                return False
            self.state = "closed" if close else "new"
            return True

    def created(self, transfer_id: str) -> None:
        with self._lock:
            self.state, self.transfer_id = "created", transfer_id

    def close(self) -> bool:
        """Retires the key before you stop sending without a 201. Pending resends of
        this record are then refused. Returns False for a created record."""
        with self._lock:
            if self.state == "created":
                return False
            self.state = "closed"
            return True


def is_unknown_outcome(error: ApiError) -> bool:
    return not error.is_api_error or error.error_message in RETRY_WITH_SAME_KEY


def create_with_recovery(record: TransferRecord, max_attempts: int,
                         backoff_base_seconds: float) -> dict:
    last_unknown = None
    for attempt in range(1, max_attempts + 1):
        try:
            number, reserved_before = record.before_send()
        except RecordClosed:
            raise
        except Exception as error:  # the record could not be written: nothing was sent
            if record.state == "reserved":
                raise OutcomeUnknown(record.idempotency_key, error) from error
            raise
        try:
            created = create_transfer(record.body, record.idempotency_key)
        except requests.RequestException as error:
            last_unknown = error  # no readable response: the transfer may exist
        except ApiError as error:
            if is_unknown_outcome(error):
                last_unknown = error
            elif not reserved_before and record.no_transfer_created(
                    number, close=error.error_message not in FIX_AND_RESEND):
                raise
            elif (error.error_message in FIX_AND_RESEND
                  or error.error_message == "REQUEST_IDEMPOTENCY_KEY_REUSED"):
                # Fix the cause, or load the original request, before resending.
                raise OutcomeUnknown(record.idempotency_key, error) from error
            else:
                closed = record.close()  # stop: retire the key before checking Transaction History
                raise OutcomeUnknown(record.idempotency_key, error, record_closed=closed) from error
        except Exception as error:  # the request may have been sent
            raise OutcomeUnknown(record.idempotency_key, error) from error
        else:
            record.created(created["transfer_id"])
            return created
        if attempt < max_attempts:
            time.sleep(backoff_base_seconds * 2 ** (attempt - 1))
    raise OutcomeUnknown(record.idempotency_key, last_unknown)

if __name__ == "__main__":
    import sys

    body = transfer_body("own_account", os.environ["EXN_SOURCE_ACCOUNT"],
                         os.environ["EXN_DESTINATION_ACCOUNT"], os.environ["EXN_AMOUNT"])
    resume_key = os.environ.get("EXN_IDEMPOTENCY_KEY")
    if resume_key == "":
        sys.exit("EXN_IDEMPOTENCY_KEY must not be empty when resuming a transfer.")
    key = resume_key if resume_key is not None else new_idempotency_key()
    record = TransferRecord(body, key, state="reserved" if resume_key is not None else "new")
    # In your integration, load or store the complete record in your database.
    print(f"Idempotency key: {key}", flush=True)
    try:
        created = create_with_recovery(record, max_attempts=5, backoff_base_seconds=1.0)  # example values
    except OutcomeUnknown as unknown:
        if unknown.record_closed:
            sys.exit(f"Outcome unknown for key {key}: {unknown.last_error}. Do not run the "
                     "script again for this transfer; check Transaction History.")
        if (isinstance(unknown.last_error, ApiError)
                and unknown.last_error.error_message == "REQUEST_IDEMPOTENCY_KEY_REUSED"):
            sys.exit(f"Outcome unknown for key {key}: the request differs from the original. "
                     "Recover the original accounts and amount before resending with this key. "
                     "If you cannot, stop using the key and check Transaction History.")
        sys.exit(f"Outcome unknown for key {key}: {unknown.last_error}. Run the script again "
                 f"only with EXN_IDEMPOTENCY_KEY={key} and the same accounts and amount, "
                 "after fixing authentication or permission errors, or waiting after rate limiting.")
    except ApiError as error:
        if record.state == "new":
            sys.exit(f"No transfer created: {error}. Fix the cause or wait after rate limiting, "
                     f"then retry with EXN_IDEMPOTENCY_KEY={key} and the same accounts and amount.")
        sys.exit(f"No transfer created: {error}. This key is closed; fix the cause before "
                 "starting a new transfer with a new key.")
    transfer_id = created["transfer_id"]
    print(f"Transfer ID: {transfer_id}", flush=True)
    try:
        final = wait_for_final_status(
            transfer_id,
            check_interval_seconds=2.0,  # recommended interval between status checks
            max_wait_seconds=30.0,  # example total waiting time
        )
    except (ApiError, requests.RequestException) as error:
        sys.exit(f"Transfer {transfer_id} was created, but its status could not be read: {error}. "
                 f"Check GET /v1/transfers/{transfer_id} later; do not create another transfer.")
    print(final)

For the retry rules of every error code, see Troubleshooting Your API Requests.

Related pages

Recovery and idempotency · Request signing · API Reference