{
  "markdown": "# Affinity Python SDK\n\n[![CI](https://github.com/yaniv-golan/affinity-sdk/actions/workflows/ci.yml/badge.svg)](https://github.com/yaniv-golan/affinity-sdk/actions/workflows/ci.yml)\n[![Coverage](https://codecov.io/gh/yaniv-golan/affinity-sdk/branch/main/graph/badge.svg)](https://codecov.io/gh/yaniv-golan/affinity-sdk)\n[![PyPI version](https://img.shields.io/pypi/v/affinity-sdk.svg)](https://pypi.org/project/affinity-sdk/)\n[![Python versions](https://img.shields.io/pypi/pyversions/affinity-sdk.svg)](https://pypi.org/project/affinity-sdk/)\n[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)\n[![Typed](https://img.shields.io/badge/typed-mypy-blue.svg)](https://mypy-lang.org/)\n[![Pydantic v2](https://img.shields.io/badge/Pydantic-v2-orange.svg)](https://docs.pydantic.dev/)\n[![Documentation](https://img.shields.io/badge/docs-GitHub%20Pages-blue.svg)](https://yaniv-golan.github.io/affinity-sdk/latest/)\n[![MCP](https://img.shields.io/badge/MCP-server-green?logo=modelcontextprotocol)](https://yaniv-golan.github.io/affinity-sdk/latest/mcp/)\n[![MCP Bash Framework](https://img.shields.io/badge/MCP-MCP_Bash_Framework-green?logo=modelcontextprotocol)](https://github.com/yaniv-golan/mcp-bash-framework)\n[![Claude Code](https://img.shields.io/badge/Claude%20Code-plugins-blueviolet.svg)](https://yaniv-golan.github.io/affinity-sdk/latest/guides/claude-code-plugins/)\n[![Built with Skill Creator Plus](https://img.shields.io/badge/Built_with-Skill_Creator_Plus-4ecdc4?style=flat-square)](https://github.com/yaniv-golan/skill-creator-plus)\n\nA modern, strongly-typed Python wrapper for the [Affinity CRM API](https://api-docs.affinity.co/).\n\nDisclaimer: 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.\n\nMaintainer: GitHub: `yaniv-golan`\n\nDocumentation: https://yaniv-golan.github.io/affinity-sdk/latest/\n\n## Affinity's Official MCP Server\n\nAs of March 2026, Affinity has released an [official MCP Server (beta)](https://developer.affinity.co/pages/mcp/introduction) 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.\n\nThis 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.\n\nFor a detailed comparison, see [Affinity SDK vs. Official MCP](https://yaniv-golan.github.io/affinity-sdk/latest/guides/affinity-official-mcp-comparison/).\n\n## Table of Contents\n\n- [Features](#features)\n- [Installation](#installation)\n- [Quick Start](#quick-start)\n- [Usage Examples](#usage-examples)\n- [Type System](#type-system)\n- [API Coverage](#api-coverage)\n- [Authentication](#authentication)\n- [Configuration](#configuration)\n- [Error Handling](#error-handling)\n- [Async Support](#async-support)\n- [Development](#development)\n\n## Features\n\n- **Complete API coverage** - Full V1 + V2 support with smart routing\n- **CLI included** - Scriptable command-line interface for automation\n- **Strong typing** - Full Pydantic V2 models with typed ID classes\n- **No magic numbers** - Comprehensive enums for all API constants\n- **Automatic pagination** - Iterator support for seamless pagination\n- **Rate limit handling** - Automatic retry with exponential backoff\n- **Response caching** - Optional caching for field metadata\n- **Both sync and async** - Full support for both patterns\n\n### AI Integrations\n\n- **Claude Code plugins** - SDK and CLI knowledge for AI-assisted development\n- **MCP Server** - Connect desktop AI tools to Affinity\n\n## Installation\n\n```bash\npip install affinity-sdk\n```\n\nRequires Python 3.10+.\n\nOptional (local dev): load `.env` automatically:\n\n```bash\npip install \"affinity-sdk[dotenv]\"\n```\n\nOptional: install the CLI:\n\n```bash\npipx install \"affinity-sdk[cli]\"\n```\n\nThe 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).\n\nCLI docs: https://yaniv-golan.github.io/affinity-sdk/latest/cli/\n\n### MCP Server\n\nConnect desktop AI tools to Affinity CRM.\n\n**Claude Desktop** (easiest - MCPB bundle):\n\n1. Install CLI: `pipx install \"affinity-sdk[cli]\"`\n2. *(Optional)* Pre-configure API key: `xaffinity config setup-key`\n   - If skipped, Claude Desktop will prompt for your API key during MCPB install\n3. Download the `.mcpb` bundle from [GitHub Releases](https://github.com/yaniv-golan/affinity-sdk/releases)\n4. Double-click to install (or drag to Claude Desktop)\n\n**Other clients** (Cursor, Windsurf, VS Code + Copilot, Zed, etc.):\n\nThese require manual configuration. See the [MCP Server docs](https://yaniv-golan.github.io/affinity-sdk/latest/mcp/) for step-by-step instructions.\n\nMCP docs: https://yaniv-golan.github.io/affinity-sdk/latest/mcp/\n\n### Claude Code Plugins\n\nIf you use [Claude Code](https://docs.anthropic.com/en/docs/claude-code), install plugins for SDK/CLI knowledge:\n\n```bash\n/plugin marketplace add yaniv-golan/affinity-sdk\n/plugin install affinity-crm-sdk-unofficial@xaffinity   # SDK patterns\n/plugin install affinity-crm-cli-xaffinity-unofficial@xaffinity   # CLI patterns + hooks\n```\n\nPlugin docs: https://yaniv-golan.github.io/affinity-sdk/latest/guides/claude-code-plugins/\n\n## Documentation\n\n- [Full documentation](https://yaniv-golan.github.io/affinity-sdk/latest/)\n- [MCP Server](https://yaniv-golan.github.io/affinity-sdk/latest/mcp/)\n- [CLI Reference](https://yaniv-golan.github.io/affinity-sdk/latest/cli/)\n- [API Reference](https://yaniv-golan.github.io/affinity-sdk/latest/reference/client/)\n\n## Quick Start\n\n```python\nfrom affinity import Affinity\nfrom affinity.types import FieldType, PersonId\n\n# Recommended: read the API key from the environment (AFFINITY_API_KEY)\nclient = Affinity.from_env()\n\n# If you use a local `.env` file (requires `affinity-sdk[dotenv]`)\n# client = Affinity.from_env(load_dotenv=True)\n\n# Or pass it explicitly\n# client = Affinity(api_key=\"your-api-key\")\n\n# Or use as a context manager\nwith Affinity.from_env() as client:\n    # List all companies\n    for company in client.companies.all():\n        print(f\"{company.name} ({company.domain})\")\n\n    # Get a person with enriched data\n    person = client.persons.get(\n        PersonId(12345),\n        field_types=[FieldType.ENRICHED, FieldType.GLOBAL]\n    )\n    print(f\"{person.first_name} {person.last_name}: {person.primary_email}\")\n```\n\n## Usage Examples\n\n### Working with Companies\n\n```python\nfrom affinity import Affinity, F\nfrom affinity.models import CompanyCreate\nfrom affinity.types import CompanyId, FieldType\n\nwith Affinity(api_key=\"your-key\") as client:\n    # List companies with filtering (V2 API)\n    companies = client.companies.list(\n        filter=F.field(\"domain\").contains(\"acme\"),\n        field_types=[FieldType.ENRICHED],\n    )\n\n    # Iterate through all companies with automatic pagination\n    for company in client.companies.all():\n        print(f\"{company.name}: {company.fields}\")\n\n    # Get a specific company\n    company = client.companies.get(CompanyId(123))\n\n    # Create a company (uses V1 API)\n    new_company = client.companies.create(\n        CompanyCreate(\n            name=\"Acme Corp\",\n            domain=\"acme.com\",\n        )\n    )\n\n    # Search by name, domain, or email\n    results = client.companies.search(\"acme.com\")\n\n    # Get list entries for a company\n    entries = client.companies.get_list_entries(CompanyId(123))\n```\n\n### Working with Persons\n\n```python\nfrom affinity import Affinity\nfrom affinity.models import PersonCreate\nfrom affinity.types import PersonType\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Get all internal team members\n    for person in client.persons.all():\n        if person.type == PersonType.INTERNAL:\n            print(f\"{person.first_name} {person.last_name}\")\n\n    # Create a contact\n    person = client.persons.create(\n        PersonCreate(\n            first_name=\"Jane\",\n            last_name=\"Doe\",\n            emails=[\"jane@example.com\"],\n        )\n    )\n\n    # Search by email\n    results = client.persons.search(\"jane@example.com\")\n```\n\n### Working with Lists\n\n```python\nfrom affinity import Affinity, FieldResolver, ResolveMode\nfrom affinity.models import ListCreate\nfrom affinity.types import CompanyId, FieldId, FieldType, ListId, ListType\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Get all lists\n    for lst in client.lists.all():\n        print(f\"{lst.name} ({lst.type.name})\")\n\n    # Get a specific list with field metadata\n    pipeline = client.lists.get(ListId(123))\n    print(f\"Fields: {[f.name for f in pipeline.fields]}\")\n\n    # Create a new list\n    new_list = client.lists.create(\n        ListCreate(\n            name=\"Q1 Pipeline\",\n            type=ListType.OPPORTUNITY,\n            is_public=True,\n        )\n    )\n\n    # Work with list entries\n    entries = client.lists.entries(ListId(123))\n\n    # List entries with field data\n    for entry in entries.all(field_types=[FieldType.LIST]):\n        print(f\"{entry.entity.name}: {entry.fields}\")\n\n    # Look up field values by name (instead of raw field IDs)\n    # See docs/public/guides/performance.md for details\n    resolver = FieldResolver(pipeline.fields)\n    for entry in entries.all(field_types=[FieldType.LIST]):\n        status = resolver.get(entry, \"Status\", resolve=ResolveMode.TEXT)\n        print(f\"{entry.entity.name}: {status}\")\n\n    # Add a company to the list\n    entry = entries.add_company(CompanyId(456))\n\n    # Update field values\n    entries.update_field_value(\n        entry.id,\n        FieldId(101),\n        \"In Progress\"\n    )\n\n    # Batch update multiple fields\n    entries.batch_update_fields(\n        entry.id,\n        {\n            FieldId(101): \"Closed Won\",\n            FieldId(102): 100000,\n            FieldId(103): \"2024-03-15\",\n        }\n    )\n\n    # Use saved views\n    views = client.lists.get_saved_views(ListId(123))\n    for view in views.data:\n        results = entries.from_saved_view(view.id)\n```\n\n### Notes\n\n```python\nfrom affinity import Affinity\nfrom affinity.models import NoteCreate, NoteUpdate\nfrom affinity.types import NoteType, PersonId\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Create a note\n    note = client.notes.create(\n        NoteCreate(\n            content=\"<p>Great meeting!</p>\",\n            type=NoteType.HTML,\n            person_ids=[PersonId(123)],\n        )\n    )\n\n    # Get notes for a person\n    result = client.notes.list(person_id=PersonId(123))\n    for note_item in result.data:\n        print(note_item.content)\n\n    # Update a note\n    client.notes.update(note.id, NoteUpdate(content=\"Updated content\"))\n\n    # Delete a note\n    client.notes.delete(note.id)\n```\n\n### Reminders\n\n```python\nfrom datetime import datetime, timedelta\nfrom affinity import Affinity\nfrom affinity.models import ReminderCreate\nfrom affinity.types import PersonId, ReminderResetType, ReminderType, UserId\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Get current user\n    me = client.whoami()\n\n    # Create a follow-up reminder\n    reminder = client.reminders.create(\n        ReminderCreate(\n            owner_id=UserId(me.user.id),\n            type=ReminderType.ONE_TIME,\n            content=\"Follow up on proposal\",\n            due_date=datetime.now() + timedelta(days=7),\n            person_id=PersonId(123),\n        )\n    )\n\n    # Create a recurring reminder\n    recurring = client.reminders.create(\n        ReminderCreate(\n            owner_id=UserId(me.user.id),\n            type=ReminderType.RECURRING,\n            reset_type=ReminderResetType.INTERACTION,\n            reminder_days=30,\n            content=\"Monthly check-in\",\n            person_id=PersonId(123),\n        )\n    )\n```\n\n### Files\n\n```python\nfrom affinity import Affinity\nfrom affinity.types import FileId, PersonId\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Download into memory (bytes)\n    content = client.files.download(FileId(123))\n\n    # Stream download (for progress bars / piping / large files)\n    for chunk in client.files.download_stream(\n        FileId(123),\n        chunk_size=64_000,\n        timeout=60.0,          # per-call request timeout override (seconds)\n        deadline_seconds=300,  # total time budget (includes retries/backoff)\n    ):\n        ...\n\n    # Download to disk\n    saved_path = client.files.download_to(\n        FileId(123),\n        \"report.pdf\",\n        overwrite=False,\n        deadline_seconds=300,\n    )\n\n    # Upload (multipart form data)\n    client.files.upload(\n        files={\"file\": (\"report.pdf\", b\"hello\", \"application/pdf\")},\n        person_id=PersonId(123),\n    )\n\n    # Upload from disk / bytes (ergonomic helpers)\n    client.files.upload_path(\"report.pdf\", person_id=PersonId(123))\n    client.files.upload_bytes(b\"hello\", \"report.txt\", person_id=PersonId(123))\n\n    # Iterate all files attached to an entity\n    for f in client.files.all(person_id=PersonId(123)):\n        print(f.name, f.size)\n```\n\n### Webhooks\n\n```python\nfrom affinity import Affinity\nfrom affinity.models import WebhookCreate, WebhookUpdate\nfrom affinity.types import WebhookEvent\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Create a webhook subscription\n    webhook = client.webhooks.create(\n        WebhookCreate(\n            webhook_url=\"https://your-server.com/webhook\",\n            subscriptions=[\n                WebhookEvent.LIST_ENTRY_CREATED,\n                WebhookEvent.LIST_ENTRY_DELETED,\n                WebhookEvent.FIELD_VALUE_UPDATED,\n            ],\n        )\n    )\n\n    # List all webhooks (max 3 per instance)\n    webhooks = client.webhooks.list()\n\n    # Disable a webhook\n    client.webhooks.update(\n        webhook.id,\n        WebhookUpdate(disabled=True)\n    )\n```\n\n### Rate Limits\n\n```python\nfrom affinity import Affinity\n\nwith Affinity(api_key=\"your-key\") as client:\n    # Fetch/observe current rate limits now (one request)\n    limits = client.rate_limits.refresh()\n    print(f\"API key per minute: {limits.api_key_per_minute.remaining}/{limits.api_key_per_minute.limit}\")\n    print(f\"Org monthly: {limits.org_monthly.remaining}/{limits.org_monthly.limit}\")\n\n    # Best-effort snapshot derived from tracked response headers (no network)\n    snapshot = client.rate_limits.snapshot()\n    print(f\"Snapshot source: {snapshot.source}\")\n```\n\n## Type System\n\nThe SDK uses strongly-typed ID classes (int/str subclasses) to prevent accidental mixing:\n\n```python\nfrom affinity.types import PersonId, CompanyId, ListId\n\n# These are different types - IDE and type checker will catch mixing\nperson_id = PersonId(123)\ncompany_id = CompanyId(456)\n\n# This would be a type error:\n# client.persons.get(company_id)  # Wrong type!\n```\n\nAll magic numbers are replaced with enums:\n\n```python\nfrom affinity.types import (\n    ListType,        # PERSON, ORGANIZATION, OPPORTUNITY\n    PersonType,      # INTERNAL, EXTERNAL, COLLABORATOR\n    FieldValueType,  # \"text\", \"number\", \"datetime\", \"dropdown-multi\", etc.\n    InteractionType, # EMAIL, MEETING, CALL, CHAT\n    # ... and more\n)\n```\n\n## API Coverage\n\n| Feature | V2 | V1 | SDK |\n|---------|:--:|:--:|:---:|\n| Companies (read) | ✅ | ✅ | V2 |\n| Companies (write) | ❌ | ✅ | V1 |\n| Persons (read) | ✅ | ✅ | V2 |\n| Persons (write) | ❌ | ✅ | V1 |\n| Lists (read) | ✅ | ✅ | V2 |\n| Lists (write) | ❌ | ✅ | V1 |\n| List Entries (read) | ✅ | ✅ | V2 |\n| List Entries (write) | ❌ | ✅ | V1 |\n| Field Values (read) | ✅ | ✅ | V2 |\n| Field Values (write) | ✅ | ✅ | V2 |\n| Notes | Read-only | ✅ | V1 |\n| Reminders | ❌ | ✅ | V1 |\n| Webhooks | ❌ | ✅ | V1 |\n| Interactions | Read-only | ✅ | V1 |\n| Entity Files | ❌ | ✅ | V1 |\n| Relationship Strengths | ❌ | ✅ | V1 |\n\n## Authentication\n\nThe SDK resolves the API key through the following chain (first non-empty value wins):\n\n1. **Explicit constructor arg** — `Affinity(api_key=\"…\")` or `--api-key` CLI flag\n2. **`AFFINITY_API_KEY`** — standard environment variable\n3. **`AFFINITY_API_KEY_FILE`** — path to a file containing the key (12-factor / Docker secrets convention)\n4. **`AFFINITY_API_KEY_COMMAND`** — shell command whose stdout is the key (credential-helper convention)\n5. **`--api-key-file <path>`** or **`--api-key-stdin`** — CLI flags\n6. **`xaffinity config setup-key`** — saved to the system keychain\n\nEmpty string is treated as unset at every step (safe against stale `export AFFINITY_API_KEY=` lines).\n\n### AFFINITY_API_KEY_FILE — file-based secrets\n\nSet this env var to the path of a file containing your API key. Used by Docker secrets,\nKubernetes mounted Secrets, and Hashicorp Vault agent sidecars.\n\n```bash\n# Docker\ndocker run -e AFFINITY_API_KEY_FILE=/run/secrets/affinity_api_key …\n\n# Kubernetes — mount a Secret as a file and set the env var\n# (see your k8s Secret docs for creating the Secret object)\nenv:\n  - name: AFFINITY_API_KEY_FILE\n    value: /etc/secrets/affinity-api-key\n```\n\nOn Posix systems, a `UserWarning` is emitted if the file is group- or world-readable\n(mode `0644` or looser). Use `chmod 600` to silence it.\n\n### AFFINITY_API_KEY_COMMAND — command-based secrets\n\nSet this env var to a shell command. The SDK runs it at startup and uses its stdout as the key.\nFollows the same convention as `git credential.helper`, `gpg --passphrase-program`, and similar tools.\n\n```bash\n# 1Password CLI\nexport AFFINITY_API_KEY_COMMAND=\"op read op://Personal/Affinity/credential\"\n\n# macOS Keychain\nexport AFFINITY_API_KEY_COMMAND=\"security find-generic-password -a affinity -w\"\n\n# pass (Unix password manager)\nexport AFFINITY_API_KEY_COMMAND=\"pass show affinity/api-key\"\n\n# HashiCorp Vault\nexport AFFINITY_API_KEY_COMMAND=\"vault kv get -field=api_key secret/affinity\"\n```\n\nThe default timeout is 30 seconds; override with `AFFINITY_API_KEY_COMMAND_TIMEOUT=<seconds>`.\nA non-zero exit code or empty stdout raises an error (stderr is included, capped at 500 chars).\n\n> **Note on `.env` files.** When `load_dotenv=True` / `--dotenv` is used, a `.env`\n> file containing `AFFINITY_API_KEY=…` will silently take precedence over a\n> shell-set `AFFINITY_API_KEY_FILE` or `AFFINITY_API_KEY_COMMAND` (because the\n> resolver checks `AFFINITY_API_KEY` at step 2, before the file/command paths).\n> Don't mix dotenv with `_FILE`/`_COMMAND` unless you want this precedence. See\n> [Authentication caveats](docs/public/guides/authentication.md#caveats).\n\n## Configuration\n\n```python\nfrom affinity import Affinity\n\nclient = Affinity(\n    api_key=\"your-api-key\",\n\n    # Timeouts and retries\n    timeout=30.0,           # Request timeout (seconds)\n    max_retries=3,          # Retries for rate-limited requests\n\n    # Caching\n    enable_cache=True,      # Cache field metadata\n    cache_ttl=300.0,        # Cache TTL (seconds)\n\n    # Debugging\n    log_requests=False,     # Log all HTTP requests\n\n    # Hooks (DX-008)\n    # on_event=lambda event: print(event.type),\n    # on_request=lambda req: print(req.method, req.url),\n    # on_response=lambda resp: print(resp.status_code, resp.request.url),\n)\n```\n\n## Error Handling\n\nThe SDK provides a comprehensive exception hierarchy:\n\n```python\nfrom affinity import (\n    Affinity,\n    AffinityError,\n    AuthenticationError,\n    MergedEntityError,\n    RateLimitError,\n    NotFoundError,\n    ValidationError,\n)\n\ntry:\n    with Affinity(api_key=\"your-key\") as client:\n        person = client.persons.get(PersonId(99999999))\nexcept AuthenticationError:\n    print(\"Invalid API key\")\nexcept RateLimitError as e:\n    print(f\"Rate limited. Retry after {e.retry_after}s\")\nexcept NotFoundError:\n    print(\"Person not found\")\nexcept MergedEntityError as e:\n    print(f\"Entity {e.source_id} was merged into {e.target_id}\")\nexcept ValidationError as e:\n    print(f\"Invalid request: {e.message}\")\nexcept AffinityError as e:\n    print(f\"API error: {e}\")\n```\n\n## Async Support\n\n```python\nimport asyncio\nfrom affinity import AsyncAffinity\n\nasync def main():\n    async with AsyncAffinity(api_key=\"your-key\") as client:\n        # Async operations\n        companies = await client.companies.list()\n        async for company in client.companies.all():\n            print(company.name)\n\nasyncio.run(main())\n```\n\nAsync support mirrors the sync client surface area (including V1-only services like notes/reminders/webhooks/files).\n\nSee `docs/public/guides/sync-vs-async.md` for more details.\n\nIf you don't use `async with`, make sure to `await client.close()` (e.g., in a `finally`) to avoid leaking connections.\n\n## Development\n\n```bash\n# Install with dev dependencies\npip install -e \".[dev]\"\n\n# Run tests\npytest\n\n# Optional: live API smoke tests (requires a real API key)\nAFFINITY_API_KEY=\"...\" pytest -m integration -q\n\n# Type checking\nmypy affinity\n\n# Linting\nruff check affinity\nruff format affinity\n```\n\n## License\n\nMIT License - see [LICENSE](LICENSE) for details.\n\n## Contributing\n\nContributions welcome! Please read our contributing guidelines first.\n\n## Links\n\n- Repository: https://github.com/yaniv-golan/affinity-sdk\n- Issues: https://github.com/yaniv-golan/affinity-sdk/issues\n- [Affinity API V2 Documentation](https://api-docs.affinity.co/reference/getting-started-with-your-api)\n- [Affinity API V1 Documentation](https://api-docs.affinity.co/reference)\n\n---\n\nBuilt with [Skill Creator Plus](https://github.com/yaniv-golan/skill-creator-plus).\n",
  "bytes": 21630,
  "sha": "b29fed5cd9657ce9c5c14c8fecf271ac5637adedfe68eba1851f8991a4676045",
  "repo_slug": "yaniv-golan/affinity-sdk",
  "fonte": "repo",
  "truncated": false,
  "api": "https://agentalog.com/api/listings/mcp_io_github_yaniv_golan_xaffinity_mcp_208cd77b/readme"
}