Enterprise systems do not expose clean, well-documented REST APIs with generous rate limits and stable schemas. They expose Salesforce with governor limits, SAP with OData quirks, Oracle with SOAP endpoints, and ServiceNow with pagination edge cases. Building agent integrations that survive production requires treating enterprise APIs as inherently unreliable infrastructure: and designing clients that are resilient by default, typed for correctness, and credential-aware across credential rotation events.
OAuth 2.0 Client Credentials with Token Caching
Server-to-server authentication in enterprise systems uses the OAuth 2.0 client credentials flow: your agent presents a client ID and secret, receives a time-limited access token, and presents that token on every API call. The naive implementation re-authenticates before every call. The production implementation caches the token and refreshes it only when it expires.
sequenceDiagram
participant AG as Agent
participant C as Token Cache (Redis)
participant IDP as OAuth Identity Provider
participant API as Enterprise API
AG->>C: GET access_token
alt Cache hit (token valid)
C-->>AG: Return cached token
else Cache miss or expired
AG->>IDP: POST /token (client_id + client_secret)
IDP-->>AG: access_token, expires_in=7200
AG->>C: SET access_token TTL=7140s (expiry - 60s buffer)
end
AG->>API: API request with Bearer token
API-->>AG: Response
style AG fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style C fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style IDP fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style API fill:#f0fdf9,stroke:#0D9488,color:#0F172Aimport httpx
import redis.asyncio as aioredis
from pydantic import BaseModel
class TokenResponse(BaseModel):
access_token: str
token_type: str
expires_in: int
class EnterpriseAuthManager:
"""
Manages OAuth 2.0 client credentials tokens with Redis-backed caching.
All agent instances share one token: we avoid redundant auth calls
across the fleet while ensuring expiry is always respected.
"""
def __init__(
self,
token_url: str,
client_id: str,
client_secret: str,
redis: aioredis.Redis,
cache_key: str,
):
self._token_url = token_url
self._client_id = client_id
self._client_secret = client_secret
self._redis = redis
self._cache_key = cache_key
async def get_token(self) -> str:
# Check the shared Redis cache first: avoids re-auth across all instances
cached = await self._redis.get(self._cache_key)
if cached:
return cached.decode()
# Cache miss: fetch a fresh token from the identity provider
async with httpx.AsyncClient() as client:
response = await client.post(
self._token_url,
data={
"grant_type": "client_credentials",
"client_id": self._client_id,
"client_secret": self._client_secret,
},
timeout=10.0,
)
response.raise_for_status()
token_data = TokenResponse.model_validate(response.json())
# Cache with a 60-second safety buffer before actual expiry.
# This prevents any agent from using a token that expires mid-request.
ttl = max(token_data.expires_in - 60, 60)
await self._redis.setex(self._cache_key, ttl, token_data.access_token)
return token_data.access_tokenSalesforce REST API Integration
Salesforce enforces per-org API call limits (typically 15,000–1,000,000 daily depending on license tier) and per-user concurrent request limits. Your agent client must track these limits via response headers and back off gracefully. Use Pydantic models for all response types, Salesforce’s schema is complex and untyped dict access leads to silent KeyError bugs in production.
from pydantic import BaseModel, Field
import asyncio, random
class SalesforceRecord(BaseModel):
id: str = Field(alias="Id")
name: str = Field(alias="Name")
account_id: str | None = Field(None, alias="AccountId")
class SalesforceQueryResult(BaseModel):
total_size: int = Field(alias="totalSize")
done: bool = Field(alias="done")
records: list[SalesforceRecord] = Field(alias="records")
class SalesforceClient:
def __init__(self, instance_url: str, auth: EnterpriseAuthManager):
self._base = f"{instance_url}/services/data/v59.0"
self._auth = auth
async def query(self, soql: str, max_retries: int = 4) -> SalesforceQueryResult:
"""
Execute a SOQL query with automatic retry on rate limit and transient errors.
We use exponential backoff with full jitter, pure random delay in [0, cap]
: because synchronized retries from many agents create thundering-herd spikes.
"""
for attempt in range(max_retries):
token = await self._auth.get_token()
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{self._base}/query",
params={"q": soql},
headers={"Authorization": f"Bearer {token}"},
timeout=30.0,
)
if resp.status_code == 200:
return SalesforceQueryResult.model_validate(resp.json())
if resp.status_code == 429:
# Respect Retry-After if present; otherwise use exponential backoff
retry_after = int(resp.headers.get("Retry-After", 2 ** attempt))
jitter = random.uniform(0, retry_after)
await asyncio.sleep(retry_after + jitter)
continue
if resp.status_code in (401, 403):
# Token may have been invalidated: clear cache and retry once
await self._auth._redis.delete(self._auth._cache_key)
if attempt < max_retries - 1:
continue
resp.raise_for_status()
raise RuntimeError(f"Salesforce query failed after {max_retries} retries")SAP OData Service Integration
SAP systems expose data via OData v2/v4 services, a REST-like protocol layered on top of ATOM/XML or JSON. OData has its own query language ($filter, $expand, $select) and its own error envelope format. Agents must parse SAP error responses correctly rather than relying on HTTP status codes alone, SAP sometimes returns 200 with an error body.
flowchart TD
A["Agent SOQL/OData Query"] --> B{"Rate Limited?"}
B -->|Yes| C["Read Retry-After Header"]
C --> D["Exponential Backoff + Jitter"]
D --> A
B -->|No| E{"HTTP 2xx?"}
E -->|No| F["Parse Error Body\nSalesforce/SAP envelope"]
F --> G{"Retryable?\n5xx, timeout"}
G -->|Yes| D
G -->|No| H["Raise Domain Exception"]
E -->|Yes| I["Validate with Pydantic Model"]
I --> J["Return Typed Result"]
style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style B fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style C fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style D fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style E fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style F fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style G fill:#fff7ed,stroke:#f59e0b,color:#0F172A
style H fill:#EEF0F7,stroke:#6366F1,color:#0F172A
style I fill:#f0fdf9,stroke:#0D9488,color:#0F172A
style J fill:#f0fdf9,stroke:#0D9488,color:#0F172ACredential Rotation Without Downtime
Enterprise security policies mandate periodic credential rotation, often every 90 days. Rotating credentials with downtime is unacceptable for 24/7 agent systems. The zero-downtime pattern: (1) create new client credentials in the identity provider while old credentials remain valid; (2) deploy agents that try the new credential first and fall back to the old; (3) monitor authentication success rates until old-credential auth drops to zero; (4) revoke old credentials. The overlap window, both sets valid simultaneously, should be at least as long as your slowest deployment rollout. Store credentials in a secrets manager (Vault, AWS Secrets Manager) rather than environment variables, so agents can fetch the latest version without restarts.
In the next lesson, we extend these integration patterns to database and data warehouse agents: building SQL-executing agents with read-only permissions, parameterized query enforcement, and cost controls for expensive Snowflake and BigQuery queries.