Webhooks¶

The SDK manages webhooks through the V2 API (client.webhooks; Affinity API version 2026-07-15 or later, beta). Receiving webhooks is handled by your web server/app. The CLI has the same operations: xaffinity webhook ls|get|create|update|delete.

Who sees which webhooks

Users allowed to manage all webhooks see every webhook in the organization. Everyone else lists, reads, changes and deletes only the webhooks they created.

Supported Webhook Events¶

The SDK defines all webhook events in WebhookEvent. You can subscribe to any combination of these events:

Event	Value	Description
`LIST_CREATED`	`list.created`	A new list was created
`LIST_UPDATED`	`list.updated`	A list's properties were modified
`LIST_DELETED`	`list.deleted`	A list was deleted
`LIST_ENTRY_CREATED`	`list_entry.created`	An entity was added to a list
`LIST_ENTRY_DELETED`	`list_entry.deleted`	An entity was removed from a list
`NOTE_CREATED`	`note.created`	A note was created on an entity
`NOTE_UPDATED`	`note.updated`	A note's content was modified
`NOTE_DELETED`	`note.deleted`	A note was deleted
`FIELD_CREATED`	`field.created`	A new field was created
`FIELD_UPDATED`	`field.updated`	A field's properties were modified
`FIELD_DELETED`	`field.deleted`	A field was deleted
`FIELD_VALUE_CREATED`	`field_value.created`	A field value was set on an entity
`FIELD_VALUE_UPDATED`	`field_value.updated`	A field value was modified
`FIELD_VALUE_DELETED`	`field_value.deleted`	A field value was removed
`PERSON_CREATED`	`person.created`	A new person was created
`PERSON_UPDATED`	`person.updated`	A person's properties were modified
`PERSON_DELETED`	`person.deleted`	A person was deleted
`ORGANIZATION_CREATED`	`organization.created`	A new company/organization was created
`ORGANIZATION_UPDATED`	`organization.updated`	A company's properties were modified
`ORGANIZATION_DELETED`	`organization.deleted`	A company was deleted
`ORGANIZATION_MERGED`	`organization.merged`	Two companies were merged
`OPPORTUNITY_CREATED`	`opportunity.created`	A new opportunity was created
`OPPORTUNITY_UPDATED`	`opportunity.updated`	An opportunity's properties were modified
`OPPORTUNITY_DELETED`	`opportunity.deleted`	An opportunity was deleted
`FILE_CREATED`	`file.created`	A file was uploaded to an entity
`FILE_DELETED`	`file.deleted`	A file was deleted
`SMART_INTERACTION_VALUE_UPDATED`	`smart_interaction_value.updated`	A smart interaction field value was updated (listed by the V2 API; Affinity does not document the payload)
`REMINDER_CREATED`	`reminder.created`	A reminder was created
`REMINDER_UPDATED`	`reminder.updated`	A reminder's properties were modified
`REMINDER_DELETED`	`reminder.deleted`	A reminder was deleted

Forward Compatibility

WebhookEvent extends OpenStrEnum, meaning unknown event types from Affinity are preserved as strings rather than causing errors. This ensures the SDK gracefully handles any new events Affinity may add in the future.

Create a webhook¶

from affinity import Affinity
from affinity.models import WebhookCreate
from affinity.types import WebhookEvent

with Affinity.from_env() as client:
    webhook = client.webhooks.create(
        WebhookCreate(
            url="https://example.com/webhooks/affinity/<random-secret>",
            subscriptions=[
                WebhookEvent.FIELD_VALUE_UPDATED,
                WebhookEvent.LIST_ENTRY_CREATED,
            ],
        )
    )
    print(webhook.id, webhook.url, webhook.status)

subscriptions is required. An empty list subscribes the webhook to every event type, so every change in the organization is sent to the URL; name the events you need.

Test delivery

Affinity sends a test delivery to the URL before creating the webhook (and before changing its URL or re-activating it). A URL that is unreachable, contains credentials, or is already used by another webhook in the organization is rejected with ValidationError, as is a create once the organization reaches its webhook limit.

List, change and delete¶

from affinity.models import WebhookUpdate
from affinity.types import WebhookStatus

# One page (up to 100), or every webhook
page = client.webhooks.list(limit=50)
for webhook in client.webhooks.iter():
    print(webhook.id, webhook.url, webhook.status, webhook.subscriptions or "all events")

# Pause delivery; only the fields you set change. Returns the webhook as read back
# (two requests: the update itself has no response body).
webhook = client.webhooks.update(webhook.id, WebhookUpdate(status=WebhookStatus.DISABLED))

client.webhooks.delete(webhook.id)  # returns None

list()/iter() filter by ids, created_after/created_before and updated_after/updated_before. Webhook has id, url, subscriptions ([] = all events), status (active/disabled), creator (PersonRef or None), created_at and updated_at.

From SDK 1.x

1.x used the V1 API: WebhookSubscription is now Webhook, webhook_url is url, disabled=True is status=WebhookStatus.DISABLED, created_by is creator.id, list() returns a page (use iter() for all), delete() returns None, and subscriptions must be given on create. Old field names raise ValidationError.

Securing your webhook endpoint¶

No signature verification available

Affinity does not provide cryptographic signature verification for webhook requests. There is no HMAC header, signing secret, or other mechanism to verify that requests originate from Affinity. You must rely on defense-in-depth practices to secure your endpoint.

Since webhook authenticity cannot be cryptographically verified, treat your webhook endpoint as a semi-public entry point and apply multiple layers of protection:

Required: Secret URL path¶

Include a long, random, unguessable secret in your webhook URL path:

https://example.com/webhooks/affinity/a1b2c3d4e5f6g7h8i9j0...

Generate at least 32 characters of cryptographically random data
Reject any request where the path secret doesn't match
Rotate the secret periodically and after any suspected exposure
Never log the full URL or share it in plain text

Required: HTTPS only¶

Always use HTTPS for your webhook URL
Terminate TLS at your load balancer or reverse proxy
Reject HTTP requests at the application level as a fallback

Required: Request validation¶

Method: Only accept POST requests
Content-Type: Require application/json
Body size: Enforce a reasonable limit (e.g., 1MB)
JSON parsing: Use strict parsing; reject malformed payloads

Recommended: Replay protection¶

Use the sent_at field in the webhook payload to reject stale events:

from affinity import parse_webhook
from affinity.exceptions import WebhookInvalidSentAtError

try:
    # Reject events older than 5 minutes (300 seconds)
    envelope = parse_webhook(raw_body, max_age_seconds=300)
except WebhookInvalidSentAtError:
    # Event is too old or too far in the future
    return Response(status=400)

For stronger replay protection, store a short-lived dedupe key:

import hashlib

# Generate a dedupe key from event properties
dedupe_key = f"{envelope.type}:{envelope.sent_at_epoch}:{hashlib.sha256(raw_body).hexdigest()[:16]}"

# Check against a cache (Redis, memcached, etc.) with 5-10 minute TTL
if cache.exists(dedupe_key):
    return Response(status=200)  # Already processed, acknowledge silently
cache.set(dedupe_key, "1", ttl=600)

Recommended: IP allowlisting¶

If Affinity provides stable egress IP ranges for your account:

Configure your load balancer or WAF to only accept webhook traffic from those IPs
Contact Affinity support to request their webhook delivery IP ranges

If IP ranges are not available:

Restrict by geography or ASN where appropriate
Alert on unexpected source IPs for investigation
Apply rate limiting and bot protection at the edge

Recommended: Fast response with async processing¶

Respond with 2xx immediately after basic validation
Enqueue the actual processing for async handling
Assume retries can happen (Affinity retries with exponential backoff for up to 10 hours)
Make your processing idempotent using the dedupe key pattern above

Recommended: Logging considerations¶

Avoid logging raw webhook payloads (may contain PII)
If logging is required, redact sensitive fields or use a PII-safe pipeline
Log the event type, timestamp, and dedupe key for debugging

Parse inbound payloads (optional)¶

The SDK includes small, framework-agnostic helpers to parse the webhook envelope and (optionally) dispatch to a typed body for a few common events.

from affinity import dispatch_webhook, parse_webhook
from affinity.types import WebhookEvent

envelope = parse_webhook(raw_body_bytes)  # or raw str / dict
event = dispatch_webhook(envelope)  # typed for some events, dict otherwise

if event.type == WebhookEvent.LIST_ENTRY_CREATED:
    # event.body may be typed (or a dict, depending on the event)
    print(event.sent_at, event.body)

Minimal receiver example (FastAPI)¶

import hashlib
import secrets

from fastapi import FastAPI, HTTPException, Header, Request

from affinity import dispatch_webhook, parse_webhook
from affinity.exceptions import WebhookInvalidSentAtError, WebhookParseError

app = FastAPI()

# Generate with: secrets.token_urlsafe(32)
WEBHOOK_SECRET = "replace-with-a-long-random-string-at-least-32-chars"


@app.post("/webhooks/affinity/{secret}")
async def affinity_webhook(
    secret: str,
    request: Request,
    content_type: str = Header(default=""),
) -> dict[str, str]:
    # 1. Validate secret path
    if not secrets.compare_digest(secret, WEBHOOK_SECRET):
        raise HTTPException(status_code=404)  # 404 to avoid confirming endpoint exists

    # 2. Validate content type
    if not content_type.startswith("application/json"):
        raise HTTPException(status_code=415, detail="unsupported media type")

    # 3. Read and validate body size
    raw = await request.body()
    if len(raw) > 1_000_000:  # 1MB limit
        raise HTTPException(status_code=413, detail="payload too large")

    # 4. Parse with replay protection (rejects events older than 5 minutes)
    try:
        envelope = parse_webhook(raw, max_age_seconds=300)
    except WebhookInvalidSentAtError:
        raise HTTPException(status_code=400, detail="stale event")
    except WebhookParseError:
        raise HTTPException(status_code=400, detail="invalid payload")

    # 5. Optional: Check dedupe key against cache here

    # 6. Dispatch to typed body (if registered) and process
    event = dispatch_webhook(envelope)

    # TODO: Enqueue for async processing instead of processing inline
    # process_webhook_event.delay(event)

    return {"ok": "true"}