Affinity Python SDK

IMPORTANT: Write Operations Require Explicit User Request

Always use read-only mode by default. Only allow writes when the user explicitly requests data modification.

from affinity.policies import Policies, WritePolicy

# DEFAULT: Read-only mode (prevents accidental data modification)
with Affinity.from_env(policies=Policies(write=WritePolicy.DENY)) as client:
    ...  # Write operations will raise WriteNotAllowedError

# ONLY when user explicitly approves writes:
with Affinity.from_env() as client:
    ...

Installation

# SDK only (Python API wrapper)
pip install affinity-sdk

# SDK with .env file support
pip install "affinity-sdk[dotenv]"

Client Initialization

from affinity import Affinity, AsyncAffinity
from affinity.policies import Policies, WritePolicy

# RECOMMENDED: Read-only with .env file
with Affinity.from_env(load_dotenv=True, policies=Policies(write=WritePolicy.DENY)) as client:
    me = client.whoami()
    companies = client.companies.all()

# Async client
async with AsyncAffinity.from_env(policies=Policies(write=WritePolicy.DENY)) as client:
    companies = await client.companies.all()

API versions

Affinity versions its V2 API by date (2024-01-01, 2026-07-15, 2026-09-17, or current). Unpinned, each request uses the API key's default version, and calls that need a newer one (relationships, company/person field writes, notes per company) send it themselves. Pin when you need consistent response shapes:

client = Affinity.from_env(affinity_api_version="2026-09-17")
client.affinity_api_versions_seen        # versions that answered, e.g. frozenset({"2026-09-17"})

A pin older than a call needs raises ApiVersionTooOldError; a version Affinity rejects raises UnsupportedApiVersionError (both from affinity.exceptions). There is no silent fallback.

Multi-Source Tasks: Output Only the Summary

When a task combines data from multiple Affinity sources (e.g., person + interactions + list entries), fetch everything in one script and print only the relevant summary. Never dump raw model_dump_json() — it floods the conversation with hundreds of lines the agent must parse just to extract a few facts.

Do this when: combining entity details with interactions, cross-referencing list entries with entities, generating reports from multiple queries.

A single SDK call is fine when: fetching one entity, listing one page of results, or performing a single write.

Bad: dumping raw models

person = client.persons.get(PersonId(123))
print(person.model_dump_json(indent=2))  # 200+ lines of raw JSON

Good: extract and print only what's needed

person = client.persons.get(PersonId(123))
print(f"Name: {person.first_name} {person.last_name}")
print(f"Email: {person.primary_email}")

Full example: deals with no recent contact

"""Find pipeline deals with no contact in 30 days."""
from datetime import datetime, timedelta, timezone
from affinity import Affinity
from affinity.policies import Policies, WritePolicy
from affinity.types import InteractionType, FieldType

with Affinity.from_env(load_dotenv=True, policies=Policies(write=WritePolicy.DENY)) as client:
    pipeline = client.lists.resolve(name="Dealflow")
    entries = client.lists.entries(pipeline.id).all(
        field_types=[FieldType.LIST],
        expand=["interactionDates"],
    )
    cutoff = datetime.now(timezone.utc) - timedelta(days=30)
    stale = []
    for entry in entries:
        last = getattr(entry, "last_interaction_date", None)
        if last is None or last < cutoff:
            stale.append(entry)

    print(f"Deals with no contact in 30 days: {len(stale)}/{len(entries)}")
    for e in stale:
        days = (datetime.now(timezone.utc) - e.last_interaction_date).days if e.last_interaction_date else "never"
        print(f"  {e.entity_name}: last contact {days} days ago")

When to use bash instead

For simple 2-command pipelines (fetch an ID, then use it in a second command), a bash script with xaffinity session start + jq is lighter weight. See the CLI skill's "Multi-Source Tasks" section.

Typed IDs (ALWAYS USE)

Prevent mixing up entity types by using typed IDs:

from affinity.types import (
    PersonId, CompanyId, ListId, ListEntryId,
    OpportunityId, FieldId, NoteId, UserId
)

# CORRECT:
person = client.persons.get(PersonId(123))
company = client.companies.get(CompanyId(456))
entries = client.lists.entries(ListId(789))

# UserId is a subtype of PersonId — pass it anywhere PersonId is accepted:
creator = client.persons.get(note.creator_id)  # Works directly, no cast needed

# WRONG - will cause type errors:
person = client.persons.get(123)  # Don't do this!

Pagination Patterns

# Single page (default 100 items)
page = client.companies.list(limit=50)
for company in page.data:
    process(company)
# For next page, use pages() iterator instead

# All items as list (default max 100,000)
all_companies = client.companies.all()

# Adjust limit
companies = client.companies.all(limit=1000)

# Disable limit (use with caution!)
companies = client.companies.all(limit=None)

# Memory-efficient iterator (large datasets)
for person in client.persons.iter():
    process(person)

# Page-by-page iteration
for page in client.companies.pages():
    for company in page.data:
        process(company)

# Progress callback
from affinity import PaginationProgress

def log_progress(p: PaginationProgress) -> None:
    print(f"Page {p.page_number}: {p.items_so_far} items")

for company in client.companies.all(on_progress=log_progress):
    ...

Searching and Filtering

Note: Global-entity lists (companies.list(), persons.list(), opportunities.list()) do NOT accept filter=: the SDK raises ValueError to prevent silently-unfiltered results. To filter companies, persons or list entries on field values, use find() (Affinity's search, below). Use search_pages() for name/domain/email fuzzy search. filter= on list entries goes through Affinity's search where it can and is checked locally (ClientSideFilterWarning when every entry has to be read).

Filtering on Affinity's side: find() and affinity.criteria

client.companies.find(...), client.persons.find(...) and client.lists.entries(list_id).find(...) call Affinity's search (V2, beta): Affinity filters and sorts, and only matching records are downloaded.

from datetime import date
from affinity.criteria import Condition, Sort, all_of

page = client.companies.find(
    all_of(
        Condition("field-498220", "is-any-of", [3, 4]),          # dropdown option ids
        Condition("affinity-data-last-funding-date", "is-after", date(2025, 1, 1)),
    ),
    sort=[Sort("affinity-data-total-funding-amount", "desc")],   # up to 5
    limit=50,                                                     # 1-100 per page
)
next_page = client.companies.find(  # same criteria + cursor; a cursor alone isn't enough
    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,
    cursor=page.next_cursor,
)

for entry in client.lists.entries(ListId(123)).iter_find(   # every page
    Condition("field-456", "is-within-the-last", (6, "month"))
):
    ...

• Signature: find(where=None, *, sort=None, term=None, term_fields=None, field_ids=None, field_types=None, limit=None, total_count=False, cursor=None); also iter_find(...) (all pages) and find_body(body, ...) (a prebuilt search body).
• Condition(field_id, operator, value=..., value_type=None, attribute_id=None); groups with all_of(...) / any_of(...); Sort(field_id, "asc"|"desc"); SearchCriteria bundles them.
• Operators are Affinity's names: is-any-of, has-any-of, contains, is-greater-than, is-between, is-after, is-within-the-last, is-empty, ... The value type is read from the fields metadata when omitted (one cached request), and the operator is checked against the field.
• Values: ids for company, person, list and dropdown fields (dropdown = option ids); dates as datetime.date or "YYYY-MM-DD" (not datetimes); relative windows as (6, "month").
• Needs the "Export All Organizations directory" permission (companies, persons) or "Export data from Lists" (list entries); otherwise AuthorizationError naming the permission. Allowed in read-only mode.
• companies.search(term) / search_pages(term) remain the separate name search (no extra permission).

from affinity import F

# Global-entity fuzzy search (name / domain / email)
for page in client.companies.search_pages("Acme"):
    for company in page.data:
        ...

for page in client.persons.search_pages("[email protected]"):
    for person in page.data:
        ...

# Companies matching a description (semantic, ranked, up to 100)
result = client.companies.semantic_search("AI infrastructure startups in Berlin", limit=20)
for c in result.data:                   # SemanticCompany: id, name, domain, score
    print(c.name, c.score)
print(result.explanation)               # how Affinity read the request

# Text inside notes and files (optionally one company)
for hit in client.notes.search("pricing", company_id=CompanyId(456), limit=20):
    print(hit.note.id, hit.preview)
for hit in client.files.search("pitch deck", limit=20):
    print(hit.file.id, hit.page_number, hit.preview)

# List-entry filters go through Affinity's search where they can (checked locally too).
# list() returns the matches of ONE page read (maybe fewer than limit, maybe none):
# follow next_cursor, or use iter() for every match.
for entry in client.lists.entries(ListId(123)).iter(
    filter=F.field("Department").equals("Sales")
):
    print(entry.id)

# For filters on large lists, prefer saved views (server-side efficient)
entries = client.lists.entries(ListId(123)).list(saved_view_id=SavedViewId(456))

Do NOT pass `filter=` to `client.companies.list()` / `client.persons.list()` / `client.opportunities.list()` — it raises ValueError with a hint to use search_pages().

Duplicate prevention on create

companies.create() and persons.create() default to if_not_exists=True. On conflict they raise DuplicateEntityError carrying the existing entity ID so callers can recover without creating a duplicate:

from affinity.exceptions import DuplicateEntityError
from affinity.models import CompanyCreate, CompanyId

try:
    company = client.companies.create(
        CompanyCreate(name="Acmeco", domain="acmeco.com")
    )
except DuplicateEntityError as e:
    # e.existing_id is the pre-existing company's ID
    # e.existing_is_global is True for global Affinity directory records
    company = client.companies.get(CompanyId(e.existing_id))

Pass if_not_exists=False only when you deliberately want to create a distinct record that collides on name/domain.

Services Reference

with Affinity.from_env() as client:
    # Core entities
    client.persons.list() / .get() / .all() / .search()
    client.companies.list() / .get() / .all() / .search()
    client.opportunities.list() / .get() / .all()

    # Lists
    client.lists.list() / .get() / .all()
    client.lists.resolve(name="Pipeline Name")          # exact name, case-insensitive
    client.lists.pages(term="pipe")                       # names containing "pipe"
    client.lists.get_fields(ListId(123))

    # List entries
    entries_service = client.lists.entries(ListId(123))
    entries_service.list() / .get() / .all()
    entries_service.add_person() / .add_company()   # opportunities: opportunities.create(list_id=)
    entries_service.update_field_value() / .batch_update_fields()

    # Notes, reminders, interactions (one entity)
    client.notes.list() / .create()
    client.reminders.list() / .create()
    client.interactions.list(type=..., start_time=..., end_time=..., person_id=...)
    client.interactions.iter(type=..., start_time=..., person_id=...)  # auto-chunks, defaults end_time to now

    # Identity
    me = client.whoami()

For anything else read `references/services.md`, organised by question:

Error Handling

from affinity.exceptions import (
    AffinityError,           # Base class
    AuthenticationError,     # 401 - invalid/missing API key
    AuthorizationError,      # 403 - insufficient permissions
    NotFoundError,           # 404 - entity not found
    ValidationError,         # 400/422 - invalid parameters
    MergedEntityError,       # 422 - entity merged into another (subclass of ValidationError)
    CompanyMergedError,      # 422 - company merged (has source_id, target_id)
    PersonMergedError,       # 422 - person merged (has source_id, target_id)
    RateLimitError,          # 429 - rate limited
    ServerError,             # 500/503 - server errors
    WriteNotAllowedError,    # Write attempted in read-only mode
    TooManyResultsError,     # .all() exceeded limit
)

try:
    company = client.companies.get(CompanyId(123))
except CompanyMergedError as e:
    # Entity was merged — follow to the surviving record
    company = client.companies.get(CompanyId(e.target_id))
except NotFoundError:
    print("Company not found")
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}")
except AffinityError as e:
    print(f"Error: {e}")
    if e.diagnostics:
        print(f"Request ID: {e.diagnostics.request_id}")

Creating Records (requires explicit user approval)

from affinity.models import NoteCreate, ReminderCreate
from affinity.types import NoteType, ReminderType
from datetime import datetime, timedelta

# Add entity to list
entries_service = client.lists.entries(ListId(123))
entry = entries_service.add_person(PersonId(456))
entry = entries_service.add_company(CompanyId(789))

# Create note
note = client.notes.create(NoteCreate(
    content="<p>Meeting notes</p>",
    type=NoteType.HTML,
    person_ids=[PersonId(123)],
))

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

# Update field value on list entry
entries_service.update_field_value(
    ListEntryId(456),
    FieldId(789),
    "New Value"
)

# Batch update multiple fields
entries_service.batch_update_fields(
    ListEntryId(456),
    {FieldId(789): "Value1", FieldId(790): "Value2"}
)

Field Selection

from affinity.types import FieldType

# Request specific field types
client.companies.list(field_types=[FieldType.ENRICHED])
client.persons.get(PersonId(123), field_types=[FieldType.GLOBAL, FieldType.RELATIONSHIP_INTELLIGENCE])

# Check if fields were requested and access data
if company.fields.requested:
    for field_name, value in company.fields.data.items():
        print(f"{field_name}: {value}")

# Values Affinity masks (restricted opportunities, API version 2026-07-15+) are unknown, not empty
if entry.fields.is_hidden("field-123"):
    ...                                   # never report as blank, never overwrite
masked = entry.fields.hidden_fields()     # ids of every masked field

# Available: GLOBAL, LIST, ENRICHED, RELATIONSHIP_INTELLIGENCE

Resolving Fields by Name

Pattern A: Reading field values by name — use FieldResolver.get():

from affinity.field_resolver import FieldResolver
from affinity.types import ListId, FieldType, ResolveMode

fields = client.lists.get_fields(ListId(123))
resolver = FieldResolver(fields)

# Fields must be requested when fetching entries — they're NOT populated by default
entries = client.lists.entries(ListId(123)).all(field_types=[FieldType.LIST])

for entry in entries:
    status = resolver.get(entry, "Status")
    owner = resolver.get(entry, "Owner", resolve=ResolveMode.TEXT)

FieldResolver.get() maps names to values. FieldResolver.find_field() maps names to metadata (including FieldId).

Pattern B: Getting FieldId for write operations — use FieldResolver.find_field():

# For list-entry fields: use client.lists.get_fields(list_id)
fields = client.lists.get_fields(ListId(123))
resolver = FieldResolver(fields)

status_meta = resolver.find_field("Status")  # -> FieldMetadata | None
if status_meta:
    entries_service = client.lists.entries(ListId(123))
    # For text fields, pass the string directly:
    entries_service.update_field_value(ListEntryId(456), status_meta.id, "Active")
    # For dropdown fields, pass the dropdown option ID (int), not the text:
    #   option = next(o for o in status_meta.dropdown_options if o.text == "Active")
    #   entries_service.update_field_value(entry_id, status_meta.id, option.id,
    #                                      value_type=FieldValueType.DROPDOWN)

# For entity-level fields: use the entity-specific method
# FieldResolver(client.companies.get_fields()) / FieldResolver(client.persons.get_fields())

Enriched Fields

Enriched fields (Phone Number, Source of Introduction, Industry, Location, Description, etc.) are returned on entity.fields.data like any other field when you request them via field_types=[FieldType.ENRICHED].

Company and person fields — global, enriched (incl. "Current Organization") and Source of Introduction — are written in one all-or-nothing request with their FieldMetadata.id:

client.companies.batch_update_fields(
    company_id,
    {"dealroom-description": "AI infra", "field-123": None},   # None clears
    value_types={"field-123": "dropdown"},                     # needed for None and lists
)
client.persons.get_field_values(person_id, ids=["affinity-data-current-organization"])

These writes need API version 2026-07-15 or newer (see "API versions").

For fresh post-write reads where you want to skip the in-memory field-metadata cache, pass skip_cache=True to client.fields.list(...).

Rate Limits

# Check current status (from cached headers)
snapshot = client.rate_limits.snapshot()
print(f"Per-minute: {snapshot.api_key_per_minute.remaining}/{snapshot.api_key_per_minute.limit}")
print(f"Monthly: {snapshot.org_monthly.remaining}/{snapshot.org_monthly.limit}")

# Refresh from API
refreshed = client.rate_limits.refresh()

Retry Behavior

• GET/HEAD: Automatic retries (3 by default) for rate limits and transient errors
• POST/PUT/PATCH/DELETE: No automatic retries (to avoid duplicates)

# Configure retries
client = Affinity(api_key="key", max_retries=5)

Documentation

• Full SDK docs: https://yaniv-golan.github.io/affinity-sdk/latest/