Contracting API: Quick Start Guide

When a new producer joins your distribution network, getting them contracted quickly matters. This guide walks through the complete contracting setup — from verifying your carrier and product configuration through contract assignment and producer invitation — using the AgentSync Contracting API.

Who this is for: Engineers at insurance carriers and MGAs building automated onboarding workflows.

What you'll build: A complete onboarding flow that checks your contracting setup, creates a contract assignment for a new producer, and triggers the AgentSync invitation.

Before You Start

Sandbox credentials with scopes rino_api_agency_read and rino_api_agency_write — email support@agentsync.io to request access
Access token — follow the Authentication guide before proceeding; see Scopes for the full scope reference
Python 3.8+ with requests installed if following the Python examples: pip install requests
All Contracting API requests use a /contracting path prefix — distinct from other AgentSync APIs. See API Base URLs.

The examples in this guide assume you have a valid ACCESS_TOKEN.

Overview

Verify your carrier exists (create it if not)
Verify a product is configured under the carrier (create it if not)
Verify commission levels exist for the product (create them if not)
Check if the producer already has a contract assignment
Create the contract assignment
Send the producer invitation

Step 1: Make Your First Call

With your access token, start by fetching your carrier list. This confirms connectivity and shows the carriers already configured in your account.

curl

curl -X GET \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/carriers" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Python

import requests

base_url = "https://api.sandbox.agentsync.io/contracting"
access_token = "YOUR_ACCESS_TOKEN"

response = requests.get(
    f"{base_url}/v1/customers/carriers",
    headers={"Authorization": f"Bearer {access_token}"}
)
response.raise_for_status()
data = response.json()
carriers = data.get("_embedded", {}).get("carriers", [])
print(f"Found {data['page']['totalElements']} carrier(s)")

Example response

{
  "_embedded": {
    "carriers": [
      {
        "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "name": "Acme Insurance Company",
        "naic": "12345",
        "tin": "123456789",
        "_links": {
          "self": { "href": "/v1/customers/carriers/a1b2c3d4-e5f6-7890-abcd-ef1234567890" }
        }
      }
    ]
  },
  "_links": {
    "self": { "href": "/v1/customers/carriers" }
  },
  "page": {
    "size": 250,
    "totalElements": 1,
    "totalPages": 1,
    "number": 0
  }
}

All collection responses follow HAL+JSON format: data lives under _embedded, pagination metadata is in page, and navigation links are in _links. See Pagination.

New account? If your sandbox account has no carriers configured yet, the response will not include an _embedded key — that's expected. Proceed to Step 2 to create your first carrier.

Step 2: Set Up a Carrier

Create a carrier to represent an insurance company in your contracting network.

Carrier creation accepts several optional fields that require IDs from lookup endpoints: linesOfBusiness (the lines of business this carrier operates in) and contractSubmissionMethods (how contracts are submitted). These are reference values maintained by AgentSync — fetch them first, then pass the IDs in your carrier payload.

Note: linesOfBusiness and contractSubmissionMethods on a carrier are arrays of IDs, unlike on a product where each takes a single ID. Both are optional at carrier creation — you can also add them later.

Fetch lookup IDs

# Lines of business (e.g., "Life", "Health", "Property & Casualty")
curl -X GET "https://api.sandbox.agentsync.io/contracting/v1/linesOfBusiness" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

# Contract submission methods (e.g., "Paper", "Online")
curl -X GET "https://api.sandbox.agentsync.io/contracting/v1/contractSubmissionMethods" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

Each response returns a list under _embedded. Use the id field of the matching entries in your carrier creation request.

def get_ids_by_names(base_url, access_token, path, collection_key, names):
    """Fetch a lookup endpoint and return the ids of all items matching the given names."""
    resp = requests.get(
        f"{base_url}{path}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    resp.raise_for_status()
    items = resp.json().get("_embedded", {}).get(collection_key, [])
    return [item["id"] for item in items if item.get("name") in names]

lob_ids = get_ids_by_names(base_url, access_token, "/v1/linesOfBusiness", "linesOfBusiness", ["Life", "Health"])
csm_ids = get_ids_by_names(base_url, access_token, "/v1/contractSubmissionMethods", "contractSubmissionMethods", ["Paper", "Online"])

Tip: Cache these lookup values at startup — they change infrequently and don't need to be fetched on every request. See Best Practices: Lookup Reference Data at Startup.

Create the carrier

administrativeDivisions is the list of US states and territories where the carrier operates — pass standard two-letter state codes. userDefined1 through userDefined10 are free-form custom fields you can use for your own internal metadata.

curl

curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/carriers" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Acme Insurance Company",
    "naic": "12345",
    "tin": "123456789",
    "uplineFirmName": "Acme Holdings",
    "linesOfBusiness": ["LOB_UUID_1", "LOB_UUID_2"],
    "contractSubmissionMethods": ["CSM_UUID_1", "CSM_UUID_2"],
    "administrativeDivisions": ["AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI","IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN","MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH","OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA","WI","WV","WY"],
    "userDefined1": "my-internal-id"
  }'

Python

response = requests.post(
    f"{base_url}/v1/customers/carriers",
    headers={
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    },
    json={
        "name": "Acme Insurance Company",
        "naic": "12345",
        "tin": "123456789",
        "uplineFirmName": "Acme Holdings",
        "linesOfBusiness": lob_ids,
        "contractSubmissionMethods": csm_ids,
        "administrativeDivisions": [
            "AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI",
            "IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN",
            "MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH",
            "OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA",
            "WI","WV","WY"
        ],
        "userDefined1": "my-internal-id",
    }
)
response.raise_for_status()
carrier = response.json()
carrier_id = carrier["id"]
print(f"Created carrier: {carrier_id}")

Example response (201 Created)

{
  "createdDate": "2026-03-18T18:19:20Z",
  "modifiedDate": "2026-03-18T18:19:20Z",
  "id": "9b27b554-873e-4808-9749-45918080c277",
  "name": "Acme Insurance Company",
  "tin": "123456789",
  "typeId": "2969d43f-2846-4e0a-adef-3adc26816edc",
  "naic": "12345",
  "statusId": "ce4be1e3-6c37-4920-9b72-08f812467bf9",
  "_links": {
    "self": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277"
    },
    "linesOfBusiness": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/linesOfBusiness"
    },
    "organizationType": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizationTypes/2969d43f-2846-4e0a-adef-3adc26816edc"
    },
    "organizationStatus": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizationStatuses/ce4be1e3-6c37-4920-9b72-08f812467bf9"
    },
    "products": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/products"
    },
    "contractTiers": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/contractTiers"
    },
    "administrativeDivisions": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/administrativeDivisions"
    },
    "assignedProducts": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/assignedProducts"
    },
    "contactInfos": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/contactInfos"
    },
    "assignedOrganizations": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/assignedOrganizations"
    },
    "organizationContractDetails": {
      "href": "https://api.sandbox.agentsync.io/contracting/v1/organizations/9b27b554-873e-4808-9749-45918080c277/contractDetails"
    }
  }
}

Note: All _links URLs point to /v1/organizations/{id}, which is an internal path. Following any of these links returns 403 Forbidden. Use the id field to construct your own references using the /v1/customers/carriers/{id} path.

Field notes:

naic — 6-character max NAIC code
tin — 9-character max tax identification number
uplineFirmName — use for the upline entity name (not uplineName, which is deprecated)
linesOfBusiness — array of line-of-business UUIDs from GET /v1/linesOfBusiness; optional at creation
contractSubmissionMethods — array of submission method UUIDs from GET /v1/contractSubmissionMethods; optional at creation
administrativeDivisions — array of two-letter state/territory codes; optional at creation
userDefined1–userDefined10 — free-form string fields for your own internal metadata

Carrier names must be unique. Attempting to create a carrier with a name that already exists returns 422 Unprocessable Entity. Use GET /v1/customers/carriers to check for an existing carrier before creating.

Save the id — you'll use it in Steps 3 and 6.

Step 3: Add a Product to the Carrier

Products represent the insurance offerings a carrier makes available for contracting.

The only required fields for a product are name and lineOfBusinessId. Everything else is optional. If you fetched line-of-business IDs in Step 2, reuse the single value appropriate for this product here.

def get_id_by_name(base_url, access_token, path, collection_key, name_value):
    """Fetch a lookup endpoint and return the id of the item matching name_value."""
    resp = requests.get(
        f"{base_url}{path}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    resp.raise_for_status()
    items = resp.json().get("_embedded", {}).get(collection_key, [])
    for item in items:
        if item.get("name") == name_value:
            return item["id"]
    raise ValueError(f"No item named '{name_value}' found at {path}")

# If not already fetched in Step 2:
lob_id = get_id_by_name(base_url, access_token, "/v1/linesOfBusiness", "linesOfBusiness", "Life")

# Optional — fetch only if you want to set these on the product:
csm_id = get_id_by_name(base_url, access_token, "/v1/contractSubmissionMethods", "contractSubmissionMethods", "Paper")
ann_id = get_id_by_name(base_url, access_token, "/v1/annualizations", "annualizations", "Monthly")

Tip: Cache these lookup values at startup — they change infrequently and don't need to be fetched on every request. See Best Practices: Lookup Reference Data at Startup.

curl

curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/carriers/CARRIER_ID/products" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Term Life — Standard",
    "lineOfBusinessId": "LOB_UUID",
    "displayCommissionLevelPercentages": false,
    "administrativeDivisions": ["AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI","IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN","MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH","OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA","WI","WV","WY"],
    "contractSubmissionMethodId": "CSM_UUID",
    "annualizationId": "ANN_UUID",
    "userDefined1": "my-internal-code"
  }'

Python

response = requests.post(
    f"{base_url}/v1/customers/carriers/{carrier_id}/products",
    headers={
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    },
    json={
        "name": "Term Life — Standard",
        "lineOfBusinessId": lob_id,
        "displayCommissionLevelPercentages": False,
        "administrativeDivisions": [
            "AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI",
            "IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN",
            "MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH",
            "OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA",
            "WI","WV","WY"
        ],
        "contractSubmissionMethodId": csm_id,   # optional
        "annualizationId": ann_id,               # optional
        "userDefined1": "my-internal-code",      # optional
    }
)
response.raise_for_status()
product = response.json()
product_id = product["id"]
print(f"Created product: {product_id}")

Field notes:

lineOfBusinessId — required; single UUID from GET /v1/linesOfBusiness
displayCommissionLevelPercentages — boolean; controls whether commission percentages are shown to producers
administrativeDivisions — array of two-letter state/territory codes; optional at creation
contractSubmissionMethodId — optional; single UUID from GET /v1/contractSubmissionMethods
annualizationId — optional; single UUID from GET /v1/annualizations
userDefined1–userDefined10 — free-form string fields for your own internal metadata

Example response (201 Created)

{
  "id": "c4d5e6f7-a8b9-0123-cdef-123456789012",
  "name": "Term Life — Standard",
  "carrierId": "9b27b554-873e-4808-9749-45918080c277",
  "lineOfBusinessId": "0feda650-0ae1-4c64-ab9d-a440d799f6c1",
  "contractSubmissionMethodId": "a07f158d-d446-47a8-a76c-6f618e9da186",
  "annualizationId": "16f181c6-feb0-4b6d-a339-f33aefb8663",
  "_links": {
    "self": { "href": "/v1/customers/products/c4d5e6f7-a8b9-0123-cdef-123456789012" }
  }
}

Step 4: Add Commission Levels

Commission levels define the compensation structure for a product and must be attached to a specific product by its id.

If you're continuing from Step 3, use the id from the product creation response. If you're working with an existing setup, list all products for a carrier to find the product ID you need:

curl -X GET \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/carriers/CARRIER_ID/products" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

The type field controls how commission values are interpreted:

Type	Meaning
`PERCENT`	Commission as a percentage
`DOLLAR`	Commission as a flat dollar amount
`NUMBER`	Commission as a numeric multiplier

curl (single level)

curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/PRODUCT_ID/commissionLevels" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Standard Commission",
    "type": "PERCENT",
    "firstYearValue": 55.00,
    "level": 1,
    "administrativeDivisions": ["AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI","IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN","MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH","OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA","WI","WV","WY"],
    "userDefined1": "my-internal-code"
  }'

curl (bulk — multiple tiers at once)

curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/PRODUCT_ID/commissionLevels/bulk" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '[
    { "name": "Level 1 — 55%", "type": "PERCENT", "firstYearValue": 55.00, "level": 1 },
    { "name": "Level 2 — 60%", "type": "PERCENT", "firstYearValue": 60.00, "level": 2 },
    { "name": "Level 3 — 65%", "type": "PERCENT", "firstYearValue": 65.00, "level": 3 }
  ]'

Python

response = requests.post(
    f"{base_url}/v1/customers/{product_id}/commissionLevels",
    headers={
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    },
    json={
        "name": "Standard Commission",
        "type": "PERCENT",      # "PERCENT", "DOLLAR", or "NUMBER"
        "firstYearValue": 55.00,
        "level": 1,
        "administrativeDivisions": [
            "AK","AL","AR","AZ","CA","CO","CT","DC","DE","FL","GA","HI",
            "IA","ID","IL","IN","KS","KY","LA","MA","MD","ME","MI","MN",
            "MO","MS","MT","NC","ND","NE","NH","NJ","NM","NV","NY","OH",
            "OK","OR","PA","RI","SC","SD","TN","TX","UT","VA","VT","WA",
            "WI","WV","WY"
        ],
        "userDefined1": "my-internal-code",   # optional
    }
)
response.raise_for_status()
commission_level = response.json()
commission_level_id = commission_level["id"]
print(f"Created commission level: {commission_level_id}")

Example response (201 Created)

{
  "id": "d5e6f7a8-b9c0-1234-def0-234567890123",
  "name": "Standard Commission",
  "type": "PERCENT",
  "firstYearValue": 55.00,
  "level": 1,
  "_links": {
    "self": { "href": "/v1/customers/commissionLevels/d5e6f7a8-b9c0-1234-def0-234567890123" }
  }
}

Field notes:

firstYearValue and level have a maximum value of 99999.99
administrativeDivisions — optional; array of two-letter state/territory codes scoping where this level applies
userDefined1–userDefined10 — free-form string fields for your own internal metadata

Step 5: Create a Contract Assignment

Contract assignments link a producer (identified by NPN) to a product at a specific commission level.

Before creating, check whether the producer is already contracted to this product to avoid duplicates:

curl -X GET \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/contractAssignments?npn=15645555" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

If _embedded is present, check _embedded.contractAssignments for any record with the same productId. If _embedded is absent, no assignments exist for this NPN yet.

curl

# List available assignment statuses (~34 total)
# Use the "description" field for human-readable labels (e.g., "Approved")
# The "name" field is SCREAMING_SNAKE_CASE (e.g., "APPROVED") — use "id" in your request
curl -X GET "https://api.sandbox.agentsync.io/contracting/v1/assignmentStatuses" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN"

# Create the contract assignment
curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/customers/contractAssignments" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "npn": "15645555",
    "productId": "PRODUCT_UUID",
    "commissionLevelId": "COMMISSION_LEVEL_UUID",
    "assignmentStatusId": "STATUS_UUID"
  }'

Python

response = requests.post(
    f"{base_url}/v1/customers/contractAssignments",
    headers={
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    },
    json={
        "npn": "15645555",
        "productId": product_id,
        "commissionLevelId": commission_level_id,
        "assignmentStatusId": "STATUS_UUID",
    }
)
response.raise_for_status()
assignment = response.json()
print(f"Created contract assignment: {assignment['id']}")

Example response (201 Created)

{
  "id": "e6f7a8b9-c0d1-2345-ef01-345678901234",
  "npn": "15645555",
  "productId": "c4d5e6f7-a8b9-0123-cdef-123456789012",
  "commissionLevelId": "d5e6f7a8-b9c0-1234-def0-234567890123",
  "assignmentStatusId": "STATUS_UUID",
  "_links": {
    "self": { "href": "/v1/customers/contractAssignments/e6f7a8b9-c0d1-2345-ef01-345678901234" }
  }
}

Step 6: Invite the Producer (Optional)

Once the contract assignment is created, send the producer an invitation so they can complete their onboarding in the AgentSync portal. The response contains an inviteLink — a unique URL the producer follows to complete the process.

The invitation can either reference an existing contract assignment (created in Step 5) or create one inline as part of the invite call. The example below passes the assignment inline — useful when you want to do both in one call.

Two IDs in the contractAssignments object require some explanation:

toOrganizationId — the carrier's AgentSync id, returned when you created or fetched the carrier in Step 2
fromOrganizationId — your agency or MGA's AgentSync organization ID. This is a fixed value per account. Contact support@agentsync.io to obtain it, or find it by calling GET /v1/customers/carriers and locating your own organization in the results.

curl

curl -X POST \
  "https://api.sandbox.agentsync.io/contracting/v1/invitation" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "producer": {
      "npn": "15645555",
      "email": "producer@example.com",
      "firstName": "Jane",
      "lastName": "Smith",
      "inviter": {
        "firstName": "Admin",
        "lastName": "User",
        "email": "admin@youragency.com"
      }
    },
    "contractAssignments": [
      {
        "administrativeDivisions": ["CO", "CA"],
        "toOrganizationId": "CARRIER_UUID",
        "fromOrganizationId": "YOUR_ORG_UUID",
        "productId": "PRODUCT_UUID",
        "commissionLevelId": "COMMISSION_LEVEL_UUID",
        "uplineNpn": "UPLINE_NPN",
        "userDefined1": "my-internal-ref"
      }
    ]
  }'

Python

response = requests.post(
    f"{base_url}/v1/invitation",
    headers={
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    },
    json={
        "producer": {
            "npn": "15645555",
            "email": "producer@example.com",
            "firstName": "Jane",
            "lastName": "Smith",
            "inviter": {
                "firstName": "Admin",
                "lastName": "User",
                "email": "admin@youragency.com",
            },
        },
        "contractAssignments": [
            {
                "administrativeDivisions": ["CO", "CA"],
                "toOrganizationId": carrier_id,         # id from Step 2
                "fromOrganizationId": "YOUR_ORG_UUID",  # your agency's AgentSync org ID
                "productId": product_id,
                "commissionLevelId": commission_level_id,
                "uplineNpn": "UPLINE_NPN",              # optional; NPN of the producer's upline
                "userDefined1": "my-internal-ref",      # optional; userDefined1–userDefined10 available
            }
        ],
    }
)
response.raise_for_status()
invite_link = response.json()["inviteLink"]
print(f"Invitation link: {invite_link}")

Example response (201 Created)

{
  "inviteLink": "https://login.agentsync.io/tokens/M7F_kGNApLk0AsqsrWKl/verify"
}

Send the inviteLink to the producer via your preferred channel — they follow it to complete their contracting onboarding in the AgentSync portal.

Note: Content-Type: application/json is required. Omitting it returns a 500 error instead of 415.

Error Handling

Error	Likely cause	Fix
`400 Bad Request`	Invalid field value — wrong `type` casing, field too long	Check validation constraints below and fix the payload
`401 Unauthorized`	Token expired	Re-authenticate and retry with a new token
`403 Forbidden`	Missing scope	Ensure your token includes `rino_api_agency_write`
`404 Not Found`	Referenced ID doesn't exist (e.g., `productId`, `commissionLevelId`)	Verify the ID with a GET request first
`422 Unprocessable Entity`	Duplicate carrier name	Check for an existing carrier before creating
`429 Too Many Requests`	Rate limit exceeded	Implement exponential backoff — see Rate Limits

Validation errors return 400 Bad Request with a messages array:

{
  "status": 400,
  "timestamp": 1773867190,
  "messages": [
    "naic must have a max of 6 characters."
  ],
  "path": "/contracting/v1/customers/carriers",
  "error": "Bad Request"
}

Common validation constraints:

Field	Constraint
`naic`	Max 6 characters
`tin`	Max 9 characters
`npn`	Max 15 characters
`commissionLevel.firstYearValue`	Max 99999.99
`commissionLevel.level`	Max 99999.99
`commissionLevel.type`	Must be `"PERCENT"`, `"DOLLAR"`, or `"NUMBER"`

Contracting API Best Practices — bulk operations, sync patterns, lookup caching
Contracting API Webhook Events — subscribe to events to react to assignment and producer status changes
Authentication — full OAuth 2.0 token flow reference
Rate Limits — how to handle 429 Too Many Requests

Complete Python Reference Implementation

This script implements the full onboarding flow from Steps 1–6. It handles token reuse, checks for existing records before creating, and sends the invitation.

import time
import requests
import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

SANDBOX_TOKEN_URL = "https://auth.sandbox.agentsync.io/oauth2/token"
SANDBOX_BASE_URL  = "https://api.sandbox.agentsync.io/contracting"


class ContractingClient:
    """Thin wrapper around the AgentSync Contracting API with token reuse."""

    TOKEN_BUFFER_SECONDS = 300  # refresh 5 min before expiry

    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._token        = None
        self._token_expiry = 0.0

    # ── Auth ──────────────────────────────────────────────────────────────────

    def _refresh_token(self) -> None:
        resp = requests.post(
            self.token_url,
            data={
                "grant_type":    "client_credentials",
                "client_id":     self.client_id,
                "client_secret": self.client_secret,
                "scope":         "rino_api_agency_read rino_api_agency_write",
            },
        )
        resp.raise_for_status()
        data = resp.json()
        self._token        = data["access_token"]
        self._token_expiry = time.time() + data["expires_in"] - self.TOKEN_BUFFER_SECONDS
        logging.info("Token refreshed")

    def _headers(self) -> dict:
        if not self._token or time.time() >= self._token_expiry:
            self._refresh_token()
        return {"Authorization": f"Bearer {self._token}", "Content-Type": "application/json"}

    # ── Generic helpers ───────────────────────────────────────────────────────

    def get(self, path: str, params: dict = None) -> dict:
        resp = requests.get(f"{self.base_url}{path}", headers=self._headers(), params=params)
        resp.raise_for_status()
        return resp.json()

    def post(self, path: str, body: dict) -> dict:
        resp = requests.post(f"{self.base_url}{path}", headers=self._headers(), json=body)
        resp.raise_for_status()
        return resp.json()

    # ── Carriers ──────────────────────────────────────────────────────────────

    def find_or_create_carrier(self, name: str, naic: str, tin: str, upline_firm_name: str) -> str:
        """Return the carrier ID, creating the carrier if it doesn't exist."""
        data = self.get("/v1/customers/carriers")
        for carrier in data.get("_embedded", {}).get("carriers", []):
            if carrier.get("naic") == naic:
                logging.info(f"Carrier found: {carrier['id']}")
                return carrier["id"]

        logging.info("Carrier not found — creating")
        carrier = self.post("/v1/customers/carriers", {
            "name":           name,
            "naic":           naic,
            "tin":            tin,
            "uplineFirmName": upline_firm_name,
        })
        logging.info(f"Carrier created: {carrier['id']}")
        return carrier["id"]

    # ── Products ──────────────────────────────────────────────────────────────

    def find_or_create_product(
        self, carrier_id: str, product_name: str,
        lob_id: str, csm_id: str, ann_id: str
    ) -> str:
        """Return the product ID, creating the product if it doesn't exist."""
        data = self.get(f"/v1/customers/carriers/{carrier_id}/products")
        for product in data.get("_embedded", {}).get("products", []):
            if product.get("name") == product_name:
                logging.info(f"Product found: {product['id']}")
                return product["id"]

        logging.info("Product not found — creating")
        product = self.post(f"/v1/customers/carriers/{carrier_id}/products", {
            "name":                       product_name,
            "lineOfBusinessId":           lob_id,
            "contractSubmissionMethodId": csm_id,
            "annualizationId":            ann_id,
        })
        logging.info(f"Product created: {product['id']}")
        return product["id"]

    # ── Commission Levels ─────────────────────────────────────────────────────

    def find_or_create_commission_level(
        self, product_id: str, name: str,
        level_type: str, first_year_value: float, level: int
    ) -> str:
        """Create a commission level for the product and return its ID.

        To reuse an existing level instead, call GET /v1/customers/commissionLevels/{id}
        and pass the existing ID directly to create_contract_assignment.
        """
        commission_level = self.post(f"/v1/customers/{product_id}/commissionLevels", {
            "name":           name,
            "type":           level_type,   # Must be "Percent", "Dollar", or "Number"
            "firstYearValue": first_year_value,
            "level":          level,
        })
        logging.info(f"Commission level created: {commission_level['id']}")
        return commission_level["id"]

    # ── Contract Assignments ──────────────────────────────────────────────────

    def find_existing_assignment(self, npn: str, product_id: str) -> str | None:
        """Return the assignment ID if one already exists for this NPN + product."""
        data = self.get("/v1/customers/contractAssignments", params={"npn": npn})
        for assignment in data.get("_embedded", {}).get("contractAssignments", []):
            if assignment.get("productId") == product_id:
                logging.info(f"Existing assignment found: {assignment['id']}")
                return assignment["id"]
        return None

    def create_contract_assignment(
        self, npn: str, product_id: str,
        commission_level_id: str, status_id: str
    ) -> str:
        """Create a contract assignment and return its ID."""
        assignment = self.post("/v1/customers/contractAssignments", {
            "npn":                npn,
            "productId":          product_id,
            "commissionLevelId":  commission_level_id,
            "assignmentStatusId": status_id,
        })
        logging.info(f"Contract assignment created: {assignment['id']}")
        return assignment["id"]

    # ── Invitations ───────────────────────────────────────────────────────────

    def send_invitation(
        self, npn: str, email: str, first_name: str, last_name: str,
        inviter: dict, contract_assignments: list
    ) -> str:
        """Send a producer invitation and return the inviteLink.

        contract_assignments is a list of dicts, each containing:
          toOrganizationId, fromOrganizationId, productId, commissionLevelId,
          and optionally administrativeDivisions.
        """
        result = self.post("/v1/invitation", {
            "producer": {
                "npn":       npn,
                "email":     email,
                "firstName": first_name,
                "lastName":  last_name,
                "inviter":   inviter,
            },
            "contractAssignments": contract_assignments,
        })
        invite_link = result["inviteLink"]
        logging.info(f"Invitation sent to {email} (NPN {npn}): {invite_link}")
        return invite_link

    # ── Lookups ───────────────────────────────────────────────────────────────

    def get_first_assignment_status_id(self) -> str:
        """Fetch the first available assignment status ID."""
        data = self.get("/v1/assignmentStatuses")
        statuses = list(data.get("_embedded", {}).values())
        if not statuses or not statuses[0]:
            raise ValueError("No assignment statuses found")
        return statuses[0][0]["id"]


def onboard_producer(
    client: ContractingClient,
    npn: str,
    email: str,
    first_name: str,
    last_name: str,
    inviter: dict,
    from_organization_id: str,
    carrier_name: str,
    carrier_naic: str,
    carrier_tin: str,
    upline_firm_name: str,
    product_name: str,
    lob_id: str,
    csm_id: str,
    ann_id: str,
) -> None:
    """End-to-end producer contracting onboarding flow."""

    logging.info(f"Starting onboarding for NPN {npn}")

    # 1. Verify / create carrier
    carrier_id = client.find_or_create_carrier(
        carrier_name, carrier_naic, carrier_tin, upline_firm_name
    )

    # 2. Verify / create product
    product_id = client.find_or_create_product(
        carrier_id, product_name, lob_id, csm_id, ann_id
    )

    # 3. Ensure a commission level exists
    status_id = client.get_first_assignment_status_id()
    commission_level_id = client.find_or_create_commission_level(
        product_id, "Standard — 55%", "Percent", 55.00, 1
    )

    # 4. Check for existing assignment — skip creation if already contracted
    existing = client.find_existing_assignment(npn, product_id)
    if existing:
        logging.info(f"NPN {npn} is already contracted to product {product_id} — skipping")
    else:
        client.create_contract_assignment(npn, product_id, commission_level_id, status_id)

    # 5. Send invitation
    client.send_invitation(
        npn=npn,
        email=email,
        first_name=first_name,
        last_name=last_name,
        inviter=inviter,
        contract_assignments=[
            {
                "administrativeDivisions": [],   # e.g. ["CO", "CA"]
                "toOrganizationId":   carrier_id,
                "fromOrganizationId": from_organization_id,
                "productId":          product_id,
                "commissionLevelId":  commission_level_id,
            }
        ],
    )

    logging.info(f"Onboarding complete for NPN {npn}")


def fetch_first_id(client: ContractingClient, path: str) -> str:
    """Helper to fetch the first ID from a lookup endpoint's _embedded collection."""
    data = client.get(path)
    for items in data.get("_embedded", {}).values():
        if items:
            return items[0]["id"]
    raise ValueError(f"No items found at {path}")


if __name__ == "__main__":
    client = ContractingClient(
        client_id="YOUR_CLIENT_ID",
        client_secret="YOUR_CLIENT_SECRET",
        base_url=SANDBOX_BASE_URL,
        token_url=SANDBOX_TOKEN_URL,
    )

    # Fetch real lookup IDs from your sandbox before running
    # These reference values are stable — cache them rather than fetching every run
    lob_id = fetch_first_id(client, "/v1/linesOfBusiness")
    csm_id = fetch_first_id(client, "/v1/contractSubmissionMethods")
    ann_id = fetch_first_id(client, "/v1/annualizations")

    onboard_producer(
        client=client,
        npn="15645555",
        email="producer@example.com",
        first_name="Jane",
        last_name="Smith",
        inviter={
            "firstName": "Admin",
            "lastName":  "User",
            "email":     "admin@youragency.com",
        },
        from_organization_id="YOUR_ORG_UUID",  # your agency's AgentSync org ID
        carrier_name="Acme Life Insurance",
        carrier_naic="12345",
        carrier_tin="123456789",
        upline_firm_name="Acme Holdings",
        product_name="Term Life — Standard",
        lob_id=lob_id,
        csm_id=csm_id,
        ann_id=ann_id,
    )

Contracting API: Quick Start Guide

Before You Start

Overview

Step 1: Make Your First Call

Step 2: Set Up a Carrier

Fetch lookup IDs

Create the carrier

Step 3: Add a Product to the Carrier

Step 4: Add Commission Levels

Step 5: Create a Contract Assignment

Step 6: Invite the Producer (Optional)

Error Handling

Token Management

Deprecated Fields

Next Steps

Complete Python Reference Implementation