Skip to main content
The agentref Python package provides both synchronous and asynchronous clients for the AgentRef REST API v1. It uses httpx under the hood and returns Pydantic models for type-safe access to all API resources.

Installation

pip install agentref

Quick Start

from agentref import AgentRef

client = AgentRef(api_key="ak_live_your_api_key_here")

# List all programs
result = client.programs.list()
print(f"Found {result.meta.total} programs")

# Get conversion stats
stats = client.conversions.stats(period="30d")
print(f"Revenue: ${stats.total_revenue / 100}")

Async Usage

from agentref import AsyncAgentRef

async with AsyncAgentRef(api_key="ak_live_...") as client:
    result = await client.programs.list()
    for program in result.data:
        print(program.name)

Configuration

client = AgentRef(
    api_key="ak_live_...",                        # or set AGENTREF_API_KEY env var
    base_url="https://www.agentref.dev/api/v1",   # default
    timeout=30.0,                                  # 30s default
    max_retries=2,                                 # retries for GET + idempotent POST
)
ParameterTypeDefaultDescription
api_keystrAGENTREF_API_KEY envYour API key. Required.
base_urlstrhttps://www.agentref.dev/api/v1API base URL
timeoutfloat30.0Request timeout in seconds
max_retriesint2Max retries for safe/idempotent requests

Resource Namespaces

The client exposes 10 resource namespaces:
NamespaceDescription
client.programsCreate, update, delete, and list affiliate programs
client.affiliatesList, approve, block, and unblock affiliates
client.conversionsList conversions, get stats, and fetch recent activity
client.payoutsList payouts, pending affiliates, stats, and create payouts
client.flagsList fraud flags, get stats, and resolve flags
client.billingCheck billing status, list tiers, and subscribe
client.merchantGet and update merchant profile
client.notificationsGet and update notification preferences
client.payout_infoGet and update payout payment details
client.webhooksCreate, list, update, delete webhook endpoints and rotate secrets

Programs

# List programs (paginated)
result = client.programs.list(status="active", limit=10, page=1)

# Auto-paginate through all programs
for program in client.programs.list_all():
    print(program.name)

# Async auto-pagination
async for program in client.programs.list_all():
    print(program.name)

# Get program details (includes readiness status)
detail = client.programs.get("prog-uuid")
print(detail.readiness)  # 'setup' | 'partial' | 'ready'

# Create a program
program = client.programs.create(
    name="Acme Referrals",
    commission_type="recurring",
    commission_percent=25,
    cookie_duration=60,
    currency="USD",
    idempotency_key="create-acme-v1",
)

# Update a program
client.programs.update("prog-uuid", commission_percent=30, status="active")

# Delete a program
client.programs.delete("prog-uuid")

# Get program stats
stats = client.programs.stats("prog-uuid", period="30d")

# List program affiliates
affiliates = client.programs.list_affiliates("prog-uuid", include_blocked=False)

# Manage invites
invite = client.programs.create_invite(
    "prog-uuid",
    email="partner@example.com",
    expires_in_days=7,
    idempotency_key="invite-partner-v1",
)
invites = client.programs.list_invites("prog-uuid")

# Manage coupons
coupon = client.programs.create_coupon(
    "prog-uuid",
    affiliate_id="aff-uuid",
    code="PARTNER20",
    idempotency_key="coupon-partner20",
)
coupons = client.programs.list_coupons("prog-uuid")
client.programs.delete_coupon("coupon-uuid")

# Marketplace settings
client.programs.update_marketplace(
    "prog-uuid",
    status="public",
    category="SaaS",
    description="Earn 25% recurring for every referral.",
)

# Stripe connection
stripe = client.programs.connect_stripe("prog-uuid")
print(stripe.auth_url)  # redirect merchant to complete OAuth
client.programs.disconnect_stripe("prog-uuid")

Affiliates

# List all affiliates
result = client.affiliates.list(
    program_id="prog-uuid",
    status="pending",
    search="john",
    sort_by="totalRevenue",
    sort_order="desc",
    limit=25,
)

# Get single affiliate (with optional stats)
affiliate = client.affiliates.get("aff-uuid", include="stats")

# Approve
client.affiliates.approve("aff-uuid", idempotency_key="approve-aff-123")

# Block with reason
client.affiliates.block(
    "aff-uuid",
    reason="Suspicious click patterns detected",
    idempotency_key="block-aff-123",
)

# Unblock
client.affiliates.unblock("aff-uuid", idempotency_key="unblock-aff-123")

Conversions

# List conversions with filters
result = client.conversions.list(
    program_id="prog-uuid",
    status="pending",
    start_date="2026-01-01",
    end_date="2026-03-23",
    limit=50,
)

# Aggregate stats
stats = client.conversions.stats(program_id="prog-uuid", period="30d")
print(f"{stats.total} conversions, ${stats.total_revenue / 100} revenue")

# Recent conversions
recent = client.conversions.recent(limit=5)

Payouts

# List completed payouts
result = client.payouts.list(status="completed", start_date="2026-01-01")

# List pending affiliates (ready for payout)
pending = client.payouts.list_pending(program_id="prog-uuid")

for aff in pending.data:
    print(f"{aff.name}: ${aff.pending_amount / 100} ({aff.currency})")

# Payout stats
payout_stats = client.payouts.stats(period="30d")

# Create a payout
payout = client.payouts.create(
    affiliate_id="aff-uuid",
    program_id="prog-uuid",
    method="paypal",
    notes="March payout",
    idempotency_key="payout-march-aff123",
)

Fraud Flags

# List open flags
result = client.flags.list(status="open", limit=20)

# Get flag stats
flag_stats = client.flags.stats()
print(f"{flag_stats.open} open flags")

# Resolve a flag
client.flags.resolve(
    "flag-uuid",
    status="dismissed",
    note="Verified legitimate traffic",
    block_affiliate=False,
    idempotency_key="resolve-flag-abc",
)

Billing

# Check current billing status
billing = client.billing.current()
print(f"Tier: {billing.tier}, Revenue: ${billing.monthly_revenue / 100}")

# List available tiers
tiers = client.billing.tiers()

# Subscribe to a tier
client.billing.subscribe(tier="growth", idempotency_key="subscribe-growth-v1")

Merchant

# Get merchant profile
merchant = client.merchant.get()

# Update merchant settings
client.merchant.update(
    company_name="Acme Inc.",
    timezone="America/New_York",
    default_cookie_duration=60,
)

Notifications

# Get notification preferences
prefs = client.notifications.get()

# Update preferences
client.notifications.update(
    new_affiliate=True,
    new_conversion=True,
    weekly_digest=True,
)

Payout Info

# Get payout info
info = client.payout_info.get()

# Update payout info
client.payout_info.update(
    payout_method="paypal",
    paypal_email="payments@acme.com",
)

Webhooks

# List webhook endpoints
endpoints = client.webhooks.list(program_id="prog-uuid")

# Create endpoint
result = client.webhooks.create(
    name="Production Webhook",
    url="https://api.acme.com/webhooks/agentref",
    subscribed_events=[
        "conversion.created",
        "payout.completed",
        "affiliate.joined",
    ],
    program_id="prog-uuid",
)
signing_secret = result.signing_secret  # store securely -- shown only once

# Update endpoint
client.webhooks.update(
    "wh-uuid",
    subscribed_events=[
        "conversion.created",
        "conversion.refunded",
        "payout.completed",
    ],
)

# Rotate signing secret
new_result = client.webhooks.rotate_secret("wh-uuid")

# Delete endpoint
client.webhooks.delete("wh-uuid")

Pagination

All list endpoints return a PaginatedResponse[T] with Pydantic-typed data and metadata:
result = client.programs.list(page=1, limit=25)

print(result.data)          # List[Program]
print(result.meta.total)    # Total count
print(result.meta.has_more) # More pages available?
print(result.meta.page)     # Current page

Auto-pagination

The list_all() generator handles pagination automatically:
# Sync
for program in client.programs.list_all(page_size=100):
    print(program.name)

# Async
async for program in client.programs.list_all(page_size=100):
    print(program.name)

Idempotency

All mutation methods accept an idempotency_key keyword argument. When provided, the request is safe to retry — the server guarantees at-most-once execution.
program = client.programs.create(
    name="Acme Referrals",
    commission_type="recurring",
    commission_percent=25,
    idempotency_key="create-acme-referrals-v1",
)
The SDK automatically retries failed requests (up to max_retries) for:
  • GET/HEAD requests — always safe to retry
  • POST requests with an idempotency key — server-side deduplication
  • 429 (rate limited) and 5xx (server error) responses
POST requests without an idempotency key are never retried.

Error Handling

The SDK raises typed exceptions for all API failures:
from agentref import (
    AgentRefError,
    AuthError,
    ForbiddenError,
    ValidationError,
    NotFoundError,
    ConflictError,
    RateLimitError,
    ServerError,
)

try:
    client.programs.get("nonexistent-id")
except NotFoundError as e:
    print(f"Not found: {e}")
    print(f"Request ID: {e.request_id}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Validation failed: {e}")
    print(f"Details: {e.details}")
except AgentRefError as e:
    print(f"API error {e.status}: {e.code} - {e}")
Exception ClassHTTP StatusWhen
AuthError401Invalid or missing API key
ForbiddenError403Key lacks required scope
ValidationError400Invalid request parameters
NotFoundError404Resource does not exist
ConflictError409Duplicate resource or state conflict
RateLimitError429Too many requests
ServerError5xxServer-side error
All exceptions inherit from AgentRefError and include code, status, and request_id attributes.

Sync vs Async

The Python SDK ships with both synchronous and asynchronous clients:
from agentref import AgentRef

client = AgentRef(api_key="ak_live_...")
programs = client.programs.list()
client.close()  # optional cleanup
The async client supports async with for automatic connection cleanup. All async resource methods are identical to their sync counterparts but return coroutines.