Fivetran ships hundreds of connectors, and sooner or later you need one it does not have: an internal API, a vendor with an obscure REST interface, a partner SFTP drop. The supported way to build that today is the Fivetran Connector SDK, a Python library that lets you write the extraction logic while Fivetran runs it on its own infrastructure, schedules it, and loads the output into your destination like any other connector.
It matters which path you choose because the alternative is going away. The legacy Function connectors (AWS Lambda, Azure Functions, Google Cloud Functions) are unavailable to accounts created on or after July 22, 2025, and existing Function connections should be treated as end-of-life. The SDK has none of Lambda's timeout or payload limits and there is no serverless infrastructure to own.
This tutorial builds a small incremental connector end to end. The official references are the Connector SDK docs and the examples repository; the API names below match those as of August 2026.
1. Install
You need Python 3.10 or newer. In a fresh virtual environment:
python -m venv .venv && source .venv/bin/activate
pip install fivetran-connector-sdk requests
The package installs the fivetran command-line tool, which you will use to debug and deploy.
2. Anatomy of a connector
A connector is a single connector.py that defines two functions and hands them to a Connector object:
schema(configuration)declares the tables, primary keys, and (optionally) column types you will deliver.update(configuration, state)is called on every sync. It reads from the source, emits rows withop.upsert(orop.update/op.delete), and saves progress withop.checkpoint.
configuration is a dictionary of strings that your connection's users fill in (API keys, base URLs). state is whatever you checkpointed last time, or empty on the first sync and on a full re-sync.
3. A working incremental connector
The example below syncs an orders endpoint that supports an updated_since filter and page tokens. Replace the endpoint details with your source's.
import json
import requests
from fivetran_connector_sdk import Connector
from fivetran_connector_sdk import Logging as log
from fivetran_connector_sdk import Operations as op
def schema(configuration: dict):
return [
{
"table": "orders",
"primary_key": ["id"],
"columns": {
"id": "STRING",
"customer_id": "STRING",
"total_cents": "INT",
"status": "STRING",
"updated_at": "UTC_DATETIME",
},
}
]
def fetch_page(base_url, api_key, updated_since, page_token):
params = {"updated_since": updated_since, "limit": 500}
if page_token:
params["page_token"] = page_token
resp = requests.get(
f"{base_url}/v1/orders",
headers={"Authorization": f"Bearer {api_key}"},
params=params,
timeout=60,
)
resp.raise_for_status()
return resp.json()
def update(configuration: dict, state: dict):
base_url = configuration["base_url"]
api_key = configuration["api_key"]
# Cursor from the last checkpoint; a fixed floor for the first sync.
cursor = state.get("updated_since", "2020-01-01T00:00:00Z")
page_token = state.get("page_token")
newest_seen = cursor
log.info(f"orders: syncing from {cursor}")
while True:
page = fetch_page(base_url, api_key, cursor, page_token)
for order in page["data"]:
op.upsert(table="orders", data={
"id": order["id"],
"customer_id": order["customer"]["id"],
"total_cents": order["total_cents"],
"status": order["status"],
"updated_at": order["updated_at"],
})
if order["updated_at"] > newest_seen:
newest_seen = order["updated_at"]
page_token = page.get("next_page_token")
# Checkpoint after every page so a long sync can resume mid-way.
op.checkpoint(state={"updated_since": cursor, "page_token": page_token})
if not page_token:
break
# All pages done: advance the cursor and clear the page token.
op.checkpoint(state={"updated_since": newest_seen, "page_token": None})
connector = Connector(update=update, schema=schema)
if __name__ == "__main__":
with open("configuration.json") as f:
connector.debug(configuration=json.load(f))
Two files sit next to it:
configuration.json (local only, never committed):
{ "base_url": "https://api.example.com", "api_key": "sk_live_..." }
requirements.txt, listing anything beyond the SDK itself:
requests
A few things worth noticing in the code:
- Upserts are idempotent. Re-delivering a row with the same primary key updates it rather than duplicating it, so a sync that fails halfway and resumes is safe.
- Checkpoint inside the loop. The state you checkpoint is what the next run receives. Checkpointing per page means a 10-million-row backfill that dies at hour three resumes from page N rather than page one. This is the single biggest difference between a toy connector and a production one.
- Advance the cursor only at the end. Until every page for a given
updated_sinceis delivered, the cursor stays put. Otherwise a crash between pages silently drops rows. - Do not set
updated_sincefrom the clock. Use the newest value you actually saw, so source-side lag cannot create a gap.
4. Run it locally
fivetran debug --configuration configuration.json
fivetran debug runs your update() against a local DuckDB warehouse (files/warehouse.db) and prints the operations it emitted. Open the DuckDB file to inspect what landed. Run it twice: the second run should deliver only rows changed since the first, which proves your checkpointing works. You can also run python connector.py directly, which is what the __main__ block is for.
5. Deploy and schedule
Generate an API key in the Fivetran dashboard, then:
fivetran deploy --api-key <base64-api-key> \
--destination <destination-name> \
--connection orders_api \
--configuration configuration.json
The connection appears in your dashboard like any other. Set its sync frequency there, and from then on Fivetran runs your code on its schedule, stores state between runs, and applies the normal destination behaviour (schema drift handling, type mapping, history where enabled). Redeploy with the same command after code changes.
6. Production habits
- Log with
Logging, notprint.log.info,log.warning, andlog.severeshow up in the Fivetran dashboard and the Platform Connector's log table. - Respect source rate limits. Sleep on HTTP 429 and back off; an SDK connector that hammers a vendor API can get your whole company's key throttled.
- Declare types for anything ambiguous. Leave types unspecified and Fivetran infers them, which is fine for strings and integers and occasionally wrong for timestamps and decimals.
- Handle deletes explicitly. If the source exposes deleted records, emit
op.delete; if it does not, consider a periodic full re-sync of small tables or a soft-delete flag. - Keep secrets in
configuration. Never in code, never in the repository.
7. Migrating a Function (Lambda) connector
If you have a Function connector, port it before it becomes a problem. The mapping is mechanical:
| Function connector | Connector SDK |
|---|---|
Handler receives { state, secrets } | update(configuration, state) |
Returns insert / delete dicts per table | op.upsert / op.delete calls |
Returns state in the response | op.checkpoint(state=...) |
Returns schema with primary keys | schema() |
hasMore pagination flag | a loop with per-page checkpoints |
Most Lambda handlers were written to fit inside a 15-minute timeout and a response size cap, so they page awkwardly and keep elaborate state. In the SDK you can usually simplify: one long-running update() with checkpoints. Deploy the SDK version as a new connection, let both run side by side for a cycle, diff the destination tables, then delete the Function connection.
Where to go next
The examples repository has quickstarts for configuration forms, large datasets, pandas DataFrames, and multi-file projects, plus patterns like schema changes and parent-child endpoints. If you would rather have a connector built and maintained for you, see our connectors configuration and management service.