---
title: "ProducerSync API: Quick Start Guide"
url: "https://developer.agentsync.io/producersync-api-quick-start-guide"
image: "https://developer.agentsync.io/_og/d/c_Ocean.takumi,title_~UHJvZHVjZXJTeW5jIEFQSTogUXVpY2sgU3RhcnQgR3VpZGU,props_eyJ0aGVtZSI6eyJtb2RlIjoibGlnaHQiLCJjb2xvcnMiOnsicHJpbWFyeSI6IiMxODdFRkYifX19,p_Ii9wcm9kdWNlcnN5bmMtYXBpLXF1aWNrLXN0YXJ0LWd1aWRlIg,s_w4ZmETsE4YNcUvDx.png"
---

# ProducerSync API: Quick Start Guide

## [Before You Start](#before-you-start)

-   Sandbox credentials from AgentSync — email [support@agentsync.io](mailto:support@agentsync.io) to request access
-   Access token — follow the [Authentication guide](https://developer.agentsync.io/api-authentication) 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](#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](#_1-set-the-authorization-header)

Every request must include your access token:

```text
Authorization: Bearer <ACCESS_TOKEN>
```

### [2\. Try a Basic Request](#_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](https://api.sandbox.agentsync.io)
-   Use the Production base URL when working with live data: [https://api.agentsync.io](https://api.agentsync.io)

**Example request (Sandbox, using CURL)**

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

### [3\. Review the Response](#_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:

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

### [4\. Confirm Connectivity](#_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 finished fetching their data from NIPR yet — check your subscription job's status (see Step 3)
-   The NPN does not exist in NIPR

## [Step 2: Set Up Ongoing Sync](#step-2-set-up-ongoing-sync)

### [Part 1: Webhooks](#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](https://developer.agentsync.io/webhooks-overview) for additional details

#### [What You Need Before You Start](#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](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](#_1-get-access-to-the-webhook-portal)

Contact your AgentSync representative or [AgentSync Support](mailto:support@agentsync.io) 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](#_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](#_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](#_4-test-your-integration)

You can send test events right from the portal:

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

**Example producersync.updates\_available event**

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

See [Webhook Quick Start Guide](https://developer.agentsync.io/webhooks-quick-start-guide) for additional details

### [Part 2: Trigger Updates](#part-2-trigger-updates)

#### [1\. Respond to the Webhook](#_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](#_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](https://developer.agentsync.io/apis)

### [Recommended Daily Workflow](#recommended-daily-workflow)

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

## [Step 3: Manage Your NPN Population](#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.

> **NPN format:** NPNs are numeric strings, at most 10 digits, with no leading zeros. Lists containing duplicates or malformed NPNs are rejected — validate before sending.

> **Subscription operations are asynchronous.** PUT and POST return a `job_id` immediately — use the job status endpoint to confirm completion before relying on the updated list. DELETE returns `202 Accepted` without a job ID (and requires both a JSON body and a `Content-Type: application/json` header).

Adding thousands of NPNs at once? See [Subscribing at Scale](https://developer.agentsync.io/producersync-api-best-practices) for batch sizing, timing expectations, and how to read per-NPN job results.

### [Example: Adding NPNs (PUT — appends to existing list)](#example-adding-npns-put-appends-to-existing-list)

```bash
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:**

```json
{
  "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)](#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.

```bash
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)](#example-removing-npns-delete)

```bash
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](#checking-job-status)

Poll the status endpoint until `status` is `SUCCEEDED` or `FAILED`:

```bash
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:

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

> **Data may not be immediately available after subscribing.** The subscription job retrieves producer data from NIPR as it runs — minutes for small batches, up to a few hours for large ones. If a freshly subscribed NPN returns an empty response, check the job's per-NPN details before assuming something is wrong. A detail of `Entity skipped: already exists in database` is a **success**, not an error. An NPN that fails with `No producer data found` doesn't exist in NIPR — remove it from your list rather than re-submitting it.

## [Additional Considerations](#additional-considerations)

### [Rate Limits](#rate-limits)

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

Every API response includes rate limit headers:

```text
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](https://developer.agentsync.io/api-rate-limits) for additional details

### [Handling Pagination](#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.

```python
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](https://developer.agentsync.io/api-pagination) for additional details

### [Error Handling](#error-handling)

Most errors follow this standardized format:

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

See [API Status Codes](https://developer.agentsync.io/api-status-codes) for a full list of expected codes and additional details

### [Best Practices](#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](#complete-example)

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

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