Error Handling

{
  "error": {
    "code": "invalid_request",
    "message": "The 'content' field is required",
    "type": "validation_error",
    "param": "content",
    "status": 400,
    "request_id": "req_abc123xyz"
  }
}

{
  "error": {
    "type": "...",
    "message": "...",
    "status_code": 400
  }
}

{
  "detail": [
    {
      "loc": ["body", "content"],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

POST /v1/memories HTTP/1.1
Authorization: Bearer mem_sk_...
Idempotency-Key: 9f1c2b6e-7a3d-4e51-9c08-2b5d8e4a1f60
Content-Type: application/json

# A retry of the exact same request returns the first response:
HTTP/1.1 201 Created
X-Request-ID: 3f9a0c1d-8e2b-4a77-bd31-6c2f9e0a4b15
X-Process-Time: 0.42
X-Idempotent-Replay: true

{
  "error": "feature_not_available",
  "feature": "knowledge_graph",
  "feature_name": "Knowledge Graph",
  "message": "Knowledge Graph requires Pro plan or higher.",
  "current_tier": {
    "name": "free",
    "display_name": "Free"
  },
  "required_tier": {
    "name": "pro",
    "display_name": "Pro",
    "price": "$99/month",
    "credits": "200K/month"
  },
  "upgrade_url": "https://app.hebbrix.com/dashboard/billing",
  "pricing_url": "https://app.hebbrix.com/pricing"
}

import asyncio
from hebbrix import MemoryClient
from hebbrix import (
    HebbrixError,
    AuthenticationError,
    RateLimitError,
    ValidationError,
    NotFoundError,
)

async def main():
    async with MemoryClient(api_key="mem_sk_...") as client:
        try:
            memory = await client.memories.create(
                collection_id="col_xyz",
                content="Test memory",
            )
        except AuthenticationError as e:
            # Invalid or missing API key
            print(f"Auth error: {e}")

        except RateLimitError as e:
            # Too many requests — backend returned 429
            print(f"Rate limited: {e}")

        except ValidationError as e:
            # 422 — request body/params failed backend validation
            print(f"Validation error: {e}")

        except NotFoundError as e:
            # 404 — the collection or memory doesn't exist
            print(f"Not found: {e}")

        except HebbrixError as e:
            # Catch-all for any other API error (5xx, etc.)
            print(f"API error: {e}")

asyncio.run(main())

import {
  MemoryClient,
  HebbrixError,
  AuthenticationError,
  RateLimitError,
  ValidationError,
  NotFoundError,
} from 'hebbrix';

const client = new MemoryClient({ apiKey: 'mem_sk_...' });

try {
  await client.memories.create({
    collectionId: 'col_xyz',
    content: 'Test memory',
  });
} catch (error) {
  if (error instanceof AuthenticationError) {
    // Invalid or missing API key
    console.error('Auth error:', error.message);

  } else if (error instanceof RateLimitError) {
    // Too many requests
    console.error('Rate limited:', error.message);
    // Wait a bit and retry

  } else if (error instanceof ValidationError) {
    // 422 — backend rejected the request body
    console.error('Validation error:', error.message);

  } else if (error instanceof NotFoundError) {
    // 404 — resource doesn't exist
    console.error('Not found:', error.message);

  } else if (error instanceof HebbrixError) {
    // Catch-all for other API errors
    console.error(`API error [${error.status}]: ${error.message}`);
    console.error(`Request ID: ${error.requestId}`);

  } else {
    throw error; // Re-throw unknown errors
  }
}

HTTP/1.1 429 Too Many Requests
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1705312800
Retry-After: 45

{
  "error": {
    "code": "rate_limit_exceeded",
    "message": "Rate limit exceeded. Please retry after 45 seconds.",
    "type": "rate_limit_error",
    "status": 429,
    "request_id": "req_xyz789"
  }
}

import asyncio
import random
from hebbrix import MemoryClient
from hebbrix import RateLimitError, HebbrixError

async def create_with_retry(client: MemoryClient, max_retries: int = 5):
    """Retry with exponential backoff + jitter on 429 / 5xx."""
    for attempt in range(max_retries):
        try:
            return await client.memories.create(
                collection_id="col_xyz", content="Test"
            )
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            sleep = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limited — retry in {sleep:.1f}s")
            await asyncio.sleep(sleep)
        except HebbrixError:
            if attempt == max_retries - 1:
                raise
            sleep = (2 ** attempt) + random.uniform(0, 1)
            print(f"Transient error — retry in {sleep:.1f}s")
            await asyncio.sleep(sleep)


async def main():
    async with MemoryClient(api_key="mem_sk_...") as client:
        await create_with_retry(client)


asyncio.run(main())