"""Five9 cloud contact center connector.
Syncs call logs, dispositions, and contacts from the Five9 REST APIs.
Five9 is used by home improvement contractors for call tracking, call
recording, screen pop, and lead routing.
Auth: OAuth 2.0 authorization-code flow. Five9's regional API hosts vary
(us/eu/ca), so both the auth host and the REST API base URL are
configurable via Flask config / connector config:
* ``OAUTH_FIVE9_CLIENT_ID`` — [NEEDS_USER_CONFIG]
* ``OAUTH_FIVE9_CLIENT_SECRET`` — [NEEDS_USER_CONFIG]
* ``FIVE9_AUTH_BASE_URL`` — default ``https://cloudauthsvcs.five9.com``
* ``FIVE9_API_BASE_URL`` — default ``https://api.prod.us.five9.net``
Synced records are upserted into ``ExternalSyncRecord`` keyed by
``(company_id, external_id)`` so repeated syncs are idempotent.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ----------------------------------------------------
register_connector(
"five9",
{
"service": "five9",
"display_name": "Five9",
"name": "Five9",
"type": "oauth2",
"category": "telephony",
"description": "Cloud contact center — call logs, recordings, dispositions, and contact sync for call tracking and lead routing.",
"auth_type": "oauth2",
"auth_fields": ["client_id", "client_secret", "access_token", "refresh_token"],
"config_fields": [
{"name": "auth_base_url", "label": "Auth Base URL (region)", "type": "text", "required": False},
{"name": "api_base_url", "label": "API Base URL (region)", "type": "text", "required": False},
{"name": "domain_id", "label": "Five9 Domain ID", "type": "text", "required": False},
],
"capabilities": ["call_logs", "recordings", "contacts", "dispositions"],
"data_types": ["call_logs", "contacts", "dispositions"],
"rate_limit": "Varies by endpoint; conservative back-off applied.",
"docs_url": "https://documentation.five9.com/",
},
)
# -- Defaults ------------------------------------------------------------------
_DEFAULT_AUTH_BASE = "https://cloudauthsvcs.five9.com"
_DEFAULT_API_BASE = "https://api.prod.us.five9.net"
def _iso_to_dt(value: Optional[str]) -> Optional[datetime]:
"""Parse an ISO 8601 timestamp into a timezone-aware datetime."""
if not value:
return None
try:
dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except (ValueError, AttributeError):
return None
class Five9Connector(OAuthConnector):
"""Five9 contact center integration — call logs, dispositions, contacts."""
_SERVICE = "five9"
PAGE_LIMIT = 100
# OAuth 2.0 — hosts are region-dependent, resolved at runtime.
OAUTH_SCOPES = ["openid", "offline_access"]
# -- Config helpers --------------------------------------------------------
@property
def _auth_base(self) -> str:
base = self.config.get("auth_base_url", "")
if not base:
try:
from flask import current_app
base = current_app.config.get("FIVE9_AUTH_BASE_URL", "")
except Exception:
base = ""
return (base or _DEFAULT_AUTH_BASE).rstrip("/")
@property
def _api_base(self) -> str:
base = self.config.get("api_base_url", "")
if not base:
try:
from flask import current_app
base = current_app.config.get("FIVE9_API_BASE_URL", "")
except Exception:
base = ""
return (base or _DEFAULT_API_BASE).rstrip("/")
@property
def OAUTH_AUTHORIZE_URL(self) -> str: # type: ignore[override]
return f"{self._auth_base}/oauth2/v1/authorize"
@property
def OAUTH_TOKEN_URL(self) -> str: # type: ignore[override]
return f"{self._auth_base}/oauth2/v1/token"
@property
def _oauth_client_id(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_FIVE9_CLIENT_ID", "") # [NEEDS_USER_CONFIG]
@property
def _oauth_client_secret(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_FIVE9_CLIENT_SECRET", "") # [NEEDS_USER_CONFIG]
@property
def _access_token(self) -> str:
return self.config.get("access_token", "")
# -- OAuth flow -------------------------------------------------------------
def oauth_authorize_url(self, state: str) -> str:
"""Build the Five9 OAuth authorization URL."""
from urllib.parse import urlencode
if not self._oauth_client_id:
raise ValueError(
"Five9 OAuth credentials not configured. "
"Set OAUTH_FIVE9_CLIENT_ID and OAUTH_FIVE9_CLIENT_SECRET."
)
params = {
"client_id": self._oauth_client_id,
"response_type": "code",
"scope": " ".join(self.OAUTH_SCOPES),
"redirect_uri": self._get_redirect_uri(),
"state": state,
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange the authorization code for access/refresh tokens."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"Five9 OAuth credentials not configured. "
"Set OAUTH_FIVE9_CLIENT_ID and OAUTH_FIVE9_CLIENT_SECRET."
)
resp = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self._get_redirect_uri(),
},
auth=(client_id, client_secret),
headers={"Accept": "application/json"},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh an expired access token via the Five9 token endpoint."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("Five9 OAuth credentials not configured.")
resp = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
},
auth=(client_id, client_secret),
headers={"Accept": "application/json"},
timeout=30,
)
resp.raise_for_status()
data = resp.json()
return {
"access_token": data.get("access_token", ""),
"refresh_token": data.get("refresh_token", refresh_token),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "Bearer"),
}
# -- HTTP helpers -----------------------------------------------------------
def _api_get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
"""GET against the Five9 REST API with Bearer auth + auto token refresh."""
if not self._access_token:
raise ValueError("Five9 access_token missing — reconnect via OAuth.")
url = f"{self._api_base}{path}"
resp = requests.get(
url,
params=params or {},
headers={
"Authorization": f"Bearer {self._access_token}",
"Accept": "application/json",
},
timeout=30,
)
# Access token expired — attempt one refresh, then retry.
if resp.status_code == 401 and self.config.get("refresh_token"):
logger.info("Five9 access token expired — refreshing")
tokens = self.refresh_access_token(self.config["refresh_token"])
self.config.update({k: v for k, v in tokens.items() if v})
self._persist_tokens()
resp = requests.get(
url,
params=params or {},
headers={
"Authorization": f"Bearer {self.config.get('access_token', '')}",
"Accept": "application/json",
},
timeout=30,
)
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
logger.warning("Five9 rate limited: 429, Retry-After=%s", retry_after)
resp.raise_for_status()
return resp.json()
def _persist_tokens(self) -> None:
"""Persist refreshed tokens back to the Connector record."""
if not self.connector_id:
return
try:
from app.models import db, Connector
record = Connector.query.filter_by(id=self.connector_id).first()
if record:
record.config = dict(self.config)
db.session.commit()
except Exception:
logger.exception("Failed to persist refreshed Five9 tokens")
# -- connect / disconnect ----------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate the OAuth tokens by fetching the current user profile."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
if not self._access_token:
raise ValueError("Five9 access_token missing — complete the OAuth flow first.")
# Lightweight identity check.
me = self._retry(self._api_get, "/users/v1/me")
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={"user": me if isinstance(me, dict) else {}},
)
return {
"status": "connected",
"service": "five9",
"duration_ms": duration_ms,
}
except Exception as exc:
self._log(event_type="connect_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc)}
def test_connection(self) -> Dict[str, Any]:
"""Validate credentials without side effects — used by the test route."""
return self.connect()
def disconnect(self) -> Dict[str, Any]:
"""Drop tokens and mark the connector as disconnected."""
self._connected = False
for key in ("access_token", "refresh_token", "expires_in", "token_obtained_at"):
self.config.pop(key, None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "five9"}
# -- sync ----------------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync call logs and contacts from Five9 into ExternalSyncRecord.
Idempotent — records are upserted by (company_id, external_id).
"""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total = 0
try:
results: Dict[str, int] = {}
calls = self._sync_call_logs()
total += calls
results["call_logs"] = calls
contacts = self._sync_contacts()
total += contacts
results["contacts"] = contacts
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
self._log(event_type="sync_error", status="error", error_message=str(exc))
return {"status": "error", "error": str(exc), "record_count": total}
def _sync_call_logs(self) -> int:
"""Pull recent call log records and upsert into ExternalSyncRecord."""
from app.models import db, ExternalSyncRecord
count = 0
offset = 0
while True:
self._check_rate_limit()
data = self._retry(
self._api_get,
"/callogs/v1/call-logs",
{"limit": self.PAGE_LIMIT, "offset": offset},
)
items = data.get("items", data) if isinstance(data, dict) else data
if not isinstance(items, list) or not items:
break
for item in items:
external_id = str(
item.get("callId") or item.get("id") or item.get("sessionId") or ""
)
if not external_id:
continue
external_id = f"five9:call:{external_id}"
record = ExternalSyncRecord.query.filter_by(
company_id=self.company_id, external_id=external_id
).first()
if record is None:
record = ExternalSyncRecord(
company_id=self.company_id,
external_id=external_id,
source_service="five9",
entity_type="call_log",
)
db.session.add(record)
record.name = str(item.get("ani") or item.get("dnis") or "")[:255]
record.status = str(item.get("disposition") or item.get("callType") or "")[:100]
record.properties_json = {
"campaign": item.get("campaign", ""),
"agent": item.get("agent", ""),
"disposition": item.get("disposition", ""),
"call_type": item.get("callType", ""),
"duration_seconds": item.get("talkTime") or item.get("duration"),
"recording_url": item.get("recordingUrl", ""),
"started_at": item.get("timestamp") or item.get("startTime"),
"raw": item,
}
count += 1
db.session.commit()
if len(items) < self.PAGE_LIMIT:
break
offset += self.PAGE_LIMIT
return count
def _sync_contacts(self) -> int:
"""Pull contact records and upsert into ExternalSyncRecord."""
from app.models import db, ExternalSyncRecord
count = 0
offset = 0
while True:
self._check_rate_limit()
try:
data = self._retry(
self._api_get,
"/contacts/v1/contacts",
{"limit": self.PAGE_LIMIT, "offset": offset},
)
except requests.HTTPError as exc:
# Contacts API may not be enabled on all Five9 domains.
code = getattr(getattr(exc, "response", None), "status_code", 0)
if code in (403, 404):
logger.info("Five9 contacts API unavailable (HTTP %s) — skipping", code)
return count
raise
items = data.get("items", data) if isinstance(data, dict) else data
if not isinstance(items, list) or not items:
break
for item in items:
external_id = str(item.get("contactId") or item.get("id") or "")
if not external_id:
continue
external_id = f"five9:contact:{external_id}"
record = ExternalSyncRecord.query.filter_by(
company_id=self.company_id, external_id=external_id
).first()
if record is None:
record = ExternalSyncRecord(
company_id=self.company_id,
external_id=external_id,
source_service="five9",
entity_type="contact",
)
db.session.add(record)
first = str(item.get("firstName") or item.get("first_name") or "")
last = str(item.get("lastName") or item.get("last_name") or "")
record.name = (f"{first} {last}".strip() or str(item.get("number1", "")))[:255]
record.status = "active"
record.properties_json = {
"email": item.get("email", ""),
"phone": item.get("number1") or item.get("phone", ""),
"raw": item,
}
count += 1
db.session.commit()
if len(items) < self.PAGE_LIMIT:
break
offset += self.PAGE_LIMIT
return count
# -- status --------------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Report connection health."""
has_token = bool(self._access_token)
return {
"service": "five9",
"connected": self._connected or has_token,
"has_access_token": has_token,
"has_refresh_token": bool(self.config.get("refresh_token")),
"api_base_url": self._api_base,
}
_REGISTRY["five9"] = Five9Connector