Agent skill

api-integration

Design and implement REST API integrations with proper error handling, authentication, rate limiting, and testing. Use when building API clients, integrating third-party services, or when users mention API, REST, webhooks, HTTP requests, or service integration.

View SKILL.md on GitHub Repository
Stars 0
Forks 0

Install this agent skill to your Project

npx add-skill https://github.com/IHKREDDY/agent-skills/tree/main/api-integration

Metadata

Additional technical details for this skill

author
agent-skills-demo
version
1.0
category
integration

SKILL.md

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:

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

Bearer Token:

python
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:

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

3. Client Implementation

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

Basic structure:

python
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:

python
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:

python
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:

python
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:

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

Write unit tests:

python
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

python
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

python
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

python
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

python
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)

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

python
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

python
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:

markdown
# [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]

Recommended Agent Skills

Expand your agent's capabilities with these related and highly-rated skills.

IHKREDDY/agent-skills

work-on-ticket

Pulls ticket details from Jira, creates feature branches with proper naming conventions, and handles planning steps. Use when starting work on a Jira ticket, creating branches for tickets, or when users mention "work on ticket", "start ticket", "create branch for", or Jira ticket IDs.

0 0
Explore
IHKREDDY/agent-skills

code-review

Performs comprehensive code reviews following industry best practices. Use when reviewing pull requests, code changes, or when asked to analyze code quality, security, performance, or maintainability. Checks for common bugs, security vulnerabilities, code smells, and adherence to coding standards.

0 0
Explore
IHKREDDY/agent-skills

data-analysis

Analyze datasets using Python with pandas, numpy, and visualization libraries. Generates statistical summaries, identifies patterns, creates charts, and provides insights. Use when analyzing CSV/Excel files, exploring data, creating visualizations, or when users mention data analysis, statistics, charts, or datasets.

0 0
Explore
IHKREDDY/agent-skills-ts

tech-debt-report

Find TODOs, deprecated APIs, and outdated patterns in codebase

0 0
Explore
IHKREDDY/agent-skills-ts

sprint-summary

Generate sprint review summary from completed Jira tickets

0 0
Explore
IHKREDDY/agent-skills-ts

api-integration

Design and implement REST API integrations with proper error handling, authentication, rate limiting, and testing. Use when building API clients, integrating third-party services, or when users mention API, REST, webhooks, HTTP requests, or service integration.

0 0
Explore

Didn't find tool you were looking for?

Be as detailed as possible for better results