> ## Documentation Index > Fetch the complete documentation index at: https://docs.toorow.com/llms.txt > Use this file to discover all available pages before exploring further. # Add a connector > toorow's extensibility contract: manifest, connector.py, dbt staging, 4-layer conformance. > **Canonical reference**: this guide describes the extensibility contract as it is *actually* implemented. Read it **in full** before creating a single file. *** ## Overview ### What you get for free (the shared base) Any module that respects the contract inherits the following 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` (AD-7), `audit_log` rows, daily ledger (ok/partial/failed/missing statuses) | | **Queue / retry / quota** | `core/queue.py`, `core/quota.py` | Worker with retry, dead-letter, dedup, per-provider circuit-breaker; `RateLimitError` already 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/verification.py`, `app.monitors` | 4 universal monitors (volume, timeliness, duplication, column consistency), zero config | | **MCP flows** | `core/flows.py` | `flows_list/get/upsert/validate` — the agent and the admin UI edit your datastreams through the 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) ```text theme={null} server/modules// manifest.json # required connector.py # required (pull + transform) dbt/staging/stg_.sql # recommended (QUALIFY supersede AD-7) dbt/staging/schema.yml # recommended reports/.json # >= 1 report pack tests/fixtures/golden_pull.json # required (Layer 4) tests/fixtures/expected_facts.json # required (Layer 4) ``` Nothing in `server/core/` may be modified (AD-2). *** ## Choosing the landing kind The manifest's `module_kind` field declares the shape of the data the module produces. | kind | Row shape | Real examples | | --------- | ----------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | | `kpi` | `(project_id, date, metric, value, breakdown_dimension, breakdown_value, pull_id, loaded_at)` — long format | DV360 impressions/day, Google My Business views/day, Hotjar counts, Hootsuite metrics, Firebase events/day, CRM activity (deals created/day, revenue/day) | | `context` | `(project_id, date, event_type, event_payload)` — point-in-time events, no warehouse fact | GitHub releases, CI/CD deploys, CRM milestones, campaign launches | | `generic` | Arbitrary table mapped through `datastream_mappings` → kpi shape | CSV exports, webhooks, any tool without a dedicated module | **Decision table:** ```text theme={null} Do you have data aggregated by day and by measure (clicks, sessions, cost...)? → kpi Do you have point-in-time events (a deploy, a launch)? → context Do you have a tabular file whose metrics you don't know in advance, or a generic webhook? → generic (declare module_kind: "kpi" and let the mappings do the rest) ``` *** ## Step by step ### Step 1 — `manifest.json` **File:** `server/modules//manifest.json` **Reference schema:** `server/core/schemas/manifest.schema.json` **Excerpt adapted from gsc (reference module):** ```json theme={null} { "schema_version": "1", "name": "my-connector", "display_name": "My Connector", "auth_type": "oauth2", "module_kind": "kpi", "report_profiles": [ { "id": "daily", "display_name": "Daily report", "metrics": ["clicks", "impressions", "average_position"], "dimensions": ["date", "page"], "extraction_capabilities": { "row_limit": 25000, "filters_supported": true, "realtime": false }, "extraction_path": "custom_pull" } ], "canonical_metric_mapping": { "clicks": "clicks", "impressions": "impressions", "average_position": { "canonical": "average_position", "aggregation_rule": "impression_weighted", "non_additive": true } }, "canonical_dimension_mapping": { "page": "page", "country": "country" }, "quota": { "window_seconds": 100, "budget_points": 1200, "read_cost": 1, "write_cost": 1 }, "widget_ref": "ui://core/daily-report" } ``` **Key points:** * `name`: strict kebab-case `^[a-z0-9-]+$` (Layer 1 fails otherwise). * `auth_type`: `"oauth2"` | `"api_key"` | `"none"`. * `module_kind`: `"kpi"` | `"context"` (absent = `"kpi"` by default). * `canonical_metric_mapping`: **additive** metrics are written `"source": "canonical"` (string); **non-additive** metrics use the object form with `aggregation_rule` and `non_additive: true` (e.g. `average_position` in gsc). Never sum a non-additive metric directly (AD-4). * `quota` is optional; its absence means unlimited budget (no circuit-breaker). * `widget_ref` is optional for `context` modules. **Verification command (Layer 1):** ```bash theme={null} uv run pytest server/tests/conformance/ --module-path server/modules// -m conformance_layer_1 -v ``` *** ### Step 2 — `connector.py`: `pull()` and `transform()` **File:** `server/modules//connector.py` Two required public functions. #### 2a. `pull()` — fetch and land ```python theme={null} """ connector. AD-2: module name never hardcoded in core/. AD-3: token never stored or logged. AD-7: pull_id minted by caller, never here. AD-4: non-additive metrics stored raw per row with their weight. """ import httpx from fastmcp import FastMCP mcp_app = FastMCP("my-connector") # namespace AD-2 def pull( connection_id: str, date_from: str, date_to: str, project_id: str, pull_id: str, # AD-7: minted by the core scheduler, passed in ) -> dict: """Fetch data and land rows into the raw DuckDB table. Returns {"pull_id": str, "row_count": int, "date_from": str, "date_to": str} """ from core import nango_client # AD-2: import at call time, not module level # AD-3: token used immediately, falls out of scope after the request. token = nango_client.get_fresh_token(connection_id, provider="my-connector") resp = httpx.get( "https://api.example.com/data", headers={"Authorization": f"Bearer {token}"}, params={"startDate": date_from, "endDate": date_to}, timeout=30.0, ) if resp.status_code == 429: from core.quota import RateLimitError raise RateLimitError("my-connector", retry_after=None) if resp.status_code != 200: raise RuntimeError(f"API error {resp.status_code}: {resp.text[:200]}") raw_rows = resp.json().get("rows") or [] canonical_rows = transform(raw_rows) row_count = _insert_raw_rows(canonical_rows, pull_id, project_id) return { "pull_id": pull_id, "row_count": row_count, "date_from": date_from, "date_to": date_to, } ``` **Row shape in the raw table (long-format kpi):** ```text theme={null} project_id | date | metric | value | breakdown_dimension | breakdown_value | pull_id | loaded_at ``` This is the shape `stg_.sql` will consume and that `fact_daily_kpi` aggregates via UNION. Never store metrics as wide columns in the raw table — long format is the contract. #### 2b. `transform()` — manifest-driven (AD-2) ```python theme={null} import json from pathlib import Path def transform(raw_rows: list[dict]) -> list[dict]: """Map source field names to canonical names using manifest mappings. AD-2: field names come from the manifest, never hardcoded here. AD-4: non-additive metrics stored raw; NO aggregation in transform(). """ _manifest = json.loads( (Path(__file__).parent / "manifest.json").read_text(encoding="utf-8") ) rename_map: dict[str, str] = {} for src, val in _manifest.get("canonical_metric_mapping", {}).items(): if isinstance(val, str): rename_map[src] = val elif isinstance(val, dict): rename_map[src] = val.get("canonical", src) rename_map.update(_manifest.get("canonical_dimension_mapping", {})) _DROP_FIELDS: set[str] = set() # add ratio fields that must never be stored result: list[dict] = [] for row in raw_rows: canonical: dict = {} for key, value in row.items(): if key in _DROP_FIELDS: continue canonical[rename_map.get(key, key)] = value result.append(canonical) return result ``` **AD-2 / AD-4 rules:** * `transform()` must never contain a hardcoded source field name: always use `rename_map` derived from the manifest. * Non-additive metrics (`non_additive: true`) are stored **raw per row** with their weight (e.g. `impressions`). Never aggregated here. * Ratio metrics (`ctr`, conversion rate...) are **never stored**. They are computed at the semantic layer. **Verification command (Layer 4):** ```bash theme={null} uv run pytest server/tests/conformance/ --module-path server/modules// -m conformance_layer_4 -v ``` *** ### Step 3 — dbt staging (module-owned) **Files:** * `server/modules//dbt/staging/stg__daily.sql` * `server/modules//dbt/staging/schema.yml` **`model-paths` convention:** dbt discovers these files via the `model-paths` key of `dbt_project.yml`. The chosen option (external model-path) keeps the models inside the module folder, not in `dbt/models/`. **Adapted from the gsc staging (reference):** ```sql theme={null} -- Staging: maps raw source fields to canonical names -- AD-7: QUALIFY supersede — the latest pull per grain wins. -- ULIDs are lexicographically monotonic, ORDER BY pull_id DESC = most recent. -- AD-4: non-additive metrics stored raw with their weight. {{ config(materialized='view') }} SELECT date, page, country, clicks, impressions, average_position, -- NON-ADDITIVE: never SUM directly (AD-4) pull_id, loaded_at, project_id FROM {{ source('raw_my_connector', 'raw_my_connector_daily') }} QUALIFY ROW_NUMBER() OVER ( PARTITION BY project_id, date, page, country ORDER BY pull_id DESC ) = 1 ``` **`schema.yml` (grain-unique test required, AI-05):** ```yaml theme={null} version: 2 sources: - name: raw_my_connector schema: main tables: - name: raw_my_connector_daily description: "Immutable raw landing (AD-7: append-only)." models: - name: stg_my_connector_daily description: "Staging layer with QUALIFY supersede (AD-7). AD-4: non-additive metrics raw." columns: - name: date tests: [not_null] - name: clicks tests: [not_null] - name: impressions tests: [not_null] - name: pull_id tests: [not_null] tests: - unique: name: stg_my_connector_daily_grain_unique arguments: column_name: "project_id || '|' || date || '|' || page || '|' || country" ``` **The `grain_unique` test is MANDATORY** (AI-05: its absence is a code-review failure). *** ### Step 4 — `reports/*.json` **File:** `server/modules//reports/.json` Each file is a report pack. Conformance (Layer 3b via `test_reports.py`) validates that the declared metrics and dimensions are known. **Full structure with R6 fields (`metric_definitions`, `llm_commentary_guidelines`):** ```json theme={null} { "id": "daily_summary", "display_name": "Daily summary", "metrics": ["clicks", "impressions", "average_position"], "dimensions": ["date", "page"], "layout": { "chart_type": "line", "order_by": "date", "top_n": 20 }, "narrative_prompt": "Analyze SEO trends over the period. Identify rising and declining pages. Limit to 150 chars.", "date_window": { "default_days": 30, "min_days": 7, "max_days": 90 }, "metric_definitions": { "clicks": { "definition": "Number of clicks from search results to the page.", "unit": "clicks", "direction": "up_good", "caveats": null }, "average_position": { "definition": "Average position in Google SERPs. NON-ADDITIVE: impression-weighted aggregation.", "unit": "rank (1 = best)", "direction": "down_good", "caveats": "Do not sum directly — use the semantic_avg_position view." } }, "llm_commentary_guidelines": "Focus on weekly variations. Mention pages with position delta > 2 ranks. Avoid generalities. Analytical tone, not promotional." } ``` **Reference schema for flow overrides:** `server/core/schemas/flow.report.schema.json` Base packs (in `reports/`) do not yet include `metric_definitions`; those fields are added through a flow override (`flows_upsert` — see Step 9). *** ### Step 5 — Conformance fixtures **Files:** ```text theme={null} server/modules//tests/fixtures/golden_pull.json # raw rows as they arrive from the API server/modules//tests/fixtures/expected_facts.json # rows after transform() ``` **`golden_pull.json`** — rows in the source API's raw format. Do not pre-process: ```json theme={null} [ { "pull_id": "pull_test_001", "connector": "my-connector", "date": "2026-07-01", "page": "https://example.com/blog/", "country": "fra", "clicks": 42, "impressions": 850, "position": 7.3, "ctr": 0.049 } ] ``` **`expected_facts.json`** — the exact result of `transform(golden_pull.json)`: * Fields renamed by `canonical_metric_mapping` / `canonical_dimension_mapping`. * DROP fields (ratios like `ctr`) are absent. * Pass-through fields (`pull_id`, `connector`, `date`, unmapped dimensions) stay unchanged. ```json theme={null} [ { "pull_id": "pull_test_001", "connector": "my-connector", "date": "2026-07-01", "page": "https://example.com/blog/", "country": "fra", "clicks": 42, "impressions": 850, "average_position": 7.3 } ] ``` **Critical rule:** never hand-edit a fixture to make a test pass. The fixture **is** the contract; regenerate it from the producing code (Schema Change Checklist, CONTRIBUTING.md). *** ### Step 6 — Conformance: the 4-layer suite **Full command (all layers):** ```bash theme={null} uv run pytest server/tests/conformance/ --module-path server/modules// -v ``` **Command validated on gsc (reference):** ```bash theme={null} uv run pytest server/tests/conformance/ --module-path server/modules/gsc/ -v # Expected result: 16 passed, 3 skipped (skips are normal for kpi modules) ``` **The 4 layers (execution order):** | # | Marker | File | What it checks | | - | --------------------- | --------------------- | ------------------------------------------------------------------------------------------ | | 1 | `conformance_layer_1` | `test_manifest.py` | `manifest.json` valid against `manifest.schema.json`; `name` kebab, `auth_type` enum, etc. | | 2 | `conformance_layer_2` | `test_envelope.py` | Each MCP tool returns the canonical AD-1 envelope; summary ≤ 30 lines | | 3 | `conformance_layer_3` | `test_bundle.py` | The widget artifact exists and contains no external `http(s)://` reference (AD-11/NFR4) | | 4 | `conformance_layer_4` | `test_golden_pull.py` | `transform(golden_pull.json)` matches `expected_facts.json` field by field | Layer 1 failing = fail-fast (layers 2-4 are skipped). **Error format:** ```text theme={null} [manifest] $.name: must match pattern ^[a-z0-9-]+$, got "MyConnector" [golden_pull] row 0 field average_position: expected 7.3, got null ``` **All 4 layers must pass before the module is considered conformant.** *** ### Step 7 — Live-integration pass (AI-13) Any module that introduces a **new** external API requires a live-integration pass before it can be marked done (AI-13 DoD). **Phase 1 — Mocked tests with respx:** ```python theme={null} import respx import httpx import pytest from server.modules.my_connector.connector import pull @pytest.mark.respx(base_url="https://api.example.com") def test_pull_mocked(respx_mock): respx_mock.get("/data").mock( return_value=httpx.Response(200, json={"rows": [ {"date": "2026-07-01", "page": "/", "clicks": 10, "impressions": 100, "position": 5.0} ]}) ) result = pull( connection_id="test-conn", date_from="2026-07-01", date_to="2026-07-01", project_id="test", pull_id="pull_test_001", ) assert result["row_count"] == 1 ``` **Phase 2 — Live against the real service:** run a real pull, verify the received columns, error codes and date format. Record the evidence in the story's Dev Agent Record. Mocked tests encode assumptions; the live contract must be verified (Story 2.2 lesson: Nango contract violations — image tag, port, query params all differed from the mocks). *** ### Step 8 — Backfill into datastreams + mappings Once the module is in place, datastreams and mappings are created **automatically** by `backfill_datastreams()` in `server/core/datastreams.py`. **How it works:** 1. Reads active `app.connection_ref` rows from Postgres. 2. Cross-references the module manifest (via `core.main._loaded_modules`). 3. Creates one row in `app.datastreams` per `(connection_ref x report_profile)`. 4. Seeds `app.datastream_mappings` from the manifest's `canonical_metric_mapping` + `canonical_dimension_mapping`. * String form: `source_field` → `target_field` * Object form: `source_field` → `val["canonical"]` **Command:** ```bash theme={null} uv run python -c " import os os.environ['PLATFORM_DB_URL'] = '' import core.main # loads _loaded_modules as a side effect from core.datastreams import backfill_datastreams print(backfill_datastreams()) " ``` **Expected result:** ```python theme={null} {"created": 2, "skipped": 0, "mappings_created": 8, "errors": []} ``` **"Used-by" view in Data model:** after the backfill, the admin's Data model page automatically shows the new fields and which datastreams feed them (target\_fields "Used by" — join `datastream_mappings` → `datastreams`). No extra code required. **Date normalization (Story 8.10 / Refinement R3):** if your source delivers dates in a format or timezone other than ISO-8601/UTC, declare `date_format` and `source_timezone` in the datastream's `config`: ```json theme={null} { "schema_version": "1", "kind": "datastream", "config": { "date_format": "%d/%m/%Y", "source_timezone": "Europe/Paris" } } ``` Schema reference: `server/core/schemas/flow.datastream.schema.json` (`$.config.date_format`, `$.config.source_timezone`). *** ### Step 9 (Optional) — Edit the flow through MCP (`flows_upsert`) Once the module is live, you can modify a datastream or enrich a report directly through MCP — without touching the DB. **Modify an existing datastream:** ```python theme={null} import asyncio from fastmcp.client import Client from core.main import mcp async def main(): async with Client(mcp) as c: result = await c.call_tool("flows_upsert", { "project_id": "my-project", "kind": "datastream", "definition": { "schema_version": "1", "kind": "datastream", "id": "ds_", # id present = UPDATE "project_id": "my-project", "name": "My Connector - daily [abcd]", "module_name": "my-connector", "connection_ref_id": "ref_", "report_profile_id": "daily", "enabled": True, "schedule_mode": "nightly", "refetch_days": 3, "date_window_days": 30, "config": {}, "mappings": [ {"source_field": "clicks", "target_field": "clicks"}, {"source_field": "impressions", "target_field": "impressions"} ] } }) print(result) asyncio.run(main()) ``` **`flows_upsert` guardrails (do not bypass):** * `validate_flow()` is called **before** any DB access: an unknown field (e.g. `token`, `access_token`) is rejected immediately. * `connection_ref_id` is an opaque reference — never a token (AD-3). * Every change is audited (`flow_updated` in `audit_log`) with a before/after diff. * An idempotent upsert (empty diff) writes nothing and generates no audit. *** ## Final checklist Enforced by the conformance suite. **Every box must be checked before merge.** * **1.** `manifest.json`: valid `auth_type` enum, kebab `name`, `report_profiles` ≥ 1, complete `canonical_*_mapping`, non-additive metrics in object form with `aggregation_rule`. * **2.** `connector.py`: `pull()` returns `{pull_id, row_count, date_from, date_to}`. `transform()` is manifest-driven (no hardcoded source field name — AD-2 grep gate). Token never stored or logged (AD-3). `pull_id` received from the caller, never minted here (AD-7). * **3.** (Optional but recommended) dbt staging with `QUALIFY` supersede (AD-7) and a `grain_unique` test (AI-05 mandatory if the model is created). * **4.** Mappings auto-seeded via `backfill_datastreams()` from the manifest's `canonical_*_mapping` — nothing to write in SQL manually. * **5.** `tests/fixtures/golden_pull.json` + `expected_facts.json` present and consistent with `transform()`. * **6.** Full conformance suite: all 4 layers pass. * **7.** Live-integration pass documented (AI-13 DoD) — evidence in the Dev Agent Record. * **8.** `uv run ruff check server` passes (AI-03: ASCII-only stdout in Python scripts). * **9.** Schema Change Checklist (CONTRIBUTING.md) walked if a persisted schema or wire format changes. *** ## Known pitfalls (Epics 1-8 retros) ### AD-2 — The grep gate on module names in core/ Module names (`gsc`, `google-analytics`...) must **never** appear in `server/core/`. CI has a gate that greps `server/core/` for these strings. If your `connector.py` imports from `core/` while passing its own name as a literal, the gate fires. Correct workaround: use `provider` as a variable, not a hardcoded literal. ### AI-03 — ASCII stdout required Every Python script that writes to stdout must use ASCII characters only. Non-ASCII characters cause `UnicodeEncodeError` on some CI runners and Windows terminals. If you need Unicode, declare the encoding explicitly: ```python theme={null} import sys sys.stdout = open(sys.stdout.fileno(), 'w', encoding='utf-8', closefd=False) ``` ### AD-4 — Non-additive metrics: no naive aggregation Metrics declared `non_additive: true` (e.g. `average_position`, CTR, conversion rate) must **never** be summed directly from `fact_daily_kpi`. They must be read from a **semantic view** (`semantic_avg_position`) that applies impression-weighted aggregation: ```sql theme={null} -- CORRECT: via the semantic view SELECT average_position FROM main_marts.semantic_avg_position WHERE project_id = ? AND connector = 'my-connector' AND date BETWEEN ? AND ? -- FORBIDDEN: direct SUM from the fact (AD-4 violation) SELECT SUM(average_position) FROM main_marts.fact_daily_kpi ... ``` ### R3 — Per-datastream date normalization Do not assume the source delivers ISO-8601/UTC dates. Explicitly declare `date_format` and `source_timezone` in the datastream's `config` (Story 8.10). A wrong date-format assumption causes silent duplicates at the grain level. ### Grain-unique test (AI-05) Any dbt model that materializes a mart **must** declare a `unique` test over its entire grain key in `schema.yml`. The absence of this test is a **code-review failure** (AI-05). Silent double-counting = wrong KPIs in the widget. ### Ratio metrics: never store CTR, conversion rate, bounce rate — these are **ratios** (numerator / denominator). They are **never stored** in the raw table (example: `ctr` is dropped in `gsc/connector.py` `_DROP_FIELDS`). They are computed at the semantic layer from separately-stored numerators and denominators. ### AI-44 — Migration naming If your connector needs a new migration in `infra/nango/migrations/`, the name must follow `NNN_description.sql` with a 3-digit zero-padded numeric prefix. Never reuse or renumber an already-applied prefix.