Authentication

A package declares how it authenticates once, on its connection. After that, your sync and verify code calls ctx.fetch, ctx.json or ctx.paginate with a path, and the SDK adds the credentials, refreshes tokens and retries transient failures. Your code never reads a credential or a token.

Choose an auth mode

Set connection.auth to one of these helpers, or leave it out for a public API. The helper defines the credential fields a user fills in.

See Connections for how users save and reuse credentials.

For custom and tokenExchange, declare the fields as a z.strictObject of strings and place them with { credential: "<field>" }. Wrap sensitive fields in secret() so the form masks them. Every field must be used, and none can have a default.

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

const authentication = auth.custom({
  credentials: z.strictObject({
    accountId: z.string().min(1),
    token: secret(z.string().min(1)),
  }),
  headers: {
    "x-account": { credential: "accountId" },
    authorization: { credential: "token" },
  },
});

tokenExchange trades credentials for a short-lived session token at a path on the provider's origin, then sends it as Authorization: Bearer unless you configure session. It exchanges before a run's first request, shortly before the token expires, and after a 401. Any authenticated origin must use HTTPS.

OAuth2 authorization code

The SDK always uses PKCE with S256. All 3 URLs must use HTTPS. Each can be absolute, or a path on the connection's origin if that origin is known before sign-in (a fixed URL or one picked from connection settings).

If an API returns a per-account base URL in the token response, copy it with tokenFields and use it as the origin. The OAuth URLs must then be absolute.

connection: {
  origin: { type: "oauth", oauthTokenField: "instanceUrl" },
  auth: auth.oauth2({
    issuer: "https://login.example.com",
    authorizationUrl: "https://login.example.com/oauth/authorize",
    tokenUrl: "https://login.example.com/oauth/token",
    scopes: ["api", "refresh_token"],
    clientSecret: true,
    tokenFields: { instanceUrl: "instance_url" },
  }),
},

What the user sees

The user registers Beetl's redirect URL with the provider's OAuth app, enters the app's client ID and secret, and chooses Connect with OAuth. Sign-in opens in a popup and has to finish within 10 minutes. Each connection holds its own authorization. See Integrations for the redirect URL and steps.

How tokens are refreshed

The SDK sends the access token as Authorization: Bearer on every request. When the provider answers 401, it refreshes the token and repeats the request once, and concurrent requests share one refresh. Beetl keeps the new access token, and a rotated refresh token, for later runs.

• Without a refresh token, nothing is refreshed and the 401 reaches your code as an HttpError. Many providers issue one only for a scope such as offline_access or a parameter such as access_type=offline.
• Only a 401 triggers a refresh. If the provider signals an expired token with another status, the request fails.
• If the provider rejects the refresh token (invalid_grant or access_denied), the connection shows Reauthorization required and its syncs stop until someone chooses Reconnect with OAuth. If the token endpoint is only temporarily down, the run fails and the authorization stays.

Requests

ctx.fetch(path, init?) returns a standard Response. ctx.json(path, schema, init?) parses a successful JSON body with a Zod schema and throws an HttpError (with status, a shortened detail and any requestId) for any other status.

Paths are relative to the connection's origin. An absolute URL works only on that origin; anything else is rejected. Redirects are not followed. One response can be at most 16 MiB, so request smaller pages if needed.

Pagination

ctx.paginate fetches a page, parses it with your schema, and passes the data, response status and headers, request and page index to next. Return the next request, or undefined to stop. An absolute URL from a Link header works if it is on the connection's origin. A call fails if it repeats a request or passes 10,000 pages.

Retries and rate limits

Requests with a listed method are retried on those statuses and on network errors. The wait starts at initialDelayMs and doubles up to maxDelayMs. A Retry-After header, in seconds or as a date, replaces the wait, still capped at maxDelayMs. Set retry: false to turn retries off, and add POST only for requests that are safe to repeat.

When attempts run out, ctx.json throws and ctx.fetch returns the last response. Waits count toward the run's time limit. If the provider reports its remaining quota, slow down before you hit a 429, as the example below does.

Example

This sync reads contacts from an OAuth API with cursor pagination. It fetches only contacts changed since the last committed run, and each record carries its id and updated_at so a pipeline can keep the latest version of each contact. It is meant for the Append write mode.

import { setTimeout as sleep } from "node:timers/promises";
import { auth, defineIntegration, z } from "@beetlio/connect";

const Contact = z.object({
  id: z.string(),
  email: z.string().nullable(),
  updated_at: z.string(),
});

const ContactPage = z.object({
  data: z.array(Contact),
  next_cursor: z.string().nullable(),
});

export default defineIntegration({
  key: "acme-crm",
  displayName: "Acme CRM",
  connection: {
    origin: "https://api.acme.example",
    auth: auth.oauth2({
      issuer: "https://login.acme.example",
      authorizationUrl: "https://login.acme.example/oauth/authorize",
      tokenUrl: "https://login.acme.example/oauth/token",
      scopes: ["contacts:read", "offline_access"],
      clientSecret: true,
    }),
    retry: { maxAttempts: 5, maxDelayMs: 60_000 },
    async verify(ctx) {
      await ctx.json("/v1/me", z.object({ id: z.string() }));
    },
  },
  syncs: (sync) => ({
    contacts: sync({
      records: Contact,
      checkpoint: z.object({ updatedSince: z.string() }),
      async *run(ctx) {
        const since = ctx.checkpoint?.updatedSince ?? "1970-01-01T00:00:00Z";
        let newest = since;
        for await (const page of ctx.paginate({
          request: { path: "/v1/contacts", query: { updated_since: since, limit: 100 } },
          schema: ContactPage,
          next: ({ data, request }) =>
            data.next_cursor === null
              ? undefined
              : { ...request, query: { ...request.query, cursor: data.next_cursor } },
        })) {
          for (const contact of page.data.data) {
            if (contact.updated_at > newest) newest = contact.updated_at;
          }
          yield { records: page.data.data, checkpoint: { updatedSince: newest } };
          if (page.response.headers["x-ratelimit-remaining"] === "0") {
            await sleep(1_000, undefined, { signal: ctx.signal });
          }
        }
      },
    }),
  }),
});

Each page becomes one batch. Yielding the newest timestamp per page is safe because Beetl saves the checkpoint only when the run's write commits. Header names in page.response.headers are lowercase.

Related