Corsair Attio 插件将 Attio CRM 接入你的 Agent 与自动化工作流【免费下载链接】corsairConnect your users to their apps项目地址: https://gitcode.com/GitHub_Trending/corsa/corsair导读本文以仓库中 packages/attio/README.md 为主体结合 docs/plugins/attio 下的官方文档与 packages/attio 的源码实现系统讲解如何在 Corsair 中安装、配置并使用 Attio 插件覆盖 110 个类型安全的 API 操作、本地数据同步.search()/.list()、OAuth 2.0 与 API Key 两种认证方式以及 3 类记录级 Webhook 的接入与验签。读完本文你将掌握从“创建插件实例”到“连接租户、同步数据、接收实时事件”的完整落地链路。插件概览Attio 是什么插件能做什么Attio 是一款以记录records为核心的现代 CRM 平台围绕人、公司、交易deal、用户、工作区等对象管理关系数据。Corsair 将其封装为corsair-dev/attio插件让上层应用或 Agent 通过同一个 Corsair 客户端即可完成三类能力110 个类型安全的 API 操作覆盖记录person / company / deal / user / workspace、对象与属性object / attribute、列表list / entry、注释comment、笔记note、任务task、会议meeting、Webhook 等全部读写操作全部通过tenant.attio.api.*调用入参出参由 Zod schema 约束2 个本地同步实体records、workspaceMembersWebhook 事件到达时自动落库支持attio.db.entity.search()/.list()快速检索避免频繁打远程 API3 类传入 Webhookrecord.created、record.updated、record.deleted支持签名验签、租户路由与webhookHooks生命周期钩子。官方文档 docs/plugins/attio/overview.mdx 对插件能力的定位与此一致可作为能力边界的权威依据。安装与基础配置安装依赖使用你的包管理器安装 Corsair 核心与 Attio 插件# npm npm install corsair corsair-dev/attio # pnpmREADME 推荐 pnpm add corsair corsair-dev/attio # yarn / bun yarn add corsair corsair-dev/attio bun add corsair corsair-dev/attio从 packages/attio/package.json 可见插件将corsair0.1.0与zod^4.1.13声明为 peerDependencies因此宿主项目必须同时安装这两个依赖插件本身以 ESM 形式发布type: module入口dist/index.js。创建 Corsair 实例并注册插件// corsair.ts import Database from better-sqlite3; import { createCorsair } from corsair; import { attio } from corsair-dev/attio; export const corsair createCorsair({ plugins: [ attio(), ], database: new Database(corsair.db), kek: process.env.CORSAIR_KEK!, // 密钥加密密钥用于加密 OAuth token 等敏感数据 hub: { projectApiKey: process.env.CORSAIR_API_KEY!, signingSecret: process.env.CORSAIR_SIGNING_SECRET!, }, });注册后插件会以attio命名空间挂载到 Corsair 实例上形成corsair.attio.api.*远程 API、corsair.attio.db.*本地同步数据、corsair.attio.webhooks.*Webhook 处理三个访问入口。多租户是默认行为Corsair 默认按租户隔离数据。所有 API 调用、数据库记录与 Webhook 路由都以tenantId为边界const tenant corsair.withTenant(acme); await tenant.attio.api.generated.findRecord({...});connect / OAuth 流程中的租户绑定细节可参考 docs/concepts/multi-tenancy.mdx 与 docs/management/connect.mdx。认证方式OAuth 2.0 与 API Key插件工厂attio()通过authType选项在两种认证方式间切换默认值为oauth_2见 packages/attio/index.ts 中defaultAuthType oauth_2。README 原文为“Auth: API key, OAuth 2.0 (default OAuth 2.0). SetauthTypeon the plugin factory to pick one.”OAuth 2.0推荐attio({ authType: oauth_2, // 默认值可省略 })底层 OAuth 配置在 packages/attio/index.ts 中定义授权端点https://app.attio.com/authorize令牌端点https://app.attio.com/oauth/token请求的 scoperecord_permission:read-writeobject_configuration:read-writeuser_management:readlist_entry:read-writelist_configuration:read-writecomment:read-writetask:read-writenote:read-writewebhook:read-write授权参数prompt: consent令牌提交方式tokenAuthMethod: body令牌请求参数放在请求体中使用 OAuth 时需要在 Hubcorsair.dev dashboard中为项目填入你在 Attio 开发者后台创建的 OAuth App 的client ID与client secret之后由 Hub 托管授权页并把结果回传给应用Connect 流程。本地开发时也可以在插件选项中内联凭据attio({ authType: oauth_2, credentials: { accessToken: ..., refreshToken: ..., clientId: ..., clientSecret: ..., }, })AttioCredentials与AttioPluginOptions的完整定义见 packages/attio/index.ts。API Keyattio({ authType: api_key, })API Key 模式下无需预配置——当某个租户第一次发起请求时Corsair 会提示为该校验并存储 API Key。两种认证方式下插件的keyBuilder会按请求来源解析密钥Webhook 请求优先使用options.webhookSecret否则从ctx.keys.get_webhook_signature()读取签名密钥endpoint 请求 api_key从ctx.keys.get_api_key()读取取不到则抛出AuthMissingError(attio, api_key)endpoint 请求 oauth_2优先读取已存储的 access token其次回退到options.credentials.accessToken并经getValidAccessToken()校验。完整逻辑见 packages/attio/index.tsHTTP 层则以Authorization: Bearer key注入请求头packages/attio/client.ts。连接租户Connect 流程OAuth 与 API Key 都通过统一的 Connect 流程完成租户绑定const { connectUrl } await corsair.manage.connect.createLink({ plugin: attio, tenantId: acme, }); // 将用户浏览器重定向到 connectUrlHub 托管授权页面完成后把结果token / API key安全地回传到你的应用Corsair 再将其加密存储并绑定到acme租户。此后所有corsair.withTenant(acme).attio.*调用都会自动携带该租户的凭据。Connect / OAuth 完整说明见 docs/management/connect.mdx。110 个类型安全的 API 操作调用方式const tenant corsair.withTenant(acme); await tenant.attio.api.generated.findRecord({ ... }); await tenant.attio.api.generated.assertPerson({ ... });所有操作挂载在generated命名空间下如generated.assertPerson每个操作都有唯一的 Operation ID如attio.api.generated.assertPerson与风险等级read/write。attioEndpointSchemas为每个操作注册了 Zod 输入/输出 schemapackages/attio/index.ts因此入参出参在编译期即受类型约束。核心操作分类根据 packages/attio/README.md 的 Endpoints 表操作可按用途归为以下几类记录增删改查records操作风险说明createRecordwrite创建指定对象类型的记录需提供对象类型与 values 字典不同对象类型属性不同people 用name/email_addresses/phone_numbers/job_titleusers 用primary_email_address/user_id/person/workspacecompanies 用name/domains/descriptioncreatePerson/createCompany/createDealRecord/createUserRecord/createWorkspaceRecordwrite各对象专用创建端点唯一属性冲突时抛错person 的email_addresses、company 的domains、workspace 的workspace_id等assertPerson/assertUserRecord/assertWorkspace/putV2ObjectsObjectRecordswrite“断言式”创建或更新按唯一属性查找存在则更新、不存在则创建避免重复记录findRecord/getRecord/getCompany/getPerson/getDealRecord等read按 record_id 或唯一属性查找/读取单条记录listRecords/listCompanies/listUserRecords/listWorkspaceRecords/postV2ObjectsObjectRecordsQueryread分页列出记录支持过滤与排序queryRecordsread服务端过滤查询如“status in Y 的所有 agreement”避免大页下载后本地过滤searchRecords/postV2ObjectsRecordsSearchread跨对象模糊搜索按名称、域名、邮箱、电话、社交账号beta 接口最终一致patchRecord/putV2ObjectsObjectRecordsRecordId/updatePerson/updateCompany/updateDealRecord/updateUserRecord/updateWorkspaceRecord/updateRecordwrite更新记录注意 multiselect 属性PATCH 是前置追加PUT 是整体覆盖/删除deleteRecord/deletePerson/deleteCompany/deleteDeal/deleteUser/deleteWorkspaceRecord/deleteRecordByIdwrite永久删除不可恢复对象与属性objects attributes操作风险说明createObjectwrite创建自定义对象需api_slugsnake_case、singular_noun、plural_nounupdateObjectwrite更新对象的 api slug / 单复数名词listObjects/getObjectread列出/读取对象 schema官方文档建议在listRecords/findRecord/createRecord前先调用以确认对象 slug并强调硬编码 slug 可能不存在于某些工作区createAttributewrite在对象或列表上新增自定义字段record-reference 类型可通过 relationship 对象建立双向关联updateAttributewrite更新属性标题、描述、校验规则或配置getAttribute/listAttributesread读取单个/全部属性 schema含 slug、类型、select/status 配置用于指导过滤与写入createSelectOption/updateSelectOptionwrite新增/重命名或归档 select 选项归档选项保留历史数据createStatus/updateStatuswrite新增/更新状态属性company 与 person 对象暂不支持 status 属性listAttributeOptions/listAttributeStatusesread列出 select / status 属性的可用选项列表lists entries操作风险说明createList/updateListwrite创建/更新列表新列表必须设置workspace_access: full-access或至少一个workspace_member_access为full-accessAPI 不支持修改列表的 parent objectpostV2ListsListEntries/putV2ListsListEntrieswrite向列表添加记录PUT 按 parent record 匹配存在则更新、不存在则创建patchV2ListsListEntriesEntryId/putV2ListsListEntriesEntryIdwrite按 entry_id 更新PATCH 前置追加 multiselectPUT 整体覆盖listLists/getList/listListEntries/postV2ListsListEntriesQuery/getListEntry/getRecordEntriesread列出列表/条目、查询条目支持过滤排序、查看某记录属于哪些列表deleteEntrywrite按 entry_id 删除列表条目不可恢复注释 / 笔记 / 任务 / 会议 / Webhook操作风险说明createComment/getComment/deleteComment/listThreadsread/write线程化评论删除线程头部评论会级联删除整个线程createNote/getNote/listNotes/deleteNoteread/write记录级笔记title content可附created_at支持 plaintext 与 markdowncreateTask/getTask/updateTask/deleteTask/getV2Tasksread/write任务管理updateTask 仅支持 deadline、完成状态、关联记录、assignee 四个字段任务内容仅支持纯文本listMeetings/listCallRecordingsread会议与通话录音betacreateWebhook/getWebhook/updateWebhook/deleteWebhook/listWebhooksread/writeWebhook 订阅管理创建时返回一次性签名密钥getSelfread校验当前 token、查看其关联工作区与权限listWorkspaceMembers/getWorkspaceMember/getV2WorkspaceMembersread工作区成员actor信息用于给 actor 引用属性如记录/列表条目的 owner赋referenced_actor_type: workspace-membergetRecordAttributeValues/listPeopleAttributeValues/listCompanyAttributeValues等read读取属性的当前值或历史值show_historic参数COMINT 与 enriched 属性不支持历史值查询几个值得注意的官方提示来自 README 描述createCompany与updateCompany无法通过 API 设置logo_urlcreatePerson/updatePerson无法设置avatar_urlcreateUserRecord的必填属性可能因工作区配置而异遇到missing_value错误时先用listAttributes确认记录响应中的属性值以时间分段的数组形式返回如values[name]select 属性通过option.title访问读取时需处理嵌套结构、空数组与 nullcreateEntry、listEntries、listListEntries、getV2WorkspaceMembers已在 README 中标注为 DEPRECATED应改用对应的postV2ListsListEntries/postV2ListsListEntriesQuery/listWorkspaceMembers。本地数据同步attio.db.*插件将 Webhook 事件同步到本地 SQLite 数据库提供两个可检索实体。records记录路径attio.db.records.searchconst rows await corsair.attio.db.records.search({ data: { /* 过滤条件 */ }, limit: 100, offset: 0, });字段类型支持的运算符entity_idstringequals, contains, startsWith, endsWith, increated_atdateequals, before, after, betweenworkspaceMembers工作区成员路径attio.db.workspaceMembers.searchconst rows await corsair.attio.db.workspaceMembers.search({ data: { /* 过滤条件 */ }, limit: 100, offset: 0, });字段类型支持的运算符entity_idstringequals, contains, startsWith, endsWith, innamestringequals, contains, startsWith, endsWith, inemail_addressstringequals, contains, startsWith, endsWith, inavatar_urlstringequals, contains, startsWith, endsWith, increated_atdateequals, before, after, between每个.search()都支持limit/offset分页同一路径下也可直接使用.list()拉取数据。搜索过滤与运算符的通用语义见 docs/concepts/database.mdx。从实现看同步的落库逻辑位于 packages/attio/webhooks/records.ts收到record.created/record.updated事件时调用ctx.db.records.upsertByEntityId(entityId, record)收到record.deleted时调用ctx.db.records.deleteByEntityId(entityId)实现“事件驱动 本地镜像”的数据同步模型。Webhook接收 Attio 实时事件插件处理3 个Webhook 事件README “Handles 3 webhook events. See the reference for payloads andwebhookHooks.”事件路径映射为record→created/updated/deleted对应事件名record.created、record.updated、record.deletedpackages/attio/index.ts。HTTP handler 接入把 Attio 侧订阅的 Webhook URL 指向你的 Corsair HTTP handler例如 Next.js App Router// app/api/webhook/route.ts import { processWebhook } from corsair; import { corsair } from /server/corsair; export async function POST(request: Request) { const headers Object.fromEntries(request.headers); const body await request.json(); const result await processWebhook(corsair, headers, body); return result.response; }事件负载结构三种事件的负载结构一致Zod 定义见 packages/attio/webhooks/types.ts名称类型必填说明event_typerecord.created/record.updated/record.deleted是事件类型idobject是{ workspace_id: string, object_id?: string, record_id: string }actorany否触发事件的 actor事件处理后的响应数据corsairEntityId为record_id类型{ event_type: record.created, // 或 record.updated / record.deleted id: { workspace_id: string, object_id?: string, record_id: string, }, actor?: any, }用webhookHooks挂接业务逻辑attio({ webhookHooks: { record: { created: { before(ctx, args) { return { ctx, args }; }, after(ctx, response) { // 在这里处理创建事件 }, }, updated: { before(ctx, args) { return { ctx, args }; }, after(ctx, response) {}, }, deleted: { before(ctx, args) { return { ctx, args }; }, after(ctx, response) {}, }, }, }, })before钩子可对入参做校验或改写返回{ ctx, args }after钩子接收处理结果适合做日志、通知、二次写库等副作用。签名验签与租户路由的实现细节事件匹配createAttioMatch(record.xxx)解析请求体同时支持单个事件对象event_type字段与批量事件数组events数组两种负载格式packages/attio/webhooks/types.ts签名验签从attio-signature或x-attio-signature头读取签名用webhookSecret或存储的webhook_signature密钥对原始请求体做HMAC-SHA256再通过crypto.timingSafeEqual常量时间比较防时序攻击。验签失败返回 401packages/attio/webhooks/records.ts、packages/attio/webhooks/types.ts租户路由插件通过pluginWebhookMatcher检查是否有签名头、pluginTenantWebhookMatcher与oauthWebhookTenantLinkResolverpackages/attio/webhooks/tenant-matcher.ts 与 packages/attio/webhooks/oauth-tenant-link.ts将传入事件正确归属到对应租户事件落库处理器在验签通过后将事件按record.${kind}从负载中提取并通过ctx.db.records.upsertByEntityId/deleteByEntityId维护本地镜像同时以attio.webhook.record.${kind}记录事件日志packages/attio/webhooks/records.ts。Webhook 通用路由机制见 docs/concepts/webhooks.mdx钩子机制见 docs/concepts/hooks.mdx。错误处理内置重试策略插件内置了与 Corsair 错误处理框架集成的errorHandlerspackages/attio/error-handlers.ts处理器匹配条件行为RATE_LIMIT_ERRORHTTP 429或消息含rate_limited/429最多重试5次并透传retry-after头AUTH_ERRORHTTP 401或消息含unauthorized/invalid_auth不重试maxRetries: 0DEFAULT兜底匹配不重试你可以在插件工厂的errorHandlers选项中追加自定义处理器它们会与内置处理器合并packages/attio/index.ts。错误处理概念详见 docs/concepts/error-handling.mdx。与其他框架/适配器的联动LangChain通过 adapters/langchain 将attio.api.*操作暴露为 LangChain tool供 Agent 直接调用LlamaIndex通过 adapters/llamaindex 接入 LlamaIndexMastra通过 adapters/mastra 接入 Mastra agentMCP通过 packages/mcp 将插件的操作暴露为 MCP tools让 Claude Code、Cursor、OpenAI 等 MCP 客户端直接使用见 docs/mcp-adapters。小结corsair-dev/attio是 Corsair 生态中面向 Attio CRM 的完整集成插件安装即用一行attio()注册默认 OAuth 2.0、可切换 API Key110 个类型安全操作覆盖对象、属性、列表、记录、注释、笔记、任务、会议与 Webhook 的完整生命周期Webhook 驱动本地 SQLite 同步records、workspaceMembers支持过滤检索内置 HMAC-SHA256 验签、租户路由、限流重试与生命周期钩子适合直接嵌入 Agent 工作流或作为 MCP 工具暴露给下游。进一步资料完整 API 参考见 docs/plugins/attio/api.mdx本地数据库说明见 docs/plugins/attio/database.mdxWebhook 事件说明见 docs/plugins/attio/webhooks.mdx插件配置清单见 packages/attio/plugin-docs.yaml。【免费下载链接】corsairConnect your users to their apps项目地址: https://gitcode.com/GitHub_Trending/corsa/corsair创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考