ProducerSync API: Quick Start Guide

Before You Start

Sandbox credentials from AgentSync — email support@agentsync.io to request access
Access token — follow the Authentication guide before proceeding
The Sandbox includes two pre-loaded sample NPNs for testing: 15645555 and 19855109. These work in Step 1 without any setup. For your own producers, you must subscribe their NPNs first (Step 3) and wait for AgentSync to fetch their data from NIPR before they appear in API responses.

Step 1: Make Your First Call

With your access token, 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://api.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

Getting an empty response for your own NPNs? If _embedded is empty or missing, it means one of:

The NPN is not yet subscribed to your account (see Step 3)
The NPN is subscribed but AgentSync hasn't fetched their data from NIPR yet — data becomes available after the next daily NIPR update cycle
The NPN does not exist in NIPR

Step 2: 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 Webhooks 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 or AgentSync Support 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": "whe_000012345",
  "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.

⚠️ includeDeleted defaults to true. By default, responses include records deleted from NIPR. Pass includeDeleted=false if your system should only process active records. Omitting this parameter means you must handle deletions explicitly.

For all available v2 endpoints see the API Reference

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 3: Manage Your NPN Population

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

View current subscriptions → GET /v1/accounts/subscriptions — returns a JSON array of NPN strings
Add NPNs → PUT /v1/accounts/subscriptions — appends to your existing list without removing others
Replace entire list → POST /v1/accounts/subscriptions — sets your monitored population to exactly the provided NPNs
Remove NPNs → DELETE /v1/accounts/subscriptions — removes specific NPNs; others remain unchanged

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

Subscription operations are asynchronous. PUT, POST, and DELETE all return a job_id immediately. Use the job status endpoint to confirm completion before relying on the updated list.

Example: Adding NPNs (PUT — appends to existing list)

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"
  ]'

Response:

{
  "success": true,
  "message": "Job is in progress",
  "job_id": "0ded8367-c0a1-4b2e-9f3d-1a2b3c4d5e6f",
  "links": {
    "status": { "href": "https://api.sandbox.agentsync.io/v1/accounts/subscriptions/status/0ded8367-c0a1-4b2e-9f3d-1a2b3c4d5e6f" }
  }
}

Example: Replacing Your Entire List (POST)

Use POST when you want to set your monitored population to exactly a given list, removing any NPNs not included.

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

Example: Removing NPNs (DELETE)

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"
  ]'

Checking Job Status

Poll the status endpoint until status is SUCCEEDED or FAILED:

curl -X GET \
  "https://api.sandbox.agentsync.io/v1/accounts/subscriptions/status/{jobId}" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

For per-NPN success/failure details:

curl -X GET \
  "https://api.sandbox.agentsync.io/v1/accounts/subscriptions/status/{jobId}/details" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

⚠️ Data is not immediately available after subscribing. A successful subscription job means AgentSync has recorded the NPNs — not that their data has been fetched. NIPR data for newly subscribed NPNs becomes available after the next daily NIPR update cycle. If you query a freshly subscribed NPN and get an empty response, this is expected — check back after the next producersync.updates_available event fires.

Additional Considerations

Rate Limits

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

Every API response includes rate limit headers:

HTTP/1.1 200 OK
ratelimit-limit: 300
ratelimit-remaining: 295
ratelimit-reset: 1713510790

Header	Description
`ratelimit-limit`	Total requests allowed per window
`ratelimit-remaining`	Requests remaining in the current window
`ratelimit-reset`	Unix timestamp when the window resets

When ratelimit-remaining reaches 0, subsequent requests return 429 Too Many Requests. Implement exponential backoff and retry after ratelimit-reset.

See Rate Limits for additional details

Handling Pagination

All v2 endpoints use continuation-token–based pagination. Follow links.next.href until it is absent or contains continuationToken=0.

v1 collection endpoints (/v1/licenses, /v1/appointments, etc.) were deprecated in October 2025. Use /v2/ equivalents for all new integrations.

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; stop when continuation token is 0
        url = data.get('links', {}).get('next', {}).get('href')
        if url and 'continuationToken=0' in url:
            break

    return all_licenses

See Pagination 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:
        headers = {'Content-Type': 'application/x-www-form-urlencoded'}
        data = {
            'grant_type': 'client_credentials',
            'client_id': self.client_id,
            'client_secret': self.client_secret,
        }
        response = self._request('POST', self.token_url, headers=headers, 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]) -> str:
        """Appends NPNs to your monitored population. Returns job_id; poll status to confirm completion."""
        self.ensure_valid_token()
        headers = {
            'Authorization': f'Bearer {self.access_token}',
            'Content-Type': 'application/json'
        }
        url = f'{self.base_url}/v1/accounts/subscriptions'
        response = self._request('PUT', url, headers=headers, json=npn_list)
        job = response.json()
        job_id = job['job_id']
        logging.info(f'Subscription job started ({len(npn_list)} NPNs): {job_id}')
        return job_id

    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 using the v2 licenses endpoint.

        Note: includeDeleted defaults to true — pass includeDeleted=false if you
        want to exclude deleted records from results.
        """
        self.ensure_valid_token()
        headers = {'Authorization': f'Bearer {self.access_token}'}
        # Use v2; size max is 1,000 (default 250)
        url = f'{self.base_url}/v2/licenses?updatedSince={run_date}&size=1000'
        all_licenses = []
        while url:
            response = self._request('GET', url, headers=headers)
            data = response.json()
            all_licenses.extend(data.get('embedded', {}).get('licenses', []))
            next_href = data.get('links', {}).get('next', {}).get('href')
            url = next_href if next_href and 'continuationToken=0' not in next_href else None
        logging.info(f'Retrieved {len(all_licenses)} updated licenses for {run_date}')
        return all_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']
    job_id = client.subscribe_to_npns(npns_to_monitor)
    logging.info(f'Poll /v1/accounts/subscriptions/status/{job_id} to confirm completion')

    #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']}")