Connector Authoring Guide

How to build connectors for Kaduno PullPush — AI-generated, manually coded, or JSON-defined.

Connector types

Canonical-aware. Connectors are built around the canonical data model. Source connectors emit canonical events (NormalizedEvent carries canonicalType/canonicalVersion); destinations consume canonical payloads in deliver(); and connectors should implement the optional snapshot(connection, opts): AsyncIterable<SnapshotItem> that yields current state in canonical space ({ key, canonicalType, fields }) — this is what powers sync-preview and the reconcile→deliver loop. A snapshot must distinguish a real 0 from "no data" (never emit 0 for a missing scope, or you will wipe destination stock).

Credentials are envelope-encrypted. connection.config is empty at rest; callers hydrate via await hydrateConnectionConfig(prisma, connection) before invoking a connector. Inside a connector, read connection.config (already hydrated by the caller) — never decrypt or log raw credentials yourself.

Builtin connectors (manual TypeScript)

Source connector interface

import type { SourceConnector, RawChange } from '@pullpush/connectors';

export class MySourceConnector implements SourceConnector {
  type = 'my-source';

  async *poll(connection, cursor): AsyncIterable<RawChange> {
    // Yield raw changes since `cursor` (ISO timestamp or page token)
    yield { type: 'item.updated', externalId: '123', payload: { ... }, cursor: '...' };
  }

  supportsWebhook(): boolean {
    return true; // or false
  }

  verifyWebhook(rawBody: string, signature: string, config: Record<string, unknown>): boolean {
    // HMAC verification
  }

  parseWebhook(rawBody: string, headers: Record<string, string>): RawChange[] {
    // Parse incoming webhook payload into raw changes
  }

  async test(connection): Promise<boolean> {
    // Verify credentials work (e.g. GET /api/status)
  }
}

Destination connector interface

import type { DestinationConnector, DeliveryResult } from '@pullpush/connectors';

export class MyDestinationConnector implements DestinationConnector {
  type = 'my-destination';

  async deliver(connection, payload, event): Promise<DeliveryResult> {
    // Send the transformed payload to the destination system
    // Return { success: true, latencyMs } or { success: false, retryable: true, errorMessage: '...' }
  }

  async test(connection): Promise<boolean> {
    // Verify credentials work
  }
}

Built-in connectors are classes implementing SourceConnector / DestinationConnector (see magento/, linnworks/, abicart/). The destination signature is deliver(connection, payload, event) and returns { success, errorCode?, errorMessage?, latencyMs? }. Add the optional snapshot() to make the connector previewable/reconcilable.

snapshot(connection, opts?) contract. Yield canonical SnapshotItems ({ key, canonicalType, fields }). opts.fieldHint is a scope hint (e.g. a target stock location). opts.onlyKeys (when present) restricts the snapshot to a specific set of canonical keys — emit at most those keys. A source that supports it should resolve channel-specific keys to the destination's keys (e.g. Linnworks resolves Abicart article numbers via master SKU first, then uses the preloaded channelSkuCache for instant channel-SKU lookups) so the diff lines up and a destination that carries a subset doesn't trigger a full-catalog fan-out. The pipeline derives onlyKeys from the destination snapshot when scopeToDestination/ scopeSourceToDestination is set (reconcile / preview). opts.channelSkuCache is auto-loaded from the ChannelSkuCache table (built by the channel-sku-cache background job every 12h) — connectors that support it get O(1) channel-SKU resolution instead of per-item API calls. Stock safety: if the target location has no record for an item, skip it — never emit 0 (that would push 0 and wipe real destination stock).

Registration

Register in packages/connectors/src/index.ts inside registerBuiltinConnectors() using the factory form (a new instance per resolve):

import { registerSource, registerDestination } from './registry.js';
import { MySourceConnector } from './my-source/index.js';
import { MyDestinationConnector } from './my-destination/index.js';

registerSource('my-source', () => new MySourceConnector());
registerDestination('my-destination', () => new MyDestinationConnector());

bootstrapConnectors(prisma) calls registerBuiltinConnectors() + loads DB-backed dynamic definitions; every app entrypoint uses it.

AI-generated connectors (code)

Use the MCP tool pullpush.connector.generate:

Tool: pullpush.connector.generate
Input:
  name: "shopify"
  connectorType: "source"
  apiDescription: "Shopify Admin REST API for orders and products"
  apiDocUrl: "https://shopify.dev/docs/api/admin-rest"

This runs a generate → typecheck → test → self-correct loop (up to 3 attempts):

1. AI generates connector code from the API description/OpenAPI spec
2. Code is written to packages/connectors/src/generated/<name>.ts
3. TypeScript compiler validates the code
4. If typecheck fails, AI self-corrects and retries
5. On success, the connector is auto-registered in the registry

Generated connectors follow the same interface as builtins. You can later move them to the main source tree and customize.

Dynamic connectors (JSON definition)

For connectors that don't need custom code — just REST API calls with config.

Create via MCP

Tool: pullpush.connector.define
Input:
  name: "My REST API"
  apiDocUrl: "https://api.example.com/openapi.json"
  apiDescription: "REST API with OAuth2, polls /orders endpoint"

This generates a JSON definition and stores it as a ConnectorDefinition record in draft status.

Definition lifecycle

draft → review → active → archived

• draft: Just generated, not yet usable
• review: Validated but awaiting approval
• active: Live and usable for connections (hot-registered on activation)
• archived: Retired, not loaded on startup

Activate with:

Tool: pullpush.connector.definition.activate
Input: { id: "<definition-id>" }

Definition JSON structure

{
  "metadata": {
    "slug": "my-api",
    "type": "source",
    "description": "My REST API connector",
    "version": "1.0.0"
  },
  "protocol": "rest",
  "auth": {
    "type": "oauth2",
    "tokenUrl": "https://api.example.com/oauth/token",
    "scopes": ["read:orders"]
  },
  "endpoints": {
    "poll": {
      "method": "GET",
      "url": "https://api.example.com/orders",
      "params": { "since": "{{cursor}}", "limit": 100 },
      "pagination": { "type": "cursor", "field": "next_cursor" }
    },
    "test": {
      "method": "GET",
      "url": "https://api.example.com/status"
    }
  },
  "eventMapping": {
    "externalId": "$.id",
    "eventType": "order.updated",
    "occurredAt": "$.updated_at",
    "payload": "$"
  }
}

Base URL from connection settings. metadata.base_url may name connection settings ("https://{{cred.shop_domain}}/admin/api/2024-01"); they are filled from the connection, whose own base_url wins when it sets one. A missing setting fails before any request.

Paths and the base URL's path (PP-858). A relative path is joined onto the base URL's path: with metadata.base_url https://{{cred.shop_domain}}/admin/api/2025-01, products.json calls /admin/api/2025-01/products.json (the engine adds the trailing slash itself). A path that starts with / resolves from the host root and drops the base path, so /products.json would call the shop root. The validator therefore refuses a leading-slash path (in entities, documents and test_endpoint) when the base URL has a path and the path does not repeat it: write the path relative, or repeat the prefix (/wp-json/wp/v2/hanen_member against {{cred.base_url}}/wp-json/wp/v2 is accepted). With a host-only base URL a leading slash is fine. A connection's own base_url replaces the definition's at runtime, so relative paths are also the safe choice when a connection may carry a path.

Calling a second host (metadata.allowed_origins). Request paths resolve against the connection's base URL, values substituted into a path are percent-encoded, and the resolved URL must stay on the base URL's origin. To call an API on another host, write its origin literally at the start of the path ("path": "https://salesorder.visma.net/api/v3/SalesOrders") and list it in metadata.allowed_origins (["https://salesorder.visma.net"]). Entries are https://host[:port] origins only: no path, no wildcard, no private or internal host. A host written through a {{placeholder}} is never allowed. A refused URL fails the delivery as URL_ORIGIN_REFUSED (permanent, never retried) before the push makes anything. Poll, snapshot and lookup requests follow the same rule: a page cursor, a returned next-page link or a record's {{key}} is encoded into its place and never chooses the host.

Empty path values are refused. A placeholder before the first ? in a request path that has no value fails the request before it is sent (/items/{{id}}/inventory-overview with no id is never called as /items//inventory-overview). A push fails as URL_EMPTY_PATH_VALUE, permanent and per record; a poll or snapshot page fails the run with a message naming the placeholder. Poll and snapshot paths can only use connection settings (cred.*), the paging variables and preloaded lookup.* values, and the validator refuses anything else there.

Paging by the Link header. For an API that pages through the Link response header (Shopify REST, page_info), set "pagination": { "type": "cursor", "link_header_cursor_param": "page_info", "cursor_excludes_params": true, "page_size_param": "limit", "max_page_size": 250 }. Only the cursor value is taken from the rel="next" link and sent on the definition's own path; the link is never followed, and no next link ends the listing. cursor_excludes_params sends a cursor page without the other query parameters. A body cursor (HubSpot paging.next.after) uses next_cursor_path with cursor_param instead. A listing with no pagination reads one page only.

MCP protocol connectors

For connectors that delegate to an external MCP server:

{
  "metadata": { "slug": "ext-service", "type": "source" },
  "protocol": "mcp",
  "mcpConfig": {
    "serverUrl": "https://ext-service.example.com/mcp",
    "tools": {
      "poll": "ext.orders.list",
      "deliver": "ext.orders.create",
      "test": "ext.health"
    }
  }
}

Testing connectors

Builtin/generated (vitest)

Create a test file alongside the connector:

// packages/connectors/src/__tests__/my-source.test.ts
import { describe, it, expect } from 'vitest';
import { mySource } from '../my-source.js';

describe('my-source', () => {
  it('normalizes order payload', () => {
    const raw = { /* sample API response */ };
    const events = mySource.parseWebhook(JSON.stringify(raw), {});
    expect(events).toHaveLength(1);
    expect(events[0].eventType).toBe('order.created');
  });
});

Use golden-file recorded fixtures for deterministic tests:

• Store sample API responses in packages/connectors/src/__fixtures__/
• Assert normalized output matches expected snapshots

Dynamic definitions

Tool: pullpush.connector.definition.validate
Input: { definition: { ... } }

Tool: pullpush.connector.definition.test
Input: { definitionId: "...", connectionId: "..." }

Transform DSL

Connectors produce raw events; the transform DSL (defined in @pullpush/core) maps source payloads to destination shape. The DSL is shared between the runtime executor and the AI mapping prompt.

Key operations: map, rename, cast, default, template, lookup, flatten, filter.

Use pullpush.mapping.suggest to AI-generate a mapping from sample data, or pullpush.flow.dryRun to test a mapping against real events without delivery.

Dynamic-connector DSL (canonical)

The DynamicConnector interprets a per-entity JSON definition with: canonical_map (raw↔canonical field mapping + an extensions passthrough bag), body_template (non-REST envelopes, e.g. JSON-RPC), a declarative snapshot (paginated canonical scan), lookup (cross-API join: preload_all, on_miss skip|error|passthrough), pagination, auth, and rate_limiting (requests_per_second, per-endpoint endpoints, and host_budget {requests_per_second, burst, max_wait_seconds}: one budget shared by every connection and flow that calls the base-url host, Redis-backed across workers; each origin in metadata.allowed_origins gets a bucket of its own on its host, from origin_budgets["https://host"] (same shape, keyed by a listed origin) or else the host_budget numbers, PP-846). Build/extend definitions with pullpush.connector.compose (canonical-aware) or pullpush.connector.define, then validate/activate. Prefer a DSL definition; drop to a custom TS connector only when logic is irreducible.

Publishing to the marketplace

Once a connector definition is active and working, you can share it with other tenants:

Tool: pullpush.marketplace.publish
Input:
  definitionId: "clxyz..."
  category: "ecommerce"         # crm, ecommerce, accounting, messaging, logistics, hr, custom
  tags: ["payments", "stripe"]
  description: "Stripe payment integration with order and refund sync"

This creates a marketplace listing that other tenants can browse and install. The definition is cloned into the installing tenant's scope — they get their own versioned copy.

Other tenants install via pullpush.marketplace.install and can update to newer published versions via pullpush.marketplace.update.

Best practices

1. Always implement test() — it's used by connection.test and the health-check job
2. Return retryable vs non-retryable errors — the pipeline uses this to decide retry vs dead-letter
3. Use incremental cursors — don't re-fetch all data on every poll
4. Handle rate limits — return { retryable: true } on 429; BullMQ handles backoff
5. Normalize event types — use dot notation (order.created, product.updated)
6. Keep config in the connection — never hardcode credentials in connector code
7. Prefer pullpush.connector.compose — canonical-aware DSL generation; drop to code only when logic is irreducible
8. Publish to marketplace — if your connector is reusable, publish it so other tenants benefit