"""ServiceTitan connector.
Syncs contacts (customers), deals (jobs), and companies from the ServiceTitan
REST API v2 using OAuth 2.0 client_credentials (machine-to-machine) auth.
Data mapping:
- ServiceTitan Customer → CrmContact
- ServiceTitan Job → CrmDeal
- ServiceTitan Company → CrmCompany
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone, timedelta
from typing import Any, Dict, List, Optional
import requests
from . import BaseConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ---------------------------------------------------
register_connector(
"servicetitan",
{
"service": "servicetitan",
"name": "ServiceTitan",
"category": "crm",
"description": "CRM for home service companies — jobs, contacts, companies.",
"auth_type": "oauth2",
"auth_fields": [
"client_id",
"client_secret",
"app_key",
"tenant_id",
],
"capabilities": ["contacts", "deals", "companies"],
"rate_limit": "100 req/min",
"docs_url": "https://developer.servicetitan.com/",
},
)
# -- Constants ---------------------------------------------------------------
_TOKEN_URL = "https://auth.servicetitan.io/connect/token"
_BASE_URL = "https://api.servicetitan.io/v2"
_PAGE_LIMIT = 100
# -- ServiceTitanConnector ----------------------------------------------------
class ServiceTitanConnector(BaseConnector):
"""ServiceTitan integration — pulls customers, jobs, companies."""
_SERVICE = "servicetitan"
_BASE_URL = _BASE_URL
PAGE_LIMIT = _PAGE_LIMIT
# Token cache (per-instance)
_access_token: Optional[str] = None
_token_expires_at: Optional[datetime] = None
# -- Token management ----------------------------------------------------
def _get_access_token(self) -> str:
"""Return a valid access token, refreshing if expired or missing."""
now = datetime.now(timezone.utc)
# Already have a token and it's not expired yet
if (
self._access_token
and self._token_expires_at
and now < self._token_expires_at - timedelta(seconds=60)
):
return self._access_token
# Token expired or not cached — fetch a new one
self._refresh_token()
return self._access_token # type: ignore[return-value]
def _refresh_token(self) -> None:
"""Fetch a new OAuth2 access token via client_credentials grant."""
client_id = self.config.get("client_id", "")
client_secret = self.config.get("client_secret", "")
if not client_id or not client_secret:
raise ValueError(
"ServiceTitan client_id and client_secret are required."
)
resp = requests.post(
_TOKEN_URL,
data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
resp.raise_for_status()
data = resp.json()
token = data.get("access_token", "")
if not token:
raise ValueError("Token response did not contain access_token")
expires_in = int(data.get("expires_in", 3600))
self._access_token = token
self._token_expires_at = datetime.now(timezone.utc) + timedelta(
seconds=expires_in
)
# Persist the token in config so it survives reconnects
self.config["access_token"] = token
logger.info(
"ServiceTitan token refreshed, expires in %ds", expires_in
)
# -- connect / disconnect ------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Obtain an OAuth2 token and validate credentials."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
# Validate required fields
required = ["client_id", "client_secret", "app_key"]
missing = [f for f in required if not self.config.get(f)]
if missing:
error = f"Missing required fields: {', '.join(missing)}"
logger.warning("ServiceTitan: %s", error)
self._log(
event_type="connect_error",
status="error",
error_message=error,
)
return {"status": "error", "error": error}
try:
# Fetch token
self._refresh_token()
# Validate by making a real API call
response = self._st_request("GET", "/customers", params={"limit": 1})
_ = response # Silenced — we just need it to succeed
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
)
return {
"status": "connected",
"service": "servicetitan",
"duration_ms": duration_ms,
}
except Exception as exc:
self._connected = False
self._log(
event_type="connect_error",
status="error",
error_message=str(exc),
)
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
"""Clear local token cache and reset connection state."""
self._access_token = None
self._token_expires_at = None
self._connected = False
# Remove sensitive data from config
self.config.pop("access_token", None)
self.config.pop("refresh_token", None)
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "servicetitan"}
# -- sync ----------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Sync customers, jobs, and companies from ServiceTitan."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
if not self._connected:
error = "Not connected — call connect() first"
self._log(event_type="sync_error", status="error", error_message=error)
return {"status": "error", "error": error, "record_count": 0}
try:
results: Dict[str, Any] = {}
# Sync customers (contacts) first — jobs may reference them
contacts = self._sync_customers()
total_records += contacts
results["contacts"] = contacts
# Sync companies
companies = self._sync_companies()
total_records += companies
results["companies"] = companies
# Sync jobs (deals) — they reference contacts
deals = self._sync_jobs()
total_records += deals
results["deals"] = deals
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- status --------------------------------------------------------------
def status(self) -> Dict[str, Any]:
if not self._connected:
return {
"service": "servicetitan",
"connected": False,
"config_present": bool(self.config.get("client_id")),
}
try:
self._st_request("GET", "/customers", params={"limit": 1})
return {
"service": "servicetitan",
"connected": True,
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {
"service": "servicetitan",
"connected": False,
"error": str(exc),
}
# -- HTTP request helper -------------------------------------------------
def _st_request(
self,
method: str,
path: str,
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Make an authenticated request to the ServiceTitan API.
Handles:
- Bearer token + ST-App-Key headers
- Automatic token refresh on 401
- Retry via self._retry() for transient errors
- 429 rate-limit back-off
"""
def _do_request() -> Dict[str, Any]:
token = self._get_access_token()
app_key = self.config.get("app_key", "")
headers: Dict[str, str] = {
"Authorization": f"Bearer {token}",
"ST-App-Key": app_key,
"Content-Type": "application/json",
}
url = f"{self._BASE_URL}{path}"
resp = requests.request(
method, url, headers=headers, params=params, json=json_data, timeout=30
)
# Auto-refresh on 401 and retry once
if resp.status_code == 401:
logger.warning("ServiceTitan 401 — refreshing token")
self._refresh_token()
headers["Authorization"] = f"Bearer {self._access_token}"
resp = requests.request(
method, url, headers=headers, params=params, json=json_data, timeout=30
)
# Handle 429 rate-limit
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 2))
logger.warning(
"ServiceTitan rate limited (429), waiting %ds", retry_after
)
time.sleep(retry_after)
resp = requests.request(
method, url, headers=headers, params=params, json=json_data, timeout=30
)
resp.raise_for_status()
return resp.json()
return self._retry(_do_request)
# -- Paginated fetch helper (offset-based) ------------------------------
def _fetch_all(
self, endpoint: str, extra_params: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""Paginate through a ServiceTitan list endpoint using offset.
ServiceTitan uses ?offset=0&limit=100 style pagination.
Increments offset by PAGE_LIMIT until the page is empty.
"""
all_items: List[Dict[str, Any]] = []
offset = 0
while True:
params = {"offset": offset, "limit": self.PAGE_LIMIT}
if extra_params:
params.update(extra_params)
data = self._st_request("GET", endpoint, params=params)
# ServiceTitan v2 API wraps results in a "data" key or returns
# a list directly. Handle both shapes.
results = data.get("data", data.get("results", data))
if isinstance(results, list):
items = results
else:
items = [results] if results else []
if not items:
break
all_items.extend(items)
logger.info(
"ServiceTitan %s: fetched %d (total %d)",
endpoint,
len(items),
len(all_items),
)
offset += self.PAGE_LIMIT
# Rate-limit: 100 req/min → ~1 req per 0.6s minimum;
# sleep 0.1 between pages as a conservative throttle.
time.sleep(0.1)
logger.info(
"ServiceTitan %s: total %d records", endpoint, len(all_items)
)
return all_items
# -- Merge (upsert) helper -----------------------------------------------
@staticmethod
def _merge(
model: Any,
company_id: str,
external_id: str,
defaults: Dict[str, Any],
) -> int:
"""Find or create a record by (company_id, external_id) and update fields.
Returns 1 if a record was inserted or updated.
"""
from app.models import db
record = model.query.filter_by(
company_id=company_id, external_id=external_id
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults,
)
db.session.add(record)
return 1
# -- Customer (Contact) sync ---------------------------------------------
def _sync_customers(self) -> int:
"""Pull customers from ServiceTitan, merge into CrmContact."""
from app.models import db, CrmContact
items = self._fetch_all("/customers")
count = 0
for raw in items:
# ServiceTitan customer shape — normalise field names
# The API returns flat objects with fields like:
# id, first_name, last_name, email, phone, company_name, address
st_id = str(raw.get("id", raw.get("contact_id", "")))
if not st_id:
continue
# Build address string
address_parts = [
raw.get("address_line_1", ""),
raw.get("address_line_2", ""),
raw.get("city", ""),
raw.get("state", ""),
raw.get("postal_code", ""),
]
address = ", ".join(p for p in address_parts if p)
count += self._merge(
CrmContact,
self.company_id,
st_id,
{
"email": raw.get("email", ""),
"first_name": raw.get("first_name", raw.get("firstName", "")),
"last_name": raw.get("last_name", raw.get("lastName", "")),
"phone": raw.get("phone", raw.get("mobile_phone", "")),
"company_name": raw.get("company_name", raw.get("companyName", "")),
"lifecycle_stage": raw.get("lifecycle_stage", "active"),
"properties_json": raw,
},
)
db.session.commit()
logger.info(
"ServiceTitan customers synced: %d records merged", count
)
return count
# -- Job (Deal) sync -----------------------------------------------------
def _sync_jobs(self) -> int:
"""Pull jobs from ServiceTitan, merge into CrmDeal."""
from app.models import db, CrmDeal, CrmContact
items = self._fetch_all("/jobs")
count = 0
for raw in items:
st_id = str(raw.get("id", raw.get("job_id", "")))
if not st_id:
continue
# Parse amount
amount_raw = raw.get("total", raw.get("amount", raw.get("job_value", "")))
amount: Optional[float] = None
if amount_raw is not None:
try:
amount = float(amount_raw)
except (ValueError, TypeError):
pass
# Map ST job status → deal stage
stage = raw.get("status", raw.get("job_status", ""))
# Parse expected date
expected_date_str = raw.get("expected_date", raw.get("scheduled_date", ""))
expected_close_date: Optional[datetime] = None
if expected_date_str:
try:
expected_close_date = datetime.fromisoformat(
str(expected_date_str).replace("Z", "+00:00")
)
except (ValueError, TypeError):
pass
# Try to link to contact
contact_id: Optional[str] = None
customer_id_raw = raw.get("customer_id", raw.get("contact_id", ""))
if customer_id_raw:
contact = CrmContact.query.filter_by(
company_id=self.company_id,
external_id=str(customer_id_raw),
).first()
if contact:
contact_id = contact.id
count += self._merge(
CrmDeal,
self.company_id,
st_id,
{
"name": raw.get("job_name", raw.get("name", raw.get("description", ""))),
"amount": amount,
"stage": stage,
"pipeline": raw.get("pipeline", raw.get("job_type", "")),
"expected_close_date": expected_close_date,
"contact_id": contact_id,
"properties_json": raw,
},
)
db.session.commit()
logger.info("ServiceTitan jobs synced: %d records merged", count)
return count
# -- Company sync --------------------------------------------------------
def _sync_companies(self) -> int:
"""Pull companies from ServiceTitan, merge into CrmCompany.
ServiceTitan doesn't expose a top-level /companies endpoint in all
tenant configs. We try /companies first; if it fails with 404 we
derive unique companies from the customers list instead.
"""
from app.models import db, CrmCompany
try:
items = self._fetch_all("/companies")
except requests.HTTPError as exc:
if getattr(exc, "response", None) and exc.response is not None:
if exc.response.status_code == 404:
logger.info(
"ServiceTitan /companies not available — "
"deriving from customers"
)
items = self._derive_companies_from_customers()
else:
raise
else:
raise
count = 0
for raw in items:
st_id = str(raw.get("id", raw.get("company_id", "")))
if not st_id:
continue
count += self._merge(
CrmCompany,
self.company_id,
st_id,
{
"name": raw.get("name", raw.get("company_name", "")),
"domain": raw.get("domain", raw.get("website", "")),
"industry": raw.get("industry", ""),
"num_employees": self._safe_int(raw.get("num_employees", None)),
"properties_json": raw,
},
)
db.session.commit()
logger.info("ServiceTitan companies synced: %d records merged", count)
return count
def _derive_companies_from_customers(self) -> List[Dict[str, Any]]:
"""Derive unique companies from the customers endpoint.
Groups customers by company_name and returns one entry per unique
company. Uses a hash of the company name as the external_id
(prefixed with 'derived_') since we don't have a native ST company ID.
"""
import hashlib
customers = self._fetch_all("/customers")
seen: Dict[str, Dict[str, Any]] = {}
for c in customers:
cname = c.get("company_name", c.get("companyName", ""))
if not cname:
continue
key = cname.strip().lower()
if key not in seen:
seen[key] = {
"id": f"derived_{hashlib.sha256(cname.encode()).hexdigest()[:16]}",
"name": cname.strip(),
"domain": c.get("domain", ""),
"industry": c.get("industry", ""),
}
return list(seen.values())
@staticmethod
def _safe_int(value: Any) -> Optional[int]:
"""Parse an integer, returning None on failure."""
if value is None:
return None
try:
return int(value)
except (ValueError, TypeError):
return None
# -- Register in the framework registry --------------------------------------
_REGISTRY["servicetitan"] = ServiceTitanConnector