name: api-integration
description: Comprehensive API integration toolkit for building, testing, and maintaining REST APIs and integrations. Use when users need to build API clients, integrate with external APIs, handle authentication (OAuth2, API keys, JWT), implement error handling, test endpoints, or manage API rate limiting and caching.

API Integration

Overview

This skill provides patterns, utilities, and best practices for building robust API integrations including authentication, error handling, testing, and monitoring.

Quick Start

For common API integration tasks:
1. REST client: Use scripts/rest_client.py for HTTP requests
2. Authentication: Use scripts/auth_manager.py for OAuth2 and API keys
3. Testing: Use scripts/api_tester.py for endpoint testing
4. Monitoring: Use scripts/api_monitor.py for health checks

Core Capabilities

1. REST API Clients

• HTTP methods (GET, POST, PUT, DELETE, PATCH)
• Request/response handling
• Automatic serialization/deserialization
• Request retries and exponential backoff
• Connection pooling
• Timeout management

2. Authentication Methods

• API Key authentication
• Basic authentication
• Bearer Token (JWT)
• OAuth2 (Authorization Code, Client Credentials)
• Custom authentication schemes
• Token refresh and expiration handling

3. Error Handling

• HTTP status code handling
• Retry logic with exponential backoff
• Rate limit handling (429)
• Request/response validation
• Custom exception types
• Comprehensive logging

4. Testing & Validation

• Endpoint mocking
• Response schema validation
• Integration testing
• Load testing preparation
• Contract testing

5. Monitoring & Analytics

• Health checks
• Response time tracking
• Error rate monitoring
• Request logging
• Webhook support

REST API Client Pattern

from api_client import APIClient, APIError

# Initialize client
client = APIClient(
    base_url="https://api.example.com",
    api_key="your-api-key"
)

# GET request
response = client.get("/users")
users = response.json()

# POST with data
new_user = client.post("/users", data={
    "name": "John Doe",
    "email": "[email protected]"
})

# Error handling
try:
    response = client.get("/users/invalid")
except APIError as e:
    print(f"API Error: {e.status_code} - {e.message}")

Authentication Implementation

API Key Authentication

from auth_manager import APIKeyAuth

auth = APIKeyAuth(api_key="your-key")
client = APIClient(base_url="...", auth=auth)

OAuth2 Client Credentials

from auth_manager import OAuth2ClientCredentials

auth = OAuth2ClientCredentials(
    client_id="your-client-id",
    client_secret="your-client-secret",
    token_url="https://auth.example.com/token"
)

JWT Bearer Token

from auth_manager import JWTAuth

auth = JWTAuth(token="your-jwt-token")

Error Handling Pattern

import logging
from api_client import APIError, RateLimitError

logger = logging.getLogger(__name__)

def fetch_with_retry(client, endpoint, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.get(endpoint)
        except RateLimitError as e:
            wait_time = e.retry_after or (2 ** attempt)
            logger.warning(f"Rate limited. Waiting {wait_time}s")
            time.sleep(wait_time)
        except APIError as e:
            if e.is_retryable():
                logger.warning(f"Retryable error: {e}")
                continue
            else:
                raise
    raise Exception(f"Failed after {max_retries} retries")

Response Validation

from api_tester import validate_schema

# Define expected response schema
schema = {
    "type": "object",
    "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string"},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["id", "name", "email"]
}

# Validate response
response = client.get("/users/1")
is_valid = validate_schema(response.json(), schema)

Request Rate Limiting

from api_client import RateLimitedClient

# Create rate-limited client
client = RateLimitedClient(
    base_url="https://api.example.com",
    requests_per_second=10
)

# Requests automatically respect rate limit
for i in range(100):
    client.get(f"/resource/{i}")  # Respects rate limit

Health Checks and Monitoring

from api_monitor import HealthChecker

checker = HealthChecker(
    endpoints=[
        "https://api.example.com/health",
        "https://api.example.com/status"
    ],
    interval=60  # Check every 60 seconds
)

# Start background health checks
checker.start()

# Get health status
status = checker.get_status()

Resources

scripts/

• rest_client.py - Core REST client implementation
• auth_manager.py - Authentication handlers
• api_tester.py - Testing and validation utilities
• api_monitor.py - Health monitoring and tracking

references/

• auth-patterns.md - Authentication implementation details
• error-handling.md - Error handling patterns and codes
• testing-guide.md - API testing best practices