Beetl Connect SDK

Get started

This quickstart reads one public GitHub user with a pull sync. Use Node.js 24.2 or newer. In a new directory, create package.json:

{
  "name": "my-integration",
  "version": "0.0.0",
  "private": true,
  "type": "module",
  "files": ["integration.ts"]
}

Install the SDK:

npm install --save-dev @beetlio/connect

Create integration.ts. This integration reads one public GitHub user:

import { defineIntegration, z } from "@beetlio/connect";

const User = z.object({ id: z.number(), login: z.string() });

export default defineIntegration({
  key: "github",
  displayName: "GitHub",
  connection: { origin: "https://api.github.com" },
  syncs: (sync) => ({
    users: sync({
      mode: "replace",
      records: User,
      async *run(ctx) {
        const user = await ctx.json("/users/octocat", User);

        yield { records: [user] };
      },
    }),
  }),
});

Configure it, then write the result to a local file:

npx beetl-connect configure . users
npx beetl-connect sync . users --output users.ndjson

The output is one JSON object per line. Run only trusted integration code locally.

Configuration

Put shared settings in connection.inputs and per-sync or destination settings in its inputs. Use z.strictObject():

inputs: z.strictObject({
  workspace: z.string().min(1).meta({ title: "Workspace ID" }),
  limit: z.number().int().min(1).max(100).default(25),
}),

Read them as ctx.config.connection and ctx.config.sync or ctx.config.destination. A connection's verify(ctx) reads its inputs directly from ctx.config.

Strings, numbers, booleans, enums, arrays, nested strict objects, optional fields, and validated defaults produce portable forms. Use .meta({ title, description }) for labels and help. For free-form JSON, use z.json().meta({ "x-beetl-widget": "json" }).

Input schemas cannot use transforms or custom refinements that a form cannot represent. Credentials belong in connection.auth, not ordinary inputs.

Provider origins

Use a fixed origin such as https://api.example.com, or select it from connection inputs:

origin: {
  type: "environment",
  input: "environment",
  values: { live: "https://api.example.com", test: "https://sandbox.example.com" },
},
inputs: z.strictObject({ environment: z.enum(["live", "test"]) }),

For a tenant URL, use { type: "input", input: "baseUrl" } with a required URL input. OAuth can derive the origin from { type: "oauth", oauthTokenField: "instanceUrl" }, provided that field is mapped in auth.oauth2({ tokenFields: … }). Authenticated origins require HTTPS, except for loopback development.

Authentication

Import auth from @beetlio/connect and set connection.auth. Omit it for public APIs.

Built-in helpers define their credential fields. Custom and token-exchange credentials are strict objects of strings. Mark sensitive fields with secret(); credential defaults are not allowed.

import { auth, secret, z } from "@beetlio/connect";

const authentication = auth.tokenExchange({
  credentials: z.strictObject({ apiKey: secret(z.string().min(1)) }),
  request: {
    path: "/token",
    body: { encoding: "json", fields: { key: { credential: "apiKey" } } },
  },
  response: {
    tokenPath: "access_token",
    expiry: { type: "relative", path: "expires_in" },
  },
});

Assign authentication to connection.auth. The SDK obtains and reuses the token; sync and destination code keep calling ctx.fetch or ctx.json without handling credentials.

For OAuth, add clientSecret: true when the provider requires one. Endpoints may be absolute or provider-relative if the origin is known before authorization. The CLI prompts for credentials and prints the authorization URL.

Verify a connection

Add async verify(ctx) to the connection and make a small authenticated request. Throw if access is invalid. configure runs verification before saving the profile.

Requests & pagination

ctx.fetch(path, init?) returns a Response. ctx.json(path, schema, init?) parses successful JSON with Zod and throws HttpError for HTTP failures. The error includes status, bounded detail, and an optional requestId.

Requests accept provider-relative paths or same-origin HTTP URLs. The provider enforces origin boundaries and does not follow redirects. By default, transient failures on GET, HEAD, and OPTIONS get up to three attempts. Set connection.retry to tune the policy, or false to disable retries.

For an API returning { items, next }, add a paginated sync like this:

items: sync({
  records: z.object({ id: z.string() }),
  checkpoint: z.object({ cursor: z.string().nullable() }),
  async *run(ctx) {
    const cursor = ctx.checkpoint?.cursor;

    for await (const page of ctx.paginate({
      request: { path: "/items", query: cursor ? { cursor } : {} },
      schema: z.object({
        items: z.array(z.object({ id: z.string() })),
        next: z.string().nullable(),
      }),
      next: ({ data }) => data.next === null
        ? undefined
        : { path: "/items", query: { cursor: data.next } },
    })) {
      yield { records: page.data.items, checkpoint: { cursor: page.data.next } };
    }
  },
}),

next receives parsed data, response status and headers, the request, and the page index. Return the next request or undefined to finish. Empty pages can continue; repeated requests and more than 10,000 pages fail.

Await context calls and use ctx.signal for other async work. Log through ctx.log so hosts can route messages consistently.

Records & checkpoints

A sync's records schema describes each output row. Use a Zod object whose output can be represented as JSON Schema. Provider-response schemas may transform data before it becomes a record.

Merge keys must be required, scalar, non-null fields. Each deletion contains exactly those fields. Only merge syncs accept deletions.

Use yield* batchRecords(records) for record-only streams: it groups up to 100 records or about 1 MiB per batch. A larger record is emitted alone. Use storageRecord(providerSchema) when dynamic JSON fields need fixed storage columns.

For resumable work, declare a checkpoint schema and yield the position to resume after each batch. The host commits records and checkpoint before asking for more data. A failed commit does not advance the checkpoint.

Checkpoints are opaque JSON; absent and null are different. Validation must preserve their exact value. A host can stop after a checkpoint-bearing batch; the generator's finally runs before execution returns.

Destinations

A destination writes batches of records to a provider API. Add destinations alongside syncs, or define an integration with only destinations. Both use the same connection, authentication, and request helpers.

Destinations require SDK 0.4 or newer and run through the embedded host API. The CLI and process host currently execute pull syncs only; use the mocked example below to try destination execution locally.

Define a destination

This example upserts contacts and deletes contacts by key:

import { auth, defineIntegration, z } from "@beetlio/connect";

export default defineIntegration({
  key: "contacts",
  displayName: "Contacts",
  connection: { origin: "https://provider.example", auth: auth.bearer() },
  destinations: (destination) => ({
    contacts: destination({
      records: z.object({ id: z.string(), email: z.email() }),
      primaryKey: ["id"],
      supportsDelete: true,
      async run(ctx, batch) {
        const result = await ctx.json("/contacts/batch", z.object({
          failedIds: z.array(z.string()),
        }), {
          method: "POST",
          headers: {
            "content-type": "application/json",
            "idempotency-key": batch.batchId,
          },
          body: JSON.stringify({
            records: batch.records,
            deletedKeys: batch.deletedKeys ?? [],
          }),
        });
        if (result.failedIds.length) throw new Error("Contact write failed");
      },
    }),
  }),
});

records describes accepted input for field mapping. Every primary-key field must be required, scalar, and non-null; key coercion is unsupported. Each deletion contains exactly those fields. supportsDelete defaults to false. Optional inputs are available as ctx.config.destination.

Run a batch

Embedding hosts call runDestinationBatch from @beetlio/connect/host with the integration, input, and SDK provider. The runnable example below includes provider setup and authentication cleanup.

await runDestinationBatch(integration, {
  destination: "contacts",
  batch: {
    batchId: "delivery-42",
    records: [{ id: "one", email: "one@example.com" }],
    deletedKeys: [{ id: "gone" }],
  },
}, provider);

The executor validates the entire batch before destination code runs: at most 10,000 records plus deletions and 8 MiB of UTF-8 JSON, before and after parsing. It rejects unknown root fields, invalid rows, and duplicate or conflicting keys. Treat the supplied batch as immutable.

Successful completion acknowledges the whole batch. Throw on any failed item, including failures inside an HTTP 200 response. Some writes may already have succeeded, so make replay safe with upserts, repeatable deletes, and provider idempotency support. A retry keeps the same batch ID and contents.

The host selects and maps records, orders batches, and saves delivery progress. Destination code receives the prepared records and deletion keys; it does not manage checkpoints. See the destination contract for host requirements.

Try the example

Run the mocked destination example from an SDK checkout. It writes and replays a batch without real credentials:

npm ci
npm run build
node examples/destination/run.ts

Local development

configure saves connection settings and a sync profile. Supply inputs interactively or as JSON:

npx beetl-connect configure . users --profile demo \
  --inputs '{"connection":{},"sync":{}}'
npx beetl-connect sync . users --profile demo --output users.ndjson

Use --connection to share a named connection across profiles, --origin to override the provider origin, and --reauthorize to collect credentials again. Credentials are owner-readable files in the operating system's user configuration directory; they are not encrypted.

sync saves checkpoints under .beetl/state/ by default. Override that with --state. Append and replace write record NDJSON; replace publishes the file only after success. Merge writes batch envelopes containing records and deletions.

Package for Beetl

npx beetl-connect pack . --output github.tgz

Keep integration.ts, package.json, and the lockfile in the project. List source and assets in package.json's files allowlist. Registry dependencies are supported; local paths, workspaces, Git dependencies, and remote tarballs are not.

The archive contains source for a hosted build. Packing does not execute the integration. Check the printed file list before sharing it. Run npx beetl-connect <command> --help for all options.

Examples & reference

• Basic — a small integration with verification.
• All features — OAuth, configuration, pagination, and checkpoints.
• Wikidata — a public API integration you can run locally.
• Destination — mapped upserts, deletions, and replay against a mocked provider.

The basic and all-features examples use mocked APIs in tests. For host implementers, see the architecture, compatibility contract, and process protocol.

SDK 0.4 builds manifest v4 and host contract v4, and supports existing v3 pull artifacts and source. The process protocol remains v1.