Building Custom Connectors with AI Agents

Invoke build-connector with context about your data source in the prompt. The agent checks whether Fivetran already has a managed or Lite Connector for your source before proceeding to custom code generation. If none exists, it researches your source, scaffolds the project, and runs a local test.

/fivetran-connector-sdk:build-connector

Build a connector for the Example API. The base URL is https://api.example.com/v2.
Auth uses an API key passed in the header as X-API-Key.
Replicate the users and events endpoints with incremental sync.

Sample response (GET /users):
{"data": [{"id": "u1", "email": "user@example.com", "created_at": "2024-01-01T00:00:00Z"}], "next_cursor": "abc123"}

If Fivetran already has a managed connector for your source, the agent tells you and offers to stop. Tell it to proceed if you need a custom connector.

Before the local test runs, the agent asks you to enter credentials in a separate terminal. Run the command it provides. Credentials are never passed through the agent chat.

Extending an existing connector requires rebuilding it from scratch. Describe in plain language the customization you want:

Add a new table for the /reports endpoint. Primary key is id, incremental cursor is created_at.
Sample response: {"id": "r1", "name": "Q1 Report", "created_at": "2024-01-01", "status": "published"}

Describe the error or unexpected behavior in natural language:

The connector runs without errors but the events table only has 3 rows.
It should have hundreds. Here is the debug output: [paste output]

The agent classifies the error into one of three types:

For issues that persist after the agent's fix attempt, see the Connector SDK Troubleshooting guide.

We recommend running evaluate-connector before deploying to production. The skill performs a read-only static analysis of your connector code and does not modify any files.

/fivetran-connector-sdk:evaluate-connector

The report groups issues into two categories:

• required — fix these before deploying:

  Category	Examples
  Memory	Loading an entire dataset into memory before processing; unclosed file handles or HTTP connections
  SDK compliance	Missing op.checkpoint(); using yield op.upsert() instead of calling it directly; update() has a return statement; connector = Connector(...) not at module scope; using print() or logging.* instead of the SDK logger
  Security	Credentials written to state or logs; hardcoded secrets in code
  Data reliability	Missing raise_for_status() after API calls; cursor updated before op.upsert() (it must be updated after); no pagination handling
  Exception handling	Bare except: pass blocks; silently caught exceptions; missing error handling around I/O operations

• good-to-have — recommended improvements:

  Category	Examples
  Performance	No timeout on HTTP requests; no retry logic for transient failures
  Code quality	requests or fivetran_connector_sdk listed in requirements.txt (both are pre-installed in the SDK runtime environment); all columns explicitly typed (prevents Fivetran from inferring types automatically when new columns appear)
  Reliability	Retry logic without exponential backoff; string timestamp comparison without datetime parsing

The report produces four scores, each starting at 100: overall, SDK adherence, required, and good-to-have. required issues deduct 10–30 points each; good-to-have issues deduct 1–5. Fix all required issues before running deploy-connector.

The plugin includes three migration skills. Each one reads your source connector, maps its concepts to Connector SDK equivalents, generates new connector.py, configuration.json, requirements.txt, and README.md files, and runs a local test.

Migration skills are not deterministic converters. Read the generated code and compare it against the original logic, particularly for authentication, pagination, and state handling, before running the connector against your data.

/fivetran-connector-sdk:migrate-functions-connector

Migrate my Lambda connector at ./my-lambda-connector to Connector SDK.

/fivetran-connector-sdk:migrate-meltano-connector

Migrate my Meltano extractor at ./tap-example to Connector SDK.

/fivetran-connector-sdk:migrate-airbyte-connector

Migrate my Airbyte source connector at ./source-example to Connector SDK.

Before deploying, ensure:

• At least one destination is configured in your Fivetran account
• README.md exists in your connector directory alongside connector.py, configuration.json, and requirements.txt

/fivetran-connector-sdk:deploy-connector

The agent validates the connector, runs a final local test, and deploys to your Fivetran account. The connection name defaults to your connector directory name. The new connection starts paused. The agent asks before starting the initial sync, as it consumes MAR.

To redeploy an updated connector, specify the existing connection name:

/fivetran-connector-sdk:deploy-connector

Deploy to the same connection as before. The connection name is my_connector.

Start from an OpenAPI spec:

/fivetran-connector-sdk:build-connector

Build a connector for <api_name> using the OpenAPI spec at <openapi_spec_url>.
Replicate the <resource_1> and <resource_2> endpoints.
Primary key for <resource_1> is <primary_key_field>. Primary key for <resource_2> is <primary_key_field>.
Use incremental sync with <cursor_field> as the cursor.

OAuth 2.0 client credentials:

/fivetran-connector-sdk:build-connector

Build a connector for <api_name>. Auth uses the OAuth 2.0 client credentials grant.
Don't use refresh_token. Write custom token acquisition code.
POST to <token_endpoint_url> with <client_id>, <client_secret>, and <scope> as a space-separated string.
Response includes <access_token> and <expires_in>.
Use <expires_in> for proactive token refresh.
Replicate:
- GET [endpoint_path_1] — primary key: [field], incremental sync on [cursor_field], [pagination_type] pagination
- GET [endpoint_path_2] — primary key: [field], full refresh

Rate limit errors:

The connector gets 429 errors after the first few requests.
The API allows [N] requests per minute. Add retry logic with exponential backoff.
Here is the debug output: [paste debug output]

Pagination not working:

The connector only returns the first page of results.
The API uses [pagination type: cursor | offset | page_token] pagination.
Here is the relevant response structure: [paste response structure]
Here is the debug output: [paste debug output]

Add a missing column:

The [table] table is missing the [field] column. It is returned in the API
response but not currently included in the schema. Add it without changing
any existing column types.

Instead of specifying every column type:

// Don't do this
Generate the full schema with data types for all columns in the users table.

// Do this
Define the schema for the users table. Only declare the primary key ([primary_key_field]).
Let Fivetran infer the rest.

Specifying every column type removes Fivetran's ability to adapt automatically when the source schema changes and new columns appear.

Instead of pasting credentials in chat:

// Don't do this
Here is my API key: <api_key>. Build the connector.

// Do this
Build the connector. I'll enter credentials in a separate terminal when prompted.

The plugin enforces credential entry through a separate local process. Credentials never enter the agent conversation.