API Integration Skill

When to Use This Skill

Use this skill when:

Building a client to consume a REST API
Integrating third-party services (Stripe, Twilio, etc.)
Implementing webhooks
Creating or testing HTTP endpoints
Users mention "API", "REST", "integration", "webhook", or "HTTP"

Integration Process

1. API Discovery & Planning

Understand the API:

Review API documentation thoroughly
Identify base URL and API version
Note authentication requirements
Check rate limits and quotas
Review error response formats

Plan the integration:

List required endpoints
Map data models
Identify dependencies
Plan error handling strategy

2. Authentication Setup

Choose the appropriate authentication method:

API Key:

headers = {
    'X-API-Key': os.environ.get('API_KEY'),
    'Content-Type': 'application/json'
}

Bearer Token:

headers = {
    'Authorization': f'Bearer {os.environ.get("ACCESS_TOKEN")}',
    'Content-Type': 'application/json'
}

OAuth 2.0:

Implement token refresh logic
Store tokens securely
Handle token expiration

Basic Auth:

from requests.auth import HTTPBasicAuth
auth = HTTPBasicAuth(username, password)

3. Client Implementation

See references/API-PATTERNS.md for detailed patterns.

Basic structure:

import os
import requests
from typing import Dict, Any, Optional
import time

class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': api_key,
            'Content-Type': 'application/json'
        })
        self.rate_limit_remaining = None
        self.rate_limit_reset = None
    
    def _request(
        self, 
        method: str, 
        endpoint: str, 
        **kwargs
    ) -> Dict[str, Any]:
        """Make HTTP request with error handling."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            response = self.session.request(method, url, **kwargs)
            
            # Track rate limits
            self.rate_limit_remaining = response.headers.get('X-RateLimit-Remaining')
            self.rate_limit_reset = response.headers.get('X-RateLimit-Reset')
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.HTTPError as e:
            self._handle_http_error(e)
        except requests.exceptions.ConnectionError:
            raise APIConnectionError("Failed to connect to API")
        except requests.exceptions.Timeout:
            raise APITimeoutError("Request timed out")
        except requests.exceptions.RequestException as e:
            raise APIError(f"API request failed: {str(e)}")
    
    def _handle_http_error(self, error):
        """Handle HTTP errors with specific status codes."""
        status_code = error.response.status_code
        
        if status_code == 401:
            raise APIAuthenticationError("Invalid credentials")
        elif status_code == 403:
            raise APIAuthorizationError("Insufficient permissions")
        elif status_code == 404:
            raise APINotFoundError("Resource not found")
        elif status_code == 429:
            retry_after = error.response.headers.get('Retry-After', 60)
            raise APIRateLimitError(f"Rate limit exceeded. Retry after {retry_after}s")
        elif 500 <= status_code < 600:
            raise APIServerError(f"Server error: {status_code}")
        else:
            raise APIError(f"HTTP {status_code}: {error.response.text}")

4. Error Handling

Define custom exceptions:

class APIError(Exception):
    """Base exception for API errors."""
    pass

class APIConnectionError(APIError):
    """Network connection failed."""
    pass

class APIAuthenticationError(APIError):
    """Authentication failed."""
    pass

class APIRateLimitError(APIError):
    """Rate limit exceeded."""
    pass

class APIServerError(APIError):
    """Server-side error."""
    pass

Implement retry logic:

from functools import wraps
import time

def retry_on_failure(max_attempts=3, backoff_factor=2):
    """Decorator for retrying failed requests."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except (APIConnectionError, APIServerError) as e:
                    if attempt == max_attempts - 1:
                        raise
                    wait_time = backoff_factor ** attempt
                    time.sleep(wait_time)
            return None
        return wrapper
    return decorator

5. Rate Limiting

Implement rate limit handling:

class RateLimiter:
    def __init__(self, calls_per_second: float):
        self.calls_per_second = calls_per_second
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0
    
    def wait_if_needed(self):
        """Wait if necessary to respect rate limit."""
        elapsed = time.time() - self.last_call
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_call = time.time()

6. Testing

Test endpoints with the provided script:

python scripts/test-endpoint.py \
    --url "https://api.example.com/v1/users" \
    --method GET \
    --headers '{"Authorization": "Bearer token"}'

Write unit tests:

import pytest
from unittest.mock import Mock, patch

def test_successful_request(api_client):
    with patch.object(api_client.session, 'request') as mock_request:
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {'id': 1, 'name': 'Test'}
        mock_request.return_value = mock_response
        
        result = api_client.get_user(1)
        
        assert result['id'] == 1
        assert result['name'] == 'Test'

def test_rate_limit_error(api_client):
    with patch.object(api_client.session, 'request') as mock_request:
        mock_response = Mock()
        mock_response.status_code = 429
        mock_response.headers = {'Retry-After': '60'}
        mock_request.return_value = mock_response
        
        with pytest.raises(APIRateLimitError):
            api_client.get_user(1)

Best Practices

Configuration Management

Store API keys in environment variables
Never commit credentials to version control
Use different keys for dev/staging/production

Logging

import logging

logger = logging.getLogger(__name__)

def _request(self, method, endpoint, **kwargs):
    logger.info(f"API Request: {method} {endpoint}")
    try:
        response = self.session.request(method, url, **kwargs)
        logger.info(f"API Response: {response.status_code}")
        return response.json()
    except Exception as e:
        logger.error(f"API Error: {str(e)}")
        raise

Response Caching

from functools import lru_cache
from datetime import datetime, timedelta

class CachedAPIClient(APIClient):
    def __init__(self, *args, cache_ttl=300, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache_ttl = cache_ttl
    
    @lru_cache(maxsize=100)
    def get_user(self, user_id: int):
        """Cached user lookup."""
        return self._request('GET', f'/users/{user_id}')

Pagination

def get_all_items(self, endpoint: str) -> list:
    """Fetch all items from paginated endpoint."""
    all_items = []
    page = 1
    
    while True:
        response = self._request('GET', endpoint, params={'page': page})
        items = response.get('data', [])
        
        if not items:
            break
            
        all_items.extend(items)
        
        if not response.get('has_more', False):
            break
            
        page += 1
    
    return all_items

Webhooks

import hmac
import hashlib

def verify_webhook_signature(
    payload: bytes, 
    signature: str, 
    secret: str
) -> bool:
    """Verify webhook signature."""
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected)

Common Patterns

Async/Await (Python)

import aiohttp
import asyncio

class AsyncAPIClient:
    async def fetch(self, endpoint: str):
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{self.base_url}/{endpoint}") as response:
                return await response.json()

Batch Requests

def batch_create(self, items: list, batch_size: int = 100):
    """Create items in batches."""
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        self._request('POST', '/batch', json={'items': batch})

Troubleshooting

Debug Mode

import http.client
http.client.HTTPConnection.debuglevel = 1

Common Issues

SSL Certificate errors: Set verify=False temporarily (not for production!)
Timeout issues: Increase timeout: timeout=30
Large responses: Use streaming: stream=True
Rate limits: Implement exponential backoff

Documentation Template

Document your integration:

# [Service Name] API Integration

## Setup
1. Get API key from [service dashboard]
2. Set environment variable: `export API_KEY=your_key`

## Usage
python
from api_client import ServiceClient
client = ServiceClient(api_key=os.environ['API_KEY'])
users = client.list_users()


## Rate Limits
- 1000 requests per hour
- 10 requests per second

## Error Handling
[List common errors and solutions]

api-integration

Safety Notice

Copy this and send it to your AI assistant to learn