Extend connectors and steps

Custom TypeScript connectors and steps are build-time Envoy extensions, not runtime plugins. Add them to the product source, test them, and produce a new customer build.

Before adding code, check whether an existing connector plus call_method, JSONata, and common steps already express the workflow.

Choose the extension level

Use:

• an existing connector method for a new task workflow;
• a deployment-catalog connector for a SaaS API described by catalog metadata;
• a compiled connector for a new protocol, driver, streaming source, or specialized capability;
• a compiled step for reusable pipeline behavior that cannot be composed cleanly from existing steps.

Prefer connector methods for I/O. Steps should orchestrate item-level pipeline behavior rather than hide external credentials or create a second connection system.

Implement a connector

Extend BaseConnector and define static metadata:

import {
  BaseConnector,
  type CallMethodOptions,
  type ConnectorConfig,
  type FieldSchema,
  type MethodSchema,
} from './base-connector'
 
export default class ExampleConnector extends BaseConnector {
  static type = 'example'
  static label = 'Example'
 
  static configSchema: Record<string, FieldSchema> = {
    base_url: { type: 'string', required: true },
    token: { type: 'string', required: true, secret: true },
  }
 
  static methodSchemas: Record<string, MethodSchema> = {
    get: {
      readonly: true,
      args: { id: { type: 'string', required: true } },
      returns: 'object',
      returnsShape: '{ id: string; name: string }',
    },
  }
 
  async init(): Promise<void> {}
  async destroy(): Promise<void> {}
 
  protected async executeMethod(
    method: string,
    args: Record<string, unknown>,
    options?: CallMethodOptions,
  ): Promise<unknown> {
    this.assertMethod(method, ['get'])
    // Perform the request and pass options?.signal to abortable I/O.
  }
}

The static schemas power the UI, validation, Data Explorer, dashboard safety, and AI connector discovery. Mark a method readonly: true only when it cannot mutate upstream state.

Connector lifecycle

init() acquires or validates shared resources. destroy() closes clients and pools. executeMethod() handles a named call.

Use:

• this.logger for structured connector logs;
• this.httpError(...) for enriched non-success HTTP errors;
• this.runCheck(...) when overriding testConnection();
• options.signal for cancellable network and driver calls;
• inherited connector metrics instead of manually duplicating call counters.

Do not log full request headers, credential-bearing URLs, or secret config.

Stream collections

Collection methods should return an AsyncIterable or async generator and yield one item at a time:

private async *listRecords(signal?: AbortSignal) {
  let cursor: string | undefined
 
  do {
    const page = await this.fetchPage({ cursor, signal })
    for (const item of page.items) yield item
    cursor = page.next_cursor
  } while (cursor)
}

Do not collect an unbounded cursor into an array. The call_method step recognizes iterable results and preserves stream backpressure.

For UI discovery, expose one bounded page and a cursor rather than draining every page.

Register a connector

Import and register it in the runtime entry used by the compiled build:

import ExampleConnector from './connectors/example'
import { registerConnector } from './connectors'
 
registerConnector('example', ExampleConnector)

Keep the registration key and static type aligned. Ensure registration occurs in server, CLI, executor subprocess, and validator entry points; the shared register-all.ts path is intended to prevent drift.

Then update Connectors reference with public config, method, return, pagination, and safety behavior.

Optional connector capabilities

A connector can expose additional typed capabilities, including file upload, download, and metadata operations. Reuse the shared file-storage guards:

• accepted extension checks;
• safe relative-path resolution;
• overwrite protection;
• server upload-size limit.

Never concatenate an untrusted upload name directly onto a local base directory.

Mapping-store connectors can override getMapping() and setMapping() for source-to-destination ID and hash storage.

Implement a step

Extend BaseStep and implement only execute():

import { BaseStep, type StepOptions } from './base-step'
import type { StepContext } from './types'
 
type NormalizeConfig = { trim?: boolean }
 
export class NormalizeStep extends BaseStep {
  constructor(
    config: unknown,
    context: StepContext,
    name: string,
    options?: StepOptions,
  ) {
    super(config, context, name, options)
  }
 
  async execute(data: unknown, config: unknown): Promise<unknown> {
    const { trim = true } = config as NormalizeConfig
    if (typeof data !== 'string') return data
    return trim ? data.trim() : data
  }
}

BaseStep provides envelope context, config evaluation, conditions, error handling, concurrency, backpressure, batching, spooling, debug output, and metrics.

Step return contract

The return from execute() has exact streaming meaning:

• undefined or [] drops the item;
• an array fans out into individual downstream items;
• any other value emits one item.

Two earlier branches differ:

• when run_if or skip_if skips a step, input passes through;
• when evaluated config is null or undefined, input passes through.

A transform expression that evaluates to undefined therefore passes its input through before execute() is called. Use the built-in filter step to remove records explicitly.

See Task structure and execution model.

Register and validate a step

Step construction currently uses the createStep() dispatch in src/steps/index.ts. Add the new type there and export it. If the step has a structured config, add static validator support so errors appear before runtime.

Also update:

• TypeScript step unions as appropriate;
• Step types reference;
• task-validator tests;
• stream, error, and cancellation tests;
• build registration when the step depends on a new connector or backend.

There is no separate dynamic step registry to update.

Test an extension

Use Bun tests and cover:

bun test src/__tests__/example-connector.test.ts
bun test

Connector tests should cover:

• required config and secret redaction;
• supported and unknown methods;
• read-only classification;
• pagination and backpressure;
• upstream status and body errors;
• cancellation;
• initialization and cleanup;
• test-connection checks.

Step tests should cover:

• scalar, array, empty, and undefined returns;
• run and skip conditions;
• fail, ignore, and ignore_rest;
• concurrent ordering assumptions;
• batch and spool behavior;
• upstream and downstream backpressure.

Finally, compile the customer build and execute a real bounded task through both direct CLI and remote executor paths.