---
title: Integrating ClickHouse with Attio
description: Sync Attio CRM records, custom objects, lists, notes, tasks, and workspace members into ClickHouse with editable readers and SQL views.
sidebar:
  label: Attio
  order: 2
---

Sync Attio into ClickHouse to query CRM records, relationships, list membership, notes, and tasks alongside product or business data.


<img src="/logos/attio.svg" alt="Attio logo" width="37" height="30" style="box-sizing: content-box; padding: 12px; background: white; border-radius: 8px;" />

The [app registry](/integrations/) installs the Attio integration as editable TypeScript source. It includes nine raw tables, separate streams for each configured object type or list, workspace-wide resource streams, and three SQL views for people, companies, and deals. The default configuration creates nine streams writing to seven resource tables; entries and list-attribute readers require configured lists. Every run performs full reads through the Attio REST API and updates the latest observed versions in ClickHouse.

| Property | Included behavior |
| --- | --- |
| Registry app | `attio` (v0.2.1) |
| Install command | `bunx chkit add attio` |
| Source directory | `src/integrations/attio` |
| Authentication | Bearer token |
| Environment variables | `ATTIO_API_TOKEN` |
| ClickHouse | `>=25.3.0` |
| Coverage | 9 synced resources, 3 derived views |
| Sync strategy | Full scans |

## Install the Attio integration

Run these commands in a TypeScript project with a `package.json`. Use a chkit release that includes the registry and ingestion plugin; the examples use the beta release.

```sh
bun add -d chkit@beta
bunx chkit registry inspect attio
bunx chkit add attio --dry-run
bunx chkit add attio
```

The installer copies the provider source, installs compatible dependencies, registers the ingestion plugin, and connects the exported schemas and pipeline to the project's config. An existing project's schema definitions and connection settings remain in place. See [installation behavior](/cli/add/) for supported config shapes and conflicts.

`add` prepares the project. Migrations create the ClickHouse objects, and `ingest run` performs the API calls. The registry is only used to obtain source files; subsequent syncs execute the installed code.

### Configure the ClickHouse connection

For a new project, the generated `clickhouse.config.ts` has this shape:

```ts
import { defineConfig } from '@chkit/core'
import { ingest } from '@chkit/plugin-ingest'

export default defineConfig({
  entry: './src/integrations/attio/index.ts',
  plugins: [ingest()],
  clickhouse: {
    url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
    username: process.env.CLICKHOUSE_USER ?? 'default',
    password: process.env.CLICKHOUSE_PASSWORD ?? '',
    database: process.env.CLICKHOUSE_DB ?? 'default',
  },
})
```

Set the connection variables for the intended target. Keep the existing entry, schema paths, and plugins when adapting an existing project. Ingestion requires a direct connection; workbench-only authentication does not provide an ingestion destination.

The copied `config.ts` separately controls the database of the Attio tables and views. Set its `database` to the intended destination too; changing `CLICKHOUSE_DB` alone does not rename those schema objects.

## Create an Attio API key

A workspace admin creates a single-workspace access token in Attio. The following steps use Attio's [API-key setup instructions](https://attio.com/help/reference/apps/generating-an-api-key):

1. As a workspace admin, open the workspace-name dropdown in Attio and select Workspace settings.
2. Open Developers, select + New access token, and name the token (for example, chkit ClickHouse).
3. Grant the read scopes listed for the resources to sync.
4. Click the token on the Developers page to copy it. Set ATTIO_API_TOKEN in the project .env file or the scheduler secret environment.

[Attio credential setup](https://attio.com/help/reference/apps/generating-an-api-key)

Grant read access to the categories required by the enabled streams. The scope table covers all supported resources; list-entry and list-attribute readers run only when `lists` is configured. No write scopes are needed.

| Resource | Required scopes |
| --- | --- |
| Objects | `object_configuration:read` |
| Object attributes | `object_configuration:read` |
| Records | `object_configuration:read`, `record_permission:read` |
| Lists | `list_configuration:read` |
| List attributes | `list_configuration:read` |
| List entries | `list_configuration:read`, `list_entry:read` |
| Notes | `note:read`, `object_configuration:read`, `record_permission:read` |
| Tasks | `task:read`, `object_configuration:read`, `record_permission:read`, `user_management:read` |
| Workspace members | `user_management:read` |

### Configure the token in the project and scheduler

Set the copied token in the project's `.env`, replacing the placeholder:

```sh
ATTIO_API_TOKEN=replace-with-the-access-token-from-attio
```

Bun loads the project's `.env`; a scheduler must supply `ATTIO_API_TOKEN` through its own secret or environment configuration, together with the ClickHouse connection variables. The installer writes required names to `.env.example`, which does not load credentials. Keep real tokens out of source control.

The reader sends the token in an `Authorization: Bearer` header when a request runs. Schema inspection, migration generation, and `ingest list` do not need an Attio token. An existing OAuth access token also works, but this integration does not obtain or refresh OAuth tokens; see [Attio authentication](https://docs.attio.com/rest-api/guides/authentication).

A token's permissions also determine which data the API returns. A `401` or `403` fails the affected stream with a diagnostic; the reader does not turn a permission error into an empty result.

## Choose objects, lists, and destination names

The installed `src/integrations/attio/config.ts` starts with:

```ts
export interface AttioConfig {
  sourceId: string
  database: string
  tablePrefix: string
  objects: readonly string[]
  lists: readonly string[]
  pageSize: number
  notesPageSize: number
}

export const attioConfig: AttioConfig = {
  sourceId: 'attio.primary',
  database: 'default',
  tablePrefix: 'attio',
  objects: ['people', 'companies'],
  lists: [],
  pageSize: 500,
  notesPageSize: 50,
}
```

| Setting | Effect |
|---|---|
| `sourceId` | Stable, non-secret identity used in stream IDs and row IDs |
| `database` | Database containing the nine raw tables and three views |
| `tablePrefix` | Prefix used before `_records_raw`, `_people`, and the other suffixes |
| `objects` | Explicit object API slugs or UUIDs; default `['people', 'companies']`; creates records and attribute streams per type |
| `lists` | Explicit list API slugs or UUIDs; default `[]`; creates entries and attribute streams per list |
| `pageSize` | Positive integer, default `500`, for attributes, records, entries, and tasks |
| `notesPageSize` | Positive integer up to `50`, default `50`, for notes |

Add `deals` or a custom slug to `objects` to collect that type, and add list slugs or IDs to `lists` to collect their entries. Empty arrays disable the corresponding child streams. The objects and lists catalog streams retain all accessible metadata independently of these selections. An unknown or inaccessible configured slug or ID fails its stream visibly. Notes, tasks, and members remain workspace-wide; their readers need separate changes to narrow their coverage.

The pipeline creates its stream graph from these explicit arrays without API discovery. People and Companies have independent readers and journal progress, while both write to `attio_records_raw`. An object stream's ID ends in `.records.<ref>` or `.object_attributes.<ref>`; a list stream's ID ends in `.entries.<ref>` or `.list_attributes.<ref>`. Keep each configured reference stable: switching a slug to an equivalent UUID changes the stream ID.

Keep `sourceId` stable after ingestion starts. For a second workspace, use a distinct source identity and a separate table prefix or database. Copying the directory alone does not isolate pipeline IDs or destination tables.

## Create tables and run the first sync

```sh
bunx chkit check
bunx chkit ingest list --tag provider:attio
bunx chkit generate --name add-attio
bunx chkit migrate
```

Review the generated SQL, then apply it and run ingestion:

```sh
bunx chkit migrate --apply
bunx chkit ingest run --tag provider:attio
bunx chkit query "SELECT count() FROM default.attio_records_raw FINAL"
bunx chkit query "SELECT name, domains FROM default.attio_companies LIMIT 10"
```

Queries in this guide use the default database and prefix. Substitute the configured names when these differ.

## All synced resources and records

Each object type has its own records and attribute streams, and each configured list has its own entries and attribute streams. Catalogs, notes, tasks, and members have workspace-wide streams. Streams for the same resource share a raw table. This table lists the returned entity data retained in `raw.data`; the complete API entity is preserved, including additional fields returned by Attio. Fields may be absent or null depending on the resource and workspace configuration.

| Resource | Default ClickHouse table | Records synced | API reference |
| --- | --- | --- | --- |
| Objects (`objects`) | `attio_objects_raw` | Accessible standard and custom object definitions, IDs, API slug, singular and plural names, creation time | [`GET /objects`](https://docs.attio.com/rest-api/endpoint-reference/objects/list-objects) |
| Object attributes (`object_attributes`) | `attio_object_attributes_raw` | Attribute IDs, slug, title, description, type, flags, defaults, relationship definitions, configuration, creation time; includes archived definitions | [`GET /objects/{object_id}/attributes`](https://docs.attio.com/rest-api/endpoint-reference/attributes/list-attributes), [`GET /objects/{object_id}`](https://docs.attio.com/rest-api/endpoint-reference/objects/get-an-object) |
| Records (`records`) | `attio_records_raw` | Independent records streams for configured object types (People and Companies by default), IDs, creation time, Attio URL, and the complete values object with all returned attribute arrays | [`POST /objects/{object_id}/records/query`](https://docs.attio.com/rest-api/endpoint-reference/records/list-records), [`GET /objects/{object_id}`](https://docs.attio.com/rest-api/endpoint-reference/objects/get-an-object) |
| Lists (`lists`) | `attio_lists_raw` | List IDs, slug, name, parent objects, workspace and member access configuration, creator, creation time | [`GET /lists`](https://docs.attio.com/rest-api/endpoint-reference/lists/list-all-lists) |
| List attributes (`list_attributes`) | `attio_list_attributes_raw` | Attribute definitions for selected lists, including types, configuration, flags, defaults, relationships, and archived definitions | [`GET /lists/{list_id}/attributes`](https://docs.attio.com/rest-api/endpoint-reference/attributes/list-attributes), [`GET /lists/{list_id}`](https://docs.attio.com/rest-api/endpoint-reference/lists/get-a-list) |
| List entries (`entries`) | `attio_entries_raw` | Entry IDs, parent object and record ID, creation time, and the complete entry_values object | [`POST /lists/{list_id}/entries/query`](https://docs.attio.com/rest-api/endpoint-reference/entries/list-entries), [`GET /lists/{list_id}`](https://docs.attio.com/rest-api/endpoint-reference/lists/get-a-list) |
| Notes (`notes`) | `attio_notes_raw` | Note IDs, parent record links, title, plaintext and Markdown content, tags, creator, creation time, and meeting reference when returned | [`GET /notes`](https://docs.attio.com/rest-api/endpoint-reference/notes/list-notes) |
| Tasks (`tasks`) | `attio_tasks_raw` | Task IDs, plaintext content, deadline, completion state and time, linked records, assignees, creator, creation time | [`GET /tasks`](https://docs.attio.com/rest-api/endpoint-reference/tasks/list-tasks) |
| Workspace members (`members`) | `attio_members_raw` | Member IDs, first and last name, email address, avatar URL, access level, creation time | [`GET /workspace_members`](https://docs.attio.com/rest-api/endpoint-reference/workspace-members/list-workspace-members) |

The default records streams read People and Companies separately. Deals and custom objects require explicit slugs or IDs in `objects`; configured types all land in the same records table. A `users` object, when configured, contains CRM records; workspace members are collected separately. The object catalog exposes accessible types without automatically adding readers for them.

Record and entry attributes remain arrays, including multivalued custom fields and relationship references. Returned attribute values keep their accompanying metadata. The integration does not issue separate requests for historical attribute values, select-option catalogs, or status catalogs. See the [record API](https://docs.attio.com/rest-api/endpoint-reference/records/list-records), [entry API](https://docs.attio.com/rest-api/endpoint-reference/entries/list-entries), and [attribute API](https://docs.attio.com/rest-api/endpoint-reference/attributes/list-attributes) for the provider's response shapes.

Notes retain the content and references in the [notes response](https://docs.attio.com/rest-api/endpoint-reference/notes/list-notes); a meeting reference does not fetch the meeting itself. Task links and assignees remain in the [task payload](https://docs.attio.com/rest-api/endpoint-reference/tasks/list-tasks). Member identity fields come from the [workspace members endpoint](https://docs.attio.com/rest-api/endpoint-reference/workspace-members/list-workspace-members).

### API requests and pagination

Every API path in the resource table is relative to `https://api.attio.com/v2`. An object/list child reader first resolves its single configured slug or ID with an individual parent request, then fetches that parent's collection.

| Resource | Pagination and filters |
|---|---|
| `objects` | Single collection request retaining all accessible objects |
| `object_attributes` | `limit` and `offset` query parameters; `show_archived=true` |
| `records` | `limit` and `offset` in JSON body; no record filter or custom sort |
| `lists` | Single collection request retaining all accessible lists |
| `list_attributes` | `limit` and `offset` query parameters; `show_archived=true` |
| `entries` | `limit` and `offset` in JSON body; no entry filter or custom sort |
| `notes` | `limit` and `offset` query parameters; no parent filter |
| `tasks` | `limit` and `offset` query parameters; `sort=created_at:asc`; no completion filter |
| `members` | Single collection request |

Paginated reads use the ingestion plugin's `paginate()` helper: they start at offset zero, advance by the number of returned entities, and stop when a page is shorter than the requested limit. Each reader resolves only its configured object/list, so running People records does not depend on the objects catalog having run first. The pipeline's stream array does not establish dependencies.

## How the sync works

| Behavior | Details |
| --- | --- |
| Sync | People, Companies, each configured custom object/list, and workspace resources have independent full-sync streams. Each reader paginates one collection through chkit-ingest; interrupted mutable offset reads replay that collection from zero. Completed and empty reads are recorded by the executor journal without a custom scan checkpoint. Custom types and lists must be configured explicitly. |
| Schedule | Run chkit ingest run through cron, CI, or an existing scheduler; the template does not start a background worker. |
| Deletions | Records absent from a later API response remain in ClickHouse. The template does not propagate deletions or consume webhooks. |

1. The CLI loads the installed pipeline and selects streams by their tags.
2. Each selected reader resolves its one configured object/list where needed and fetches that collection's pages.
3. The client validates the response's `data` array and required entity IDs. Record `values` and entry `entry_values` must contain arrays.
4. The reader wraps each entity with its source identity and relevant parent slug, then assigns a stable composite row ID.
5. The runtime batches rows and writes them to ClickHouse. The journal records execution and load progress.
6. SQL views project the raw records and reconcile repeated row IDs at query time with `FINAL`.

### Full reads and independent recovery

Every stream uses `fullSync()` and begins its next execution at offset zero. Attio's offsets describe positions in mutable collections and are not durable change cursors. A failed Companies reader therefore replays Companies independently of People; it does not maintain an outer loop over object types or a completed-parent ledger. The executor journal records execution and load completion, including successful empty reads, without custom provider checkpoint state.

Child response workspace and parent IDs are checked against the reader's resolved object/list before rows are published. Adding a configured type creates a new stream; removing it stops future reads without removing existing rows. Every scheduled run rereads the selected collections, so edits remain observable.

Version `0.2.0` changes the former combined records, object-attributes, entries, and list-attributes stream IDs to per-type/per-list IDs. Those streams begin with fresh journal progress; the raw table schemas, row IDs for the same `sourceId`, and views are unchanged. The `0.1.3` release artifact remains available. When upgrading from its checkpointed-scan preview, use a new `sourceId` to avoid reusing the old custom strategy for unchanged workspace stream IDs. A new source ID also changes raw row IDs, so review retained observations as part of that migration. Installed files are not automatically overwritten by a newer registry release.

The destination stores the latest **observed** entity. New observations replace earlier observations for the same row identity when ClickHouse reconciles versions. Freshness depends on the external schedule, scan duration, and successful API responses. There are no webhooks or continuous change capture in this integration.

### Batching, retries, and failures

| Control | Attio integration default |
|---|---|
| Active streams, fetches, and loads | One of each per pipeline within a run |
| Destination batch threshold | `500` rows; pages are accumulated without splitting a source chunk |
| HTTP request timeout | `30` seconds |
| Delay before a request | `25` ms; `125` ms for notes |
| Source retry budget | Five retries after the initial attempt |
| Backoff | Exponential, starting at `1` second, capped at `30` seconds, with jitter |
| Retry time budget | `10` minutes per retry boundary, subject to the overall execution budget |
| Default run duration | `3600` seconds; override with `--max-duration` |

HTTP `429` responses honor `Retry-After`. Network failures and HTTP `408`, `425`, and `5xx` responses are retried within the budget. Other `4xx` responses, invalid configuration, invalid JSON, and invalid entity shapes fail without retries. Exhausting a page request's retries fails the stream; restarting the reader does not multiply that request's retry budget.

The request delays keep the default serial pipeline below 40 requests per second and notes below 8 requests per second, before network time. Attio can apply lower limits, including query-complexity limits; repeatedly expensive record or entry queries may require simpler filters or sorts after customization. See [Attio rate limits](https://docs.attio.com/rest-api/guides/rate-limiting).

Destination writes have a separate retry policy and reuse an insert deduplication token. Physical duplicate suppression depends on ClickHouse's deduplication settings and window. The logical row identity and `FINAL` queries reconcile repeated observations; this is not an exactly-once delivery guarantee.

Writes become visible batch by batch. A failed or interrupted run can leave already loaded pages visible, with no rollback to the previous complete workspace snapshot. A normal stream failure allows other selected streams to be attempted. `ingest run` exits with `0` only when all selected streams succeed; an incomplete execution exits with `1`.

## Raw tables and record identity

Every raw table uses the same five columns:

| Column | ClickHouse type | Meaning |
|---|---|---|
| `id` | `String` | JSON-encoded composite identity for the source, resource, and entity IDs |
| `raw` | `JSON` | Envelope containing `source_id`, optional parent slug, and the complete entity under `data` |
| `_chkit_batch_id` | `String` | Runtime batch identity |
| `_chkit_run_id` | `String` | Ingestion run identity |
| `_chkit_ingested_at` | `DateTime64(6, 'UTC')` | ClickHouse publication time, populated with `now64(6)` |

Tables use `ReplacingMergeTree(_chkit_ingested_at)` with `id` as both the primary key and ordering key. Later ingestion time wins during replacement; the integration does not order rows by an Attio source version. Physical rows can include earlier observations until merges occur. Use `FINAL` when directly querying raw tables for the reconciled result.

Each ID is built as `JSON.stringify([sourceId, resource, ...entityIds])`. The resource names and entity ID fields are:

| Resource | Entity ID fields, in order | Extra envelope field |
|---|---|---|
| `objects` | `workspace_id`, `object_id` | None |
| `object_attributes` | `workspace_id`, `object_id`, `attribute_id` | `object_slug` |
| `records` | `workspace_id`, `object_id`, `record_id` | `object_slug` |
| `lists` | `workspace_id`, `list_id` | None |
| `list_attributes` | `workspace_id`, `object_id`, `attribute_id` | `list_slug` |
| `entries` | `workspace_id`, `list_id`, `entry_id` | `list_slug` |
| `notes` | `workspace_id`, `note_id` | None |
| `tasks` | `workspace_id`, `task_id` | None |
| `members` | `workspace_id`, `workspace_member_id` | None |

Attio's shared attribute response calls the parent list's ID `object_id`; the list-attribute key follows that response shape. Names, email addresses, and domains are never row keys. The resource name is part of `id`, not an additional field in `raw`.

A record envelope has this shape, with illustrative IDs:

```json
{
  "source_id": "attio.primary",
  "object_slug": "people",
  "data": {
    "id": {
      "workspace_id": "workspace-id",
      "object_id": "people-object-id",
      "record_id": "person-record-id"
    },
    "created_at": "2026-01-01T00:00:00Z",
    "web_url": "https://app.attio.com/example/person/person-record-id",
    "values": {
      "name": [{ "full_name": "Ada Example" }],
      "email_addresses": [{ "email_address": "ada@example.test" }]
    }
  }
}
```

The original entity lives under `data`, so custom provider fields cannot overwrite chkit's envelope metadata. Nested arrays and fields omitted from the packaged views remain available for new SQL projections.

## Included people, company, and deal views

The integration creates ordinary SQL views over `attio_records_raw FINAL`, filtered by `object_slug`. These views project the collected record rows; they do not make additional API requests.

| Derived view | Source resource | Included projection |
| --- | --- | --- |
| `attio_people` | `records` | Names, email addresses, and linked companies for people records. |
| `attio_companies` | `records` | Names, domains, and descriptions for company records. |
| `attio_deals` | `records` | Deal names, stages, values, currencies, and linked people and companies. |

Each view shares these columns:

| Column | Meaning |
|---|---|
| `id` | Stable composite raw row ID |
| `source_id` | Configured source identity |
| `workspace_id` | Attio workspace ID |
| `object_id` | Attio object ID |
| `record_id` | Attio record ID |
| `created_at` | Source creation time parsed as a nullable UTC timestamp with microsecond precision |
| `web_url` | Record URL in Attio |
| `_chkit_ingested_at` | Time this observation was published to ClickHouse |

The additional projections use these paths relative to `raw.data.values`. `[1]` means the first array element, following ClickHouse indexing.

| View | Column | Source value |
|---|---|---|
| `attio_people` | `name` | `name[1].full_name` |
| `attio_people` | `email_addresses` | All `email_addresses[].email_address` values |
| `attio_people` | `company_record_ids` | All `company[].target_record_id` values |
| `attio_companies` | `name` | `name[1].value` |
| `attio_companies` | `domains` | All `domains[].domain` values |
| `attio_companies` | `description` | `description[1].value` |
| `attio_deals` | `name` | `name[1].value` |
| `attio_deals` | `stage` | `stage[1].status.title` |
| `attio_deals` | `value` | `value[1].currency_value`, parsed as `Nullable(Float64)` |
| `attio_deals` | `currency_code` | `value[1].currency_code` |
| `attio_deals` | `company_record_ids` | All `associated_company[].target_record_id` values |
| `attio_deals` | `people_record_ids` | All `associated_people[].target_record_id` values |

Missing optional strings become empty strings; absent arrays become empty arrays. Missing or unparseable dates and amounts become null. On an initial sync without a selected `deals` object, the deals view is empty. Deselecting an object later leaves previously observed rows intact. Custom object records remain in the raw table and do not receive automatic typed views.

### Query Attio records in ClickHouse

**Count the latest observed records by object:**

```sql
SELECT
  JSONExtractString(toJSONString(raw), 'object_slug') AS object_slug,
  count() AS records,
  max(_chkit_ingested_at) AS last_observed_at
FROM default.attio_records_raw FINAL
GROUP BY object_slug
ORDER BY records DESC;
```

**Aggregate deal value by stage and currency:**

```sql
SELECT
  stage,
  currency_code,
  count() AS deals,
  sum(value) AS total_value
FROM default.attio_deals
GROUP BY stage, currency_code
ORDER BY currency_code, total_value DESC;
```

Grouping by currency avoids combining amounts from different currencies. The packaged amount is a floating-point projection; edit it to use a suitable decimal representation when exact monetary arithmetic is required.

**Inspect list membership and retain custom entry attributes:**

```sql
WITH toJSONString(raw) AS payload
SELECT
  JSONExtractString(payload, 'list_slug') AS list_slug,
  JSONExtractString(payload, 'data', 'id', 'entry_id') AS entry_id,
  JSONExtractString(payload, 'data', 'parent_object') AS parent_object,
  JSONExtractString(payload, 'data', 'parent_record_id') AS parent_record_id,
  JSONExtractRaw(payload, 'data', 'entry_values') AS entry_values
FROM default.attio_entries_raw FINAL
LIMIT 20;
```

**Project a custom company field:**

```sql
WITH toJSONString(raw) AS payload
SELECT
  JSONExtractString(payload, 'data', 'id', 'record_id') AS record_id,
  JSONExtractString(payload, 'data', 'values', 'name', 1, 'value') AS company,
  JSONExtractString(payload, 'data', 'values', 'customer_tier', 1, 'value') AS customer_tier
FROM default.attio_records_raw FINAL
WHERE JSONExtractString(payload, 'object_slug') = 'companies';
```

The last example assumes a text attribute with API slug `customer_tier`. Adapt the slug and value shape to the attribute definitions in `attio_object_attributes_raw`. Preserve `source_id` and `workspace_id` when joining records across sources, and use the provider's record IDs for relationships.

## Scheduling, verification, and recovery

After the first successful run, schedule this finite command through cron, CI, or a job runner:

```sh
bunx chkit ingest run --tag provider:attio --max-duration 3600 --json
```

The scheduler supplies the project directory, credentials, runtime, cadence, and concurrency control. Run at most one ingestion process per project and ClickHouse target at a time, including runs that select different tags. Pipeline limits apply within one process and do not lock out another process.

For a narrower run, select all configured records streams:

```sh
bunx chkit ingest list --tag provider:attio --tag resource:records
bunx chkit ingest run --tag provider:attio --tag resource:records --json
bunx chkit ingest status --tag provider:attio --json
```

Repeated tags use AND matching. Add `--tag object:people` to select only People records. Its record stream carries `resource:records` and `object:people`; its attribute stream carries `resource:object_attributes` and `object:people`. With the default source ID, `--tag stream:attio.primary.records.people` selects the People record stream. List readers carry `list:<ref>` together with `resource:entries` or `resource:list_attributes`.

Inspect the run's per-stream outcomes and error messages. `status` reports committed execution state rather than complete run history. Keep the ingestion journal as durable operational state. After fixing a token, permission, or destination issue, rerun the affected stream; its collection starts at offset zero.

Verify a known Attio record against its retained `raw.data`, update that record in a test workspace, rerun the sync, and confirm its view reflects the new observation. A row count alone does not establish full API coverage. These full-sync readers ignore date bounds and do not support date-range backfills; implementing historical selection requires a source-specific strategy.

## Coverage and consistency limits

| Situation | Current behavior |
|---|---|
| Source record deleted or merged | Previously observed row remains; no deletion or merge reconciliation |
| Token loses access or a parent is deselected | Existing rows remain in ClickHouse |
| Source changes during offset pagination | A scan may miss or repeat entities; it is not a provider snapshot |
| Run stops after some batches | Loaded batches remain visible; the affected stream restarts its collection at offset zero |
| Older source data loaded later | Later ingestion time wins; no source-version ordering |
| Complete change history needed | Raw tables reconcile observations and do not form a permanent change log |
| Meetings, call recordings, transcripts, emails, files, comments, or sequences needed | No readers are included for these resources |
| Historical attribute values or separate option/status catalogs needed | No separate requests are made for these resources |
| Continuous sync or OAuth lifecycle needed | Supply scheduling or authentication lifecycle outside this integration |

Deleting or deselecting a source entity does not remove it from a view. Applications requiring an exact current snapshot must add a deliberate reconciliation process. A failed or permission-limited scan must not be used as evidence that unseen source records were deleted.

## Customize the installed integration

| File | Responsibility |
|---|---|
| `config.ts` | Source identity, destination names, explicit object/list references, page sizes |
| `client.ts` | Authentication, response validation, single-parent lookup, pagination, row identity, request pacing |
| `sources/objects.ts`, `sources/object-attributes.ts` | Object and attribute readers with their raw table schemas |
| `sources/records.ts` | Record reader, raw records table, people/company/deal views |
| `sources/lists.ts`, `sources/list-attributes.ts`, `sources/entries.ts` | List resource readers with their raw table schemas |
| `sources/notes.ts`, `sources/tasks.ts`, `sources/members.ts` | One reader and raw table schema per resource |
| `tests/attio.test.ts`, `tests/fixtures.ts` | Optional installed tests and mocked API responses |
| `pipeline.ts` | Per-type/per-list streams, resource tags, retries, batching, concurrency |
| `index.ts` | Exports that make the pipeline and schema objects discoverable |

Remove a stream entry from `pipeline.ts` to stop collecting a resource. Keep its schema export to preserve existing tables under migration management. Removing a schema export changes the desired database schema and can generate a drop operation; review that migration separately.

Edit the view SQL in `sources/records.ts` to expose custom fields, then generate and apply a reviewed migration. Existing raw payloads support new projections without another API fetch. Reader changes apply to subsequent ingestion runs. Revisit pacing when raising concurrency, and test modified readers with fixtures before scheduling them.

Installed files belong to the project. Reinstalling does not overwrite local edits or automatically merge a newer integration version. See [provider templates](/api-sync/templates/) for version pinning and updates.

### Install and run the fixture tests

Include the integration's tests when installing it:

```sh
bunx chkit add attio --with-tests
bun test src/integrations/attio/tests/attio.test.ts
```

The tests use fixture responses, a mocked HTTP client, and an in-memory destination. They exercise reader behavior without Attio credentials or a ClickHouse server. Edit the fixtures and assertions alongside customized readers. With `--path`, adjust the test command to the chosen installation directory.

Repository-only packaging and live-database tests remain outside the installed test set. The manifest selects the portable tests explicitly.

## Related pages

- [Integration list](/integrations/): available apps with dedicated ClickHouse integration guides.
- [`chkit registry`](/cli/registry/): discover and inspect installable apps.
- [`chkit add`](/cli/add/): installation flags, project wiring, and ownership.
- [`chkit ingest`](/cli/ingest/): stream selection, outcomes, and execution budgets.
- [Destinations and transformations](/api-sync/destinations/): raw JSON, replacement semantics, and deletion handling.
- [Scheduling and recovery](/api-sync/operations/): operational limits and retry behavior.
- [Test a source](/api-sync/testing/): fixture-based checks for customized readers.