featuredImage: /assets/integrate-apis-part-1.png

Integrating AI Agents with External APIs - Part 1: API Selection & Authentication

AI agents turn into business systems the moment they touch external APIs. Pick the wrong endpoint or mismanage tokens and your agent becomes a flaky liability. This part of the series locks down your evaluation and authentication decisions so the later, richer integrations (Slack, Discord, Zapier, automation workflows) are built on a clean foundation.

Scenario: Escalation Concierge

You are building an AI concierge that escalates customer incidents. It must:

• Pull the latest customer context from a CRM API.
• Post threaded updates to Slack and Discord.
• Notify Zapier to trigger payment adjustments.

The agent runs 24x7, so token refreshes, consistent rate limiting, and deep observability matter as much as the LLM prompts. We will design the API layer once and reuse it across platforms.

---

Section 1: API Selection Scorecard

Use a scorecard that reflects autonomy, compliance, and operational constraints. Fill it out before you write a line of integration code.

Criterion	Questions to ask	Weight	Notes
Functional reach	Does the API expose the workflows the agent must automate end to end?	25%	Prefer APIs with webhook callbacks plus bulk search.
Reliability	What is the published uptime, historical incident volume, and SDK maturity?	20%	Dig into status pages and GitHub issue velocity.
Performance	How strict are rate limits, and can you request higher quotas?	15%	Burst-friendly tiers matter for LLM retries.
Security	Are OAuth scopes granular? How is tenant isolation handled?	20%	Avoid APIs that require broad admin scopes.
Data governance	Where is data stored, logged, and encrypted in transit and at rest?	10%	Map to your compliance objectives.
Tooling ecosystem	Are there official SDKs, typed models, and CLI tools?	10%	Saves weeks of glue code.

Score each API from 1 to 5 per criterion, multiply by the weight, and pick the highest scoring API that satisfies compliance constraints. When two APIs tie, bias toward the one with stronger security guarantees because auth rewrites are expensive later.

---

Section 2: Authentication Architecture

AI agents usually need both OAuth (acting on behalf of users or workspaces) and server-to-server API keys (service automation). Design auth as a separate service with vault-backed storage, automated rotation, and audit logs.

OAuth 2.0 workspace install in Node.js

// auth/oauth-server.js
import express from "express";
import axios from "axios";
import crypto from "crypto";
import qs from "querystring";

const app = express();
const stateStore = new Map();

app.get("/auth/slack", (req, res) => {
  const state = crypto.randomBytes(16).toString("hex");
  stateStore.set(state, Date.now());
  const params = qs.stringify({
    client_id: process.env.SLACK_CLIENT_ID,
    scope: "channels:history chat:write users:read",
    redirect_uri: `${process.env.APP_URL}/auth/slack/callback`,
    state
  });
  res.redirect(`https://slack.com/oauth/v2/authorize?${params}`);
});

app.get("/auth/slack/callback", async (req, res) => {
  const { code, state } = req.query;
  if (!stateStore.has(state)) {
    return res.status(400).send("Invalid state");
  }
  stateStore.delete(state);

  const response = await axios.post("https://slack.com/api/oauth.v2.access", qs.stringify({
    client_id: process.env.SLACK_CLIENT_ID,
    client_secret: process.env.SLACK_CLIENT_SECRET,
    code,
    redirect_uri: `${process.env.APP_URL}/auth/slack/callback`
  }), { headers: { "Content-Type": "application/x-www-form-urlencoded" } });

  if (!response.data.ok) {
    return res.status(400).send(response.data.error);
  }

  await saveTokens({
    workspaceId: response.data.team.id,
    accessToken: response.data.access_token,
    refreshToken: response.data.refresh_token,
    expiresAt: Date.now() + response.data.expires_in * 1000,
    scopes: response.data.authed_user.scope
  });

  res.send("Workspace connected");
});

async function saveTokens(payload) {
  // Replace with your vault or encrypted database write
  console.log("Persist tokens", payload.workspaceId);
}

app.listen(4000, () => console.log("OAuth server listening on 4000"));

Key practices:

1. Store state to block CSRF.
2. Persist tokens in a vault (HashiCorp Vault, AWS Secrets Manager) and encrypt at rest.
3. Track expiresAt and queue refresh jobs before the token lapses.

OAuth 2.0 workspace install in Python

# auth/oauth_server.py
import os
import time
import secrets
import httpx
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse

app = FastAPI()
state_cache = {}

@app.get("/auth/discord")
def start_discord_oauth():
    state = secrets.token_hex(16)
    state_cache[state] = time.time()
    params = {
        "client_id": os.environ["DISCORD_CLIENT_ID"],
        "response_type": "code",
        "scope": "bot applications.commands",
        "redirect_uri": f"{os.environ['APP_URL']}/auth/discord/callback",
        "state": state,
        "permissions": "2147483648"
    }
    query = "&".join(f"{k}={v}" for k, v in params.items())
    return RedirectResponse(f"https://discord.com/api/oauth2/authorize?{query}")

@app.get("/auth/discord/callback")
async def discord_callback(code: str, state: str):
    if state not in state_cache:
        raise HTTPException(status_code=400, detail="invalid state")
    state_cache.pop(state)

    async with httpx.AsyncClient() as client:
        token_resp = await client.post(
            "https://discord.com/api/oauth2/token",
            data={
                "client_id": os.environ["DISCORD_CLIENT_ID"],
                "client_secret": os.environ["DISCORD_CLIENT_SECRET"],
                "grant_type": "authorization_code",
                "code": code,
                "redirect_uri": f"{os.environ['APP_URL']}/auth/discord/callback"
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"}
        )
    token_resp.raise_for_status()
    data = token_resp.json()
    persist_secret("discord", data)
    return {"status": "workspace connected"}

def persist_secret(provider: str, token_payload: dict):
    # Write to KMS-backed store
    print(f"[vault] {provider} token saved with scopes {token_payload.get('scope')}")

---

Section 3: API Keys and Service Credentials

Server-to-server automation (Zapier NLA, internal CRMs, search services) relies on API keys or service accounts. Treat keys as short-lived secrets with rotation metadata.

Node.js service account flow

// auth/service-credential-manager.js
import crypto from "crypto";
import axios from "axios";

export async function getServiceClient() {
  const key = await fetchKeyFromVault("crm-api-key");
  return axios.create({
    baseURL: "https://api.example-crm.com/v1",
    timeout: 10000,
    headers: {
      "Authorization": `Bearer ${key.value}`,
      "x-key-version": key.version
    }
  });
}

async function fetchKeyFromVault(secretName) {
  // Replace with AWS Secrets Manager, GCP Secret Manager, etc.
  return {
    value: process.env.CRM_API_KEY,
    version: process.env.CRM_API_KEY_VERSION,
    rotatedAt: Number(process.env.CRM_API_KEY_ROTATED_AT || Date.now())
  };
}

export async function rotateKey() {
  const newKey = crypto.randomBytes(32).toString("hex");
  // Write to vault and notify downstream services
  console.log("Rotate CRM key:", newKey.slice(0, 6), "...");
}

Python signature enforcement for API keys

# auth/api_key_guard.py
import hashlib
import hmac
from fastapi import FastAPI, Header, HTTPException

app = FastAPI()
ACTIVE_KEYS = {"agent-key-1": "super-secret"}

def verify_signature(key: str, payload: bytes, signature: str):
    digest = hmac.new(ACTIVE_KEYS[key].encode(), payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(digest, signature):
        raise HTTPException(status_code=401, detail="invalid signature")

@app.post("/agent/proxy")
async def proxy_request(x_agent_key: str = Header(...), x_signature: str = Header(...), body: bytes = b""):
    if x_agent_key not in ACTIVE_KEYS:
        raise HTTPException(status_code=401, detail="unknown key")
    verify_signature(x_agent_key, body, x_signature)
    # Forward request to downstream API
    return {"status": "accepted"}

Practices:

1. Store key metadata (issuer, scope, created_at, rotated_at).
2. Enforce signatures for payload integrity.
3. Rotate keys on a schedule and immediately if compromised.

---

Section 4: Rate Limiting, Retries, and Backoff

LLM-driven agents spike traffic. Build rate limiting that protects quota while keeping latency consistent.

Node.js token bucket

// transport/rate-limiter.js
const buckets = new Map();

export function configureBucket(name, { capacity, refillRate }) {
  buckets.set(name, { tokens: capacity, capacity, refillRate, lastRefill: Date.now() });
}

export async function callWithBudget(name, fn) {
  const bucket = buckets.get(name);
  refill(bucket);
  if (bucket.tokens <= 0) {
    await sleep(1000);
    return callWithBudget(name, fn);
  }
  bucket.tokens -= 1;
  try {
    return await fn();
  } catch (err) {
    if (err.response?.status === 429) {
      bucket.tokens = 0;
      await sleep(2000);
      return callWithBudget(name, fn);
    }
    throw err;
  }
}

function refill(bucket) {
  const now = Date.now();
  const elapsed = now - bucket.lastRefill;
  const tokensToAdd = Math.floor(elapsed / bucket.refillRate);
  if (tokensToAdd > 0) {
    bucket.tokens = Math.min(bucket.capacity, bucket.tokens + tokensToAdd);
    bucket.lastRefill = now;
  }
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

Python adaptive limiter

# transport/rate_limiter.py
import asyncio
import time

class AdaptiveLimiter:
    def __init__(self, rate_per_minute):
        self.capacity = rate_per_minute
        self.tokens = rate_per_minute
        self.updated_at = time.time()

    async def run(self, coro):
        await self._refill()
        if self.tokens <= 0:
            await asyncio.sleep(1)
            return await self.run(coro)
        self.tokens -= 1
        try:
            return await coro
        except Exception as exc:
            if getattr(exc, "status_code", None) == 429:
                self.tokens = 0
                await asyncio.sleep(2)
                return await self.run(coro)
            raise exc

    async def _refill(self):
        now = time.time()
        elapsed = now - self.updated_at
        refill = elapsed * (self.capacity / 60)
        if refill >= 1:
            self.tokens = min(self.capacity, self.tokens + refill)
            self.updated_at = now

Guidelines:

1. Configure separate buckets per API and per workflow.
2. Log every 429 and include the Retry-After header in the backoff logic.
3. Keep a budget dashboard (Grafana, DataDog) that alarms when usage exceeds 80% of quota.

---

Section 5: Observability, Auditing, and Runbooks

Authentication bugs rarely show up in local tests. Instrument every token exchange, quota spike, and credential rotation event.

Node.js instrumentation hook

// observability/instrumentation.js
import winston from "winston";

export const log = winston.createLogger({
  level: "info",
  transports: [new winston.transports.Console()],
  format: winston.format.json()
});

export function trackAuthEvent(event, payload) {
  log.info({ event, ...payload, timestamp: new Date().toISOString() });
}

export function wrapClient(client, serviceName) {
  client.interceptors.request.use((config) => {
    trackAuthEvent("api_request", {
      service: serviceName,
      path: config.url,
      method: config.method
    });
    return config;
  });
  client.interceptors.response.use(
    (response) => response,
    (error) => {
      trackAuthEvent("api_error", {
        service: serviceName,
        status: error.response?.status,
        retryAfter: error.response?.headers?.["retry-after"]
      });
      throw error;
    }
  );
  return client;
}

Python OpenTelemetry traces

# observability/otel.py
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("agent.integrations")

async def traced_call(service: str, request_name: str, func, *args, **kwargs):
    with tracer.start_as_current_span(f"{service}.{request_name}") as span:
        try:
            result = await func(*args, **kwargs)
            span.set_status(Status(StatusCode.OK))
            return result
        except Exception as exc:
            span.set_status(Status(StatusCode.ERROR, str(exc)))
            span.set_attribute("error.service", service)
            span.set_attribute("error.request", request_name)
            raise

Build runbooks that include:

1. How to manually refresh tokens if automation fails.
2. How to request rate limit increases.
3. Steps to revoke compromised credentials.
4. Dashboards to inspect when latency spikes.

---

Implementation Checklist

• Fill out the API scorecard and capture the decision rationale in the repo.
• Stand up an auth service (Node or Python) with OAuth flows plus API key vault integration.
• Implement rate limiting as a shared utility and test under load.
• Instrument every auth and transport action with logs and traces.
• Draft runbooks for token refresh, rotation, and quota escalations.

Next up: Part 2 applies this architecture to Slack, including event subscriptions, slash commands, and structured messaging. Keep the scorecard handy; we will reuse it for each platform.