Python SDK

The official Python client for the Syndicate Links API. Type-checked with Pydantic v2, supports both sync and async, auto-paginates, retries transient failures, and ships every endpoint the API exposes — merchant, publisher, and agent.

Use it to build affiliate programs for SaaS or physical goods, give AI agents first-class attribution, and automate commission payouts from a script or a production service.

pip install syndicate-links

Python 3.9 or newer. Depends only on httpx and pydantic.


Authentication

Every request uses an API key on the Authorization header. Key prefixes determine which endpoints you can call:

PrefixRoleEndpoints
mk_live_Merchantclient.merchant.*
ak_live_Affiliate (human or agent)client.affiliate.*
aff_agent_AI agent publisherclient.affiliate.*

The SDK auto-routes based on the key prefix. Calling a merchant method with an affiliate key raises ForbiddenError before the request leaves your machine.

from syndicate_links import SyndicateClient

client = SyndicateClient(api_key="mk_live_...")
# → client.merchant is available
# → client.affiliate raises ForbiddenError

Quick start — Merchant

Launch a program, add products, and list your affiliates.

from syndicate_links import SyndicateClient

client = SyndicateClient(api_key="mk_live_...")

# Create a program
program = client.merchant.programs.create(
    name="Acme Pro",
    default_commission_pct=20,
    cookie_days=30,
    auto_approve=True,
    category="saas",
)

# Add a product
client.merchant.products.create(
    program_id=program.id,
    name="Acme Pro (Monthly)",
    url="https://acme.com/pro",
    price=49.00,
    commission_pct=25,  # overrides the program default
)

# Bulk import a catalog
client.merchant.products.bulk_create(
    program_id=program.id,
    products=[
        {"name": "Starter", "url": "https://acme.com/starter", "price": 9},
        {"name": "Team",    "url": "https://acme.com/team",    "price": 29},
        {"name": "Pro",     "url": "https://acme.com/pro",     "price": 49},
    ],
)

# Approve pending partnerships
for page in client.merchant.affiliates.iter_all(status="pending"):
    for partnership in page:
        client.merchant.affiliates.approve(partnership.partnership.id)

Quick start — Publisher (human affiliate)

Join a program, generate a tracking link, and check your balance.

from syndicate_links import SyndicateClient

client = SyndicateClient(api_key="ak_live_...")

# Browse programs
page = client.affiliate.programs.list(sort="commission_desc")
program = page.data[0]

# Apply and get a tracking link
client.affiliate.programs.apply(program.id)
link = client.affiliate.links.create(
    program_id=program.id,
    destination_url="https://acme.com/pro",
    source_tag="blog-post-1",
)

print(f"Share this: https://syndicatelinks.co/r/{link.code}")

# Check earnings
balance = client.affiliate.me.balance()
print(f"Available: ${balance.balance} {balance.currency}")

Quick start — AI agent

The differentiator. Mint an attribution token, record a conversion with ai_referral and ai_surface, and get paid via Lightning or USDC.

from syndicate_links import SyndicateClient

agent = SyndicateClient(api_key="aff_agent_...")

# Create a tracking link for a product recommendation
link = agent.affiliate.links.create(
    program_id="prog_...",
    destination_url="https://acme.com/pro",
    source_tag="chatgpt-plugin",
)

# Mint an attribution token before recommending
token = agent.affiliate.attribution_token.create(
    program_id="prog_...",
    tracking_code=link.code,
    agent_id="acme-research-bot",
    surface="chatgpt",
    ttl_seconds=3600,
)

# When the customer buys, record the conversion
conversion = agent.affiliate.events.conversion(
    tracking_code=link.code,
    order_id="ord_42",
    sale_amount=49.00,
    ai_referral="ai",
    ai_surface="chatgpt",
)

print(f"Earned ${conversion.commission_amount} on order {conversion.order_id}")

Pagination

Every list endpoint supports cursor-based pagination. You have two options.

Manual pagination — explicit control for batch jobs:

page = client.merchant.products.list(limit=50)
while True:
    for product in page.data:
        process(product)
    if not page.has_more:
        break
    page = client.merchant.products.list(cursor=page.cursor, limit=50)

Auto-pagination — iterate through every result transparently:

for product in client.merchant.products.iter_all():
    process(product)

iter_all() is available on every list endpoint. Filters (like status="pending") are forwarded automatically.


Async usage

Every client has an async twin. The method signatures are identical.

import asyncio
from syndicate_links import AsyncSyndicateClient

async def main():
    async with AsyncSyndicateClient(api_key="mk_live_...") as client:
        # Single request
        programs = await client.merchant.programs.list()

        # Async iteration
        async for product in client.merchant.products.iter_all():
            print(product.name)

asyncio.run(main())

Error handling

The SDK maps HTTP status codes to specific exception types. Everything inherits from SyndicateError.

SyndicateError
├── APIError
│   ├── ValidationError      # 400
│   ├── AuthenticationError  # 401
│   ├── ForbiddenError       # 403
│   ├── NotFoundError        # 404
│   ├── ConflictError        # 409
│   ├── RateLimitError       # 429
│   └── ServerError          # 5xx
└── NetworkError             # Timeouts, DNS failures
from syndicate_links import SyndicateClient
from syndicate_links.errors import (
    NotFoundError,
    ValidationError,
    RateLimitError,
    NetworkError,
)

client = SyndicateClient(api_key="mk_live_...")

try:
    program = client.merchant.programs.update(
        program_id="prog_does_not_exist",
        name="Updated",
    )
except NotFoundError:
    print("Program not found")
except ValidationError as e:
    print(f"Bad request: {e.message}")
except RateLimitError:
    print("Slow down")
except NetworkError as e:
    print(f"Couldn't reach API: {e}")

Retries

The client retries 5xx responses and network errors up to 3 times by default, with jittered exponential backoff (0.5s → 16s cap). 4xx errors are never retried.

client = SyndicateClient(
    api_key="mk_live_...",
    max_retries=5,       # default: 3
    timeout=60.0,        # default: 30.0
)

Configuration

from syndicate_links import SyndicateClient

client = SyndicateClient(
    api_key="mk_live_...",
    base_url="https://api.syndicatelinks.co",  # override for dev
    timeout=30.0,
    max_retries=3,
    user_agent="my-app/1.0",
)

Webhook testing

You can exercise webhook endpoints without leaving your Python console. Create a webhook, then run the built-in trigger to validate your signature check.

webhook = client.merchant.webhooks.create(
    url="https://myapp.example.com/webhooks/syndicate",
    events=["conversion.created", "conversion.refunded", "payout.paid"],
)
print(f"Webhook ID: {webhook.id}")
print(f"Signing secret: {webhook.secret}")

Check merchant.webhooks.list() to see every active subscription and their delivery history.


API coverage

Every endpoint the REST API exposes is implemented. Highlights:

Merchant

NamespaceMethods
merchant.programslist, create, update, iter_all
merchant.productslist, create, bulk_create, update, delete, iter_all
merchant.affiliateslist, approve, reject, iter_all
merchant.conversionslist, create, iter_all
merchant.refundscreate
merchant.reportssummary, ai_endorsements
merchant.billingget
merchant.profileget, update
merchant.payoutslist, stats, iter_all
merchant.webhookslist, create, delete

Affiliate / Publisher / Agent

NamespaceMethods
affiliate.programslist, mine, get, products, apply, iter_all
affiliate.productssearch
affiliate.partnershipslist, iter_all
affiliate.linkslist, create, iter_all
affiliate.eventslist, click, conversion
affiliate.reportsearnings, clicks, conversions
affiliate.meget, update, balance, payouts
affiliate.payoutsclaim
affiliate.dashboardsummary, earnings_chart
affiliate.keyscreate
affiliate.attribution_tokencreate

Type safety

All models are Pydantic v2. Numeric fields (prices, commission rates, sale amounts) are parsed to Decimal — never floats — so you never lose precision.

product = client.merchant.products.create(
    program_id="...",
    name="Pro",
    url="https://acme.com/pro",
    price="49.99",  # string or Decimal or float — all work
)

assert isinstance(product.price, Decimal)  # True

Models also allow extra fields, so the SDK stays compatible when the API adds new response fields.


Installation from source

git clone https://github.com/syndicate-links/syndicate-links-python
cd syndicate-links-python
pip install -e ".[dev]"
pytest

94 tests covering every endpoint, pagination, auth, retries, error mapping, and Decimal coercion.