Attio Lists, Notes & Tasks (Core Workflow B)

Overview

Lists are Attio's pipeline/board primitive -- they contain entries (records added to the list with list-specific attributes like stage or owner). This skill also covers notes and tasks, which attach to records for activity tracking.

Prerequisites

• attio-install-auth completed
• Scopes: objectconfiguration:read, recordpermission:read, listentry:read, listentry:read-write, note:read-write, task:read-write, user_management:read

Instructions

Step 1: List All Lists

// GET /v2/lists
const lists = await client.get<{
  data: Array<{
    id: { list_id: string };
    api_slug: string;
    name: string;
    parent_object: string[];  // Which object types can be added
  }>;
}>("/lists");

console.log(lists.data.map((l) => `${l.api_slug}: ${l.name}`));
// Output: ["sales_pipeline: Sales Pipeline", "hiring: Hiring Pipeline"]

Step 2: Query List Entries

// POST /v2/lists/{list_slug}/entries/query
const entries = await client.post<{
  data: Array<{
    entry_id: string;
    record_id: string;
    created_at: string;
    values: Record<string, any[]>;
  }>;
}>("/lists/sales_pipeline/entries/query", {
  filter: {
    stage: { status: { $eq: "In Progress" } },
  },
  sorts: [
    { attribute: "created_at", field: "created_at", direction: "desc" },
  ],
  limit: 50,
});

Step 3: Add a Record to a List (Create Entry)

// POST /v2/lists/{list_slug}/entries
const entry = await client.post<{ data: { entry_id: string } }>(
  "/lists/sales_pipeline/entries",
  {
    data: {
      // The record to add to the list
      parent_record_id: companyRecordId,
      parent_object: "companies",
      // List-specific attribute values
      values: {
        stage: [{ status: "Qualified" }],
        deal_value: [{ currency_code: "USD", currency_value: 50000 }],
        owner: [{ referenced_actor_id: workspaceMemberId, referenced_actor_type: "workspace-member" }],
      },
    },
  }
);

Step 4: Update an Entry

// PATCH /v2/lists/{list_slug}/entries/{entry_id} -- append multiselect
await client.patch(
  `/lists/sales_pipeline/entries/${entryId}`,
  {
    data: {
      values: {
        stage: [{ status: "Won" }],
      },
    },
  }
);

// PUT /v2/lists/{list_slug}/entries/{entry_id} -- overwrite multiselect
await client.put(
  `/lists/sales_pipeline/entries/${entryId}`,
  {
    data: {
      values: {
        stage: [{ status: "Lost" }],
      },

Attio Cost Tuning

Overview

Attio pricing is based on workspace seats, not API calls. However, API rate limits effectively cap throughput, so optimizing request volume improves both performance and cost efficiency. This skill covers practical strategies to reduce unnecessary API calls.

Attio Pricing Model

API access requires Plus plan or higher. Rate limits are per-workspace, not per-seat.

Instructions

Step 1: Audit Current API Usage

// Instrument all API calls to measure usage patterns
class AttioUsageTracker {
  private calls: Array<{
    method: string;
    path: string;
    timestamp: number;
    durationMs: number;
    cached: boolean;
  }> = [];

  async track<T>(
    method: string,
    path: string,
    operation: () => Promise<T>,
    cached = false
  ): Promise<T> {
    const start = Date.now();
    try {
      const result = await operation();
      this.calls.push({
        method, path, timestamp: start,
        durationMs: Date.now() - start, cached,
      });
      return result;
    } catch (err) {
      this.calls.push({
        method, path, timestamp: start,
        durationMs: Date.now() - start, cached: false,
      });
      throw err;
    }
  }

  report(windowMs = 3600_000): {
    totalCalls: number;
    cachedCalls: number;
    topEndpoints: Array<{ path: string; count: number }>;
  } {
    const cutoff = Date.now() - windowMs;
    const recent = this.calls.filter((c) => c.timestamp > cutoff);
    const cached = recent.filter((c) => c.cached).length;

    const endpointCounts = new Map<string, number>();
    for (const call of recent) {
      const key = `${call.method} ${call.path}`;
      endpointCounts.set(key, (endpointCounts.get(key) || 0) + 1);
    }

    const topEndpoints = [...endpointCounts.entries()]
      .sort((a, b) => b[1] - a[1])
      .slice(0, 10)
      .map(([path, count]) => ({ path, count }));

    return { totalCalls: recent.length, cachedCalls: cached, topEndpoints };
  }
}

Step 2: Reduce Request Volume

The five biggest cost/rate-limit savers:

#!/bin/bash
# attio-debug-bundle.sh -- Collects Attio diagnostic evidence
set -euo pipefail

BUNDLE_DIR="attio-debug-$(date +%Y%m%d-%H%M%S)"
mkdir -p "$BUNDLE_DIR"
SUMMARY="$BUNDLE_DIR/summary.txt"

echo "=== Attio Debug Bundle ===" | tee "$SUMMARY"
echo "Generated: $(date -u +%Y-%m-%dT%H:%M:%SZ)" | tee -a "$SUMMARY"
echo "" | tee -a "$SUMMARY"

# 1. Check token is set (never log the actual key)
echo "--- Token Status ---" | tee -a "$SUMMARY"
if [ -z "${ATTIO_API_KEY:-}" ]; then
  echo "ATTIO_API_KEY: NOT SET" | tee -a "$SUMMARY"
  echo "Cannot proceed without API key" | tee -a "$SUMMARY"
  exit 1
else
  echo "ATTIO_API_KEY: SET (${#ATTIO_API_KEY} chars, starts with ${ATTIO_API_KEY:0:3}...)" | tee -a "$SUMMARY"
fi

# 2. API connectivity and auth test
echo "" | tee -a "$SUMMARY"
echo "--- API Connectivity ---" | tee -a "$SUMMARY"
HTTP_CODE=$(curl -s -o "$BUNDLE_DIR/objects-response.json" -w "%{http_code}" \
  -H "Authorization: Bearer ${ATTIO_API_KEY}" \
  https://api.attio.com/v2/objects)
echo "GET /v2/objects: HTTP $HTTP_CODE" | tee -a "$SUMMARY"

if [ "$HTTP_CODE" = "200" ]; then
  echo "AUTH: OK" | tee -a "$SUMMARY"
  jq -r '.data[] | "  - " + .api_slug + " (" + .singular_noun + ")"' \
    "$BUNDLE_DIR/objects-response.json" | tee -a "$SUMMARY"
elif [ "$HTTP_CODE" = "401" ]; then
  echo "AUTH: FAILED -- token invalid or revoked" | tee -a "$SUMMARY"
elif [ "$HTTP_CODE" = "403" ]; then
  echo "AUTH: INSUFFICIENT SCOPES" | tee -a "$SUMMARY"
  jq '.message' "$BUNDLE_DIR/objects-response.json" 2>/dev/null | tee -a "$SUMMARY"
fi

# 3. Rate limit headers
echo "" | tee -a "$SUMMARY"
echo "--- Rate Limit Status ---" | tee -a "$SUMMARY"
curl -s -D "$BUNDLE_DIR/headers.txt" -o /dev/null \
  -H "Authorization: Bearer ${ATTIO_API_KEY}" \
  https://api.attio.com/v2/objects
grep -i "x-ratelimit\|retry-after" "$BUNDLE_DIR/headers.txt" 2>/dev/null | tee -a "$SUMMARY" || echo 

# Add secrets
vercel env add ATTIO_API_KEY production
vercel env add ATTIO_WEBHOOK_SECRET production

# Deploy
vercel --prod

// vercel.json
{
  "env": {
    "ATTIO_API_KEY": "@attio_api_key",
    "ATTIO_WEBHOOK_SECRET": "@attio_webhook_secret"
  },
  "functions": {
    "api/webhooks/attio.ts": {
      "maxDuration": 30
    }
  }
}

// api/webhooks/attio.ts (Vercel serverless function)
import type { VercelRequest, VercelResponse } from "@vercel/node";
import crypto from "crypto";

export const config = { api: { bodyParser: false } };

export default async function handler(req: VercelRequest, res: VercelResponse) {
  if (req.method !== "POST") return res.status(405).end();

  const chunks: Buffer[] = [];
  for await (const chunk of req) chunks.push(chunk as Buffer);
  const rawBody = Buffer.concat(chunks);

  // Verify webhook signature before processing
  const signature = req.headers["x-attio-signature"] as string;
  const timestamp = req.headers["x-attio-timestamp"] as string;
  if (!verifySignature(rawBody, signature, timestamp)) {
    return res.status(401).json({ error: "Invalid signature" });
  }

  const event = JSON.parse(rawBody.toString());
  // Process async -- return 200 immediately
  res.status(200).json({ received: true });

  // Handle event in background
  await processAttioEvent(event);
}

# fly.toml
app = "my-attio-app"
primary_region = "iad"

[env]
  NODE_ENV = "production"

[http_service]
  internal_port = 3000
  force_https = true
  auto_stop_machines = "suspend"
  auto_start_machines = true

[[http_service.checks]]
  interval = "30s"
  timeout = "5s"
  grace_period = "10s"
  method = "GET"
  path = "/api/health"

# Set secrets
fly secrets set ATTIO_API_KEY=sk_prod_xyz
fly secrets set ATTIO_WEBHOOK_SECRET=whsec_prod_abc

# Deploy
fly deploy

# Verify
fly status
curl -s https://my-attio-app.fly.dev/api/health | jq .

# Store secret i

curl -s https://api.attio.com/v2/objects \
  -H "Authorization: Bearer ${ATTIO_API_KEY}" | jq '.data[] | {slug: .api_slug, singular: .singular_noun}'

{"slug": "people", "singular": "Person"}
{"slug": "companies", "singular": "Company"}
{"slug": "deals", "singular": "Deal"}

curl -s "https://api.attio.com/v2/objects/people/attributes" \
  -H "Authorization: Bearer ${ATTIO_API_KEY}" | jq '.data[] | {slug: .api_slug, type: .type}'

const person = await attioFetch<{ data: { id: { record_id: string } } }>({
  method: "POST",
  path: "/objects/people/records",
  body: {
    data: {
      values: {
        email_addresses: ["ada@example.com"],
        name: [
          {
            first_name: "Ada",
            last_name: "Lovelace",
            full_name: "Ada Lovelace",
          },
        ],
        description: ["First computer programmer"],
      },
    },
  },
});

console.log("Created person:", person.data.id.record_id);

// List people whose email contains "example.com"
const results = await attioFetch<{
  data: Array<{ id: { record_id: string }; values: Record<string, any> }>;
}>({
  method: "POST",
  path: "/

# .env (add to .gitignore immediately)
ATTIO_API_KEY=sk_your_token_here

# .gitignore
.env
.env.local
.env.*.local

// src/attio/client.ts
const ATTIO_BASE = "https://api.attio.com/v2";

interface AttioRequestOptions {
  method?: string;
  path: string;
  body?: Record<string, unknown>;
}

export async function attioFetch<T>({
  method = "GET",
  path,
  body,
}: AttioRequestOptions): Promise<T> {
  const res = await fetch(`${ATTIO_BASE}${path}`, {
    method,
    headers: {
      Authorization: `Bearer ${process.env.ATTIO_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: body ? JSON.stringify(body) : undefined,
  });

  if (!res.ok) {
    const error = await res.json();
    throw new Error(
      `Attio ${res.status}: ${er

Attio Local Dev Loop

Overview

Set up a fast, reproducible local development workflow for Attio REST API integrations. Includes project structure, typed client, mock server for offline work, and integration test harness.

Prerequisites

• Completed attio-install-auth setup
• Node.js 18+ with npm or pnpm
• TypeScript 5+

Instructions

Step 1: Project Structure

my-attio-integration/
├── src/
│   ├── attio/
│   │   ├── client.ts          # Typed fetch wrapper (see attio-install-auth)
│   │   ├── types.ts           # Attio response types
│   │   └── config.ts          # Env-based configuration
│   ├── services/
│   │   ├── people.ts          # People record operations
│   │   ├── companies.ts       # Company record operations
│   │   └── lists.ts           # List entry operations
│   └── index.ts
├── tests/
│   ├── mocks/
│   │   └── attio-fixtures.ts  # Realistic API response fixtures
│   ├── unit/
│   │   └── people.test.ts
│   └── integration/
│       └── attio-live.test.ts # Runs against real API (CI only)
├── .env.example
├── .env.local                 # Git-ignored, real credentials
├── tsconfig.json
└── package.json

Step 2: Type the Attio Response Model

// src/attio/types.ts

/** Attio record identifier */
export interface AttioRecordId {
  object_id: string;
  record_id: string;
}

/** Attio attribute value wrapper */
export interface AttioValue<T = unknown> {
  active_from: string;
  active_until: string | null;
  created_by_actor: { type: string; id: string };
  attribute_type: string;
  [key: string]: T | unknown;
}

/** Generic Attio record */
export interface AttioRecord {
  id: AttioRecordId;
  created_at: string;
  values: Record<string, AttioValue[]>;
}

/** Paginated list response */
export interface AttioListResponse<T> {
  data: T[];
  pagination?: {
    next_cursor?: string;
    has_more?: boolean;
  };
}

/** Attio API error response */
export interface AttioError {
  status_code: number;
  type: string;
  code: string;
  message: string;
}

Step 3: Environment Configuration

// src/attio/config.ts
export interface AttioConfig {
  apiKey: string;
  baseUrl: string;
  timeout: number;
  environment: "development" | "staging" | "production";
}

export function loadConfig(): AttioConfig {
  const env = process.env.NODE_ENV || "development";
  return {
    apiKey: process.env.ATTIO_API_KEY || "",
    baseUrl: process.env.ATTIO_BASE_URL || "https://api.attio.com/v2",
    timeout: parseInt(process.env.ATTIO_TIMEOUT || "30000", 10),
    environment: env as AttioConfig["environment"],
  };
}

Step 4: Package Scripts for Dev Loop

{
  "scripts": {
    "dev": &

Attio Performance Tuning

Overview

Attio's REST API returns JSON over HTTPS. Performance optimization focuses on reducing request count (batching, caching), maximizing throughput (connection reuse, pagination), and minimizing latency (selective field fetching, parallel queries).

Key Performance Facts

Instructions

Strategy 1: Response Caching with LRU

Cache read-heavy data (object schemas, attribute definitions) that rarely change:

import { LRUCache } from "lru-cache";

const cache = new LRUCache<string, unknown>({
  max: 500,              // Max entries
  ttl: 5 * 60 * 1000,   // 5 minutes for schema data
});

async function cachedGet<T>(
  client: AttioClient,
  path: string,
  ttlMs?: number
): Promise<T> {
  const cached = cache.get(path) as T | undefined;
  if (cached) return cached;

  const result = await client.get<T>(path);
  cache.set(path, result, { ttl: ttlMs });
  return result;
}

// Schema data: cache for 30 minutes (rarely changes)
const objects = await cachedGet(client, "/objects", 30 * 60 * 1000);
const attrs = await cachedGet(client, "/objects/people/attributes", 30 * 60 * 1000);

// Record data: cache for 1-5 minutes (changes more often)
const person = await cachedGet(client, `/objects/people/records/${id}`, 60 * 1000);

Strategy 2: Batch Queries Instead of N+1

// BAD: N+1 pattern -- one request per email lookup
const people = [];
for (const email of customerEmails) {
  const res = await client.post("/objects/people/records/query", {
    filter: { email_addresses: email },
    limit: 1,
  });
  people.push(res.data[0]);
}
// Cost: N requests

// GOOD: Single query with $in operator
const allPeople = await client.post<{ data: AttioRecord[] }>(
  "/objects/people/records/query",
  {
    filter: {
      email_addresses: {
        email_address: { $in: customerEmails },
      },
    },
    limit: customerEmails.length,
  }
);
// Cost: 1 request

Strategy 3: Parallel Independent Queries

// Fetch multiple object types in parallel
const [people, companies, deals] = await Promise.all([
  client.post<{ data: AttioRecord[] }>(
    "/objects/people/records/query",
    { limit: 100 }
  ),
  client.post<{ data: AttioRecord[] }>(
    "/objects/comp

Attio Production Checklist

Overview

Systematic checklist for launching Attio API integrations in production. Covers the real failure modes observed in Attio integrations.

Prerequisites

• Staging environment tested
• Production API token created with minimal scopes
• Monitoring infrastructure available

Instructions

Phase 1: Authentication & Secrets

[ ] Production token created with minimal scopes (see attio-security-basics)
[ ] Token stored in platform secrets manager (not env file on disk)
[ ] Separate tokens for dev/staging/prod environments
[ ] .env files in .gitignore
[ ] No tokens in logs, error messages, or client-side bundles
[ ] Token rotation procedure documented

Verify:

# Confirm production token works
curl -s -o /dev/null -w "%{http_code}" \
  https://api.attio.com/v2/objects \
  -H "Authorization: Bearer ${ATTIO_API_KEY_PROD}"
# Must return 200

Phase 2: Error Handling

[ ] All API calls wrapped in try/catch
[ ] AttioApiError class distinguishes retryable (429, 5xx) from fatal errors
[ ] Exponential backoff with jitter on 429 responses
[ ] Retry-After header honored (Attio sends a date, not seconds)
[ ] 5xx errors retried (Attio may have transient issues)
[ ] 400/422 validation errors logged with request body for debugging
[ ] 403 scope errors produce actionable log messages
[ ] 404 errors handled gracefully (records can be deleted/merged)

Phase 3: Rate Limiting

[ ] Queue-based throttling implemented (p-queue or similar)
[ ] Concurrency limited to 5-10 parallel requests
[ ] Bulk operations use query endpoint (1 POST) instead of N GETs
[ ] Batch imports use offset-based pagination, not individual fetches
[ ] Rate limit monitor logs approaching-limit warnings

Key fact: Attio uses a 10-second sliding window. Rate limit scores are summed across all tokens in the workspace.

Phase 4: Data Integrity

[ ] Record creation uses PUT (assert) for idempotent upserts where possible
[ ] Email/domain values validated before sending to API
[ ] Phone numbers formatted in E.164 ("+14155551234")
[ ] Record-reference attributes use verified target_record_ids
[ ] Pagination handles all pages (check data.length === limit to know if more)
[ ] Webhook events processed idempotently (deduplicate by event ID)

Phase 5: Health Check Endpoint

// api/health.ts -- include Attio in your health check
export async function GET() {
  const start = Date.now();
  let attioStatus: "healthy" | "degraded" | "down" = "down";
  let attioLatency = 0;

  try {
    const res = await fetch("https://api.attio.com/v2/objects", {
      headers: { Authorization: `Bear

Attio Rate Limits

Overview

Attio uses a sliding window algorithm with a 10-second window. Rate limit scores are summed across all apps and access tokens hitting the API. When exceeded, you get HTTP 429 with a Retry-After header containing a date (usually the next second). Attio may temporarily reduce limits during incidents.

Rate Limit Response

HTTP/1.1 429 Too Many Requests
Retry-After: Sat, 22 Mar 2025 14:30:01 GMT
Content-Type: application/json

{
  "status_code": 429,
  "type": "rate_limit_error",
  "code": "rate_limit_exceeded",
  "message": "Rate limit exceeded, please try again later"
}

Key fact: The Retry-After header is a date string (not seconds). Parse it as a Date to calculate wait time.

Instructions

Step 1: Parse Retry-After Header

function parseRetryAfter(headers: Headers): number {
  const retryAfter = headers.get("Retry-After");
  if (!retryAfter) return 1000; // Default 1s

  // Attio sends a date string
  const retryDate = new Date(retryAfter);
  const waitMs = retryDate.getTime() - Date.now();
  return Math.max(waitMs, 100); // Minimum 100ms
}

Step 2: Exponential Backoff with Retry-After Awareness

import { AttioApiError } from "./client";

interface RetryConfig {
  maxRetries: number;
  baseMs: number;
  maxMs: number;
}

async function withRateLimitRetry<T>(
  operation: () => Promise<{ data: T; headers?: Headers }>,
  config: RetryConfig = { maxRetries: 5, baseMs: 1000, maxMs: 30000 }
): Promise<T> {
  for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
    try {
      const result = await operation();
      return result.data;
    } catch (err) {
      if (attempt === config.maxRetries) throw err;

      if (err instanceof AttioApiError) {
        if (!err.retryable) throw err; // Only retry 429 and 5xx

        // Use Retry-After if available, otherwise exponential backoff
        const backoff = config.baseMs * Math.pow(2, attempt);
        const jitter = Math.random() * 500;
        const delay = Math.min(backoff + jitter, config.maxMs);

        console.warn(
          `Attio ${err.statusCode} on attempt ${attempt + 1}/${config.maxRetries}. ` +
          `Retrying in ${delay.toFixed(0)}ms...`
        );
        await new Promise((r) => setTimeout(r, delay));
      } else {
        throw err;
      }
    }
  }
  throw new Error("Unreachable");
}

Step 3: Queue-Based Throttling

Prevent 429s proactively by limiting concurrency and request rate:

import PQueue from "p-queue";

// Attio: sliding 10-second window. Stay well under the limit.
const attioQueue = new PQue

Attio Reference Architecture

Overview

Production architecture for CRM integrations with the Attio REST API (https://api.attio.com/v2). Designed for contact enrichment pipelines, deal tracking across custom lists, bi-directional activity sync with external systems, and workspace isolation for multi-tenant deployments. Key design drivers: webhook-driven data freshness, idempotent upserts via PUT assertions, schema-aware caching, and layered separation between API client, business logic, and infrastructure.

Architecture Diagram

Your App ──→ Service Layer ──→ Cache (Redis) ──→ Attio REST API v2
                  ↓                               /objects/people/records
             Queue (p-queue) ──→ Sync Worker      /lists/{slug}/entries
                  ↓                               /notes, /tasks
             Webhook Handler ←── Attio Events     /webhooks
                  ↓
             External CRM Sync ──→ HubSpot/Salesforce

Service Layer

class ContactService {
  constructor(private client: AttioClient, private cache: CacheLayer) {}

  async findByEmail(email: string): Promise<AttioRecord | null> {
    const res = await this.client.post('/objects/people/records/query', { filter: { email_addresses: email }, limit: 1 });
    return res.data[0] || null;
  }

  async upsertPerson(data: { email: string; firstName: string; lastName: string }): Promise<AttioRecord> {
    const res = await this.client.put('/objects/people/records', {
      data: { values: { email_addresses: [data.email], name: [{ first_name: data.firstName, last_name: data.lastName }] } }
    });
    await this.cache.invalidate(`person:${data.email}`);
    return res.data;
  }

  async addToPipeline(recordId: string, listSlug: string, stage: string): Promise<void> {
    await this.client.post(`/lists/${listSlug}/entries`, {
      data: { parent_record_id: recordId, parent_object: 'people', values: { stage: [{ status: stage }] } }
    });
  }
}

Caching Strategy

const CACHE_CONFIG = {
  schema:  { ttl: 1800, prefix: 'schema' },   // 30 min — object/attribute definitions change rarely
  records: { ttl: 300,  prefix: 'record' },    // 5 min — webhook-driven invalidation handles freshness
  lists:   { ttl: 120,  prefix: 'list' },      // 2 min — deal pipeline stages need near-real-time
  notes:   { ttl: 60,   prefix: 'note' },      // 1 min — activity feed freshness
};
// Webhook events (record.updated, list-entry.created) flush matching cache keys immediately

Event Pipeline

class AttioEventPipeline {
  private queue = new Bull('attio-events', { redis: process.env.REDIS_URL });

  async onWebhook(event: AttioWebhookEvent): Promise<void> {
    await this.queue.add(event.even

Attio SDK Patterns

Overview

There is no official Attio Node.js SDK. The API is a clean REST/JSON interface at https://api.attio.com/v2. These patterns wrap fetch into a production-grade typed client with retry, pagination, and error normalization.

Prerequisites

• Node.js 18+ (native fetch)
• TypeScript 5+
• Completed attio-install-auth

Instructions

Pattern 1: Typed Client with Error Normalization

// src/attio/client.ts
const ATTIO_BASE = "https://api.attio.com/v2";

export class AttioApiError extends Error {
  constructor(
    public statusCode: number,
    public type: string,
    public code: string,
    message: string
  ) {
    super(message);
    this.name = "AttioApiError";
  }

  get retryable(): boolean {
    return this.statusCode === 429 || this.statusCode >= 500;
  }
}

export class AttioClient {
  constructor(private apiKey: string) {}

  async request<T>(
    method: string,
    path: string,
    body?: Record<string, unknown>
  ): Promise<T> {
    const res = await fetch(`${ATTIO_BASE}${path}`, {
      method,
      headers: {
        Authorization: `Bearer ${this.apiKey}`,
        "Content-Type": "application/json",
      },
      body: body ? JSON.stringify(body) : undefined,
    });

    if (!res.ok) {
      const err = await res.json().catch(() => ({}));
      throw new AttioApiError(
        res.status,
        err.type || "unknown",
        err.code || "unknown",
        err.message || `HTTP ${res.status}`
      );
    }

    return res.json() as Promise<T>;
  }

  // Convenience methods for common HTTP verbs
  get<T>(path: string) { return this.request<T>("GET", path); }
  post<T>(path: string, body: Record<string, unknown>) { return this.request<T>("POST", path, body); }
  patch<T>(path: string, body: Record<string, unknown>) { return this.request<T>("PATCH", path, body); }
  put<T>(path: string, body: Record<string, unknown>) { return this.request<T>("PUT", path, body); }
  delete<T>(path: string) { return this.request<T>("DELETE", path); }
}

Pattern 2: Retry with Exponential Backoff

// src/attio/retry.ts
export async function withRetry<T>(
  operation: () => Promise<T>,
  config = { maxRetries: 4, baseMs: 1000, maxMs: 30000 }
): Promise<T> {
  for (let attempt = 0; attempt <= config.maxRetries; attempt++) {
    try {
      return await operation();
    } catch (err) {
      if (attempt === config.maxRetries) throw err;

      // Only retry on rate limits (429) and server errors (5xx)
      if (err instanceof AttioApiError && !err.retryable) throw err;

Attio Security Basics

Overview

Attio access tokens never expire and have no scopes by default. This makes scoping, rotation, and secret management critical. This skill covers practical security controls for Attio REST API integrations.

Token Properties

Instructions

Step 1: Apply Least-Privilege Scopes

Tokens should have only the scopes needed for their use case:

# Read-only analytics integration
object_configuration:read
record_permission:read

# CRM sync (needs write)
object_configuration:read
record_permission:read-write
list_entry:read-write

# Webhook receiver (just needs to verify, no API calls)
# No scopes needed -- webhook signature uses a separate secret

# Full admin (avoid in production)
object_configuration:read
record_permission:read-write
list_entry:read-write
note:read-write
task:read-write
user_management:read
webhook:read-write

Step 2: Environment Variable Management

# .env.local (development -- git-ignored)
ATTIO_API_KEY=sk_dev_abc123

# .env.example (committed -- template for team)
ATTIO_API_KEY=sk_your_token_here
# ATTIO_WEBHOOK_SECRET=whsec_your_secret_here

# .gitignore (mandatory)
.env
.env.local
.env.*.local

Platform-specific secrets management:

# Vercel
vercel env add ATTIO_API_KEY production

# Fly.io
fly secrets set ATTIO_API_KEY=sk_prod_xyz

# Google Cloud (Secret Manager)
echo -n "sk_prod_xyz" | gcloud secrets create attio-api-key --data-file=-

# GitHub Actions
gh secret set ATTIO_API_KEY --body "sk_prod_xyz"

# AWS Systems Manager
aws ssm put-parameter --name /app/attio-api-key \
  --value "sk_prod_xyz" --type SecureString

Step 3: Token Rotation Procedure

Attio tokens cannot be rotated in-place. You must create a new token and delete the old one.

# 1. Generate new token in Settings > Developers > Access tokens
#    Match the scopes of the old token exactly

# 2. Update the secret in your deployment platform
vercel env rm ATTIO_API_KEY production
vercel env add ATTIO_API_KEY production
# Enter new token value

# 3. Deploy with new token
vercel --prod

# 4. Verify the new token works
curl -s -o /dev/null -w "%{http_code}" \
  https://api.attio.com/v2/objects \
  -H "Authorization: Bearer ${NEW_TOKEN}"
# Should return 200

# 5. Delete old token in Atti

Attio Upgrade & Migration

Overview

Attio has two API generations: v1 (legacy, deprecated) and v2 (current). This skill covers the v1-to-v2 migration, community SDK upgrade paths, and how to detect and adapt to API changes since Attio does not publish a traditional SDK changelog.

V1 to V2 Migration Reference

Endpoint Changes

Key Differences

Step 1: Update Base URL

// Before
const BASE = "https://api.attio.com/v1";

// After
const BASE = "https://api.attio.com/v2";

Step 2: Migrate Record Queries

// V1: GET with query params
const v1 = await fetch(
  `${BASE}/objects/${objectId}/records?page=1&per_page=50`,
  { headers: { Authorization: `Bearer ${token}` } }
);

// V2: POST with filter body, using slug instead of UUID
const v2 = await fetch(
  `${BASE}/objects/people/records/query`,
  {
    method: "POST",
    headers: {
      Authorization: `Bearer ${token}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      filter: {
   

Attio Webhooks & Events

Overview

Attio v2 webhooks deliver real-time CRM event notifications to your HTTPS endpoint. Subscribe to record, list-entry, note, and task events with optional object or attribute filters to reduce volume. Webhooks are managed via POST /v2/webhooks and verified with HMAC-SHA256 signatures using a timestamp-prefixed payload.

Webhook Registration

const webhook = await fetch("https://api.attio.com/v2/webhooks", {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${process.env.ATTIO_API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    target_url: "https://yourapp.com/webhooks/attio",
    subscriptions: [
      { event_type: "record.created" },
      { event_type: "record.updated", filter: { object: { $eq: "deals" } } },
      { event_type: "note.created" },
      { event_type: "task.completed" },
    ],
  }),
});

Signature Verification

import crypto from "crypto";
import { Request, Response, NextFunction } from "express";

function verifyAttioSignature(req: Request, res: Response, next: NextFunction) {
  const signature = req.headers["x-attio-signature"] as string;
  const timestamp = req.headers["x-attio-timestamp"] as string;
  const age = Date.now() - parseInt(timestamp) * 1000;
  if (age > 300_000) return res.status(401).json({ error: "Timestamp too old" });
  const payload = `${timestamp}.${req.body.toString()}`;
  const expected = crypto.createHmac("sha256", process.env.ATTIO_WEBHOOK_SECRET!)
    .update(payload).digest("hex");
  if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
    return res.status(401).json({ error: "Invalid signature" });
  }
  next();
}

Event Handler

import express from "express";
const app = express();

app.post("/webhooks/attio", express.raw({ type: "application/json" }), verifyAttioSignature, (req, res) => {
  const event = JSON.parse(req.body.toString());
  res.status(200).json({ received: true });

  switch (event.event_type) {
    case "record.created":
      syncRecordToCRM(event.object?.api_slug, event.record?.id?.record_id); break;
    case "record.updated":
      reindexRecord(event.object?.api_slug, event.record?.id?.record_id); break;
    case "note.created":
      forwardToNotionSync(event.id.event_id); break;
    case "task.completed":
      closeProjectTask(event.id.event_id); break;
  }
});

Event Types