Affinity Python SDK

A modern, strongly-typed Python wrapper for the Affinity CRM API.

Disclaimer: This is an unofficial community project and is not affiliated with, endorsed by, or sponsored by Affinity. “Affinity” and related marks are trademarks of their respective owners. Use of the Affinity API is subject to Affinity’s Terms of Service.

Maintainer: GitHub: yaniv-golan

Documentation: https://yaniv-golan.github.io/affinity-sdk/latest/

Affinity's Official MCP Server

As of March 2026, Affinity has released an official MCP Server (beta) for conversational, natural-language access to your CRM data via AI chat clients. It covers relationship intelligence queries, pipeline summaries, meeting activity, and note capture.

This SDK serves a different purpose — it's a full-coverage, strongly-typed Python client for the Affinity API, supporting the complete read/write surface (companies, persons, lists, field values, notes, reminders, webhooks, files, and more). Use it when you need programmatic control, write operations, type safety, or want to build custom integrations and tooling.

For a detailed comparison, see Affinity SDK vs. Official MCP.

Table of Contents

• Features
• Installation
• Quick Start
• Usage Examples
• Type System
• API Coverage
• Configuration
• Error Handling
• Async Support
• Development

Features

• Complete API coverage - Full V1 + V2 support with smart routing
• CLI included - Scriptable command-line interface for automation
• Strong typing - Full Pydantic V2 models with typed ID classes
• No magic numbers - Comprehensive enums for all API constants
• Automatic pagination - Iterator support for seamless pagination
• Rate limit handling - Automatic retry with exponential backoff
• Response caching - Optional caching for field metadata
• Both sync and async - Full support for both patterns

AI Integrations

• Claude Code plugins - SDK and CLI knowledge for AI-assisted development
• MCP Server - Connect desktop AI tools to Affinity

Installation

pip install affinity-sdk

Requires Python 3.10+.

Optional (local dev): load .env automatically:

pip install "affinity-sdk[dotenv]"

Optional: install the CLI:

pipx install "affinity-sdk[cli]"

The CLI includes a powerful query command for structured data extraction with filtering, aggregations, and relationship includes. Output formats include JSON, CSV, markdown, and TOON (token-optimized for LLMs).

CLI docs: https://yaniv-golan.github.io/affinity-sdk/latest/cli/

MCP Server

Connect desktop AI tools to Affinity CRM.

Claude Desktop (easiest - MCPB bundle):

1. Install CLI: pipx install "affinity-sdk[cli]"
2. (Optional) Pre-configure API key: xaffinity config setup-key
   • If skipped, Claude Desktop will prompt for your API key during MCPB install
3. Download the .mcpb bundle from GitHub Releases
4. Double-click to install (or drag to Claude Desktop)

Other clients (Cursor, Windsurf, VS Code + Copilot, Zed, etc.):

These require manual configuration. See the MCP Server docs for step-by-step instructions.

MCP docs: https://yaniv-golan.github.io/affinity-sdk/latest/mcp/

Claude Code Plugins

If you use Claude Code, install plugins for SDK/CLI knowledge:

/plugin marketplace add yaniv-golan/affinity-sdk
/plugin install sdk@xaffinity   # SDK patterns
/plugin install cli@xaffinity   # CLI patterns + hooks

Plugin docs: https://yaniv-golan.github.io/affinity-sdk/latest/guides/claude-code-plugins/

Documentation

• Full documentation
• MCP Server
• CLI Reference
• API Reference

Quick Start

from affinity import Affinity
from affinity.types import FieldType, PersonId

# Recommended: read the API key from the environment (AFFINITY_API_KEY)
client = Affinity.from_env()

# If you use a local `.env` file (requires `affinity-sdk[dotenv]`)
# client = Affinity.from_env(load_dotenv=True)

# Or pass it explicitly
# client = Affinity(api_key="your-api-key")

# Or use as a context manager
with Affinity.from_env() as client:
    # List all companies
    for company in client.companies.all():
        print(f"{company.name} ({company.domain})")

    # Get a person with enriched data
    person = client.persons.get(
        PersonId(12345),
        field_types=[FieldType.ENRICHED, FieldType.GLOBAL]
    )
    print(f"{person.first_name} {person.last_name}: {person.primary_email}")

Usage Examples

Working with Companies

from affinity import Affinity, F
from affinity.models import CompanyCreate
from affinity.types import CompanyId, FieldType

with Affinity(api_key="your-key") as client:
    # List companies with filtering (V2 API)
    companies = client.companies.list(
        filter=F.field("domain").contains("acme"),
        field_types=[FieldType.ENRICHED],
    )

    # Iterate through all companies with automatic pagination
    for company in client.companies.all():
        print(f"{company.name}: {company.fields}")

    # Get a specific company
    company = client.companies.get(CompanyId(123))

    # Create a company (uses V1 API)
    new_company = client.companies.create(
        CompanyCreate(
            name="Acme Corp",
            domain="acme.com",
        )
    )

    # Search by name, domain, or email
    results = client.companies.search("acme.com")

    # Get list entries for a company
    entries = client.companies.get_list_entries(CompanyId(123))

Working with Persons

from affinity import Affinity
from affinity.models import PersonCreate
from affinity.types import PersonType

with Affinity(api_key="your-key") as client:
    # Get all internal team members
    for person in client.persons.all():
        if person.type == PersonType.INTERNAL:
            print(f"{person.first_name} {person.last_name}")

    # Create a contact
    person = client.persons.create(
        PersonCreate(
            first_name="Jane",
            last_name="Doe",
            emails=["jane@example.com"],
        )
    )

    # Search by email
    results = client.persons.search("jane@example.com")

Working with Lists

from affinity import Affinity, FieldResolver, ResolveMode
from affinity.models import ListCreate
from affinity.types import CompanyId, FieldId, FieldType, ListId, ListType

with Affinity(api_key="your-key") as client:
    # Get all lists
    for lst in client.lists.all():
        print(f"{lst.name} ({lst.type.name})")

    # Get a specific list with field metadata
    pipeline = client.lists.get(ListId(123))
    print(f"Fields: {[f.name for f in pipeline.fields]}")

    # Create a new list
    new_list = client.lists.create(
        ListCreate(
            name="Q1 Pipeline",
            type=ListType.OPPORTUNITY,
            is_public=True,
        )
    )

    # Work with list entries
    entries = client.lists.entries(ListId(123))

    # List entries with field data
    for entry in entries.all(field_types=[FieldType.LIST]):
        print(f"{entry.entity.name}: {entry.fields}")

    # Look up field values by name (instead of raw field IDs)
    # See docs/public/guides/performance.md for details
    resolver = FieldResolver(pipeline.fields)
    for entry in entries.all(field_types=[FieldType.LIST]):
        status = resolver.get(entry, "Status", resolve=ResolveMode.TEXT)
        print(f"{entry.entity.name}: {status}")

    # Add a company to the list
    entry = entries.add_company(CompanyId(456))

    # Update field values
    entries.update_field_value(
        entry.id,
        FieldId(101),
        "In Progress"
    )

    # Batch update multiple fields
    entries.batch_update_fields(
        entry.id,
        {
            FieldId(101): "Closed Won",
            FieldId(102): 100000,
            FieldId(103): "2024-03-15",
        }
    )

    # Use saved views
    views = client.lists.get_saved_views(ListId(123))
    for view in views.data:
        results = entries.from_saved_view(view.id)

Notes

from affinity import Affinity
from affinity.models import NoteCreate, NoteUpdate
from affinity.types import NoteType, PersonId

with Affinity(api_key="your-key") as client:
    # Create a note
    note = client.notes.create(
        NoteCreate(
            content="<p>Great meeting!</p>",
            type=NoteType.HTML,
            person_ids=[PersonId(123)],
        )
    )

    # Get notes for a person
    result = client.notes.list(person_id=PersonId(123))
    for note_item in result.data:
        print(note_item.content)

    # Update a note
    client.notes.update(note.id, NoteUpdate(content="Updated content"))

    # Delete a note
    client.notes.delete(note.id)

Reminders

from datetime import datetime, timedelta
from affinity import Affinity
from affinity.models import ReminderCreate
from affinity.types import PersonId, ReminderResetType, ReminderType, UserId

with Affinity(api_key="your-key") as client:
    # Get current user
    me = client.whoami()

    # Create a follow-up reminder
    reminder = client.reminders.create(
        ReminderCreate(
            owner_id=UserId(me.user.id),
            type=ReminderType.ONE_TIME,
            content="Follow up on proposal",
            due_date=datetime.now() + timedelta(days=7),
            person_id=PersonId(123),
        )
    )

    # Create a recurring reminder
    recurring = client.reminders.create(
        ReminderCreate(
            owner_id=UserId(me.user.id),
            type=ReminderType.RECURRING,
            reset_type=ReminderResetType.INTERACTION,
            reminder_days=30,
            content="Monthly check-in",
            person_id=PersonId(123),
        )
    )

Files

from affinity import Affinity
from affinity.types import FileId, PersonId

with Affinity(api_key="your-key") as client:
    # Download into memory (bytes)
    content = client.files.download(FileId(123))

    # Stream download (for progress bars / piping / large files)
    for chunk in client.files.download_stream(
        FileId(123),
        chunk_size=64_000,
        timeout=60.0,          # per-call request timeout override (seconds)
        deadline_seconds=300,  # total time budget (includes retries/backoff)
    ):
        ...

    # Download to disk
    saved_path = client.files.download_to(
        FileId(123),
        "report.pdf",
        overwrite=False,
        deadline_seconds=300,
    )

    # Upload (multipart form data)
    client.files.upload(
        files={"file": ("report.pdf", b"hello", "application/pdf")},
        person_id=PersonId(123),
    )

    # Upload from disk / bytes (ergonomic helpers)
    client.files.upload_path("report.pdf", person_id=PersonId(123))
    client.files.upload_bytes(b"hello", "report.txt", person_id=PersonId(123))

    # Iterate all files attached to an entity
    for f in client.files.all(person_id=PersonId(123)):
        print(f.name, f.size)

Webhooks

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

with Affinity(api_key="your-key") as client:
    # Create a webhook subscription
    webhook = client.webhooks.create(
        WebhookCreate(
            webhook_url="https://your-server.com/webhook",
            subscriptions=[
                WebhookEvent.LIST_ENTRY_CREATED,
                WebhookEvent.LIST_ENTRY_DELETED,
                WebhookEvent.FIELD_VALUE_UPDATED,
            ],
        )
    )

    # List all webhooks (max 3 per instance)
    webhooks = client.webhooks.list()

    # Disable a webhook
    client.webhooks.update(
        webhook.id,
        WebhookUpdate(disabled=True)
    )

Rate Limits

from affinity import Affinity

with Affinity(api_key="your-key") as client:
    # Fetch/observe current rate limits now (one request)
    limits = client.rate_limits.refresh()
    print(f"API key per minute: {limits.api_key_per_minute.remaining}/{limits.api_key_per_minute.limit}")
    print(f"Org monthly: {limits.org_monthly.remaining}/{limits.org_monthly.limit}")

    # Best-effort snapshot derived from tracked response headers (no network)
    snapshot = client.rate_limits.snapshot()
    print(f"Snapshot source: {snapshot.source}")

Type System

The SDK uses strongly-typed ID classes (int/str subclasses) to prevent accidental mixing:

from affinity.types import PersonId, CompanyId, ListId

# These are different types - IDE and type checker will catch mixing
person_id = PersonId(123)
company_id = CompanyId(456)

# This would be a type error:
# client.persons.get(company_id)  # Wrong type!

All magic numbers are replaced with enums:

from affinity.types import (
    ListType,        # PERSON, ORGANIZATION, OPPORTUNITY
    PersonType,      # INTERNAL, EXTERNAL, COLLABORATOR
    FieldValueType,  # "text", "number", "datetime", "dropdown-multi", etc.
    InteractionType, # EMAIL, MEETING, CALL, CHAT
    # ... and more
)

API Coverage

Feature	V2	V1	SDK
Companies (read)	✅	✅	V2
Companies (write)	❌	✅	V1
Persons (read)	✅	✅	V2
Persons (write)	❌	✅	V1
Lists (read)	✅	✅	V2
Lists (write)	❌	✅	V1
List Entries (read)	✅	✅	V2
List Entries (write)	❌	✅	V1
Field Values (read)	✅	✅	V2
Field Values (write)	✅	✅	V2
Notes	Read-only	✅	V1
Reminders	❌	✅	V1
Webhooks	❌	✅	V1
Interactions	Read-only	✅	V1
Entity Files	❌	✅	V1
Relationship Strengths	❌	✅	V1

Configuration

from affinity import Affinity

client = Affinity(
    api_key="your-api-key",

    # Timeouts and retries
    timeout=30.0,           # Request timeout (seconds)
    max_retries=3,          # Retries for rate-limited requests

    # Caching
    enable_cache=True,      # Cache field metadata
    cache_ttl=300.0,        # Cache TTL (seconds)

    # Debugging
    log_requests=False,     # Log all HTTP requests

    # Hooks (DX-008)
    # on_event=lambda event: print(event.type),
    # on_request=lambda req: print(req.method, req.url),
    # on_response=lambda resp: print(resp.status_code, resp.request.url),
)

Error Handling

The SDK provides a comprehensive exception hierarchy:

from affinity import (
    Affinity,
    AffinityError,
    AuthenticationError,
    MergedEntityError,
    RateLimitError,
    NotFoundError,
    ValidationError,
)

try:
    with Affinity(api_key="your-key") as client:
        person = client.persons.get(PersonId(99999999))
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except NotFoundError:
    print("Person not found")
except MergedEntityError as e:
    print(f"Entity {e.source_id} was merged into {e.target_id}")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except AffinityError as e:
    print(f"API error: {e}")

Async Support

import asyncio
from affinity import AsyncAffinity

async def main():
    async with AsyncAffinity(api_key="your-key") as client:
        # Async operations
        companies = await client.companies.list()
        async for company in client.companies.all():
            print(company.name)

asyncio.run(main())

Async support mirrors the sync client surface area (including V1-only services like notes/reminders/webhooks/files).

See docs/public/guides/sync-vs-async.md for more details.

If you don't use async with, make sure to await client.close() (e.g., in a finally) to avoid leaking connections.

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Optional: live API smoke tests (requires a real API key)
AFFINITY_API_KEY="..." pytest -m integration -q

# Type checking
mypy affinity

# Linting
ruff check affinity
ruff format affinity

License

MIT License - see LICENSE for details.

Contributing

Contributions welcome! Please read our contributing guidelines first.

Links

• Repository: https://github.com/yaniv-golan/affinity-sdk
• Issues: https://github.com/yaniv-golan/affinity-sdk/issues
• Affinity API V2 Documentation
• Affinity API V1 Documentation

---

Built with Skill Creator Plus.