API Client & SDK Design

Role

You are an elite API integration engineer. You build resilient HTTP clients, design

developer-friendly SDKs, and handle every edge case of API communication including

pagination, rate limits, retries, and authentication refresh.

Part 1: Production httpx Client

import httpx
from typing import Any

class APIClient:
    def __init__(self, base_url: str, api_key: str | None = None,
                 timeout: float = 30, max_retries: int = 3):
        self.base_url = base_url.rstrip("/")
        self.max_retries = max_retries
        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=httpx.Timeout(timeout, connect=10),
            headers={"Authorization": f"Bearer {api_key}"} if api_key else {},
            follow_redirects=True,
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        last_error = None
        for attempt in range(self.max_retries + 1):
            try:
                resp = await self._client.request(method, path, **kwargs)
                if resp.status_code == 429:
                    retry_after = int(resp.headers.get("Retry-After", 5))
                    await asyncio.sleep(retry_after)
                    continue
                resp.raise_for_status()
                return resp.json() if resp.content else {}
            except httpx.HTTPStatusError as e:
                if e.response.status_code < 500:
                    raise  # don't retry client errors
                last_error = e
            except (httpx.ConnectError, httpx.ReadTimeout) as e:
                last_error = e
            if attempt < self.max_retries:
                await asyncio.sleep(2 ** attempt)
        raise last_error

    async def get(self, path: str, params: dict | None = None) -> dict:
        return await self._request("GET", path, params=params)

    async def post(self, path: str, data: dict | None = None) -> dict:
        return await self._request("POST", path, json=data)

    async def put(self, path: str, data: dict | None = None) -> dict:
        return await self._request("PUT", path, json=data)

    async def delete(self, path: str) -> dict:
        return await self._request("DELETE", path)

    async def close(self):
        await self._client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

Part 2: Pagination Handling

from typing import AsyncIterator

class PaginatedClient(APIClient):

    async def paginate_offset(self, path: str, page_size: int = 100,
                              params: dict | None = None) -> AsyncIterator[dict]:
        """Offset-based pagination."""
        offset = 0
        base_params = params or {}
        while True:
            page_params = {**base_params, "limit": page_size, "offset": offset}
            resp = await self.get(path, params=page_params)
            items = resp.get("data", resp.get("results", []))
            for item in items:
                yield item
            if len(items) < page_size:
                break
            offset += page_size

    async def paginate_cursor(self, path: str, page_size: int = 100,
                              cursor_field: str = "next_cursor") -> AsyncIterator[dict]:
        """Cursor-based pagination."""
        cursor = None
        while True:
            params = {"limit": page_size}
            if cursor:
                params["cursor"] = cursor
            resp = await self.get(path, params=params)
            for item in resp.get("data", []):
                yield item
            cursor = resp.get(cursor_field)
            if not cursor:
                break

    async def paginate_link(self, path: str) -> AsyncIterator[dict]:
        """Link header pagination (GitHub style)."""
        url = path
        while url:
            resp = await self._client.get(url)
            resp.raise_for_status()
            for item in resp.json():
                yield item
            # Parse Link header
            link = resp.headers.get("Link", "")
            url = None
            for part in link.split(","):
                if 'rel="next"' in part:
                    url = part.split(";")[0].strip(" <>")

# Usage
async for user in client.paginate_cursor("/api/users"):
    process(user)

Part 3: Rate Limit Management

import asyncio, time

class RateLimiter:
    """Token bucket rate limiter."""
    def __init__(self, max_requests: int, per_seconds: float):
        self.max_tokens = max_requests
        self.tokens = max_requests
        self.per_seconds = per_seconds
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.max_tokens, self.tokens + elapsed * (self.max_tokens / self.per_seconds))
            self.last_refill = now

            if self.tokens < 1:
                wait = (1 - self.tokens) * (self.per_seconds / self.max_tokens)
                await asyncio.sleep(wait)
                self.tokens = 0
            else:
                self.tokens -= 1

class RateLimitedClient(APIClient):
    def __init__(self, *args, requests_per_second: int = 10, **kwargs):
        super().__init__(*args, **kwargs)
        self.limiter = RateLimiter(requests_per_second, 1.0)

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        await self.limiter.acquire()
        return await super()._request(method, path, **kwargs)

Part 4: OAuth Token Refresh

import time

class OAuthClient(APIClient):
    def __init__(self, base_url: str, client_id: str, client_secret: str,
                 token_url: str, **kwargs):
        super().__init__(base_url, **kwargs)
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.access_token: str | None = None
        self.refresh_token: str | None = None
        self.token_expires_at: float = 0

    async def _ensure_token(self):
        if self.access_token and time.time() < self.token_expires_at - 60:
            return
        if self.refresh_token:
            await self._refresh()
        else:
            await self._client_credentials()

    async def _client_credentials(self):
        resp = await self._client.post(self.token_url, data={
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
        })
        resp.raise_for_status()
        self._set_tokens(resp.json())

    async def _refresh(self):
        resp = await self._client.post(self.token_url, data={
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token,
            "client_id": self.client_id,
        })
        if resp.status_code == 401:
            self.refresh_token = None
            return await self._client_credentials()
        resp.raise_for_status()
        self._set_tokens(resp.json())

    def _set_tokens(self, data: dict):
        self.access_token = data["access_token"]
        self.refresh_token = data.get("refresh_token", self.refresh_token)
        self.token_expires_at = time.time() + data.get("expires_in", 3600)
        self._client.headers["Authorization"] = f"Bearer {self.access_token}"

    async def _request(self, method: str, path: str, **kwargs) -> dict:
        await self._ensure_token()
        try:
            return await super()._request(method, path, **kwargs)
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 401:
                self.access_token = None
                await self._ensure_token()
                return await super()._request(method, path, **kwargs)
            raise

Part 5: Webhook Handling

import hmac, hashlib, json
from fastapi import FastAPI, Request, Header, HTTPException

app = FastAPI()

def verify_webhook_signature(payload: bytes, signature: str, secret: str,
                             algorithm: str = "sha256") -> bool:
    expected = hmac.new(secret.encode(), payload, getattr(hashlib, algorithm)).hexdigest()
    return hmac.compare_digest(f"{algorithm}={expected}", signature)

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request, stripe_signature: str = Header()):
    body = await request.body()
    if not verify_webhook_signature(body, stripe_signature, STRIPE_SECRET, "sha256"):
        raise HTTPException(401, "Invalid signature")

    event = json.loads(body)
    match event["type"]:
        case "payment_intent.succeeded":
            await handle_payment_success(event["data"]["object"])
        case "customer.subscription.deleted":
            await handle_subscription_cancel(event["data"]["object"])

    return {"received": True}

# Idempotent webhook processing
from functools import lru_cache

processed_events: set[str] = set()

@app.post("/webhooks/generic")
async def generic_webhook(request: Request):
    body = await request.json()
    event_id = body.get("id") or body.get("event_id")

    if event_id in processed_events:
        return {"status": "already_processed"}

    await process_event(body)
    processed_events.add(event_id)
    return {"status": "processed"}

Part 6: SDK Design Pattern

class MyServiceSDK:
    """Developer-friendly SDK wrapping REST API."""

    def __init__(self, api_key: str, base_url: str = "https://api.myservice.com/v1"):
        self._client = RateLimitedClient(base_url, api_key=api_key, requests_per_second=50)
        self.users = UserResource(self._client)
        self.orders = OrderResource(self._client)

    async def close(self):
        await self._client.close()

    async def __aenter__(self): return self
    async def __aexit__(self, *a): await self.close()

class UserResource:
    def __init__(self, client: APIClient):
        self._client = client

    async def list(self, page_size: int = 50) -> AsyncIterator[dict]:
        async for item in self._client.paginate_cursor("/users", page_size=page_size):
            yield item

    async def get(self, user_id: str) -> dict:
        return await self._client.get(f"/users/{user_id}")

    async def create(self, email: str, name: str, **kwargs) -> dict:
        return await self._client.post("/users", data={"email": email, "name": name, **kwargs})

    async def update(self, user_id: str, **kwargs) -> dict:
        return await self._client.put(f"/users/{user_id}", data=kwargs)

    async def delete(self, user_id: str) -> dict:
        return await self._client.delete(f"/users/{user_id}")

# Usage
async with MyServiceSDK("sk-xxx") as sdk:
    user = await sdk.users.create("[email protected]", "James")
    async for order in sdk.orders.list(user_id=user["id"]):
        print(order)

Part 7: Idempotency Keys

import uuid

class IdempotentClient(APIClient):
    async def post_idempotent(self, path: str, data: dict,
                              idempotency_key: str | None = None) -> dict:
        key = idempotency_key or str(uuid.uuid4())
        return await self._request("POST", path, json=data,
                                   headers={"Idempotency-Key": key})

# Server-side idempotency
from datetime import datetime, timedelta

idempotency_store: dict[str, tuple[dict, datetime]] = {}

async def check_idempotency(key: str) -> dict | None:
    if key in idempotency_store:
        result, created = idempotency_store[key]
        if datetime.utcnow() - created < timedelta(hours=24):
            return result
        del idempotency_store[key]
    return None

async def store_idempotency(key: str, result: dict):
    idempotency_store[key] = (result, datetime.utcnow())

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)