api-integration-specialist（API 集成 specialist）

API Integration Specialist

Expert assistant for reliable, secure, and efficient API integrations.

Core Integration Workflow

1. Discovery and Planning

Before integrating any API:

• Review official documentation thoroughly
• Identify required endpoints and authentication methods
• Map out data flow and transformation needs
• Check rate limits and quotas
• Plan error handling strategy
• Document dependencies

2. Authentication Setup

Choose and implement appropriate authentication:

OAuth 2.0 (Recommended for user-delegated access):

1. Register application with provider
2. Obtain client_id and client_secret
3. Implement authorization flow:
   - Authorization Code Flow (web apps)
   - PKCE Flow (mobile/SPA)
   - Client Credentials (server-to-server)
4. Handle token refresh automatically
5. Store tokens securely (never in code)

API Keys (Simple public APIs):

- Store in environment variables
- Rotate keys regularly
- Use different keys per environment
- Monitor usage and revoke if compromised

Bearer Tokens (JWT-based):

- Validate token signature
- Check expiration (exp claim)
- Verify issuer and audience
- Implement token refresh logic

3. Request Construction

Build reliable API requests:

Headers:

- Content-Type: application/json (for JSON APIs)
- Authorization: Bearer {token} or API-Key {key}
- Accept: application/json
- User-Agent: YourApp/version
- X-Request-ID: {unique-id} (for tracking)

Query Parameters:

• URL encode all values
• Use consistent naming (camelCase or snake_case)
• Validate before sending
• Document required vs optional params

Request Body:

• Validate data structure before sending
• Use schema validation (JSON Schema, Zod, Joi)
• Handle nested objects correctly
• Respect payload size limits

4. Response Handling

Process API responses systematically:

Status Code Patterns:

2xx Success:
  200 OK - Request successful, data returned
  201 Created - Resource created successfully
  202 Accepted - Request accepted, processing async
  204 No Content - Success with no response body

4xx Client Errors:
  400 Bad Request - Invalid input data
  401 Unauthorized - Missing/invalid auth
  403 Forbidden - Valid auth but no permission
  404 Not Found - Resource doesn't exist
  422 Unprocessable Entity - Validation failed
  429 Too Many Requests - Rate limit exceeded

5xx Server Errors:
  500 Internal Server Error - Server issue
  502 Bad Gateway - Upstream server error
  503 Service Unavailable - Temporary outage
  504 Gateway Timeout - Request timeout

Response Validation:

# Always validate response structure
def validate_response(response):
    # Check status code first
    if response.status_code >= 500:
        raise ServerError("API server error")

    # Validate JSON structure
    try:
        data = response.json()
    except ValueError:
        raise ParseError("Invalid JSON response")

    # Validate expected fields
    if 'data' not in data:
        raise ValidationError("Missing data field")

    return data['data']

Error Handling Patterns

Retry Strategy with Exponential Backoff

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    """
    Retry failed requests with exponential backoff
    """
    for attempt in range(max_retries):
        try:
            return func()
        except (ConnectionError, TimeoutError, ServerError) as e:
            if attempt == max_retries - 1:
                raise

            # Calculate backoff: base_delay * 2^attempt + jitter
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)

            print(f"Attempt {attempt + 1} failed: {e}")
            print(f"Retrying in {delay:.2f} seconds...")
            time.sleep(delay)

Circuit Breaker Pattern

class CircuitBreaker:
    """
    Prevent cascading failures by breaking circuit after threshold
    """
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN

    def call(self, func):
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.timeout:
                self.state = 'HALF_OPEN'
            else:
                raise CircuitOpenError("Circuit breaker is OPEN")

        try:
            result = func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise

    def on_success(self):
        self.failure_count = 0
        self.state = 'CLOSED'

    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = 'OPEN'

Error Response Format

Always return structured errors:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid email format",
    "details": [
      {
        "field": "email",
        "issue": "Must be valid email address"
      }
    ],
    "correlation_id": "req-abc-123",
    "timestamp": "2025-11-03T10:30:00Z",
    "documentation_url": "https://docs.api.com/errors/validation"
  }
}

Rate Limiting and Throttling

Respect Rate Limits

import time
from collections import deque

class RateLimiter:
    """
    Token bucket algorithm for rate limiting
    """
    def __init__(self, max_requests, time_window):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()

    def allow_request(self):
        now = time.time()

        # Remove expired requests
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()

        # Check if we can make request
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True

        return False

    def wait_time(self):
        """Calculate wait time until next request allowed"""
        if not self.requests:
            return 0

        oldest = self.requests[0]
        wait = (oldest + self.time_window) - time.time()
        return max(0, wait)

Handle 429 Responses

def handle_rate_limit(response):
    """
    Handle rate limit using Retry-After header
    """
    if response.status_code == 429:
        # Check for Retry-After header
        retry_after = response.headers.get('Retry-After')

        if retry_after:
            # Can be seconds or HTTP date
            try:
                wait_seconds = int(retry_after)
            except ValueError:
                # Parse HTTP date
                retry_time = parse_http_date(retry_after)
                wait_seconds = (retry_time - datetime.now()).total_seconds()

            print(f"Rate limited. Waiting {wait_seconds} seconds...")
            time.sleep(wait_seconds)
            return True

        # No Retry-After, use exponential backoff
        return False

    return False

Security Best Practices

Secret Management

NEVER hardcode secrets:

# ❌ WRONG - Never do this
API_KEY = "sk-1234567890abcdef"
API_SECRET = "secret_abc123"

# ✅ CORRECT - Use environment variables
import os
API_KEY = os.getenv('API_KEY')
API_SECRET = os.getenv('API_SECRET')

# Validate secrets are present
if not API_KEY or not API_SECRET:
    raise ValueError("Missing required API credentials")

Input Validation

Always validate and sanitize:

def validate_api_input(data):
    """
    Validate data before sending to API
    """
    # Check required fields
    required = ['email', 'name']
    for field in required:
        if field not in data:
            raise ValidationError(f"Missing required field: {field}")

    # Sanitize strings
    data['name'] = data['name'].strip()
    data['email'] = data['email'].lower().strip()

    # Validate email format
    if not is_valid_email(data['email']):
        raise ValidationError("Invalid email format")

    # Remove unexpected fields
    allowed_fields = ['email', 'name', 'age']
    data = {k: v for k, v in data.items() if k in allowed_fields}

    return data

HTTPS Only

# Always use HTTPS
BASE_URL = "https://api.example.com"  # ✅ Secure
BASE_URL = "http://api.example.com"   # ❌ Insecure

# Verify SSL certificates
import requests
response = requests.get(url, verify=True)  # ✅ Verify SSL
response = requests.get(url, verify=False) # ❌ Dangerous

Data Transformation

Request Transformation

def transform_for_api(internal_data):
    """
    Transform internal data format to API format
    """
    return {
        'firstName': internal_data['first_name'],
        'lastName': internal_data['last_name'],
        'emailAddress': internal_data['email'],
        'phoneNumber': format_phone(internal_data['phone']),
        'dateOfBirth': internal_data['dob'].isoformat(),
        'metadata': {
            'source': 'internal_system',
            'version': '1.0'
        }
    }

Response Transformation

def transform_from_api(api_response):
    """
    Transform API response to internal format
    """
    return {
        'id': api_response['userId'],
        'first_name': api_response['firstName'],
        'last_name': api_response['lastName'],
        'email': api_response['emailAddress'],
        'phone': parse_phone(api_response['phoneNumber']),
        'dob': datetime.fromisoformat(api_response['dateOfBirth']),
        'created_at': datetime.fromisoformat(api_response['createdAt'])
    }

Pagination Patterns

Offset-Based Pagination

def fetch_all_pages_offset(base_url, limit=100):
    """
    Fetch all pages using offset pagination
    """
    all_data = []
    offset = 0

    while True:
        response = requests.get(
            base_url,
            params={'limit': limit, 'offset': offset}
        )
        response.raise_for_status()

        data = response.json()
        items = data['items']

        if not items:
            break

        all_data.extend(items)
        offset += limit

        # Check if we got all data
        if len(items) < limit:
            break

    return all_data

Cursor-Based Pagination

def fetch_all_pages_cursor(base_url):
    """
    Fetch all pages using cursor pagination
    """
    all_data = []
    cursor = None

    while True:
        params = {'cursor': cursor} if cursor else {}
        response = requests.get(base_url, params=params)
        response.raise_for_status()

        data = response.json()
        all_data.extend(data['items'])

        # Check for next page
        cursor = data.get('next_cursor')
        if not cursor:
            break

    return all_data

Webhook Integration

Webhook Security

import hmac
import hashlib

def verify_webhook_signature(payload, signature, secret):
    """
    Verify webhook came from trusted source
    """
    # Compute expected signature
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()

    # Compare signatures (timing-safe)
    return hmac.compare_digest(signature, expected)

# Usage in webhook handler
@app.route('/webhooks/payment', methods=['POST'])
def handle_payment_webhook():
    signature = request.headers.get('X-Webhook-Signature')
    payload = request.get_data(as_text=True)

    if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401

    # Process webhook
    data = request.json
    process_payment_event(data)

    return jsonify({'status': 'received'}), 200

Testing Strategy

Unit Tests

def test_api_client():
    """Test API client with mocked responses"""
    with requests_mock.Mocker() as m:
        # Mock successful response
        m.get(
            'https://api.example.com/users/123',
            json={'id': 123, 'name': 'John'},
            status_code=200
        )

        client = APIClient()
        user = client.get_user(123)

        assert user['id'] == 123
        assert user['name'] == 'John'

def test_error_handling():
    """Test error handling"""
    with requests_mock.Mocker() as m:
        # Mock error response
        m.get(
            'https://api.example.com/users/999',
            status_code=404,
            json={'error': 'User not found'}
        )

        client = APIClient()
        with pytest.raises(NotFoundError):
            client.get_user(999)

Integration Tests

Run against sandbox/test environments:

def test_real_api_integration():
    """Test against sandbox environment"""
    client = APIClient(
        api_key=TEST_API_KEY,
        base_url='https://sandbox.api.example.com'
    )

    # Test create
    user = client.create_user({
        'name': 'Test User',
        'email': 'test@example.com'
    })
    assert user['id'] is not None

    # Test read
    fetched = client.get_user(user['id'])
    assert fetched['name'] == 'Test User'

    # Test cleanup
    client.delete_user(user['id'])

Monitoring and Logging

Request Logging

import logging

def log_api_request(method, url, status_code, duration, error=None):
    """Log API request with relevant details"""
    log_data = {
        'method': method,
        'url': url,
        'status_code': status_code,
        'duration_ms': duration * 1000,
        'timestamp': datetime.utcnow().isoformat()
    }

    if error:
        log_data['error'] = str(error)
        logging.error(f"API request failed: {log_data}")
    else:
        logging.info(f"API request completed: {log_data}")

Metrics to Track

Monitor these key metrics:

• Request success rate (%)
• Average response time (ms)
• Error rate by status code
• Rate limit hit frequency
• Circuit breaker state changes
• Retry attempts
• Timeout occurrences

Common Integration Patterns

1. Idempotency

Use idempotency keys for retries:

import uuid

def create_payment(amount, idempotency_key=None):
    """Create payment with idempotency support"""
    if not idempotency_key:
        idempotency_key = str(uuid.uuid4())

    response = requests.post(
        f'{BASE_URL}/payments',
        json={'amount': amount},
        headers={
            'Authorization': f'Bearer {API_KEY}',
            'Idempotency-Key': idempotency_key
        }
    )
    return response.json()

2. Batch Operations

Batch requests when supported:

def batch_create_users(users, batch_size=100):
    """Create users in batches"""
    results = []

    for i in range(0, len(users), batch_size):
        batch = users[i:i + batch_size]

        response = requests.post(
            f'{BASE_URL}/users/batch',
            json={'users': batch}
        )
        response.raise_for_status()

        results.extend(response.json()['created'])

    return results

3. Caching

Implement smart caching:

from functools import lru_cache
from datetime import datetime, timedelta

class CachedAPIClient:
    def __init__(self):
        self.cache = {}
        self.cache_ttl = timedelta(minutes=5)

    def get_user(self, user_id):
        cache_key = f'user:{user_id}'

        # Check cache
        if cache_key in self.cache:
            data, timestamp = self.cache[cache_key]
            if datetime.now() - timestamp < self.cache_ttl:
                return data

        # Fetch from API
        response = requests.get(f'{BASE_URL}/users/{user_id}')
        response.raise_for_status()
        data = response.json()

        # Update cache
        self.cache[cache_key] = (data, datetime.now())

        return data

Resources

For detailed guidance, see:

• references/ERROR_CODES.md - Complete error code reference
• references/AUTHENTICATION_PATTERNS.md - Auth implementation details
• references/RATE_LIMITING.md - Advanced rate limiting strategies
• scripts/validate_api_spec.py - Validate OpenAPI specifications
• scripts/test_api_health.py - Health check automation

Validation Tools

Run validation before deployment:

# Validate API specification
python scripts/validate_api_spec.py openapi.yaml

# Test API health
python scripts/test_api_health.py --base-url https://api.example.com

Quick Reference

Authentication Priority:

1. OAuth 2.0 (user delegation)
2. JWT Bearer tokens
3. API Keys (public APIs only)

Error Handling:

• Implement retry with exponential backoff
• Use circuit breaker for cascading failures
• Log all errors with correlation IDs
• Return structured error responses

Security Checklist:

• ✅ Use HTTPS only
• ✅ Secrets in environment variables
• ✅ Validate all inputs
• ✅ Verify SSL certificates
• ✅ Implement rate limiting
• ✅ Use webhook signatures

Testing Levels:

1. Unit tests (mocked responses)
2. Integration tests (sandbox)
3. Load tests (performance)
4. Security tests (penetration)

This skill provides comprehensive API integration guidance. For advanced topics and detailed examples, refer to the references directory.