api-integration

# API Integration

This skill enables an AI agent to integrate applications with external APIs reliably. The agent selects the right integration pattern (REST client, webhook consumer, polling, SDK wrapper), implements authentication (API keys, OAuth, JWT), handles errors with retries and circuit breakers, and respects rate limits. The result is production-grade integration code that handles real-world failure modes.

## Workflow

1. **Analyze the target API:** Review the API documentation, OpenAPI spec, or SDK reference to understand available endpoints, authentication requirements, rate limits, and response formats. Identify whether the API supports webhooks for push-based updates or requires polling. Note any idiosyncrasies like non-standard error formats or pagination schemes.

2. **Choose an integration pattern:** Select the appropriate pattern based on the use case. Use a REST client for on-demand request/response interactions. Use webhook consumers for real-time event-driven data. Use polling when the API has no webhook support but you need near-real-time updates. Wrap official SDKs when they exist to add retry logic, logging, and a consistent interface.

3. **Implement authentication:** Configure the correct authentication method—API key in headers, OAuth 2.0 bearer tokens, JWT-based service auth, or basic auth. Store credentials securely using environment variables or a secrets manager. For OAuth flows, implement token refresh logic so long-running integrations don't break when access tokens expire.

4. **Build the client with error handling:** Write the integration code with structured error handling. Catch HTTP errors by status code category: 4xx for client errors (don't retry), 429 for rate limiting (retry with backoff), 5xx for server errors (retry with exponential backoff). Parse error response bodies for actionable messages. Log all requests and responses at debug level for troubleshooting.

5. **Add retry and circuit breaker logic:** Implement exponential backoff with jitter for transient failures. Set a maximum retry count (typically 3-5). Implement a circuit breaker that opens after consecutive failures and periodically allows a test request through. This prevents cascading failures when a downstream API is degraded.

6. **Test and monitor:** Write integration tests using recorded HTTP fixtures (VCR pattern) so tests don't hit live APIs. Monitor integration health with metrics for request latency, error rates, and rate limit headroom. Set up alerts for sustained error rates above threshold.

## Supported Technologies

- **HTTP clients:** requests (Python), httpx (Python async), axios (Node.js), fetch, HttpClient (.NET), OkHttp (Java)
- **Authentication:** API keys, OAuth 2.0, JWT, Basic Auth, HMAC signatures
- **Resilience:** tenacity (Python), retry (Node.js), Polly (.NET), resilience4j (Java)
- **Testing:** responses (Python), nock (Node.js), VCR.py, WireMock
- **GraphQL clients:** gql (Python), graphql-request (Node.js), Apollo Client

## Usage

Provide the agent with the target API name or documentation URL, the operations you need to perform, and the programming language. Specify authentication method and any constraints (rate limits, data volume). The agent will produce a complete integration module with error handling, retries, and usage examples.

## Examples

### Example 1: Stripe API Integration in Python

```python
import os
import time
import logging
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

class StripeClient:
    """Production-ready Stripe API client with retries and error handling."""

    BASE_URL = "https://api.stripe.com/v1"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.environ["STRIPE_SECRET_KEY"]
        self.session = self._build_session()

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/x-www-form-urlencoded",
            "Stripe-Version": "2024-06-20",
        })
        retry_strategy = Retry(
            total=4,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "DELETE"],
            respect_retry_after_header=True,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        return session

    def create_customer(self, email: str, name: str, metadata: Optional[dict] = None) -> dict:
        """Create a Stripe customer."""
        payload = {"email": email, "name": name}
        if metadata:
            for key, value in metadata.items():
                payload[f"metadata[{key}]"] = value
        response = self._request("POST", "/customers", data=payload)
        return response

    def create_payment_intent(self, amount_cents: int, currency: str = "usd",
                              customer_id: Optional[str] = None) -> dict:
        """Create a payment intent for a given amount."""
        payload = {"amount": amount_cents, "currency": currency}
        if customer_id:
            payload["customer"] = customer_id
        return self._request("POST", "/payment_intents", data=payload)

    def list_charges(self, customer_id: str, limit: int = 10) -> list:
        """List charges for a customer with automatic pagination."""
        charges = []
        params = {"customer": customer_id, "limit": limit}
        while True:
            data = self._request("GET", "/charges", params=params)
            charges.extend(data["data"])
            if not data["has_more"]:
                break
            params["starting_after"] = data["data"][-1]["id"]
        return charges

    def _request(self, method: str, path: str, **kwargs) -> dict:
        url = f"{self.BASE_URL}{path}"