Python Example

Using PyJWT with cryptography for JWT signing.

Install

pip install PyJWT cryptography requests

Complete example

import jwt
import uuid
import time
import requests

# --- Configuration ---
PARTNER_ID = "your-partner-id-uuid"
API_KEY_ID = "your-api-key-id-uuid"
TOKEN_ENDPOINT = "https://prod.rynopay.io/auth/partner/v1/token"
API_BASE_URL = "https://prod.rynopay.io"

with open("private-key.pem", "r") as f:
    PRIVATE_KEY = f.read()


# --- Token Manager ---
class TokenManager:
    def __init__(self):
        self._token = None
        self._expires_at = 0

    def get_token(self):
        """Get a valid access token, refreshing if needed."""
        # Return cached token if still valid (with 10-minute buffer)
        if self._token and time.time() < self._expires_at - 600:
            return self._token

        self._token = self._exchange_token()
        return self._token

    def _exchange_token(self):
        now = int(time.time())

        # Build the JWT assertion
        headers = {
            "alg": "ES256",
            "typ": "JWT",
            "kid": API_KEY_ID,
        }

        payload = {
            "iss": PARTNER_ID,
            "sub": PARTNER_ID,
            "aud": "https://prod.rynopay.io/auth/partner/v1/token",
            "iat": now,
            "exp": now + 300,  # 5 minutes
            "jti": str(uuid.uuid4()),
        }

        assertion = jwt.encode(
            payload, PRIVATE_KEY, algorithm="ES256", headers=headers
        )

        # Exchange for an access token
        response = requests.post(
            TOKEN_ENDPOINT,
            json={
                "grantType": "urn:ietf:params:oauth:grant-type:jwt-bearer",
                "assertion": assertion,
                "scope": "partner.read partner.write",
            },
        )

        if not response.ok:
            error = response.json()
            raise Exception(
                f"Token exchange failed: {error.get('error', {}).get('code')} "
                f"- {error.get('error', {}).get('message')}"
            )

        data = response.json()
        self._expires_at = time.time() + data["expiresIn"]
        return data["accessToken"]

    def clear(self):
        """Clear the cached token (e.g., after a 401)."""
        self._token = None
        self._expires_at = 0


# --- Making API calls ---
token_manager = TokenManager()


def call_api(path, method="GET", json_data=None):
    """Make an authenticated API call with automatic token refresh."""
    token = token_manager.get_token()

    response = requests.request(
        method,
        f"{API_BASE_URL}{path}",
        headers={"Authorization": f"Bearer {token}"},
        json=json_data,
    )

    # Handle expired token
    if response.status_code == 401:
        token_manager.clear()
        token = token_manager.get_token()
        response = requests.request(
            method,
            f"{API_BASE_URL}{path}",
            headers={"Authorization": f"Bearer {token}"},
            json=json_data,
        )

    response.raise_for_status()
    return response.json()


# --- Usage ---
data = call_api("/api/partner/some-endpoint")
print(data)