ProducerSync API: Quick Start Guide

Step 1: Access & Authentication

Before you can call the ProducerSync API, you'll need API credentials. These credentials identify your system and enable secure authentication.

To request Sandbox or Production credentials, email support@agentsync.io.

1. Request Sandbox Credentials

Always start in the Sandbox environment. This is a safe, no-cost place to test and explore without impacting your production data.

⚠️ Always complete testing in Sandbox before using Production

2. Understand Your Credentials

Your credentials will include:

client_id – identifies your application
client_secret – used with your client ID to authenticate
scope (if applicable) – defines the level of access granted

🔐 Keep your client_secret safe. Never hard-code it in source code or share it publicly.

3. Authenticate with OAuth 2.0

ProducerSync uses OAuth 2.0 Client Credentials Flow for authentication:

You'll send your client_id and client_secret to the authentication endpoint (also called access token url).
- Sandbox: https://auth.sandbox.agentsync.io/oauth2/token
- Production: https://auth.agentsync.io/oauth2/token
The API will return an access token.
Include this token in the header of every API request.

Example request (CURL)

curl --request POST \
  --url https://auth.sandbox.agentsync.io/oauth2/token \
  --header 'Content-Type: application/x-www-form-urlencoded' \
  --data grant_type=client_credentials \
  --data client_id=YOUR_CLIENT_ID \
  --data client_secret=YOUR_CLIENT_SECRET \
  --data scope=YOUR_REQUIRED_SCOPE

Example Response

{
  "token_type": "Bearer",
  "expires_in": 3600,
  "access_token": "xyz123",
  "scope": "https://api.agentsync.io/a_valid_scope"
}

Token Management Best Practices

Reuse tokens – Each access token is valid for 60 minutes. Use the same token for multiple API calls instead of requesting a new one every time.
Refresh before expiration – Set up your system to automatically request a fresh token a few minutes before the old one expires.
Handle errors gracefully – If you receive a 401 Unauthorized response, request a new token and retry the call.

Base URLs Best Practices

After authentication, all API requests must use the correct Base URL for your environment.

Do not send API calls to the auth. domain — that is only for obtaining tokens.

Environment	API Base URL
Sandbox	https://api.sandbox.agentsync.io
Production	https://api.agentsync.io

Step 2: Make Your First Call

With your access token from Step 1, you're ready to make your first API request. We'll start with the /v2/entities endpoint, which returns basic producer data.

Test in Sandbox with the sample NPNs: 15645555, 19855109

1. Set the Authorization Header

Every request must include your access token:

Authorization: Bearer <ACCESS_TOKEN>

2. Try a Basic Request

Every API request starts with a Base URL — this is the root web address for the environment you're connecting to.

Use the Sandbox base URL when testing: https://api.sandbox.agentsync.io
Use the Production base URL when working with live data: https://api.agentsync.io

Example request (Sandbox, using CURL)

curl -X GET \
  "https://sandbox.agentsync.io/v2/entities/15645555" \
  -H "Authorization: Bearer <ACCESS_TOKEN>"

3. Review the Response

If your request is successful, you'll see data come back from the API. For example, a response might include producer records:

{
    "id": 158583,
    "npn": "15645555",
    "type": "INDIVIDUAL",
    "firstName": "Joe",
    "middleName": "A",
    "lastName": "Producer",
    "updatedAt": "2025-06-13",
    "noLicenses": false,
    "niprDeleted": false,
    ...
}

4. Confirm Connectivity

If you get a valid response (2XX), you're connected and ready to explore more endpoints.

If you see an error (like 401 Unauthorized), double-check:

Your access token hasn't expired
You include it in the Authorization header
You're pointing to the correct environment base URL

Step 3: Set Up Ongoing Sync

Part 1: Webhooks

Polling the API for updates can be inefficient. Instead, ProducerSync provides webhooks so you're notified as soon as new data is available and complete, enabling efficient daily synchronization.

What is a Webhook? A webhook is like a push notification for your system. When ProducerSync has new data, it sends an event to a URL you control. Your system can then call the API to fetch the latest updates.

See Webhook Overview for additional details

What You Need Before You Start

An endpoint URL - this is simply a web address in your system where we will send notifications (example: https://api.yourcompany.com/webhooks/psapi)

This URL should accept POST requests with JSON payloads.
We suggest the path /webhooks/psapi to clearly identify that this endpoint is dedicated to PSAPI events.

1. Get Access to the Webhook Portal

Contact your AgentSync representative to request webhook access. You'll receive a one-time login link (valid for 7 days). Need more time? Request a new link!

2. Register the Webhook

Log into the portal.

Navigate to the Endpoints page and click Add Endpoint.

Enter your HTTPS endpoint in the Endpoint URL field.

Under Subscribe to events, choose what you'd like to receive:

producersync.updates_available – Daily signal that new NIPR data is ready.
producersync.npn.activated – Fires when a producer is added to your monitored population.
producersync.npn.deactivated – Fires when a producer is removed from your monitored population.
Or, select the high-level producersync to receive all events.

Click Create to save your endpoint.

3. Verify Your Endpoint

Once you save, AgentSync will send a test request to confirm your endpoint works.

Verify your endpoint:

Responds with 200 OK within 5 seconds.
Handles retries (AgentSync will use exponential backoffs, meaning retry intervals increase over time until a maximum limit is reached).

4. Test Your Integration

You can send test events right from the portal:

Select the Endpoints page in the portal from the menu on the left
Click into the endpoint you would like to test
Select the Testing tab (options should be Overview, Testing, Advanced)
Choose an event type from the Send event dropdown
Send the sample payload to your endpoint by clicking Send Example
Check logs (can be found on the same page under Message Attempts) to confirm receipt and processing

Example producersync.updates_available event

{
  "data": {
    "runDate": "2025-07-24"
  },
  "id": "12345",
  "timestamp": "2025-07-24T22:51:05.206Z",
  "type": "producersync.updates_available"
}

See Webhook Quick Start Guide for additional details

Part 2: Trigger Updates

1. Respond to the Webhook

Receiving a producersync.updates_available event should immediately kick off your update process, replacing the need for a scheduled job. Use the value from data.runDate in the event payload to know what date to pull updates from.

2. Fetch the Latest Changes

Call the relevant endpoints with the updatedSince parameter set to the data.runDate value. This ensures you only receive the most recent changes for your monitored NPN population — no missed data and less data to process.

For all available v2 endpoints see ProducerSync API Spec

Recommended Daily Workflow

Listen for webhook notification (producersync.updates_available)
- This replaces scheduled jobs by kicking off processing as soon as updates are ready.
Request only new data
- Call API endpoints with the updatedSince value from the webhook event.
- This ensures no missed data and less to process.
Fetch updates for your NPNs
- Use the relevant /v2 endpoints for your subscribed population.
Process and store results
- Update your database, trigger automations, or notify downstream systems.

Step 4: Manage Your NPN Population

Subscriptions control which NPNs (producers) you're actively monitoring.

View current subscriptions → GET /v1/accounts/subscriptions
Add NPNs → PUT /v1/accounts/subscriptions with your desired list of NPNs
Remove NPNs: DELETE /v1/accounts/subscriptions with only the NPNs you want to remove

⚠️ Important: Avoid rapidly subscribing and unsubscribing NPNs within the same billing period, as this may affect billing and data accuracy.

Example: Subscribing to NPNs

Use a PUT request with your full list of NPNs you want to monitor.

curl --request PUT \
  --url https://api.sandbox.agentsync.io/v1/accounts/subscriptions \
  --header 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
  --header 'Content-Type: application/json' \
  --data '[
    "12345678",
    "87654321",
    "11223344"
  ]'

Example: Unsubscribing from NPNs

Use a DELETE request with only the NPNs you want to remove. Your other active subscriptions will remain unchanged.

curl --request DELETE \
  --url https://api.sandbox.agentsync.io/v1/accounts/subscriptions \
  --header 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
  --header 'Content-Type: application/json' \
  --data '[
    "87654321"
  ]'

Additional Considerations

Rate Limits

Token Endpoint: 200 requests per minute
API Endpoints: 300 requests per minute

See Rate Limiting for additional details

Handling Pagination

v1 Endpoints (Page-Based)

def get_all_licenses(access_token, api_base):
    all_licenses = []
    page = 0
    
    while True:
        url = f'{api_base}/v1/licenses?page={page}&size=250'
        response = requests.get(url, headers={'Authorization': f'Bearer {access_token}'})
        data = response.json()
        
        licenses = data['embedded']['licenses']
        all_licenses.extend(licenses)
        
        # Check if there's a next page
        if 'next' not in data.get('links', {}):
            break
            
        page += 1
    
    return all_licenses

v2 Endpoints (Continuation Tokens)

def get_all_v2_licenses(access_token, api_base):
    all_licenses = []
    url = f'{api_base}/v2/licenses'
    
    while url:
        response = requests.get(url, headers={'Authorization': f'Bearer {access_token}'})
        data = response.json()
        
        if 'embedded' in data:
            licenses = data['embedded']['licenses']
            all_licenses.extend(licenses)
        
        # Get next page URL
        url = data.get('links', {}).get('next', {}).get('href')
        
        # Stop if continuation token is 0 (end of results)
        if url and 'continuationToken=0' in url:
            break
    
    return all_licenses

See API Response Standards for additional details

Error Handling

Most errors follow this standardized format:

{
  "timestamp": 1613510729601,
  "status": 500,
  "error": "Internal Server Error",
  "message": "Error message describing why this was an error.",
  "path": "/v2/{endpoint}"
}

See API Status Codes for a full list of expected codes and additional details

Best Practices

Reuse access tokens (valid for 60 minutes)
Implement exponential backoff for 429 errors
Use updatedSince parameter to minimize payload size
Run daily synchronization consistently
Monitor rate limit headers:
- ratelimit-remaining: Requests left in current window
- ratelimit-reset: When the limit resets

Complete Example

Here's a complete Python example implementing the recommended workflow for license updates:

import requests
from datetime import datetime
import time
import logging
from typing import List, Optional

logging.basicConfig(level=logging.INFO)

class ProducerSyncClient:
    TOKEN_BUFFER_SECONDS = 300  # 5-minute buffer

    def __init__(self, client_id: str, client_secret: str, base_url: str, token_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_url = token_url
        self.access_token: Optional[str] = None
        self.token_expiry: Optional[float] = None

    def _request(self, method: str, url: str, **kwargs) -> requests.Response:
        """Generic request handler with error checking."""
        response = requests.request(method, url, **kwargs)
        try:
            response.raise_for_status()
        except requests.HTTPError as e:
            raise Exception(f"HTTP error on {method} {url}: {e}\nResponse: {response.text}")
        return response

    def get_access_token(self) -> str:
        data = {
            'Content-Type': 'application/x-www-form-urlencoded',
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        }
        response = self._request('POST', self.token_url, data=data)
        token_data = response.json()
        self.access_token = token_data['access_token']
        self.token_expiry = time.time() + token_data['expires_in'] - self.TOKEN_BUFFER_SECONDS
        return self.access_token

    def ensure_valid_token(self) -> str:
        if not self.access_token or time.time() >= self.token_expiry:
            logging.info("Refreshing access token...")
            return self.get_access_token()
        return self.access_token

    def subscribe_to_npns(self, npn_list: List[str]) -> None:
        self.ensure_valid_token()
        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'Content-Type': 'application/json'
        }
        url = f'{self.base_url}/v1/accounts/subscriptions'
        self._request('PUT', url, headers=headers, json=npn_list)
        logging.info(f'Successfully subscribed to {len(npn_list)} NPNs')

    def handle_webhook_event(self, event: dict) -> List[dict]:
        """Parses the webhook event and uses the runDate to fetch updates."""
        try:
            run_date = event['data']['runDate']
        except KeyError:
            raise ValueError("Invalid webhook payload: 'runDate' missing")

        logging.info(f"Received webhook for runDate: {run_date}")
        return self.get_updates_for_date(run_date)
    
    def get_updates_for_date(self, run_date: str) -> List[dict]:
        """Fetches updates for a specific date."""
        self.ensure_valid_token()
        headers = {'Authorization': f'Bearer {self.access_token}'}
        url = f'{self.base_url}/v1/licenses?updatedSince={run_date}'
        response = self._request('GET', url, headers=headers)
        data = response.json()
        licenses = data.get('embedded', {}).get('licenses', [])
        logging.info(f'Retrieved {len(licenses)} updated licenses for {run_date}')
        return licenses

if __name__ == '__main__':
    client = ProducerSyncClient(
        client_id='YOUR_CLIENT_ID',
        client_secret='YOUR_CLIENT_SECRET',
        base_url='https://api.sandbox.agentsync.io',
        token_url='https://auth.sandbox.agentsync.io/oauth2/token',
    )

    npns_to_monitor = ['15645555', '19855109']
    client.subscribe_to_npns(npns_to_monitor)

    #Simulated webhook payload
    webhook_event = {
        "data": {
            "runDate": "2025-08-18"
        },
        "id": "12345",
        "timestamp": "2025-08-18T22:51:05.206Z",
        "type": "producersync.updates_available"
    }

    # Process license updates
    updates = client.handle_webhook_event(webhook_event)
    for license_update in updates:
        print(f"License update for NPN: {license_update['npn']}")