Affinity Python SDK
  ](https://pypi.org/project/affinity-sdk/) ](https://pypi.org/project/affinity-sdk/)       
一个现代的强类型Python包装器 Affinity CRM API.
免责声明：这是一个非官方的社区项目，与Affinity无关，也没有得到Affinity的认可或赞助。“Affinity”和相关标记是其各自所有者的商标。Affinity API的使用须遵守Affinity的服务条款。
维护者：GitHub： yaniv-golan
文档：https://yaniv-golan.github.io/affinity-sdk/latest/
目录
特性
- 全面覆盖API -通过智能路由提供完整的V1+V2支持
- 包含CLI -用于自动化的脚本化命令行界面
- 强类型 -带有键入ID类的完整Pydantic V2模型
- 没有神奇的数字 -所有API常量的综合枚举
- 自动分页 -迭代器支持无缝分页
- 费率限制处理 -指数回退自动重试
- 响应缓存 -字段元数据的可选缓存
- 同步和异步 -完全支持这两种模式
AI集成
- Claude代码插件 -AI辅助开发的SDK和CLI知识
- MCP服务器 -将桌面AI工具连接到Affinity
安装
pip install affinity-sdk需要Python 3.10+。
可选（本地开发）：加载 .env 自动：
pip install "affinity-sdk[dotenv]"可选：安装CLI：
pipx install "affinity-sdk[cli]"CLI包括一个强大的 query 用于提取具有过滤、聚合和关系的结构化数据的命令包括。输出格式包括JSON、CSV、markdown和TOON（针对LLM优化的令牌）。
可选：MCP服务器（用于AI集成）：
pip install "affinity-sdk[mcp]"
xaffinity-mcp # starts MCP server on stdio或者在不通过uvx安装的情况下运行（请参阅 MCP服务器 部分）。
CLI文档：https://yaniv-golan.github.io/affinity-sdk/latest/cli/
MCP服务器
将桌面AI工具连接到Affinity CRM。
选项1:Python MCP服务器（uvx-无需安装）
添加到您的Claude桌面配置(~/Library/Application Support/Claude/claude_desktop_config.json):
{
"mcpServers": {
"affinity": {
"command": "uvx",
"args": [
"--from", "affinity-sdk[mcp] @ git+https://github.com/oneryalcin/affinity-sdk",
"xaffinity-mcp"
],
"env": {"AFFINITY_API_KEY": "your-api-key"}
}
}
}这完全通过uvx运行，不需要手动安装。
选项2:MCPB捆绑包（仅限Claude Desktop）
- 安装CLI：
pipx install "affinity-sdk[cli]" - *（可选）* 预先配置API密钥：
xaffinity config setup-key - 下载
.mcpb捆绑从 - 双击安装
其他客户 （光标、风帆、VS码+复制码、Zed等）：
使用上面的uvx配置，或参阅 MCP服务器文档 用于基于bash的设置。
MCP文件：https://yaniv-golan.github.io/affinity-sdk/latest/mcp/
Claude代码插件
如果你使用 克劳德代码，安装SDK/CLI知识插件：
/plugin marketplace add yaniv-golan/affinity-sdk
/plugin install sdk@xaffinity # SDK patterns
/plugin install cli@xaffinity # CLI patterns + /affinity-help插件文档：https://yaniv-golan.github.io/affinity-sdk/latest/guides/claude-code-plugins/
文档
快速开始
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}")使用示例
与公司合作
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))与人合作
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")使用列表
from affinity import Affinity
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_SPECIFIC]):
print(f"{entry.entity.name}: {entry.fields}")
# 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)备注
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="
Great meeting!
",
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)提醒事项
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),
)
)文件
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)网络钩子
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)
)速率限制
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}")类型系统
SDK使用强类型ID类（int/str子类）来防止意外混合：
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!所有幻数都被替换为枚举：
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覆盖范围
|功能|V2|V1|SDK| |---------|:--:|:--:|:---:| |公司（阅读）|✅ | ✅ | V2| |公司（写）|❌ | ✅ | V1| |人员（阅读）|✅ | ✅ | V2| |人（写）|❌ | ✅ | V1| |列表（已读）|✅ | ✅ | V2| |列表（写入）|❌ | ✅ | V1| |列表条目（已读）|✅ | ✅ | V2| |列表条目（写入）|❌ | ✅ | V1| |字段值（读取）|✅ | ✅ | V2| |字段值（写入）|✅ | ✅ | V2| |备注|只读|✅ | V1| |提醒|❌ | ✅ | V1| |Webhooks |❌ | ✅ | V1| |交互|只读|✅ | V1| |实体文件|❌ | ✅ | V1| |关系优势|❌ | ✅ | V1|
配置
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),
)错误处理
SDK提供了一个全面的异常层次结构：
from affinity import (
Affinity,
AffinityError,
AuthenticationError,
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 ValidationError as e:
print(f"Invalid request: {e.message}")
except AffinityError as e:
print(f"API error: {e}")异步支持
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())异步支持反映了同步客户端的表面区域（包括仅限V1的服务，如笔记/提醒/webhooks/文件）。
看 docs/public/guides/sync-vs-async.md 了解更多详情。
如果你不使用 async with，请确保 await client.close() （例如，在a finally)以避免连接泄漏。
发展
# 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许可证
MIT许可证-请参阅 许可证 了解详情。
贡献
欢迎投稿！请先阅读我们的投稿指南。
链接
- 存储库：https://github.com/yaniv-golan/affinity-sdk
- 问题：https://github.com/yaniv-golan/affinity-sdk/issues
- Affinity API V2文档
- Affinity API V1文档