next link in the response envelope to walk through large data sets.The REST API for SAP SuccessFactors Incentive Management is the integration layer between the ICM platform and everything else: HR systems feeding participant data, CRM systems pushing credit transactions, BI platforms pulling results, and automation scripts triggering pipeline runs. This article covers how to authenticate, which endpoints matter, and how to write reliable integrations in Python.
Authentication: OAuth 2.0 Client Credentials
SAP SuccessFactors Incentive Management uses OAuth 2.0 with the client credentials grant type for server-to-server integrations. You register an API client in the admin console and receive a client ID and client secret. Exchange these for a bearer token, which you then include in every API request.
import requests import time # Configuration â store in environment variables, not code BASE_URL = "https://api.sap.com/successfactors/im" CLIENT_ID = "your_client_id" CLIENT_SECRET = "your_client_secret" TOKEN_URL = f"{BASE_URL}/oauth/token" # Token cache: store token + expiry time _token_cache = {"token": None, "expires_at": 0} def get_access_token(): """Return cached token or fetch a new one if expired.""" now = time.time() if (_token_cache["token"] and now < _token_cache["expires_at"] - 60): return _token_cache["token"] response = requests.post(TOKEN_URL, data={ "grant_type": "client_credentials", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "scope": "im.read im.write" }) response.raise_for_status() data = response.json() _token_cache["token"] = data["access_token"] _token_cache["expires_at"] = (now + data["expires_in"]) return _token_cache["token"] def api_headers(): """Return headers with a valid bearer token.""" return { "Authorization": f"Bearer {get_access_token()}", "Content-Type": "application/json", "Accept": "application/json" }
Core API Endpoints
The SAP SuccessFactors Incentive Management REST API organises resources around the core data objects of the platform. These are the endpoints you'll use most in real integrations.
| Resource | Endpoint | Common Use Cases |
|---|---|---|
| Participants | GET /participants | Sync org hierarchy from HR system, validate participant existence |
| Plans | GET /comp-plans | List active plans, validate plan assignments before processing |
| Results | GET /results | Extract calculated incentive amounts for payroll, BI reporting |
| Credits | POST /credits | Push credit transactions from CRM/ERP into the pipeline |
| Quotas | GET /quotas PUT /quotas/{id} | Read quota allocations, update quotas from planning systems |
| Pipeline Runs | POST /pipeline-runs | Trigger calculation runs programmatically |
| Statements | GET /statements | Check statement status, extract statement data for external portals |
Reading Participants with Pagination
Most collection endpoints in SAP SuccessFactors Incentive Management are paginated. The response envelope contains a value array of records and a @odata.nextLink (or next) field when more pages exist. Always walk the full page set â never assume the first page is all the data.
def get_all_participants(status="ACTIVE", page_size=100): """Fetch all participants, walking pagination links.""" participants = [] url = f"{BASE_URL}/participants" params = { "$filter": f"status eq '{status}'", "$top": page_size, "$select": "id,name,position,managerId,status" } while url: resp = requests.get(url, headers=api_headers(), params=params) resp.raise_for_status() data = resp.json() participants.extend(data.get("value", [])) # Follow next page link; clear params (they're in the link) url = data.get("@odata.nextLink") params = None return participants # Usage all_participants = get_all_participants() print(f"Total participants: {len(all_participants)}")
Pushing Credit Transactions
The most common write operation: pushing sales transactions (credits) from your CRM or ERP into SAP SuccessFactors IM for pipeline processing. Credits are the raw input that the calculation engine processes into incentive results.
def post_credits(credits: list): """ Post a batch of credit transactions. Each credit dict needs: participantId, transactionDate, amount, currencyCode, periodId, sourceRef """ url = f"{BASE_URL}/credits/batch" resp = requests.post(url, headers=api_headers(), json={"value": credits}) resp.raise_for_status() result = resp.json() succeeded = [r for r in result["results"] if r["status"] == "CREATED"] failed = [r for r in result["results"] if r["status"] != "CREATED"] print(f"Posted: {len(succeeded)} " f"Failed: {len(failed)}") if failed: for f in failed: print(f" ERROR [{f['sourceRef']}]: " f"{f['errorMessage']}") return succeeded, failed # Example credit payload sample_credits = [ { "participantId": "P001234", "transactionDate": "2026-03-15", "amount": 45000.00, "currencyCode": "USD", "periodId": "Q1-2026", "sourceRef": "CRM-ORDER-98765" } ] post_credits(sample_credits)
Triggering a Pipeline Run
Pipeline runs can be triggered via the API â useful for automation workflows where you want to push credits and then immediately kick off calculation without manual intervention in the UI.
import time def trigger_pipeline(pipeline_id: str, period_id: str): """Trigger a pipeline run and poll until complete.""" url = f"{BASE_URL}/pipeline-runs" resp = requests.post(url, headers=api_headers(), json={ "pipelineId": pipeline_id, "periodId": period_id }) resp.raise_for_status() run_id = resp.json()["runId"] print(f"Pipeline run started: {run_id}") # Poll for completion (every 30 seconds, max 30 min) status_url = f"{BASE_URL}/pipeline-runs/{run_id}" for _ in range(60): time.sleep(30) status_resp = requests.get(status_url, headers=api_headers()) status_resp.raise_for_status() status = status_resp.json()["status"] print(f" Status: {status}") if status in ("COMPLETED", "FAILED", "CANCELLED"): return status raise TimeoutError(f"Pipeline {run_id} did not " f"complete within 30 minutes")
Extracting Results for Payroll
After calculation completes, results need to be extracted and passed to payroll. Filter by period and status â only pull APPROVED results for payroll processing, not in-progress or disputed records.
def get_approved_results(period_id: str): """Extract approved incentive results for payroll.""" results = [] url = f"{BASE_URL}/results" params = { "$filter": ( f"periodId eq '{period_id}' and " f"status eq 'APPROVED'" ), "$select": ("participantId,participantName," "planId,periodId," "incentiveAmount,currencyCode," "approvedDate"), "$orderby": "participantId asc" } while url: resp = requests.get(url, headers=api_headers(), params=params) resp.raise_for_status() data = resp.json() results.extend(data.get("value", [])) url = data.get("@odata.nextLink") params = None return results # Usage payroll_data = get_approved_results("Q1-2026") total_payout = sum(r["incentiveAmount"] for r in payroll_data) print(f"Records: {len(payroll_data)}, " f"Total: {total_payout:,.2f}")
Error Handling and Retry Logic
Production integrations need robust error handling. SAP SuccessFactors IM returns standard HTTP status codes â 401 means your token expired (refresh and retry), 429 means you've hit the rate limit (back off), 5xx means a server-side problem (retry with exponential backoff).
import time from requests.exceptions import HTTPError def api_get_with_retry(url, params=None, max_retries=3): """GET with retry on 429/5xx, token refresh on 401.""" for attempt in range(max_retries): try: resp = requests.get(url, headers=api_headers(), params=params) if resp.status_code == 401: # Force token refresh and retry once _token_cache["expires_at"] = 0 resp = requests.get( url, headers=api_headers(), params=params) if resp.status_code == 429: retry_after = int( resp.headers.get( "Retry-After", 60)) print(f"Rate limited. " f"Waiting {retry_after}s...") time.sleep(retry_after) continue resp.raise_for_status() return resp.json() except HTTPError as e: if resp.status_code >= 500: wait = 2 ** attempt print(f"Server error {resp.status_code}. " f"Retry {attempt+1} in {wait}s...") time.sleep(wait) else: raise raise RuntimeError( f"API call failed after {max_retries} retries")
curl Reference: Quick API Checks
For quick manual checks â verifying a token works, inspecting a specific participant, spot-checking a result â curl is faster than writing Python. These patterns are also useful in CI/CD pipeline health checks.
# 1. Get an access token TOKEN=$(curl -s -X POST \ "${BASE_URL}/oauth/token" \ -d "grant_type=client_credentials" \ -d "client_id=${CLIENT_ID}" \ -d "client_secret=${CLIENT_SECRET}" \ | python3 -c \ "import sys,json; print(json.load(sys.stdin)['access_token'])") # 2. Fetch a single participant by ID curl -s \ -H "Authorization: Bearer ${TOKEN}" \ -H "Accept: application/json" \ "${BASE_URL}/participants/P001234" \ | python3 -m json.tool # 3. Get approved results for Q1-2026 curl -s -G \ -H "Authorization: Bearer ${TOKEN}" \ --data-urlencode "\$filter=periodId eq 'Q1-2026' and status eq 'APPROVED'" \ --data-urlencode "\$select=participantId,incentiveAmount" \ "${BASE_URL}/results" \ | python3 -m json.tool # 4. Check pipeline run status curl -s \ -H "Authorization: Bearer ${TOKEN}" \ "${BASE_URL}/pipeline-runs/RUN-20260401-001" \ | python3 -m json.tool
/credits/batch) exist precisely to reduce request volume â always prefer a single batch POST over looping individual POSTs.