HTTP client

The whole client — key loading, login, acting as a client, signing, retries and polling — in Python, Node and Go.

Every example in these guides calls the same small client. This page is that client, in Python, Node and Go. Copy it once and each guide fits in three to eight lines.

The surface every guide assumes

What you want	Python	Node	Go
Act for a client, or for yourself	hevn.acting_as(client_id)	hevn.actingAs(clientId)	api.ActingAs(clientID)
Read	hevn.get(path, params={…})	hevn.get(path, {…})	api.Get(path, hevn.Query{…})
Write	hevn.post(path, body, idempotency_key="…")	hevn.post(path, body, { idempotencyKey })	api.Post(path, hevn.Body{…}, hevn.IdempotencyKey("…"))
Read a response field	payout["approval"]["payload"]	payout.approval.payload	payout.Str("approval.payload")
Walk a response list	page["items"]	page.items	page.List("items")
Sign an approval	key.sign_payload(b64)	key.signPayload(b64)	key.SignPayload(b64)
Name the signing key	key.public_key	key.publicKey	key.PublicKey
Confirm to settlement	hevn.confirm_until_settled(path, sig)	hevn.confirmUntilSettled(path, sig)	api.ConfirmUntilSettled(path, sig)
Poll a read	poll_until(read, done)	pollUntil(read, done)	hevn.PollUntil(read, done, deadline)
Download an export	hevn.download(path, params, into)	hevn.download(path, params, into)	api.Download(path, hevn.Query{…}, into)

It reads three environment variables — HEVN_API, HEVN_KEY_PEM and HEVN_EMAIL — set up in the Quickstart. Nothing below hardcodes a host. HEVN_API ends in /dapi/v1 (https://sandbox-api.hevn.finance/dapi/v1); the old /v1 base answers 404 rather than redirecting, so a helper that asserts the suffix at start-up turns a whole run of mystery 404s into one clear failure.

In the Go tabs the client is a variable called api and hevn is the package: api.Post(…) is a call on your client, hevn.Body{…} is a type from the package. Naming the variable hevn shadows the package and stops the file compiling.

Binding an account

acting_as is the only thing that sets X-Hevn-Account, and three kinds of route read it differently:

Routes	The header
/banks*, /payins*, /payouts*, /contacts*, /transactions*, /documents*, /sandbox/*	optional — bound client, or your own account when unbound
/client, /client/balance, /client/kyb, /client/kyb/complete	required — the header is the only selector
POST /clients, GET /clients, every /escrow*	refused — 400 account_scope_conflict

The middle row is the one that changes how you use the client. There is no client id in any path any more, so reading your own account means binding your own cl_… too — the userId the login answered:

Python

me = hevn.acting_as(os.environ["HEVN_USER_ID"])   # the userId login answered
mine = me.get("/client/balance")

An unbound client calling /client* sends no header at all, which is a 422 validation_failed with details.location: "header" and X-Hevn-Account in details.fields — not a read of your own account. A bound client calling POST /clients or any escrow route is the mirror image, 400 account_scope_conflict with details.reason: "selfScopedRoute", so keep the unbound client for those.

Install

Python

python -m venv .venv && source .venv/bin/activate
pip install cryptography httpx

Load the developer key

The private half comes from HEVN_KEY_PEM and never leaves the process.

Python

import base64, os
from cryptography.hazmat.primitives import serialization


def load_key():
    with open(os.environ["HEVN_KEY_PEM"], "rb") as handle:
        return serialization.load_pem_private_key(handle.read(), password=None)


def public_key_b64(key) -> str:
    spki = key.public_key().public_bytes(
        serialization.Encoding.DER,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    return base64.b64encode(spki).decode()

Sign bytes

Two functions, used by login and by every payment: ECDSA P-256 over SHA-256, DER-encoded, base64 out. Signing is why each rule is there.

Python

import base64
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec


def sign_bytes(key, message: bytes) -> str:
    """base64(DER ECDSA P-256 / SHA-256) over exactly these bytes."""
    return base64.b64encode(key.sign(message, ec.ECDSA(hashes.SHA256()))).decode()


def sign_payload(key, payload_b64: str) -> str:
    return sign_bytes(key, base64.b64decode(payload_b64))

Wrap the key

The guides call key.sign_payload(...) and read key.public_key. That is these two functions bound to one loaded key.

Python

class Key:
    def __init__(self):
        self.raw = load_key()                     # the loaded key login() takes
        self.public_key = public_key_b64(self.raw)

    def sign_bytes(self, message: bytes) -> str:
        return sign_bytes(self.raw, message)

    def sign_payload(self, payload_b64: str) -> str:
        return sign_payload(self.raw, payload_b64)

Log in

Two calls, two signatures. The result is an access token good for an hour and a refresh token good for sixty days — see Sessions.

Python

import os, secrets, time, httpx


def login(key) -> dict:
    api, email = os.environ["HEVN_API"], os.environ["HEVN_EMAIL"]
    nonce = secrets.randbits(48)
    expiry = int(time.time() * 1000) + 120_000
    proof = f"hevn-developer-key-login:{email}:{nonce}:{expiry}".encode()

    started = httpx.post(f"{api}/auth/challenge", json={
        "email": email,
        "publicKey": public_key_b64(key),
        "nonce": nonce,
        "requestExpiry": expiry,
        "signature": sign_bytes(key, proof),
    })
    started.raise_for_status()
    challenge = started.json()

    minted = httpx.post(f"{api}/auth/token", json={
        "challengeId": challenge["challengeId"],
        "signature": sign_payload(key, challenge["payload"]),
    })
    minted.raise_for_status()
    return minted.json()  # accessToken, refreshToken, expiresIn, userId

Hold the session

One object owns the tokens and re-mints the access token on demand, so the request layer below never thinks about expiry.

Python

class Session:
    def __init__(self, key):
        self.key = key
        self._tokens, self._expires_at = None, 0.0

    def token(self, refresh: bool = False) -> str:
        if refresh or self._tokens is None or time.time() > self._expires_at - 60:
            self._tokens = login(self.key.raw) if self._tokens is None else self._refresh()
            self._expires_at = time.time() + self._tokens["expiresIn"]
        return self._tokens["accessToken"]

    def _refresh(self) -> dict:
        response = httpx.post(
            f"{os.environ['HEVN_API']}/auth/refresh",
            json={},
            headers={"Authorization": f"Bearer {self._tokens['refreshToken']}"},
        )
        response.raise_for_status()
        return {**self._tokens, **response.json()}

Send a request

One place turns the error envelope into a typed error, honours Retry-After on a 429, re-mints once on a 401, and adds X-Hevn-Account when the client is bound.

Python

import os, time, httpx


class HevnError(Exception):
    def __init__(self, status: int, body: dict):
        error = (body or {}).get("error", {})
        self.status, self.code = status, error.get("code", "internal_error")
        self.details = error.get("details") or {}
        super().__init__(f"{status} {self.code}: {error.get('message', '')}")


class Hevn:
    def __init__(self, session, client_id=None, http=None):
        self.session, self.client_id = session, client_id
        self.http = http or httpx.Client(base_url=os.environ["HEVN_API"], timeout=30)

    def acting_as(self, client_id):
        """A view of the same session that sends X-Hevn-Account on every call."""
        return Hevn(self.session, client_id, self.http)

    def get(self, path, params=None):
        return self.call("GET", path, params=params)

    def post(self, path, body=None, *, idempotency_key=None):
        return self.call("POST", path, body=body, idempotency_key=idempotency_key)

    def put(self, path, body=None, *, idempotency_key=None):
        return self.call("PUT", path, body=body, idempotency_key=idempotency_key)

    def patch(self, path, body=None):
        return self.call("PATCH", path, body=body)

    def delete(self, path):
        return self.call("DELETE", path)

    def call(self, method, path, *, body=None, params=None, idempotency_key=None, retried=False):
        headers = {"Authorization": f"Bearer {self.session.token()}"}
        if self.client_id:
            headers["X-Hevn-Account"] = self.client_id
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key
        again = dict(body=body, params=params, idempotency_key=idempotency_key)
        response = self.http.request(method, path, json=body, params=params, headers=headers)
        if response.status_code == 429:
            time.sleep(float(response.headers.get("Retry-After", "1")))
            return self.call(method, path, retried=retried, **again)
        if response.status_code == 401 and not retried:
            self.session.token(refresh=True)
            return self.call(method, path, retried=True, **again)
        payload = response.json() if response.content else {}
        if response.status_code >= 400:
            raise HevnError(response.status_code, payload)
        return payload

Download an export

GET /dapi/v1/transactions/export streams a file rather than JSON, so it gets its own method: same headers, same typed error, bytes to disk instead of a parsed body.

Python

def download(self, path, params, into):
    """A method on Hevn. An export streams a file; every other route answers JSON."""
    headers = {"Authorization": f"Bearer {self.session.token()}"}
    if self.client_id:
        headers["X-Hevn-Account"] = self.client_id
    with self.http.stream("GET", path, params=params, headers=headers) as response:
        if response.status_code >= 400:
            response.read()
            raise HevnError(response.status_code, response.json())
        with open(into, "wb") as file:
            for chunk in response.iter_bytes():
                file.write(chunk)
    return into

Confirm until it settles

The whole retry policy as one method: already_funded is success, funding_attempt_expired means re-open, a 202 and the bundler codes mean confirm again, anything else propagates.

Python

CONFIRM_AGAIN = {"funding_in_progress", "bundler_rejected", "bundler_unavailable"}


class ReopenRequired(Exception):
    """The approval expired. Re-open the payment with the same Idempotency-Key."""


def confirm_until_settled(self, confirm_path: str, signature: str, *, deadline=180.0):
    """A method on Hevn. confirm_path is "/payouts/po_…/confirm" or an escrow action's confirm."""
    started, delay = time.monotonic(), 1.0
    while True:
        try:
            receipt = self.post(confirm_path, {"signature": signature})
            if receipt["status"] != "submitted":
                return receipt
            reason = "submitted"
        except HevnError as error:
            if error.code == "already_funded":
                return {"status": "settled", "transactionHash": error.details.get("transactionHash")}
            if error.code == "funding_attempt_expired":
                raise ReopenRequired(confirm_path) from error
            if error.code not in CONFIRM_AGAIN:
                raise
            reason = error.code
        if time.monotonic() - started > deadline:
            raise TimeoutError(f"{confirm_path} still {reason} after {deadline:.0f}s")
        time.sleep(delay)
        delay = min(delay * 2, 8.0)

Poll a read

Client provisioning, KYB review, rail activation and payin settlement all end in a read that changes on someone else’s clock.

Python

import random, time


def poll_until(read, done, *, deadline=120.0, first=1.0, cap=8.0):
    """Call read() until done(value) is true. Returns the last value or raises TimeoutError."""
    started, delay = time.monotonic(), first
    while True:
        value = read()
        if done(value):
            return value
        if time.monotonic() - started > deadline:
            raise TimeoutError(f"still {value.get('status')} after {deadline:.0f}s")
        time.sleep(delay + random.uniform(0, delay / 2))
        delay = min(delay * 2, cap)

The Go plumbing

Python and Node are complete above. Go’s helpers are worth spelling out once.

Go

type Client struct {
	API      string
	ClientID string
	HTTP     *http.Client
	Session  *Session
}

type Opts struct {
	Query          Query
	IdempotencyKey string
}

type Opt func(*Opts)

func IdempotencyKey(key string) Opt { return func(o *Opts) { o.IdempotencyKey = key } }

func options(opts []Opt) Opts {
	var o Opts
	for _, apply := range opts {
		apply(&o)
	}
	return o
}

func first(queries []Query) Query {
	if len(queries) == 0 {
		return nil
	}
	return queries[0]
}

// at walks a dotted path and returns nil when any step is missing.
func (p Payload) at(path string) any {
	var node any = map[string]any(p)
	for _, step := range strings.Split(path, ".") {
		object, ok := node.(map[string]any)
		if !ok {
			return nil
		}
		node = object[step]
	}
	return node
}

func (p Payload) Str(path string) string { value, _ := p.at(path).(string); return value }
func (p Payload) Bool(path string) bool  { value, _ := p.at(path).(bool); return value }

// List walks a dotted path to an array of objects: catalogue.List("rails").
func (p Payload) List(path string) []Payload {
	items, _ := p.at(path).([]any)
	page := make([]Payload, 0, len(items))
	for _, item := range items {
		if object, ok := item.(map[string]any); ok {
			page = append(page, object)
		}
	}
	return page
}

func (c *Client) request(method, path string, body Body, o Opts) *http.Request {
	var reader io.Reader
	if body != nil {
		encoded, _ := json.Marshal(body)
		reader = bytes.NewReader(encoded)
	}
	request, _ := http.NewRequest(method, c.API+path, reader)
	token, _ := c.Session.Token(false)
	request.Header.Set("Authorization", "Bearer "+token)
	if body != nil {
		request.Header.Set("Content-Type", "application/json")
	}
	if c.ClientID != "" {
		request.Header.Set("X-Hevn-Account", c.ClientID)
	}
	if o.IdempotencyKey != "" {
		request.Header.Set("Idempotency-Key", o.IdempotencyKey)
	}
	if len(o.Query) > 0 {
		query := request.URL.Query()
		for name, value := range o.Query {
			query.Set(name, value)
		}
		request.URL.RawQuery = query.Encode()
	}
	return request
}

func decode(res *http.Response) (Payload, int) {
	defer res.Body.Close()
	var payload Payload
	_ = json.NewDecoder(res.Body).Decode(&payload)
	return payload, res.StatusCode
}

func retryAfter(res *http.Response) time.Duration {
	seconds, err := strconv.Atoi(res.Header.Get("Retry-After"))
	if err != nil || seconds <= 0 {
		return time.Second
	}
	return time.Duration(seconds) * time.Second
}

func asAPIError(status int, payload Payload) *APIError {
	body, _ := payload["error"].(map[string]any)
	details, _ := body["details"].(map[string]any)
	code, _ := body["code"].(string)
	message, _ := body["message"].(string)
	if code == "" {
		code = "internal_error"
	}
	return &APIError{Status: status, Code: code, Message: message, Details: details}
}

Put it together

Python

key = Key()
hevn = Hevn(Session(key))                      # unbound: /clients and /escrow
northwind = hevn.acting_as("cl_7YQ2Kf3mN8")    # bound to one client
me = hevn.acting_as(os.environ["HEVN_USER_ID"])  # bound to your own account, for /client*

Back to the Quickstart

Nine steps in the sandbox, from an empty account to money that moved on chain.