xaffinity MCP

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

Affinity runs an official MCP Server: a hosted server you connect to from Claude, ChatGPT, Copilot, Notion and other AI clients with OAuth sign-in. It reads and updates most CRM records and is the easiest way to work with Affinity in a chat.

This SDK serves a different purpose — it's a strongly-typed Python client and CLI for the Affinity API (companies, persons, opportunities, lists, field values, notes, reminders, interactions, webhooks, files, and more). Use it when you work with Affinity in code, scripts or a terminal: bulk updates, exports, data pipelines, webhooks, or 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
• Authentication
• Configuration
• Error Handling
• Async Support
• Development

Features

• Broad API coverage - V1 and V2, routed per operation (6 V2 operations still planned)
• 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 affinity-crm-sdk-unofficial@xaffinity   # SDK patterns
/plugin install affinity-crm-cli-xaffinity-unofficial@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 datetime import date

from affinity import Affinity
from affinity.criteria import Condition, Sort, all_of
from affinity.models import CompanyCreate
from affinity.types import CompanyId

with Affinity(api_key="your-key") as client:
    # Filter and sort on Affinity's side (V2 search, beta)
    page = client.companies.find(
        all_of(
            Condition("field-498220", "is-any-of", [3, 4]),
            Condition("affinity-data-last-funding-date", "is-after", date(2025, 1, 1)),
        ),
        sort=[Sort("affinity-data-total-funding-amount", "desc")],
        limit=50,
    )

    # 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, ReminderUpdate
from affinity.types import PersonId, ReminderResetType, ReminderStatus, 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),
        )
    )

    # Complete it (a recurring reminder moves to its next due date)
    client.reminders.update(recurring.id, ReminderUpdate(is_completed=True))

    # Overdue reminders for many people, one request per 50 people
    overdue = client.reminders.list_batch(
        person_ids=[PersonId(123), PersonId(456)], status=ReminderStatus.OVERDUE
    )

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))

    # The upload helpers return a bool; the *_returning_files variants return the
    # created file records (EntityFile), e.g. to read the new file id
    created = client.files.upload_path_returning_files("report.pdf", person_id=PersonId(123))
    file_ids = [f.id for f in created]

    # 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, WebhookStatus

with Affinity(api_key="your-key") as client:
    # Create a webhook (subscriptions is required; [] means every event type)
    webhook = client.webhooks.create(
        WebhookCreate(
            url="https://your-server.com/webhook",
            subscriptions=[
                WebhookEvent.LIST_ENTRY_CREATED,
                WebhookEvent.LIST_ENTRY_DELETED,
                WebhookEvent.FIELD_VALUE_UPDATED,
            ],
        )
    )

    # All webhooks you can see (all of them if you may manage all webhooks)
    for hook in client.webhooks.iter():
        print(hook.id, hook.url, hook.status)

    # Pause a webhook (returns it as read back)
    client.webhooks.update(webhook.id, WebhookUpdate(status=WebhookStatus.DISABLED))

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

Which Affinity API the SDK uses for each feature. The SDK uses V2 wherever V2 can do the job and V1 where it can't. The API coverage page lists every operation and the SDK call that sends it.

Still planned (Affinity documents them, but the API didn't serve them when tested): V2 person create/delete, opportunity update/delete, list-entry create, and transcript delete.

Authentication

The SDK and CLI resolve the API key through the following chain (first non-empty value wins):

1. The key you pass explicitly — Affinity(api_key="…") in the SDK; --api-key-file <path> or --api-key-stdin in the CLI (the CLI has no --api-key flag)
2. AFFINITY_API_KEY — standard environment variable
3. AFFINITY_API_KEY_FILE — path to a file containing the key (12-factor / Docker secrets convention)
4. AFFINITY_API_KEY_COMMAND — shell command whose stdout is the key (credential-helper convention)
5. User config file — api_key in config.toml, written by xaffinity config setup-key --scope user (file mode 0600 on Unix; xaffinity config path prints the location). CLI only.

The SDK uses steps 1–4; step 5 is CLI only. When the CLI uses a key flag while AFFINITY_API_KEY, AFFINITY_API_KEY_FILE or AFFINITY_API_KEY_COMMAND is also set, it warns that those are ignored (the file is not read and the command is not run). With load_dotenv=True (SDK) or --dotenv (CLI), a .env file is loaded into the environment before step 2, so its AFFINITY_API_KEY counts as step 2. That is how a key saved by xaffinity config setup-key --scope project (a .env in the current directory) is used: run commands with --dotenv.

Empty string is treated as unset at every step (safe against stale export AFFINITY_API_KEY= lines).

AFFINITY_API_KEY_FILE — file-based secrets

Set this env var to the path of a file containing your API key. Used by Docker secrets, Kubernetes mounted Secrets, and Hashicorp Vault agent sidecars.

# Docker
docker run -e AFFINITY_API_KEY_FILE=/run/secrets/affinity_api_key …

# Kubernetes — mount a Secret as a file and set the env var
# (see your k8s Secret docs for creating the Secret object)
env:
  - name: AFFINITY_API_KEY_FILE
    value: /etc/secrets/affinity-api-key

On Posix systems, a UserWarning is emitted if the file is group- or world-readable (mode 0644 or looser). Use chmod 600 to silence it.

AFFINITY_API_KEY_COMMAND — command-based secrets

Set this env var to a shell command. The SDK runs it at startup and uses its stdout as the key. Follows the same convention as git credential.helper, gpg --passphrase-program, and similar tools.

# 1Password CLI
export AFFINITY_API_KEY_COMMAND="op read op://Personal/Affinity/credential"

# macOS Keychain
export AFFINITY_API_KEY_COMMAND="security find-generic-password -a affinity -w"

# pass (Unix password manager)
export AFFINITY_API_KEY_COMMAND="pass show affinity/api-key"

# HashiCorp Vault
export AFFINITY_API_KEY_COMMAND="vault kv get -field=api_key secret/affinity"

The default timeout is 30 seconds; override with AFFINITY_API_KEY_COMMAND_TIMEOUT=<seconds>. A non-zero exit code or empty stdout raises an error (stderr is included, capped at 500 chars).

Note on .env files. When load_dotenv=True / --dotenv is used, a .env file containing AFFINITY_API_KEY=… will silently take precedence over a shell-set AFFINITY_API_KEY_FILE or AFFINITY_API_KEY_COMMAND (because the resolver checks AFFINITY_API_KEY at step 2, before the file/command paths). Don't mix dotenv with _FILE/_COMMAND unless you want this precedence. See Authentication caveats.

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

    # Affinity V2 API version (default: your API key's default version)
    # affinity_api_version="2026-09-17",   # or "current"; CLI: --api-version / AFFINITY_API_VERSION

    # Hooks
    # 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),
)

Affinity has three dated V2 API versions (2024-01-01, 2026-07-15, 2026-09-17). See Affinity API versions for pinning one, client.affinity_api_versions_seen, and the CLI's meta.affinityApiVersion.

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 the V1 services like notes and files).

See Sync vs async 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 the 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.