Add a connector

Canonical reference: This guide describes the extensibility contract as it is actually implemented in the repository. Read it in full before creating a single file.

Eight traps, each one measured

Every item below is a place where an earlier version of this guide described something the repository does not do. Each carries the command that re-establishes the fact and the gate that catches it — or the honest admission that nothing does.

#	The trap	What happens if you ignore it	Proof / gate
1	The manifest is `"1.2"` with `source_capabilities`, never `"1"`	Layer 1 fails immediately	`grep -h '"schema_version"' server/modules/*/manifest.json \| sort \| uniq -c` → 39 × `"1.2"` (measured 2026-08-03). Gate: `test_all_module_capabilities.py`
2	The worker does not call `pull` by convention — it resolves `source_capabilities.reports[].dispatch.callable` and fails closed	The connector is installed, cataloged, green — and silent: no pull ever leaves	`core/main.py` `get_module_pull_fn`. Gate: `server/tests/conformance/test_pull_contract.py`
3	`pull` has a closed signature: `(connection_id, date_from, date_to, project_id, pull_id)`	`TypeError: pull() missing 1 required positional argument: 'site_url'` on the first real call — this happened to `gsc` on 2026-07-31. A reporting account is resolved by `account_topology`; if the connector needs it, it takes an optional kwarg whose name is declared in `account_topology.pull_parameter`	`test_pull_contract.py`; the worker side is `core/queue.py:712`
4	Raw rows are not landed with a bare `duckdb.connect`	A bare connection writes into schema `main` — i.e. into everyone’s raw zone once `TOOROW_ORG_SCHEMAS=1`	37 of 39 connectors go through `core.warehouse_write.open_raw_writer` (measured 2026-08-03). No gate: this is a review item
5	Only `core.raw_landing` honours `TOOROW_DB_MODE=bigquery`	A hand-rolled `_insert_raw_rows` raises `unsupported db_mode`: the mode is accepted by config and silently not kept	`core/raw_landing.py` + the manifest’s `raw_landing` block (reference: `server/modules/bigquery/`)
6	Never a `RuntimeError` on a non-200	The worker cannot tell `auth_expired` from a 400: no reconnection offered, no breaker, no provider evidence kept	`classify_http_error(status, body, error_map)` — `core/pull_errors.py`
7	A Google connector must be declared in the consent screen	Measured 2026-07-31: 10 connectors declared `google_direct`, 7 scopes were requested. Three connectors were installed, cataloged, tested — and unreachable in production	Manifest `auth` block. Gate: `server/tests/conformance/test_google_consent.py`
8	Without `seeds/load_.py`, the staging model has never produced a row*	`dbt run` without `--select` breaks on `stg_<name>_daily`, and the connector counts as covered while its warehouse half was never exercised	Reference: `server/modules/brevo/seeds/load_brevo_seed.py`

There are six conformance layers, not four (see the table at the end of this page).

Overview

What You Get for Free (The Shared Base)

Any connector that respects the contract inherits the following mechanisms without writing a single line in server/core/:

Mechanism	Core file	What you get
Journaling	`core/datastreams.py`, `core/audit.py`	Append-only ULID `pull_id` — a run’s provenance is written once and never rewritten — plus `audit_log` rows and a daily ledger (ok/partial/failed/missing statuses).
Queue / Retry / Quota	`core/queue.py`, `core/quota.py`	Worker with retry, dead-letter, deduplication, per-provider circuit-breaker; `RateLimitError` automatically handled.
Dictionary / Mappings	`app.target_fields`, `app.datastream_mappings`	Auto-seeded from the manifest’s `canonical_metric_mapping` + `canonical_dimension_mapping` on the first `backfill_datastreams()`.
Quality Monitors	`core/dq_monitors.py`	10 universal monitors — see Data Quality.
MCP Flows	`core/flows.py`	`flows_list/get/upsert/validate` — the agent and the admin UI edit your datastreams through the exact same interface.
Admin Screens	`ui/`	Overview, Data model (used-by), extraction calendar, source→target mapping, Quality page.

What You Must Write (The Cost of a New Connector)

server/modules/<kebab-name>/
  manifest.json                      # Required (schema version 1.2, with source_capabilities)
  connector.py                       # Required (pull + transform functions)
  api_catalog.json                   # Required — GENERATED by scripts/build_api_catalog.py
  catalog_sources/                   # Required — official + enrichment sources, pinned api_version
  dbt/staging/stg_<name>_daily.sql   # Recommended: QUALIFY keeps the latest pull per key
  dbt/staging/schema.yml             # Recommended (grain_unique test)
  seeds/load_<name>_seed.py          # Required if there is a staging model (trap 8)
  reports/<id>.json                  # ≥ 1 report pack
  tests/fixtures/golden_pull.json    # Required (Layer 4 conformance)
  tests/fixtures/expected_facts.json # Required (Layer 4 conformance)

No file inside server/core/ may be modified: a connector is added by its own folder and its manifest, never by editing the platform. A new daily connector is registered in two places, and skipping the second one produces rows nothing reads:

dbt/dbt_project.yml → model-paths: add ../server/modules/<name>/dbt, then dbt parse.
dbt/models/marts/fact_daily_kpi.sql: add the connector’s stg_<name>_daily to the UNION.

Choosing the Landing Kind

The module_kind field in manifest.json declares the structure of rows produced by the module:

Kind	Row Structure	Typical Examples
`kpi`	`(project_id, date, metric, value, breakdown_dimension, breakdown_value, pull_id, loaded_at)` — long format	GA4, Meta Ads, TikTok, LinkedIn, Shopify, Stripe, Brevo.
`context`	`(project_id, date, event_type, event_payload)` — point-in-time events	GitHub releases, CI/CD deployments, CRM milestones, campaign launches.
`generic`	Arbitrary tabular data mapped via `datastream_mappings`	CSV exports, generic webhooks.

Walkthrough: read a shipped connector, then copy its shape

The fastest way to write a connector is to open one that works. server/modules/brevo/ is complete and passes every conformance layer — manifest 1.2, connector.py, dbt staging model, seed loader, fixtures, generated catalog. Every excerpt below is copied from it.

server/modules/README.md carries the Connector Standard Playbook — the exhaustive field catalog, the doc↔code contract, account topology and the error taxonomy. Read it alongside this page: it is what separates a connector that runs from one that can be operated. server/modules/meta-ads/ is its most complete implementation.

Step 1 — What lives on disk

server/modules/brevo/
  manifest.json                    # contract: auth, profiles, mappings, dispatch
  connector.py                     # pull + transform
  api_catalog.json                 # GENERATED — scripts/build_api_catalog.py
  catalog_sources/                 # official + enrichment sources, pinned api_version
  dbt/staging/stg_brevo_daily.sql  # QUALIFY supersede
  dbt/staging/schema.yml           # grain_unique test
  seeds/load_brevo_seed.py         # proves the warehouse half offline
  tests/fixtures/golden_pull.json
  tests/fixtures/expected_facts.json

No file under server/core/ may be touched.

Step 2 — The manifest is the contract

The header declares who you are and how you authenticate:

{
  "schema_version": "1.2",
  "name": "brevo",
  "display_name": "Brevo",
  "auth_type": "oauth2",
  "module_kind": "kpi",
  "provider_api_version": "v3",
  "provider_api_base": "https://api.brevo.com/v3"
}

source_capabilities.reports[] declares what can actually be asked for — the metrics, the dimensions, the supported grains, and the function the worker will call:

{
  "id": "email_campaign_daily",
  "selection_mode": "exact_bundle",
  "metrics": ["sent", "delivered", "opens", "clicks", "hard_bounces", "unsubscribed"],
  "dimensions": ["date", "account_id", "source_id", "channel"],
  "supported_grains": [["date", "account_id", "source_id", "channel"]],
  "filters": [{ "field_id": "date", "operators": ["gte", "lte"] }]
}

canonical_metric_mapping translates the provider’s vocabulary into the platform’s, so the same figure is comparable across sources:

{ "sent": "messages_sent", "delivered": "messages_delivered", "opens": "opens", "clicks": "clicks" }

The worker resolves source_capabilities.reports[].dispatch.callable and fails closed if that function does not exist. A connector can be installed, cataloged and green while no pull ever leaves — this is trap 2 above, and test_pull_contract.py is what catches it.

Step 3 — `connector.py`: a closed signature

def pull(connection_id, date_from, date_to, project_id, pull_id, selection=None):
    ...

Those five positional parameters are the whole contract. Anything a connector needs beyond them — a reporting account, a site URL — arrives as an optional keyword argument whose name is declared in account_topology.pull_parameter. Make it required and the first real call dies with TypeError, which is exactly what happened to gsc on 2026-07-31. Two more rules the shared base enforces:

land rows through core.warehouse_write.open_raw_writer, never a bare duckdb.connect — a bare connection writes into everyone’s raw zone;
raise through classify_http_error(status, body, error_map) from core/pull_errors.py. A bare RuntimeError on a non-200 means the worker cannot tell an expired token from a bad request: no reconnection offered, no breaker, no evidence kept.

Step 4 — The staging model deduplicates by pull

Re-running an extraction must not double a figure. The staging model keeps the newest pull per grain, and that is the whole of it:

{{ config(materialized='view') }}
SELECT * FROM {{ source('raw_brevo', 'raw_brevo_daily') }}
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY project_id, profile, account_id, source_id, date, channel,
    event_type, protected_identifier, metric
  ORDER BY pull_id DESC
) = 1

dbt/staging/schema.yml carries the grain_unique test that fails if that partition ever lets a duplicate through.

Step 5 — Register in two places

Skipping the second produces rows nothing reads:

dbt/dbt_project.yml → model-paths: add ../server/modules/<name>/dbt, then dbt parse.
dbt/models/marts/fact_daily_kpi.sql: add stg_<name>_daily to the UNION.

Step 6 — Prove it offline

seeds/load_<name>_seed.py loads a fixture into the local warehouse so the staging model runs without a single API call. Without it, dbt run breaks on your model and the connector counts as covered while its warehouse half was never exercised.

uv run python server/modules/brevo/seeds/load_brevo_seed.py
cd dbt && dbt build --select stg_brevo_daily

tests/fixtures/golden_pull.json and expected_facts.json are the recorded provider response and the facts it must produce. The conformance suite replays one into the other, so a change in your transform that alters a figure fails before it reaches a warehouse.

cd server && uv run python -m pytest tests/conformance -q

Self-Hosted Deployment

Self-Hosted Engineering & Connectors

Add a connector

Eight traps, each one measured

Overview

What You Get for Free (The Shared Base)

What You Must Write (The Cost of a New Connector)

Choosing the Landing Kind

Walkthrough: read a shipped connector, then copy its shape

Step 1 — What lives on disk

Step 2 — The manifest is the contract

Step 3 — `connector.py`: a closed signature

Step 4 — The staging model deduplicates by pull

Step 5 — Register in two places

Step 6 — Prove it offline

Final Conformance Checklist

Next Steps & Cross-References

Universal Datastreams

Semantic Layer

Platform Constraints

Quickstart Guide

​Eight traps, each one measured

​Overview

​What You Get for Free (The Shared Base)

​What You Must Write (The Cost of a New Connector)

​Choosing the Landing Kind

​Walkthrough: read a shipped connector, then copy its shape

​Step 1 — What lives on disk

​Step 2 — The manifest is the contract

​Step 3 — connector.py: a closed signature

​Step 4 — The staging model deduplicates by pull

​Step 5 — Register in two places

​Step 6 — Prove it offline

​Final Conformance Checklist

​Next Steps & Cross-References

Universal Datastreams

Semantic Layer

Platform Constraints

Quickstart Guide

Eight traps, each one measured

Overview

What You Get for Free (The Shared Base)

What You Must Write (The Cost of a New Connector)

Choosing the Landing Kind

Walkthrough: read a shipped connector, then copy its shape

Step 1 — What lives on disk

Step 2 — The manifest is the contract

Step 3 — `connector.py`: a closed signature

Step 4 — The staging model deduplicates by pull

Step 5 — Register in two places

Step 6 — Prove it offline

Final Conformance Checklist

Next Steps & Cross-References