Api Integrations

API Integrations

Agent-native integration patterns for external APIs, OAuth, webhooks, and rate limiting.

When to Use

• Connecting to third-party APIs (Stripe, Slack, GitHub, Twilio, etc.)
• Implementing OAuth 2.0 / OIDC authentication flows
• Building webhook receivers and event handlers
• Adding rate limiting and retry logic to external calls
• Creating agent-callable tool wrappers for APIs

OAuth 2.0 Lifecycle

Authorization Code Flow (with PKCE)

# FastAPI OAuth2 with PKCE
from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name="provider",
    client_id=settings.CLIENT_ID,
    client_secret=settings.CLIENT_SECRET,
    server_metadata_url="https://provider.com/.well-known/openid-configuration",
    client_kwargs={"scope": "openid email profile"},
)

@app.route("/login")
async def login(request):
    redirect_uri = request.url_for("auth")
    return await oauth.provider.authorize_redirect(request, redirect_uri)

@app.route("/auth")
async def auth(request):
    token = await oauth.provider.authorize_access_token(request)
    user = await oauth.provider.parse_id_token(request, token)
    # Store token, redirect with session

Token Refresh Pattern

async def get_valid_token(client):
    token = load_stored_token()
    if token.expires_at  bool:
    expected = hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

Webhook Reliability

• Respond 200 within 5 seconds
• Process asynchronously (BackgroundTasks, Celery, SQS)
• Log all events with idempotency key
• Retry on failure with exponential backoff
• Dead letter queue for unrecoverable events

Rate Limiting

Token Bucket Pattern

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.buckets: dict[str, list[float]] = defaultdict(list)

    async def check(self, key: str) -> bool:
        now = time.time()
        window = [t for t in self.buckets[key] if now - t = self.max_calls:
            return False
        self.buckets[key].append(now)
        return True

limiter = RateLimiter(max_calls=100, period=60.0)

@app.get("/api/external")
async def call_external(request: Request):
    ip = request.client.host
    if not await limiter.check(ip):
        raise HTTPException(429, "Rate limit exceeded")
    return await external_api.fetch()

API Provider Rate Limits

• Track provider limits in config: { provider: { calls_per_minute, daily_limit } }
• Enforce client-side before sending request
• Return 429 with Retry-After header
• Implement request queue for burst traffic

Retry with Backoff

import asyncio
from functools import wraps

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    jitter: bool = True,
):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except (ConnectionError, TimeoutError) as exc:
                    last_exc = exc
                    if attempt == max_retries:
                        raise
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    if jitter:
                        delay *= (0.5 + random.random())
                    await asyncio.sleep(delay)
            raise last_exc
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3)
async def call_api():
    return await httpx.get("https://api.provider.com/v1/data")

Agent-Callable Tool Wrapper

# Wrap API as a tool for agent calling
from pydantic import BaseModel, Field

class CreateInvoiceInput(BaseModel):
    customer_id: str
    amount: int
    currency: str = "usd"
    description: str = Field(..., max_length=500)

class CreateInvoiceResult(BaseModel):
    invoice_id: str
    status: str
    url: str

async def create_invoice(
    input: CreateInvoiceInput,
) -> CreateInvoiceResult:
    """Create an invoice in Stripe and return the payment URL."""
    # Implementation calls Stripe API
    # Agent can call this tool with structured input
    ...

Pagination

async def fetch_all_pages(
    client: httpx.AsyncClient,
    url: str,
    params: dict,
    max_pages: int = 50,
) -> list[dict]:
    """Fetch all pages with cursor-based pagination."""
    results = []
    page = 0
    while page  self.reset_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("circuit breaker is open")

        try:
            result = await func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "open"
            raise

Verification

• Test OAuth flow with mock token endpoint
• Verify webhook signature validation rejects bad signatures
• Confirm rate limiter returns 429 after threshold
• Check retry logic with failing mock server
• Validate circuit breaker opens and resets correctly

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: noah-sheldon
• Source: noah-sheldon/ai-dev-kit
• License: MIT
• Homepage: https://noahsheldon.dev

Install and usage instructions live in the source repository linked above.