Error Handling Patterns

Production-grade error handling for the Mavera API: retries, backoff, user-facing messages, and structured logging

When to Use This

You’re building a production integration and need to:

• Retry transient errors (429, 500, 503) with exponential backoff and jitter
• Distinguish between retryable and non-retryable errors
• Show appropriate messages to end users (don’t expose internal details)
• Log errors in a structured way for debugging and support
• Handle credits (402), auth (401), and validation (400) without retries

---

Error Categories: Retry vs Don’t Retry

Status	Code (typical)	Retry?	Action
429	rate_limit_exceeded	✅ Yes	Back off, respect Retry-After
500	internal_error	✅ Yes	Transient; retry with backoff
503	service_unavailable	✅ Yes	Transient; retry with backoff
401	invalid_api_key, missing_api_key	❌ No	Fix key, don’t retry
402	credits_exhausted, budget_exceeded	❌ No	Add credits or wait
400	invalid_model, invalid_persona, etc.	❌ No	Fix request
404	resource_not_found	❌ No	Fix resource ID
403	forbidden	❌ No	Fix permissions

---

Centralized Error Handler (Python)

A reusable wrapper that parses Mavera errors, decides whether to retry, and surfaces structured data for logging and user messages.

Python

import time
import random
import logging
from typing import Optional, Callable, TypeVar, Any

from openai import OpenAI, APIError, AuthenticationError, RateLimitError, APIConnectionError

logger = logging.getLogger(__name__)

T = TypeVar("T")

# Mavera base URL
MAVERA_BASE = "https://app.mavera.io/api/v1"


def parse_mavera_error(error: Exception) -> dict:
    """Extract structured error info from Mavera/OpenAI SDK errors."""
    result = {
        "retryable": False,
        "status_code": None,
        "code": None,
        "message": str(error),
        "param": None,
    }

    if hasattr(error, "status_code"):
        result["status_code"] = error.status_code
    if hasattr(error, "response") and error.response is not None:
        try:
            body = error.response.json()
            err = body.get("error", {})
            result["code"] = err.get("code")
            result["message"] = err.get("message", result["message"])
            result["param"] = err.get("param")
        except Exception:
            pass

    # Determine retryability
    if result["status_code"] in (429, 500, 503):
        result["retryable"] = True
    if isinstance(error, APIConnectionError):
        result["retryable"] = True

    return result


def retry_with_backoff(
    func: Callable[[], T],
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
) -> T:
    """
    Call func(); on retryable errors, wait and retry.
    Uses exponential backoff with optional jitter.
    """
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            return func()
        except (APIError, APIConnectionError, RateLimitError) as e:
            last_error = e
            info = parse_mavera_error(e)

            if not info["retryable"] or attempt == max_retries:
                raise

            delay = base_delay * (2 ** attempt)
            if jitter:
                delay += random.uniform(0, base_delay)
            delay = min(delay, max_delay)

            retry_after = None
            if hasattr(e, "response") and e.response is not None:
                retry_after = e.response.headers.get("Retry-After")
            if retry_after is not None:
                try:
                    delay = max(delay, float(retry_after))
                except ValueError:
                    pass

            logger.warning(
                "Mavera API error (retryable), attempt %d/%d, waiting %.1fs",
                attempt + 1,
                max_retries + 1,
                delay,
                extra={"mavera_error": info},
            )
            time.sleep(delay)
        except AuthenticationError:
            raise  # Never retry auth errors
    raise last_error


def user_facing_message(error: Exception) -> str:
    """Return a safe, user-facing message for display in UI."""
    info = parse_mavera_error(error)
    status = info.get("status_code")
    code = info.get("code")

    if status == 401 or code in ("missing_api_key", "invalid_api_key", "revoked_api_key"):
        return "Authentication failed. Please check your API key."
    if status == 402 or code in ("credits_exhausted", "budget_exceeded"):
        return "You've run out of credits. Please add more or try again later."
    if status == 429 or code == "rate_limit_exceeded":
        return "Too many requests. Please wait a moment and try again."
    if status in (500, 503):
        return "The service is temporarily unavailable. Please try again in a few minutes."
    if status == 404:
        return "The requested resource was not found."
    if status == 400:
        return "Invalid request. Please check your input and try again."

    return "Something went wrong. Please try again later."


# Usage with OpenAI client
client = OpenAI(api_key="mvra_live_xxx", base_url=MAVERA_BASE)


def chat_with_retry(messages, persona_id: str):
    def _call():
        return client.responses.create(
            model="mavera-1",
            input=messages,
            extra_body={"persona_id": persona_id},
        )

    try:
        return retry_with_backoff(_call)
    except APIError as e:
        logger.error("Mavera API error", extra={"mavera_error": parse_mavera_error(e)})
        raise  # Or return a fallback / show user_facing_message(e)

---

Centralized Error Handler (JavaScript/TypeScript)

JavaScript

const MAVERA_BASE = "https://app.mavera.io/api/v1";

/**
 * Parse Mavera error response into structured info.
 */
function parseMaveraError(error) {
  const result = {
    retryable: false,
    statusCode: null,
    code: null,
    message: error?.message ?? String(error),
    param: null,
  };

  if (error?.status) result.statusCode = error.status;
  if (error?.error?.code) result.code = error.error.code;
  if (error?.error?.message) result.message = error.error.message;
  if (error?.error?.param) result.param = error.error.param;

  if ([429, 500, 503].includes(result.statusCode)) result.retryable = true;

  return result;
}

/**
 * Sleep for ms milliseconds.
 */
function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

/**
 * Call fn(); on retryable errors, wait and retry with backoff.
 */
async function retryWithBackoff(fn, { maxRetries = 5, baseDelay = 1000, maxDelay = 60000, jitter = true } = {}) {
  let lastError;
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (e) {
      lastError = e;
      const info = parseMaveraError(e);

      if (!info.retryable || attempt === maxRetries) throw e;

      let delay = baseDelay * Math.pow(2, attempt);
      if (jitter) delay += Math.random() * baseDelay;
      delay = Math.min(delay, maxDelay);

      const retryAfter = e?.response?.headers?.get?.("Retry-After");
      if (retryAfter) delay = Math.max(delay, parseInt(retryAfter, 10) * 1000);

      console.warn(`Mavera API error (retryable), attempt ${attempt + 1}/${maxRetries + 1}, waiting ${delay}ms`, info);
      await sleep(delay);
    }
  }
  throw lastError;
}

/**
 * Safe user-facing message for UI.
 */
function userFacingMessage(error) {
  const info = parseMaveraError(error);
  const { statusCode, code } = info;

  if (statusCode === 401 || ["missing_api_key", "invalid_api_key", "revoked_api_key"].includes(code)) {
    return "Authentication failed. Please check your API key.";
  }
  if (statusCode === 402 || ["credits_exhausted", "budget_exceeded"].includes(code)) {
    return "You've run out of credits. Please add more or try again later.";
  }
  if (statusCode === 429 || code === "rate_limit_exceeded") {
    return "Too many requests. Please wait a moment and try again.";
  }
  if ([500, 503].includes(statusCode)) {
    return "The service is temporarily unavailable. Please try again in a few minutes.";
  }
  if (statusCode === 404) return "The requested resource was not found.";
  if (statusCode === 400) return "Invalid request. Please check your input and try again.";

  return "Something went wrong. Please try again later.";
}

// Usage with OpenAI client
import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.MAVERA_API_KEY,
  baseURL: MAVERA_BASE,
});

async function chatWithRetry(messages, personaId) {
  return retryWithBackoff(async () => {
    return client.responses.create({
      model: "mavera-1",
      input: messages,
      persona_id: personaId,
    });
  });
}

---

REST API Error Handling (Non-SDK)

For Mave, Focus Groups, Video Analysis, etc., you’re using raw HTTP. Parse the JSON error body and branch on status code.

Python

import requests

def mavera_request(method: str, path: str, **kwargs) -> dict:
    """Make a request to Mavera REST API with error handling."""
    url = f"https://app.mavera.io/api/v1{path}"
    headers = kwargs.pop("headers", {})
    headers.setdefault("Authorization", f"Bearer {os.environ['MAVERA_API_KEY']}")
    headers.setdefault("Content-Type", "application/json")

    resp = requests.request(method, url, headers=headers, **kwargs)

    if resp.status_code in (429, 500, 503):
        # Retryable — raise for retry logic to catch
        raise MaveraRetryableError(resp.status_code, resp.json())

    if not resp.ok:
        err = resp.json().get("error", {})
        raise MaveraAPIError(
            status_code=resp.status_code,
            code=err.get("code"),
            message=err.get("message", resp.text),
            param=err.get("param"),
        )

    return resp.json()


class MaveraAPIError(Exception):
    def __init__(self, status_code, code, message, param=None):
        self.status_code = status_code
        self.code = code
        self.message = message
        self.param = param


class MaveraRetryableError(MaveraAPIError):
    pass

---

Structured Logging for Support

When contacting support, include structured error data. Never log full API keys — only the prefix (e.g. mvra_live_abcd...).

Python

import logging
import json

def log_mavera_error(error: Exception, context: dict = None):
    """Log error in a format useful for debugging and support."""
    info = parse_mavera_error(error)
    log_entry = {
        "event": "mavera_api_error",
        "retryable": info["retryable"],
        "status_code": info["status_code"],
        "code": info["code"],
        "param": info["param"],
        "message": info["message"],
        "context": context or {},
    }
    # Redact sensitive fields in context
    if "api_key" in str(log_entry.get("context", {})):
        log_entry["context"] = {...}
    logging.error("Mavera API error: %s", json.dumps(log_entry))

---

Circuit Breaker (Advanced)

For high-volume systems, consider a circuit breaker: after N consecutive failures, stop calling the API for a cooldown period to avoid cascading failures.

Python

class CircuitOpenError(Exception):
    pass


class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("Circuit breaker is open")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.failures = 0
                self.state = "closed"
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "open"
            raise

---

See Also

Error Handling Guide

Error format, status codes, and error types

Rate Limits

Rate limit headers and Retry-After

Credits

Handling 402 and credit exhaustion

Rate Limits in Production

Proactive throttling and queuing