API Integration Skill

Overview

Build reliable integrations with REST APIs: authentication, pagination, rate limiting, retries, and error handling. Covers the full lifecycle from first call to production-ready client.

Step-by-Step Process

Step 1: Read the API Docs

Before writing any code, identify:

• Base URL and API version (https://api.example.com/v2)
• Authentication method (API key header, Bearer token, OAuth 2.0, Basic auth)
• Rate limits (requests per minute/hour, burst limits)
• Pagination style (offset/limit, cursor, page number, Link header)
• Error response format (status codes, error body schema)

Step 2: Set Up Authentication

API Key (header)

import httpx

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}
client = httpx.Client(base_url="https://api.example.com/v2", headers=headers)

OAuth 2.0 Client Credentials

import httpx

def get_access_token(client_id: str, client_secret: str, token_url: str) -> str:
    response = httpx.post(token_url, data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
    })
    response.raise_for_status()
    return response.json()["access_token"]

Basic Auth

client = httpx.Client(auth=(username, password))

Step 3: Make Requests with Retry Logic

import httpx
import time
from typing import Any

def api_request(
    client: httpx.Client,
    method: str,
    path: str,
    max_retries: int = 3,
    **kwargs: Any,
) -> dict:
    for attempt in range(max_retries):
        try:
            response = client.request(method, path, timeout=30, **kwargs)

            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 60))
                time.sleep(retry_after)
                continue

            response.raise_for_status()
            return response.json()

        except httpx.HTTPStatusError as e:
            if e.response.status_code >= 500 and attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # exponential backoff
                continue
            raise

    raise RuntimeError(f"Failed after {max_retries} attempts")

Step 4: Handle Pagination

Offset/limit pagination

def paginate_offset(client, path: str, page_size: int = 100):
    offset = 0
    while True:
        data = api_request(client, "GET", path, params={"limit": page_size, "offset": offset})
        items = data.get("items", data.get("data", []))
        if not items:
            break
        yield from items
        if len(items) < page_size:
            break
        offset += page_size

Cursor pagination

def paginate_cursor(client, path: str):
    cursor = None
    while True:
        params = {"cursor": cursor} if cursor else {}
        data = api_request(client, "GET", path, params=params)
        yield from data["items"]
        cursor = data.get("next_cursor")
        if not cursor:
            break

Link header (GitHub-style)

import re

def paginate_link_header(client, url: str):
    while url:
        response = client.get(url)
        response.raise_for_status()
        yield from response.json()
        link_header = response.headers.get("Link", "")
        next_url = re.search(r'<([^>]+)>;\s*rel="next"', link_header)
        url = next_url.group(1) if next_url else None

Step 5: Handle Errors

def handle_api_error(response: httpx.Response) -> None:
    try:
        error_body = response.json()
    except Exception:
        error_body = {"message": response.text}

    status = response.status_code
    if status == 400:
        raise ValueError(f"Bad request: {error_body}")
    elif status == 401:
        raise PermissionError("Authentication failed - check API key")
    elif status == 403:
        raise PermissionError(f"Forbidden: {error_body}")
    elif status == 404:
        raise KeyError(f"Resource not found: {response.url}")
    elif status == 422:
        raise ValueError(f"Validation error: {error_body}")
    elif status == 429:
        raise RuntimeError("Rate limit exceeded")
    elif status >= 500:
        raise RuntimeError(f"Server error {status}: {error_body}")

Step 6: Webhook Handling

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

app = FastAPI()

@app.post("/webhooks/example")
async def handle_webhook(request: Request):
    # Verify signature
    signature = request.headers.get("X-Webhook-Signature")
    body = await request.body()
    expected = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(f"sha256={expected}", signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = await request.json()
    event_type = payload.get("type")
    # Route to handler...
    return {"status": "ok"}

Production Checklist

• [ ] Secrets in environment variables, never in code
• [ ] Retry with exponential backoff on 5xx and 429
• [ ] Timeout set on every request (default: 30s)
• [ ] Rate limit tracking to avoid hitting limits proactively
• [ ] Structured logging with request ID for debugging
• [ ] Webhook signature verification before processing