Corsair × Agenty 集成实战：79 个类型化 API 操作、浏览器自动化与本地数据同步

导读

@corsair-dev/agenty 是 Corsair 生态中面向 Agenty（网页抓取与数据提取平台）的一等插件，它将 Agenty 的完整 API 封装为 79 个类型安全的端点，并自动把 agents、jobs、lists 三类核心实体同步到本地 SQLite，支持 search() / list() 快速检索。本文以该插件的 README 与官方文档为主体，结合仓库源码剖析其认证、请求管线、错误重试与本地缓存同步的实现原理，帮助你在一套代码里完成从「创建抓取 Agent」到「调度运行、下载结果、组织项目」的完整数据提取闭环。

插件定位与安装

Agenty 是一个面向网页抓取、变更检测、整站爬取、地图监控与品牌监控的数据提取平台。Corsair 通过插件机制让开发者以统一的客户端形态接入这类外部服务：一次配置、类型化调用、自动本地同步。@corsair-dev/agenty 的定位可从 packages/agenty/plugin-docs.yaml 中的描述得到印证——"Web scraping and data extraction platform for monitoring and harvesting online data"。

安装（npm / pnpm / yarn / bun 均可）：

pnpm add corsair @corsair-dev/agenty

包本身以 @corsair-dev/agenty 发布，其 package.json 声明了 corsair >= 0.1.0 与 zod ^4.1.13 两个 peer 依赖，即运行时依赖 Corsair 核心与 Zod 校验层，构建产物为 ESM（dist/index.js），并提供完整的 .d.ts 类型声明。

快速接入：把 Agenty 装进 Corsair

在 docs/plugins/agenty/overview.mdx 中给出了标准接入示例。创建一个 corsair.ts，初始化数据库并注册插件：

import Database from 'better-sqlite3';
import { createCorsair } from 'corsair';
import { agenty } from '@corsair-dev/agenty';

export const corsair = createCorsair({
	plugins: [
		agenty(),
	],
	database: new Database('corsair.db'),
	kek: process.env.CORSAIR_KEK!,
	hub: {
		projectApiKey: process.env.CORSAIR_API_KEY!,
		signingSecret: process.env.CORSAIR_SIGNING_SECRET!,
	},
});

要点说明：

• plugins: [agenty()] 即完成注册，无需额外配置即可用 API Key 认证；
• database 传入 better-sqlite3 实例，是本地同步的落盘位置（corsair.db）；
• kek（Key Encryption Key）与 hub 配置用于多租户凭据托管，KEK + Hub 密钥的获取方式见 Quick start，租户隔离机制见 Multi-tenancy。

多租户是默认行为，租户级调用通过 corsair.withTenant(id) 完成作用域切换：

const tenant = corsair.withTenant('acme');

从源码看，agenty() 工厂函数（packages/agenty/index.ts）在无参调用时会把 authType 默认合并为 api_key，并绑定 schema、endpoints、errorHandlers 与 keyBuilder，插件 ID 固定为 'agenty'，webhooks 为空对象（与 README「No webhooks」一致）。

认证：API Key 模式与首次连接

Agenty 插件的认证方式为 API Key。README 明确说明：首次使用时，Corsair 会提示租户（tenant）提供凭据。连接流程分两步：

1. 签发连接链接：后端调用 corsair.manage.connect.createLink 生成一个 connect URL，引导用户浏览器跳转完成授权，Hub 托管该页面并把结果回传应用——详见 Connect / OAuth：

const { connectUrl } = await corsair.manage.connect.createLink({
	plugin: 'agenty',
	tenantId: 'acme',
});
// 将用户的浏览器重定向到 connectUrl

2. 首次请求触发凭据提示：当租户发起第一个 API 请求时，Corsair 提示输入 Agenty API Key，之后密钥被加密托管。

底层实现位于 packages/agenty/index.ts 的 keyBuilder：当调用来源为 endpoint 且插件选项提供了 options.key 时直接使用该静态密钥；否则通过 ctx.keys.get_api_key() 从租户密钥库读取；若取不到则打印 [AGENTY] API key missing 并抛出 AuthMissingError('agenty', 'api_key')。这意味着你既可以在多租户场景下让每个租户自带密钥，也可以在单租户场景下直接 agenty({ key: 'xxx' }) 注入固定密钥。

79 个类型化端点总览

插件共封装 79 个 API 操作（docs/plugins/agenty/overview.mdx 中记录为 79 个），按资源分为 11 组。每个操作带有唯一 Operation ID 与 风险等级（read / write / destructive），这是 Corsair 权限模型的基础。下表完整列出全部端点：

Agents（抓取智能体生命周期）

Operation ID	风险	说明
agenty.api.agents.agentsControllerCreateAgent	write	创建抓取 Agent，type 支持 scraping、changedetection、crawling、mapmonitoring、brandmonitoring；config 含 url、browser 与 collections 字段定义；start=true 可创建后立即运行
agenty.api.agents.agentsControllerGetTemplates	read	获取公开 Agent 模板与示例
agenty.api.agents.agentsDeleteById	destructive	按 ID 删除 Agent
agenty.api.agents.agentsGetAll	read	分页/排序列出账号下所有活跃 Agent
agenty.api.agents.agentsGetById	read	获取 Agent 完整配置、输入设置、调度器与元数据
agenty.api.agents.agentsUpdateById	write	更新 Agent 配置（名称、类型、config、tags、scheduler、scripts、可见性）；响应只返回被更新字段，需用 GetById 获取完整对象
agenty.api.agents.copyAgent	write	复制现有 Agent，可选新名称
agenty.api.agents.transferAgentOwnership	write	通过邮箱把 Agent 所有权转移给其他 Agenty 账号

Api Keys（API 密钥管理）

Operation ID	风险	说明
agenty.api.apiKeys.apiKeysControllerCreateApiKeys	write	创建新 API Key，支持 Owner / Admin / Manager 三种权限级别
agenty.api.apiKeys.apiKeysDeleteById	destructive	按 ID 永久吊销密钥，不可撤销
agenty.api.apiKeys.apiKeysDownload	read	以 CSV 导出账号下全部密钥
agenty.api.apiKeys.apiKeysGetAll	read	分页/排序列出全部密钥
agenty.api.apiKeys.apiKeysGetById	read	获取密钥详情（值、角色、状态）
agenty.api.apiKeys.apiKeysResetById	destructive	轮换密钥 Secret，旧值立即失效；注意：此操作不返回新 Secret，需再调用 GetById 获取
agenty.api.apiKeys.apiKeysUpdateById	write	更新密钥名称与角色（仅这两个字段可改）
agenty.api.apiKeys.changeApiKeyStatusById	write	切换密钥启用/禁用状态

Browser（浏览器自动化与页面提取）

Operation ID	风险	说明
agenty.api.browser.captureScreenshot	read	整页或可视区域截图，默认参数
agenty.api.browser.captureScreenshotWithOptions	write	高度可定制截图：整页、图片格式、质量、viewport、后处理
agenty.api.browser.convertUrlToPdf	read	URL 转 PDF
agenty.api.browser.convertUrlToPdfWithOptions	write	URL 或原始 HTML 转 PDF，支持页面尺寸、边距、页眉页脚、方向
agenty.api.browser.extractBrowserStructuredData	write	自动提取结构化数据（schema.org / RDFa / Microdata / JSON-LD）
agenty.api.browser.extractStructuredData	read	从 URL 自动提取结构化数据
agenty.api.browser.getBrowserRedirects	read	获取完整重定向链（含服务端 3xx 与客户端 JS/meta 跳转）
agenty.api.browser.getPageContent	read	获取完整 HTML，含 JS 渲染后内容，走代理导航
agenty.api.browser.getPageContentWithOptions	write	获取 HTML 且支持广告拦截，加速加载、减少噪音
agenty.api.browser.getRedirectsWithOptions	write	带自定义导航选项的重定向链追踪，支持超时与等待条件
agenty.api.browser.scrapeWebpageData	write	用 jQuery/CSS 选择器抓取数据，每个 query 字段映射一个 jQuery 表达式（如 $('h1').text()）

Connections / Dashboard / Inputs

Operation ID	风险	说明
agenty.api.connections.connectionsGetAll	read	列出账号下全部连接，支持 limit/offset 分页与排序
agenty.api.dashboard.dashboardGetReportsUsage	read	按 Agent、日期、产品维度获取用量报表
agenty.api.inputs.inputsGetByAgentId	read	获取指定 Agent 的输入源配置（URL、手动列表、已存列表引用、其他 Agent 输出）
agenty.api.inputs.inputsUpdateByAgentId	write	更新输入源：type='url'（URL 源）、type='manual'（手动 URL 列表）、type='list'（引用 Agenty 列表）、type='agent'（复用其他 Agent 输出）

Jobs（任务执行与结果获取）

Operation ID	风险	说明
agenty.api.jobs.downloadAgentResult	read	按 Agent ID 导出结果，支持 CSV / TSV / JSON
agenty.api.jobs.getAgentResult	read	获取 Agent 最近一次执行结果，支持分页
agenty.api.jobs.getJobResult	read	获取已完成 Job 的结果数据，支持分页
agenty.api.jobs.jobsDownload	read	以 CSV 导出全部 Job
agenty.api.jobs.jobsDownloadFilesById	read	按 Job ID 下载输出文件
agenty.api.jobs.jobsDownloadResultById	read	按 Job ID 下载最终输出（CSV / TSV / JSON）
agenty.api.jobs.jobsGetAll	read	分页/排序列出账号下全部 Job
agenty.api.jobs.jobsGetById	read	获取 Job 详情：状态、进度（处理/成功/失败页数）、时间信息、页额度消耗与错误信息
agenty.api.jobs.jobsGetLogsById	read	获取 Job 执行日志，支持分页
agenty.api.jobs.jobsListFilesById	read	列出 Job 产出的全部文件（名称与大小）
agenty.api.jobs.jobsStart	write	启动既有 Agent 的新 Job
agenty.api.jobs.jobsStopById	write	停止运行中的 Job，需先确认 Job ID

Lists（列表数据管理）

Operation ID	风险	说明
agenty.api.lists.addListRows	write	向列表插入数据行，列名须匹配列表 schema
agenty.api.lists.deleteListRow	destructive	按行 ID 删除单行
agenty.api.lists.deleteListRows	destructive	按行 ID 批量删除
agenty.api.lists.downloadListRows	read	以 CSV 导出列表全部行
agenty.api.lists.getListById	read	获取列表详情（名称、描述、元数据、创建/更新时间）
agenty.api.lists.getListRowById	read	按行 ID 获取单行数据
agenty.api.lists.listsClearRows	destructive	清空列表全部行
agenty.api.lists.listsControllerCreateList	write	创建新列表
agenty.api.lists.listsDeleteById	destructive	永久删除列表
agenty.api.lists.listsDownload	read	以 CSV 导出全部列表
agenty.api.lists.listsGetAll	read	分页/排序列出全部列表
agenty.api.lists.listsGetRowsById	read	获取列表全部行，支持分页排序
agenty.api.lists.listsUpdateById	write	更新列表名称与描述，name 必填
agenty.api.lists.listsUploadCsv	write	上传 CSV 批量导入数据，目标列表须先存在
agenty.api.lists.updateListRow	write	更新指定行，row_data 须含 _id 及要修改的列字段

Projects / Scheduler / Users / Workflows

Operation ID	风险	说明
agenty.api.projects.deleteProject	destructive	永久删除项目，不可撤销
agenty.api.projects.getProjectById	read	获取项目详情（名称、描述、创建者、时间戳）
agenty.api.projects.projectsAddAgents	write	向项目添加 Agent 进行分组组织
agenty.api.projects.projectsControllerCreateProject	write	创建项目
agenty.api.projects.projectsGetAll	read	分页列出项目，支持按 name / created_at 排序
agenty.api.projects.removeAgentFromProject	destructive	将 Agent 移出项目（Agent 与项目本身保留）
agenty.api.projects.updateProject	write	更新项目名称与描述
agenty.api.scheduler.deleteSchedule	destructive	按 Agent ID 删除调度配置
agenty.api.scheduler.getSchedule	read	获取 Agent 当前调度配置
agenty.api.scheduler.toggleSchedule	write	开/关调度，不影响其他设置
agenty.api.scheduler.updateSchedule	write	更新调度频率配置
agenty.api.users.downloadUsers	read	以 CSV 导出团队成员列表
agenty.api.users.getAllTeamMembers	read	分页/排序/搜索列出团队成员
agenty.api.users.getUserById	read	获取用户档案（邮箱、角色、状态、活动时间）
agenty.api.users.updateUserById	write	更新用户信息（email、role、status 必填）
agenty.api.workflows.createWorkflow	write	基于 Agent 事件创建自动化工作流（Job 完成/失败时发邮件、触发 Webhook、通知）
agenty.api.workflows.deleteWorkflow	destructive	按 ID 删除工作流
agenty.api.workflows.downloadWorkflows	read	以 CSV 导出全部工作流，通用输出结构：data(string)、error(string 可选)、successful(boolean)
agenty.api.workflows.getWorkflowById	read	获取工作流配置（agents、triggers、actions）
agenty.api.workflows.patchWorkflow	write	部分更新工作流（PATCH），当前支持更新名称
agenty.api.workflows.updateWorkflow	write	完整更新工作流（名称、Agent 选择、触发条件、执行动作）

典型调用：创建 Agent 并立即运行

以文档中的两个示例为起点（docs/plugins/agenty/overview.mdx）：

const tenant = corsair.withTenant('acme');

// 1. 浏览可用模板
await tenant.agenty.api.agents.agentsControllerGetTemplates({});

// 2. 创建抓取 Agent（含采集字段定义）
await tenant.agenty.api.agents.agentsControllerCreateAgent({
	name: 'Competitor Price Monitor',
	type: 'scraping',
	config: {
		url: 'https://example.com/products',
		browser: {},
		collections: [
			{
				// 每个字段定义要提取什么数据以及提取方式
				name: 'title',
				selector: 'h1.product-title',
				extract: 'TEXT',
			},
			{
				name: 'price',
				selector: '.price',
				extract: 'TEXT',
			},
		],
	},
	start: true, // 创建后立即执行一次
});

agentsControllerCreateAgent 的输入结构（来自 docs/plugins/agenty/api.mdx）中，name 与 config 为必填，其余如 tags、scheduler、scripts、is_public、start、version 等均为可选；创建成功后会返回带唯一 agent_id 的 Agent 对象，供后续任务引用。type 的取值即 README 中列出的五种：scraping、changedetection、crawling、mapmonitoring、brandmonitoring。

在 packages/agenty/endpoints/routes.ts 中可以看到该操作的底层路由定义：POST /agents，hostType: 'main'，风险等级 write——这正是上层调用最终映射到的 HTTP 请求。

底层请求管线：路径、查询与双宿主

所有端点最终汇聚到统一的请求执行管线，其实现分布在 packages/agenty/endpoints/factory.ts 与 packages/agenty/client.ts：

• 路径参数解析：resolvePath 负责把路由模板中的 {agent_id}、{job_id} 等占位符替换为实际值；PATH_PARAM_ALIASES 支持 snake_case 与 camelCase 双写法（如 agent_id / agentId），并对缺失参数抛错，避免拼出残缺 URL。
• 查询参数构建：buildQuery 从输入中按路由声明的 queryParams（如分页的 sort、limit、order、offset）自动拾取值；scrapeWebpageData 特殊处理为顶层 query 即 jQuery 选择器映射。
• 请求体构建：requestBody 优先使用显式 body 字段，其次处理数组体路由（projectsAddAgents 需要原始 JSON 数组而非 {agent_ids: [...]}，源码注释注明已用真实请求验证），最后把非控制字段（排除 body/query/headers/baseUrl）打包为 JSON body。
• 双宿主路由：resolveBaseUrl 根据 hostType 区分主 API（https://api.agenty.com/v2）与浏览器 API（https://browser.agenty.com/api），例如所有 browser.* 操作走浏览器宿主。

packages/agenty/client.ts 中的 resolveAgentyBase 执行严格的主机白名单校验：仅允许 api.agenty.com 与 browser.agenty.com，且强制 HTTPS、禁止非常规端口，否则直接抛出 AgentyAPIError。请求头构造上，主 API 使用 Authorization: Bearer <key>，而浏览器宿主额外附带 X-Agenty-ApiKey（注释说明浏览器 API 文档要求该头部而非 Bearer）；同时刻意不把密钥放进 query string，避免 URL 泄漏——源码注释将其标记为历史 P1 级安全教训。

错误处理与自动重试策略

插件内置了一套分层错误处理与重试策略，实现在 packages/agenty/error-handlers.ts，并在 agenty() 工厂中与用户自定义 handler 合并：

错误类型	匹配条件	处理策略
RATE_LIMIT_ERROR	HTTP 429	最多重试 3 次，指数退避；若响应头或 body 带 Retry-After / retry_after 则优先采用（秒值自动换算毫秒）
AUTH_ERROR	401 / 403	不重试，打印 [AGENTY] Authentication failed 提示检查密钥
NOT_FOUND_ERROR	404	不重试
SERVER_ERROR	5xx	最多重试 2 次，指数退避
DEFAULT	兜底	不重试，打印错误信息

所有网络异常会被包装为 AgentyAPIError（packages/agenty/client.ts），保留 HTTP status、statusText 与原始响应体，便于上层诊断。

本地同步：agents / jobs / lists 三类实体的 search 与 list

插件的一个关键能力是把远端数据同步到本地 SQLite，让检索无需反复打 API。

实体 schema

三类同步实体的校验 schema 定义在 packages/agenty/schema/database.ts：AgentyAgent（agent_id、name、type、status）、AgentyJob（job_id、agent_id、status）、AgentyList（list_id、name），均以 catchall(z.unknown()) 兜底，保证远端返回的新增字段不会被丢弃。

同步规则

packages/agenty/endpoints/cache-sync.ts 中定义了缓存策略：

• 每个资源组（agents / jobs / lists）有对应的 idKeys（如 agent_id/id）与 listKeys（如 data/items/results），用于从响应中抽取实体并 upsertByEntityId 写入本地；
• DELETE 类操作会根据输入中的 ID 调用 deleteByEntityId 同步删除本地缓存；
• 明确的排除清单：listsGetRowsById、getJobResult、getAgentResult、jobsGetLogsById、deleteListRow、deleteListRows、listsClearRows 这类子资源/结果型接口不会污染父实体缓存；
• 缓存失败不会抛出，而是记录 agenty.cache.<entity>.failed 事件后静默降级。

检索 API

依据 docs/plugins/agenty/database.mdx，统一使用 corsair.agenty.db.<entity>.search({ data, limit?, offset? })（limit、offset 用于分页；同路径下也有 .list() 可用，详见 database operations）：

// 按状态与名称模糊检索 Agent
const agents = await corsair.agenty.db.agents.search({
	data: {
		name: { contains: 'price' },
		status: { equals: 'active' },
	},
	limit: 100,
	offset: 0,
});

// 数值范围检索 Job
const jobs = await corsair.agenty.db.jobs.search({
	data: { job_id: { gt: 1000 } },
});

// 列表检索
const lists = await corsair.agenty.db.lists.search({
	data: { name: { startsWith: 'leads' } },
});

各实体可用的过滤字段与操作符如下：

Agents（agenty.db.agents.search）

字段	类型	操作符
entity_id	string	equals, contains, startsWith, endsWith, in
agent_id	string	equals, contains, startsWith, endsWith, in
name	string	equals, contains, startsWith, endsWith, in
type	string	equals, contains, startsWith, endsWith, in
status	string	equals, contains, startsWith, endsWith, in

Jobs（agenty.db.jobs.search）

字段	类型	操作符
entity_id	string	equals, contains, startsWith, endsWith, in
job_id	number	equals, gt, gte, lt, lte, in
agent_id	string	equals, contains, startsWith, endsWith, in
status	string	equals, contains, startsWith, endsWith, in

Lists（agenty.db.lists.search）

字段	类型	操作符
entity_id	string	equals, contains, startsWith, endsWith, in
name	string	equals, contains, startsWith, endsWith, in

工作流与 Webhooks 边界

• 事件自动化：workflows.* 系列操作支持基于 Agent 事件（如 Job 完成、失败、变更检测命中）触发邮件、Webhook 或通知，形成"抓取 → 检测 → 动作"的自动化链路。
• Webhooks：README 明确指出该插件无 Webhook（No webhooks），插件工厂中 webhooks: {} 与 pluginWebhookMatcher: undefined 也从源码侧印证了这一点——如果你需要被动接收事件，应优先使用 workflows 主动触发，或结合 Corsair 的 Webhooks 概念 另行设计回调通道。

将端点暴露给 Agent：MCP 适配

79 个操作天然适合作为 LLM Agent 的工具面。Corsair 的 MCP 适配层（见 mcp-adapters）可以把这些 agenty.api.* 操作直接暴露为 MCP tools，让 Claude、Cursor、OpenCode 等编码 Agent 或 LangChain / LlamaIndex / Mastra 等框架（对应仓库 adapters 目录下的适配器）直接调用抓取与监控能力——README 中的 Operation ID 命名（如 agenty.api.agents.agentsControllerCreateAgent）即为工具调用而设计，风险等级则用于权限控制。

许可与文档索引

• 插件以 Apache-2.0 开源（见 README 与 package.json 的 license 字段）；
• 完整 API 参考（每个操作的输入输出类型、Zod schema）见 docs/plugins/agenty/api.mdx，本地同步与过滤操作符见 docs/plugins/agenty/database.mdx；
• 相关测试见 packages/agenty/api.test.ts，可用 pnpm --filter @corsair-dev/agenty test 运行。

从安装、认证、79 个类型化端点，到错误重试与本地缓存同步，@corsair-dev/agenty 将 Agenty 的抓取能力完整收编进 Corsair 的统一客户端。下一步可以基于 withTenant 做多租户隔离，或借助 MCP 适配把整套抓取管线交给 Agent 编排。