Add a connector

server/modules/<kebab-name>/
  manifest.json                  # required
  connector.py                   # required (pull + transform)
  dbt/staging/stg_<name>.sql      # recommended (QUALIFY supersede AD-7)
  dbt/staging/schema.yml          # recommended
  reports/<id>.json               # >= 1 report pack
  tests/fixtures/golden_pull.json    # required (Layer 4)
  tests/fixtures/expected_facts.json # required (Layer 4)

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)

{
  "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"
}

uv run pytest server/tests/conformance/ --module-path server/modules/<name>/ -m conformance_layer_1 -v

"""<Name> 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,
    }

project_id | date | metric | value | breakdown_dimension | breakdown_value | pull_id | loaded_at

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

uv run pytest server/tests/conformance/ --module-path server/modules/<name>/ -m conformance_layer_4 -v

-- Staging: maps raw <name> 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

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"

{
  "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."
}

server/modules/<name>/tests/fixtures/golden_pull.json    # raw rows as they arrive from the API
server/modules/<name>/tests/fixtures/expected_facts.json # rows after transform()

[
  {
    "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
  }
]

[
  {
    "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
  }
]

uv run pytest server/tests/conformance/ --module-path server/modules/<name>/ -v

uv run pytest server/tests/conformance/ --module-path server/modules/gsc/ -v
# Expected result: 16 passed, 3 skipped (skips are normal for kpi modules)

[manifest] $.name: must match pattern ^[a-z0-9-]+$, got "MyConnector"
[golden_pull] row 0 field average_position: expected 7.3, got 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

uv run python -c "
import os
os.environ['PLATFORM_DB_URL'] = '<your_postgres_dsn>'
import core.main  # loads _loaded_modules as a side effect
from core.datastreams import backfill_datastreams
print(backfill_datastreams())
"

{"created": 2, "skipped": 0, "mappings_created": 8, "errors": []}

{
  "schema_version": "1", "kind": "datastream",
  "config": {
    "date_format": "%d/%m/%Y",
    "source_timezone": "Europe/Paris"
  }
}

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_<ULID>",       # id present = UPDATE
                "project_id": "my-project",
                "name": "My Connector - daily [abcd]",
                "module_name": "my-connector",
                "connection_ref_id": "ref_<ULID>",
                "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())

import sys
sys.stdout = open(sys.stdout.fileno(), 'w', encoding='utf-8', closefd=False)

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