third-party-integration

name	third-party-integration
description	Integrate third-party services via OAuth flows, webhook handling, and API clients. Outputs OAuth implementation, webhook verification, idempotent event handling, and graceful degradation patterns.
argument-hint	["third-party service","integration type","auth method","data flow direction"]
allowed-tools	Read, Write, Bash

Third-Party Integration

Integrating with external services is one of the highest-risk activities in software engineering. External APIs change, go down, send duplicate events, and have rate limits. Design integrations to be resilient, verifiable, and replaceable.

Process

1. Map the integration — what data flows in/out, what auth is needed, what events trigger what.
2. Implement OAuth (if required) — authorization code flow with PKCE for user-delegated access.
3. Secure webhook endpoints — signature verification before processing any payload.
4. Idempotent event handlers — webhook redelivery is guaranteed; your handler must be safe to call twice.
5. Graceful degradation — what happens when the third party is down?
6. Test with real sandbox — don't just mock; validate against the real API.
7. Monitor integration health — rate limit approaching, error rate, webhook lag.

Output Format

OAuth 2.0 Authorization Code Flow

# oauth/client.py
import secrets
import hashlib
import base64
from urllib.parse import urlencode, urlparse, parse_qs
import httpx
from dataclasses import dataclass
from datetime import datetime, timezone, timedelta

@dataclass
class OAuthToken:
    access_token: str
    refresh_token: str | None
    expires_at: datetime
    scope: str
    token_type: str = "Bearer"
    
    @property
    def is_expired(self) -> bool:
        return datetime.now(timezone.utc) >= self.expires_at - timedelta(minutes=5)

class OAuthClient:
    """
    OAuth 2.0 Authorization Code flow with PKCE.
    Supports: Stripe, GitHub, Google, Slack, Salesforce, etc.
    """
    
    def __init__(
        self,
        client_id: str,
        client_secret: str,
        authorization_url: str,
        token_url: str,
        redirect_uri: str,
        scopes: list[str],
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.authorization_url = authorization_url
        .token_url = token_url
        .redirect_uri = redirect_uri
        .scopes = scopes
        ._http = httpx.Client(timeout=)
    
     () -> [, ]:
        
        
        code_verifier = secrets.token_urlsafe()
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode()).digest()
        ).rstrip().decode()
        
        
        state = secrets.token_urlsafe()
        
        params = {
            : .client_id,
            : .redirect_uri,
            : .join(.scopes),
            : ,
            : state,
            : code_challenge,
            : ,
        }
        
        
        ._store_pkce_state(state, code_verifier, user_id)
        
        url = 
         url, state
    
     () -> OAuthToken:
        
        
        
        pkce_data = ._get_pkce_state(state)
          pkce_data:
             ValueError()
        
        response = ._http.post(
            .token_url,
            data={
                : ,
                : code,
                : .redirect_uri,
                : .client_id,
                : .client_secret,
                : pkce_data[],
            },
            headers={: }
        )
        response.raise_for_status()
        
        data = response.json()
         ._parse_token_response(data)
    
     () -> OAuthToken:
        
        response = ._http.post(
            .token_url,
            data={
                : ,
                : refresh_token,
                : .client_id,
                : .client_secret,
            },
            headers={: }
        )
        response.raise_for_status()
         ._parse_token_response(response.json())
    
     () -> OAuthToken:
        
          stored_token.is_expired:
             stored_token
        
          stored_token.refresh_token:
             TokenExpiredError()
        
        new_token = .refresh_token(stored_token.refresh_token)
        ._store_token(new_token)
         new_token
    
     () -> OAuthToken:
        expires_in = data.get(, )
         OAuthToken(
            access_token=data[],
            refresh_token=data.get(),
            expires_at=datetime.now(timezone.utc) + timedelta(seconds=expires_in),
            scope=data.get(, ),
        )



 fastapi  APIRouter, Request, HTTPException
 fastapi.responses  RedirectResponse

router = APIRouter(prefix=)

github_oauth = OAuthClient(
    client_id=settings.GITHUB_CLIENT_ID,
    client_secret=settings.GITHUB_CLIENT_SECRET,
    authorization_url=,
    token_url=,
    redirect_uri=,
    scopes=[, , ],
)


  ():
    url, state = github_oauth.get_authorization_url(user_id)
    request.session[] = state
     RedirectResponse(url)


  ():
     request.session.get() != state:
         HTTPException(, )
    
    :
        token = github_oauth.exchange_code(code, state)
        user_id = request.session.get()
         token_store.save(user_id, , token)
         RedirectResponse()
     Exception  e:
         HTTPException(, )

Webhook Handling

# webhooks/handler.py
import hmac
import hashlib
import json
import time
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks
from functools import wraps

router = APIRouter(prefix="/webhooks")

def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
    """Verify Stripe webhook signature (HMAC-SHA256)."""
    # Stripe format: t=timestamp,v1=signature
    parts = {k: v for k, v in (item.split("=", 1) for item in signature.split(","))}
    timestamp = parts.get("t", "")
    sig = parts.get("v1", "")
    
    # Reject timestamps older than 5 minutes (replay protection)
    if abs(time.time() - int(timestamp)) > 300:
        return False
    
    expected = hmac.new(
        secret.encode(),
        f"{timestamp}.{payload.decode()}".encode(),
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, sig)


def verify_github_signature(payload: bytes, signature: str, secret: ) -> :
    
    expected =  + hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
     hmac.compare_digest(expected, signature)



  ():
    payload =  request.body()
    signature = request.headers.get(, )
    
      verify_stripe_signature(payload, signature, settings.STRIPE_WEBHOOK_SECRET):
         HTTPException(status_code=, detail=)
    
    event = json.loads(payload)
    
    
    
    background_tasks.add_task(process_stripe_event, event)
     {: }


  ():
    
    event_id = event[]
    event_type = event[]
    
    
      idempotency_store.exists():
        logger.info()
        
    
    :
        handlers = {
            : handle_payment_succeeded,
            : handle_payment_failed,
            : handle_subscription_cancelled,
            : handle_invoice_failed,
        }
        
        handler = handlers.get(event_type)
         handler:
             handler(event[][])
        :
            logger.info()
        
        
         idempotency_store.(, , ex= * )
    
     Exception  e:
        logger.error()
        
        
         alert(, severity=)


  ():
    order_id = payment_intent.get(, {}).get()
      order_id:
        logger.warning()
        
    
     order_service.mark_paid(
        order_id=order_id,
        payment_id=payment_intent[],
        amount_cents=payment_intent[],
    )

Rate Limit Aware API Client

# integrations/base_client.py
import time
import asyncio
from collections import deque
from threading import Lock

class RateLimitedClient:
    """API client with automatic rate limit handling."""
    
    def __init__(self, base_url: str, token: str, requests_per_second: float = 10):
        self._base_url = base_url
        self._token = token
        self._rps = requests_per_second
        self._request_times = deque()
        self._lock = Lock()
        self._http = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30.0,
        )
    
    async def _throttle(self):
        """Client-side rate limiting to avoid hitting API limits."""
        with self._lock:
            now = time.monotonic()
            window = 1.0
            
            # Remove old timestamps
            while self._request_times and now - self._request_times[0] > window:
                self._request_times.popleft()
            
            if (._request_times) >= ._rps:
                sleep_time = window - (now - ._request_times[])
                 sleep_time > :
                     asyncio.sleep(sleep_time)
            
            ._request_times.append(time.monotonic())
    
      () -> :
         ._throttle()
        
        response =  ._http.get(path, **kwargs)
        
        
         response.status_code == :
            retry_after = (response.headers.get(, ))
            logger.warning()
             asyncio.sleep(retry_after)
              .get(path, **kwargs)  
        
        
        remaining = response.headers.get()
        reset = response.headers.get()
         remaining  (remaining) < :
            logger.warning()
        
        response.raise_for_status()
         response.json()

Integration Health Monitoring

# monitoring/integration_health.py
from prometheus_client import Counter, Histogram, Gauge

# Metrics
webhook_received = Counter("webhooks_received_total", "Webhooks received", ["provider", "event_type"])
webhook_processed = Counter("webhooks_processed_total", "Webhooks processed", ["provider", "event_type", "status"])
api_call_duration = Histogram("third_party_api_duration_seconds", "Third-party API call duration", ["provider", "endpoint"])
api_errors = Counter("third_party_api_errors_total", "Third-party API errors", ["provider", "status_code"])
token_refresh_count = Counter("oauth_token_refreshes_total", "OAuth token refreshes", ["provider"])

# Health check endpoint
@app.get("/health/integrations")
async def integration_health():
    results = {}
    
    # Check Stripe connectivity
    try:
        await stripe_client.get("/v1/balance")
        results["stripe"] = {"status": "ok"}
    except Exception as e:
        results["stripe"] = {"status": "degraded", "error": str(e)}
    
    # Check GitHub API
    :
        resp =  github_client.get()
        remaining = resp[][]
        results[] = {
            :   remaining >   ,
            : remaining,
        }
     Exception  e:
        results[] = {: , : (e)}
    
    overall = 
     (r[] ==   r  results.values()):
        overall = 
    
     {: overall, : results}

Rules

• Verify webhook signatures before processing — never trust payload content alone.
• Reject timestamps older than 5 minutes — prevents replay attacks.
• Return 200 immediately, process asynchronously — slow webhook handlers cause provider retries.
• Idempotent event handlers always — webhook redelivery is guaranteed by all providers.
• Log event IDs, not payloads — event payloads may contain PII; log the ID for debugging.
• Client-side rate limiting — stay under API limits proactively rather than handling 429s reactively.
• Token storage with encryption — OAuth tokens in DB must be encrypted at rest.
• Graceful degradation when third-party is down — queue actions, cache responses, serve stale data.
• Use provider sandbox/test environments — never test webhook logic against production webhooks.
• Rotate webhook secrets regularly — treat webhook secrets like API keys; rotate on team changes.