Nobody writes about this because it's not glamorous. It's also where most enterprise AI projects actually live or die.
Every enterprise AI agent demo I've seen looks the same: clean API calls, mock data, instant responses. Then someone tries to connect it to actual SAP S/4HANA running in a client's data centre with a service account that was provisioned in 2019, and the demo timeline becomes a different project entirely.
ERP integration is the unglamorous 70% of enterprise AI agent work that determines whether your agent actually functions in production. This is the architecture guide I wish existed when we built our first finance agent integration.
The Authentication Layer: Three Patterns, Three ERPs
SAP S/4HANA - OAuth 2.0 with SAML bridging
SAP's modern API layer (via SAP Business Technology Platform) uses OAuth 2.0, but most enterprise SAP environments still run a hybrid setup with legacy RFC connections behind the scenes.
import requests
from datetime import datetime, timedelta
class SAPAuthManager:
def __init__(self, client_id, client_secret, token_url):
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self._token = None
self._expiry = None
def get_token(self) -> str:
if self._token and datetime.now() < self._expiry:
return self._token
response = requests.post(self.token_url, data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
})
response.raise_for_status()
data = response.json()
self._token = data["access_token"]
# SAP tokens typically expire in 3600s - refresh at 90% lifetime
self._expiry = datetime.now() + timedelta(
seconds=data["expires_in"] * 0.9
)
return self._token
def get_headers(self) -> dict:
return {
"Authorization": f"Bearer {self.get_token()}",
"Content-Type": "application/json"
}
Oracle Fusion - Service accounts with JWT
Oracle Fusion Cloud uses OAuth 2.0 with JWT bearer tokens, typically via a dedicated service account with scoped REST API roles.
import jwt
import time
import requests
class OracleFusionAuth:
def __init__(self, client_id, private_key, token_url, scope):
self.client_id = client_id
self.private_key = private_key
self.token_url = token_url
self.scope = scope
def build_jwt_assertion(self) -> str:
now = int(time.time())
payload = {
"iss": self.client_id,
"sub": self.client_id,
"aud": self.token_url,
"exp": now + 300,
"iat": now
}
return jwt.encode(payload, self.private_key, algorithm="RS256")
def get_token(self) -> str:
assertion = self.build_jwt_assertion()
response = requests.post(self.token_url, data={
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"assertion": assertion,
"scope": self.scope
})
response.raise_for_status()
return response.json()["access_token"]
NetSuite - Token-based authentication (TBA)
NetSuite uses OAuth 1.0a-style token-based auth, which is older but well-documented.
import hashlib
import hmac
import base64
import time
import secrets
class NetSuiteTBA:
def __init__(self, account_id, consumer_key, consumer_secret,
token_id, token_secret):
self.account_id = account_id
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
self.token_id = token_id
self.token_secret = token_secret
def build_auth_header(self, method: str, url: str) -> str:
nonce = secrets.token_hex(16)
timestamp = str(int(time.time()))
base_string = "&".join([
method, url,
f"oauth_consumer_key={self.consumer_key}",
f"oauth_nonce={nonce}",
f"oauth_signature_method=HMAC-SHA256",
f"oauth_timestamp={timestamp}",
f"oauth_token={self.token_id}",
f"oauth_version=1.0"
])
signing_key = f"{self.consumer_secret}&{self.token_secret}"
signature = base64.b64encode(
hmac.new(signing_key.encode(), base_string.encode(),
hashlib.sha256).digest()
).decode()
return (f'OAuth realm="{self.account_id}", '
f'oauth_consumer_key="{self.consumer_key}", '
f'oauth_token="{self.token_id}", '
f'oauth_signature_method="HMAC-SHA256", '
f'oauth_timestamp="{timestamp}", '
f'oauth_nonce="{nonce}", '
f'oauth_version="1.0", '
f'oauth_signature="{signature}"')
RBAC matters across all three: provision the service account with the minimum scope needed for your agent's actions, never broad admin access. We've seen integrations provisioned with full ERP admin rights because it was "easier to set up", this is a security finding waiting to happen in any audit.
Data Sync: Event-Driven vs Polling
The architecture decision that affects everything downstream is whether your agent reacts to ERP events or polls for changes.
# POLLING PATTERN - simpler, higher latency, works everywhere
class ERPPollingSync:
def __init__(self, erp_client, poll_interval_seconds=60):
self.erp_client = erp_client
self.poll_interval = poll_interval_seconds
self.last_sync_timestamp = None
async def poll_for_changes(self):
while True:
changes = await self.erp_client.get_changed_records(
since=self.last_sync_timestamp
)
for record in changes:
await self.process_change(record)
self.last_sync_timestamp = datetime.utcnow()
await asyncio.sleep(self.poll_interval)
# EVENT-DRIVEN PATTERN - lower latency, requires ERP webhook support
class ERPEventSync:
def __init__(self, webhook_secret):
self.webhook_secret = webhook_secret
async def handle_webhook(self, request_body: bytes, signature: str):
if not self.verify_signature(request_body, signature):
raise ValueError("Invalid webhook signature")
event = json.loads(request_body)
# Idempotency check - ERP webhooks can fire duplicates
if await self.already_processed(event["event_id"]):
return {"status": "duplicate_ignored"}
await self.process_event(event)
await self.mark_processed(event["event_id"])
SAP S/4HANA supports event-driven architecture via SAP Event Mesh. Oracle Fusion supports business event subscriptions. NetSuite's webhook support is more limited, polling is often the pragmatic choice there.
Idempotency: Non-Negotiable for Financial Transactions
Every write operation your agent performs against an ERP needs an idempotency key. Network failures, retries, and webhook duplicates will otherwise create duplicate financial records.
class IdempotentERPWriter:
def __init__(self, erp_client, idempotency_store):
self.erp_client = erp_client
self.store = idempotency_store
async def create_invoice(self, invoice_data: dict,
idempotency_key: str) -> dict:
existing = await self.store.get(idempotency_key)
if existing:
return existing # Already processed - return cached result
try:
result = await self.erp_client.create_invoice(invoice_data)
await self.store.set(idempotency_key, result, ttl_hours=24)
return result
except Exception as e:
await self.store.set(idempotency_key,
{"status": "failed", "error": str(e)},
ttl_hours=1)
raise
Audit Logging: Built In, Not Bolted On
Every agent action against an ERP needs an immutable audit record, what changed, who (or which agent) initiated it, and why.
async def execute_erp_action(action: dict, agent_context: dict):
audit_record = {
"agent_id": agent_context["agent_id"],
"action_type": action["type"],
"erp_system": action["target_system"],
"record_affected": action["record_id"],
"reasoning": agent_context["decision_rationale"],
"timestamp": datetime.utcnow().isoformat(),
"idempotency_key": action["idempotency_key"]
}
await audit_store.append(audit_record) # append-only, immutable
result = await erp_client.execute(action)
audit_record["result"] = result
audit_record["completed_at"] = datetime.utcnow().isoformat()
await audit_store.append(audit_record)
return result
ERP Downtime: Plan for It
ERP systems have maintenance windows and occasional outages. Your agent needs graceful degradation, not crashes.
class ResilientERPClient:
def __init__(self, erp_client, circuit_breaker):
self.erp_client = erp_client
self.circuit_breaker = circuit_breaker
async def call_with_resilience(self, operation, *args):
if self.circuit_breaker.is_open():
return {"status": "deferred",
"reason": "ERP circuit breaker open"}
try:
result = await operation(*args)
self.circuit_breaker.record_success()
return result
except ERPTimeoutError:
self.circuit_breaker.record_failure()
return {"status": "queued_for_retry"}
What This Means for Your Project Timeline
ERP integration is consistently underestimated in enterprise AI projects. The authentication setup alone, coordinating with the client's ERP admin team, provisioning service accounts, testing RBAC scopes, regularly takes longer than the AI agent's core logic development.
The full ERP integration architecture for AI agent finance workflows covers the complete reference architecture for all three ERP systems, including the specific gotchas we've hit with each.
ERP integration is one piece of enterprise AI agent deployment. The full enterprise AI agents in finance guide covers the rest of the stack, orchestration, compliance, monitoring, for teams building the complete system.
Published by Dextra Labs | AI Consulting & Enterprise Agent Development