# Extend connectors and steps

> Source: https://docs.clonepartner.com/extending/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:

```typescript
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:

```typescript
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:

```typescript
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](/reference/connectors) 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()`:

```typescript
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](/reference/tasks/task-structure).

## 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](/reference/tasks/step-types);
- 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:

```bash
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.