diff --git a/backend/config.py b/backend/config.py
index 22662172..1155b9e6 100644
--- a/backend/config.py
+++ b/backend/config.py
@@ -268,6 +268,12 @@ class Settings:
"GONG_OAUTH_REDIRECT_URI", AUTH0_ENABLED
)
+ ATTIO_OAUTH_CLIENT_ID: str | None = os.getenv("ATTIO_OAUTH_CLIENT_ID")
+ ATTIO_OAUTH_CLIENT_SECRET: str | None = os.getenv("ATTIO_OAUTH_CLIENT_SECRET")
+ ATTIO_OAUTH_REDIRECT_URI: str | None = parse_oauth_redirect_uri(
+ "ATTIO_OAUTH_REDIRECT_URI", AUTH0_ENABLED
+ )
+
SLACK_OAUTH_CLIENT_ID: str | None = os.getenv("SLACK_OAUTH_CLIENT_ID")
SLACK_OAUTH_CLIENT_SECRET: str | None = os.getenv("SLACK_OAUTH_CLIENT_SECRET")
SLACK_OAUTH_REDIRECT_URI: str | None = parse_oauth_redirect_uri(
diff --git a/backend/integrations/__init__.py b/backend/integrations/__init__.py
index 8685f2f0..89b3f140 100644
--- a/backend/integrations/__init__.py
+++ b/backend/integrations/__init__.py
@@ -16,6 +16,7 @@
from . import (
asana, # noqa: F401
+ attio, # noqa: F401
github, # noqa: F401 — import for side-effect (provider self-registration)
gmail, # noqa: F401
gong, # noqa: F401
diff --git a/backend/integrations/attio/__init__.py b/backend/integrations/attio/__init__.py
new file mode 100644
index 00000000..96c80496
--- /dev/null
+++ b/backend/integrations/attio/__init__.py
@@ -0,0 +1,11 @@
+"""Attio integration: OAuth provider (standard authorization-code app).
+
+Connected Attio call recordings are indexed into attio_documents by
+backend/integrations/attio/indexer.py (dispatched from backend/tasks/sources) —
+each recording's transcript becomes a searchable document.
+"""
+
+from ..registry import register_provider
+from .provider import AttioIntegration
+
+register_provider(AttioIntegration())
diff --git a/backend/integrations/attio/indexer.py b/backend/integrations/attio/indexer.py
new file mode 100644
index 00000000..e0e292f2
--- /dev/null
+++ b/backend/integrations/attio/indexer.py
@@ -0,0 +1,142 @@
+"""Attio → attio_documents indexer (copied content; FTS-searchable).
+
+One Attio connection = one source (external_ref "calls"). We page meetings from
+a rolling window (last LOOKBACK_DAYS), then for each meeting pull its call
+recordings and each recording's transcript, and write every recording as a text
+document (title + date + speaker-labelled transcript). Idempotent re-sync via
+source_service; recordings that age out of the window are soft-deleted.
+
+Attio ids are composite ({workspace_id, meeting_id, call_recording_id}); the
+URL path segments are the bare uuids. The recording is keyed by
+call_recording_id, which is unique across meetings.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+from uuid import UUID
+
+import httpx
+
+from ...services import source_service
+from ..storage import get_valid_token
+
+logger = logging.getLogger(__name__)
+
+BASE_URL = "https://api.attio.com"
+LOOKBACK_DAYS = 90
+MAX_MEETINGS = 5000
+PAGE_LIMIT = 200
+
+
+def _parse_time(value: str | None) -> datetime | None:
+ """Attio returns ISO-8601 ('...Z'); the column is timestamptz. A recording
+ never changes after the call, so its creation time is its modified time."""
+ if not value:
+ return None
+ return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def _render_call(title: str, created_at: str, transcript: list[dict]) -> str:
+ lines = [f"# {title}", f"Date: {created_at}", ""]
+ speaker_num: dict[str, int] = {}
+ for segment in transcript:
+ name = (segment.get("speaker") or {}).get("name") or "?"
+ if name not in speaker_num:
+ speaker_num[name] = len(speaker_num) + 1
+ speech = (segment.get("speech") or "").strip()
+ if speech:
+ lines.append(f"[Speaker {speaker_num[name]}]: {speech}")
+ return "\n".join(lines)
+
+
+async def _fetch_meetings(client: httpx.AsyncClient, ends_from: str) -> list[dict]:
+ meetings: list[dict] = []
+ cursor: str | None = None
+ while len(meetings) < MAX_MEETINGS:
+ params = {"limit": PAGE_LIMIT, "sort": "start_desc", "ends_from": ends_from}
+ if cursor:
+ params["cursor"] = cursor
+ resp = await client.get("/v2/meetings", params=params)
+ resp.raise_for_status()
+ payload = resp.json()
+ meetings.extend(payload.get("data", []))
+ cursor = (payload.get("pagination") or {}).get("next_cursor")
+ if not cursor:
+ break
+ return meetings
+
+
+async def _fetch_recordings(client: httpx.AsyncClient, meeting_id: str) -> list[dict]:
+ recordings: list[dict] = []
+ cursor: str | None = None
+ while True:
+ params = {"limit": PAGE_LIMIT}
+ if cursor:
+ params["cursor"] = cursor
+ resp = await client.get(f"/v2/meetings/{meeting_id}/call_recordings", params=params)
+ resp.raise_for_status()
+ payload = resp.json()
+ recordings.extend(payload.get("data", []))
+ cursor = (payload.get("pagination") or {}).get("next_cursor")
+ if not cursor:
+ break
+ return recordings
+
+
+async def _fetch_transcript(
+ client: httpx.AsyncClient, meeting_id: str, recording_id: str
+) -> list[dict]:
+ segments: list[dict] = []
+ cursor: str | None = None
+ while True:
+ params = {"limit": PAGE_LIMIT}
+ if cursor:
+ params["cursor"] = cursor
+ resp = await client.get(
+ f"/v2/meetings/{meeting_id}/call_recordings/{recording_id}/transcript",
+ params=params,
+ )
+ resp.raise_for_status()
+ payload = resp.json()
+ segments.extend((payload.get("data") or {}).get("transcript", []))
+ cursor = (payload.get("pagination") or {}).get("next_cursor")
+ if not cursor:
+ break
+ return segments
+
+
+async def index_attio(source: dict) -> str | None:
+ source_id = UUID(source["id"])
+ owner_user_id = UUID(source["owner_user_id"])
+ token = await get_valid_token(owner_user_id, "attio")
+ headers = {"Authorization": f"Bearer {token}"}
+ ends_from = (datetime.now(UTC) - timedelta(days=LOOKBACK_DAYS)).isoformat()
+
+ present: list[str] = []
+ async with httpx.AsyncClient(timeout=120.0, headers=headers, base_url=BASE_URL) as client:
+ meetings = await _fetch_meetings(client, ends_from)
+ for meeting in meetings:
+ meeting_id = meeting["id"]["meeting_id"]
+ title = meeting.get("title") or "Untitled call"
+ for recording in await _fetch_recordings(client, meeting_id):
+ recording_id = recording["id"]["call_recording_id"]
+ created_at = recording.get("created_at") or ""
+ transcript = await _fetch_transcript(client, meeting_id, recording_id)
+ await source_service.upsert_content_document(
+ table="attio_documents",
+ source_id=source_id,
+ owner_user_id=owner_user_id,
+ path=recording_id,
+ name=title,
+ kind="call",
+ content=_render_call(title, created_at, transcript),
+ external_ref=recording_id,
+ external_updated_at=_parse_time(created_at),
+ )
+ present.append(recording_id)
+
+ await source_service.remove_missing_documents("attio_documents", source_id, present)
+ logger.info("attio source %s: indexed %d recording(s)", source_id, len(present))
+ return None
diff --git a/backend/integrations/attio/provider.py b/backend/integrations/attio/provider.py
new file mode 100644
index 00000000..7da318f5
--- /dev/null
+++ b/backend/integrations/attio/provider.py
@@ -0,0 +1,98 @@
+"""Attio OAuth provider (standard authorization-code app).
+
+Attio access tokens do not expire and there is no refresh token
+(`supports_refresh = False`) — same shape as GitHub user tokens. If the user
+revokes the app in Attio, API calls return 401 and they must reconnect from the
+integrations page.
+
+The token exchange sends client_id/client_secret in the POST body; the returned
+access token is then sent as a Bearer to api.attio.com. GET /v2/self identifies
+the connected workspace (used for the account label).
+"""
+
+from __future__ import annotations
+
+from urllib.parse import urlencode
+
+import httpx
+
+from ...config import settings
+from ..base import AccountInfo, Integration, TokenSet
+
+AUTHORIZE_URL = "https://app.attio.com/authorize"
+TOKEN_URL = "https://app.attio.com/oauth/token"
+SELF_URL = "https://api.attio.com/v2/self"
+
+# meeting:read lists meetings; call_recording:read lists recordings and pulls
+# their transcripts. These must match the scopes enabled on the Attio app in the
+# developer dashboard, or the authorize step is rejected.
+SCOPES = ["meeting:read", "call_recording:read"]
+
+
+class AttioIntegration(Integration):
+ name = "attio"
+ display_name = "Attio"
+ scopes = SCOPES
+ supports_refresh = False
+
+ def _client_id(self) -> str:
+ if not settings.ATTIO_OAUTH_CLIENT_ID:
+ raise RuntimeError("ATTIO_OAUTH_CLIENT_ID is not set")
+ return settings.ATTIO_OAUTH_CLIENT_ID
+
+ def _client_secret(self) -> str:
+ if not settings.ATTIO_OAUTH_CLIENT_SECRET:
+ raise RuntimeError("ATTIO_OAUTH_CLIENT_SECRET is not set")
+ return settings.ATTIO_OAUTH_CLIENT_SECRET
+
+ def _redirect_uri(self) -> str:
+ if not settings.ATTIO_OAUTH_REDIRECT_URI:
+ raise RuntimeError("ATTIO_OAUTH_REDIRECT_URI is not set")
+ return settings.ATTIO_OAUTH_REDIRECT_URI
+
+ def authorize_url(self, state: str) -> str:
+ params = {
+ "client_id": self._client_id(),
+ "redirect_uri": self._redirect_uri(),
+ "response_type": "code",
+ "scope": " ".join(self.scopes),
+ "state": state,
+ }
+ return f"{AUTHORIZE_URL}?{urlencode(params)}"
+
+ async def exchange_code(self, code: str) -> TokenSet:
+ async with httpx.AsyncClient(timeout=15.0) as client:
+ resp = await client.post(
+ TOKEN_URL,
+ data={
+ "grant_type": "authorization_code",
+ "code": code,
+ "redirect_uri": self._redirect_uri(),
+ "client_id": self._client_id(),
+ "client_secret": self._client_secret(),
+ },
+ )
+ resp.raise_for_status()
+ payload = resp.json()
+ return TokenSet(
+ access_token=payload["access_token"],
+ refresh_token=None,
+ expires_at=None,
+ scopes=(payload.get("scope") or "").split(),
+ )
+
+ async def refresh(self, refresh_token: str) -> TokenSet:
+ raise RuntimeError("Attio access tokens are not refreshable")
+
+ async def revoke(self, access_token: str) -> None:
+ # Attio exposes no token-revocation endpoint; disconnect just drops our
+ # stored row (storage.revoke_stored). Nothing to call upstream.
+ return None
+
+ async def fetch_account(self, access_token: str) -> AccountInfo:
+ headers = {"Authorization": f"Bearer {access_token}"}
+ async with httpx.AsyncClient(timeout=15.0, headers=headers) as client:
+ resp = await client.get(SELF_URL)
+ resp.raise_for_status()
+ info = resp.json()
+ return AccountInfo(email=None, display_name=info.get("workspace_name") or "Attio")
diff --git a/backend/migrations/versions/0152_attio_source.py b/backend/migrations/versions/0152_attio_source.py
new file mode 100644
index 00000000..0153f65a
--- /dev/null
+++ b/backend/migrations/versions/0152_attio_source.py
@@ -0,0 +1,53 @@
+"""Per-integration source table for Attio call recordings.
+
+Copied-content source (FTS + embeddings live in the table), same shape as
+gong_documents: each Attio call recording's transcript becomes a document keyed
+by (source_id, path).
+
+Revision ID: 0152
+Revises: 0151
+"""
+
+from alembic import op
+
+revision = "0152"
+down_revision = "0151"
+branch_labels = None
+depends_on = None
+
+_COLUMNS = """
+ id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+ owner_user_id uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ source_id uuid NOT NULL REFERENCES user_sources(id) ON DELETE CASCADE,
+ path text NOT NULL,
+ name text NOT NULL,
+ kind text NOT NULL DEFAULT 'file',
+ external_ref text,
+ external_updated_at timestamptz,
+ created_at timestamptz NOT NULL DEFAULT now(),
+ updated_at timestamptz NOT NULL DEFAULT now(),
+ deleted_at timestamptz,
+ content text,
+ content_hash text,
+ embedding vector(384),
+ embed_stale boolean NOT NULL DEFAULT FALSE
+"""
+
+
+def upgrade() -> None:
+ op.execute(f"CREATE TABLE attio_documents ({_COLUMNS}, UNIQUE (source_id, path))")
+ op.execute(
+ "CREATE INDEX attio_documents_source_idx ON attio_documents (source_id, path) "
+ "WHERE deleted_at IS NULL"
+ )
+ op.execute(
+ "CREATE INDEX attio_documents_fts_idx ON attio_documents "
+ "USING gin (to_tsvector('english', coalesce(content, '')))"
+ )
+ op.execute(
+ "CREATE INDEX attio_documents_embed_stale_idx ON attio_documents (id) WHERE embed_stale"
+ )
+
+
+def downgrade() -> None:
+ op.execute("DROP TABLE IF EXISTS attio_documents")
diff --git a/backend/routers/sources.py b/backend/routers/sources.py
index 36b6acf2..652499ac 100644
--- a/backend/routers/sources.py
+++ b/backend/routers/sources.py
@@ -109,6 +109,13 @@ async def _resolve_gong_source(user_id) -> tuple[str, str]:
return "calls", "Gong"
+async def _resolve_attio_source(user_id) -> tuple[str, str]:
+ """Attio is one connection per user (all call recordings); external_ref is
+ constant. Confirm the credentials exist (raises 401 if not connected)."""
+ await integration_storage.get_valid_token(user_id, "attio")
+ return "calls", "Attio"
+
+
async def _resolve_twitter_source(user_id) -> tuple[str, str]:
"""Twitter source external_ref is the connected X account's numeric user
id, resolved once here so reads never depend on /users/me (X's most
@@ -316,6 +323,9 @@ async def add_source(
elif body.source_type == "gong_calls" and not external_ref:
external_ref, resolved_name = await _resolve_gong_source(current_user["id"])
display_name = display_name or resolved_name
+ elif body.source_type == "attio_calls" and not external_ref:
+ external_ref, resolved_name = await _resolve_attio_source(current_user["id"])
+ display_name = display_name or resolved_name
elif body.source_type == "twitter":
external_ref, username = await _resolve_twitter_source(current_user["id"])
display_name = f"Twitter / X (@{username})"
diff --git a/backend/services/source_service.py b/backend/services/source_service.py
index 1175ce0b..0bbc71f0 100644
--- a/backend/services/source_service.py
+++ b/backend/services/source_service.py
@@ -65,6 +65,7 @@
"linear": 1800,
"posthog_project": 1800,
"gong_calls": 21600,
+ "attio_calls": 21600,
"twitter_bookmarks": 21600,
# Freshness comes from extension pushes (which kick a sync); the interval
# is the retry pass for failed hydrations.
@@ -85,6 +86,7 @@
"linear": "navigable",
"posthog_project": "navigable",
"gong_calls": "searchable",
+ "attio_calls": "searchable",
"twitter": "searchable",
"twitter_bookmarks": "searchable",
"instagram_saves": "searchable",
@@ -102,6 +104,7 @@
"linear": ("linear",),
"posthog": ("posthog_project",),
"gong": ("gong_calls",),
+ "attio": ("attio_calls",),
"twitter": ("twitter", "twitter_bookmarks"),
# Provider-less grouping: there is no Instagram OAuth integration — the
# extension pushes the save list and ScrapeCreators hydrates it.
@@ -491,6 +494,7 @@ async def mark_sync_failed(source_id: UUID, error: str) -> None:
"linear": "linear_index",
"posthog_project": "posthog_index",
"gong_calls": "gong_documents",
+ "attio_calls": "attio_documents",
"twitter": "twitter_posts",
"twitter_bookmarks": "twitter_bookmark_docs",
"instagram_saves": "instagram_save_docs",
@@ -508,6 +512,7 @@ async def mark_sync_failed(source_id: UUID, error: str) -> None:
"slack_messages",
"granola_notes",
"gong_documents",
+ "attio_documents",
"notion_index",
# A picked Drive folder is bounded, so its bodies are extracted once at sync
# (OCR included) and stored. A whole-Drive source is not, and stays index-only.
diff --git a/backend/tasks/sources.py b/backend/tasks/sources.py
index 67cd85f8..c560d646 100644
--- a/backend/tasks/sources.py
+++ b/backend/tasks/sources.py
@@ -18,6 +18,7 @@
from ..celery_app import celery
from ..integrations.asana.indexer import index_asana
+from ..integrations.attio.indexer import index_attio
from ..integrations.github.indexer import index_github_repo
from ..integrations.gmail.indexer import index_gmail
from ..integrations.gong.indexer import index_gong
@@ -51,6 +52,7 @@
"asana_project": index_asana,
"linear": index_linear,
"gong_calls": index_gong,
+ "attio_calls": index_attio,
"twitter_bookmarks": index_twitter_bookmarks,
"instagram_saves": index_instagram_saves,
}
diff --git a/backend/tests/test_attio_integration.py b/backend/tests/test_attio_integration.py
new file mode 100644
index 00000000..0a1ad928
--- /dev/null
+++ b/backend/tests/test_attio_integration.py
@@ -0,0 +1,116 @@
+from contextlib import asynccontextmanager
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+
+from backend.integrations.attio import indexer
+from backend.integrations.attio.provider import AttioIntegration
+
+
+def _async(value):
+ async def result(*args, **kwargs):
+ return value
+
+ return result
+
+
+def test_authorize_url_carries_read_scopes_and_state(monkeypatch):
+ provider = AttioIntegration()
+ monkeypatch.setattr(provider, "_client_id", lambda: "client_abc")
+ monkeypatch.setattr(provider, "_redirect_uri", lambda: "https://app.example.com/cb")
+
+ url = provider.authorize_url("state_xyz")
+ params = {key: values[0] for key, values in parse_qs(urlparse(url).query).items()}
+
+ assert params["client_id"] == "client_abc"
+ assert params["response_type"] == "code"
+ assert params["state"] == "state_xyz"
+ assert params["scope"] == "meeting:read call_recording:read"
+
+
+def test_render_call_numbers_speakers_stably():
+ rendered = indexer._render_call(
+ "HeaviStash Sync",
+ "2026-07-01T10:00:00Z",
+ [
+ {"speech": "Hello there", "speaker": {"name": "Alice"}},
+ {"speech": "Hi", "speaker": {"name": "Bob"}},
+ {"speech": " ", "speaker": {"name": "Alice"}},
+ {"speech": "Back to you", "speaker": {"name": "Alice"}},
+ ],
+ )
+ assert rendered == (
+ "# HeaviStash Sync\n"
+ "Date: 2026-07-01T10:00:00Z\n"
+ "\n"
+ "[Speaker 1]: Hello there\n"
+ "[Speaker 2]: Hi\n"
+ "[Speaker 1]: Back to you"
+ )
+
+
+@pytest.mark.asyncio
+async def test_index_attio_walks_meetings_recordings_and_transcripts(monkeypatch):
+ responses = {
+ "/v2/meetings": {
+ "data": [{"id": {"meeting_id": "m1"}, "title": "HeaviStash Sync"}],
+ "pagination": {"next_cursor": None},
+ },
+ "/v2/meetings/m1/call_recordings": {
+ "data": [{"id": {"call_recording_id": "r1"}, "created_at": "2026-07-01T10:00:00Z"}],
+ "pagination": {"next_cursor": None},
+ },
+ "/v2/meetings/m1/call_recordings/r1/transcript": {
+ "data": {"transcript": [{"speech": "Hello", "speaker": {"name": "Alice"}}]},
+ "pagination": {"next_cursor": None},
+ },
+ }
+
+ class FakeResponse:
+ def __init__(self, payload):
+ self._payload = payload
+
+ def raise_for_status(self):
+ return None
+
+ def json(self):
+ return self._payload
+
+ class FakeClient:
+ async def get(self, url, params=None):
+ return FakeResponse(responses[url])
+
+ @asynccontextmanager
+ async def fake_async_client(*args, **kwargs):
+ assert kwargs["headers"] == {"Authorization": "Bearer access_token"}
+ yield FakeClient()
+
+ captured: list[dict] = []
+ removed: dict = {}
+
+ async def capture_upsert(**kwargs):
+ captured.append(kwargs)
+
+ async def capture_remove(table, source_id, present_paths):
+ removed["present"] = present_paths
+
+ monkeypatch.setattr(indexer, "get_valid_token", _async("access_token"))
+ monkeypatch.setattr(indexer.httpx, "AsyncClient", fake_async_client)
+ monkeypatch.setattr(indexer.source_service, "upsert_content_document", capture_upsert)
+ monkeypatch.setattr(indexer.source_service, "remove_missing_documents", capture_remove)
+
+ await indexer.index_attio(
+ {
+ "id": "00000000-0000-0000-0000-000000000001",
+ "owner_user_id": "00000000-0000-0000-0000-000000000002",
+ }
+ )
+
+ assert len(captured) == 1
+ doc = captured[0]
+ assert doc["table"] == "attio_documents"
+ assert doc["external_ref"] == "r1"
+ assert doc["path"] == "r1"
+ assert doc["kind"] == "call"
+ assert "[Speaker 1]: Hello" in doc["content"]
+ assert removed["present"] == ["r1"]
diff --git a/frontend/src/components/integrations/BrandIcons.tsx b/frontend/src/components/integrations/BrandIcons.tsx
index 25e8ab89..f42fffd7 100644
--- a/frontend/src/components/integrations/BrandIcons.tsx
+++ b/frontend/src/components/integrations/BrandIcons.tsx
@@ -118,6 +118,26 @@ export function GongIcon({ className, size = defaultSize }: Props) {
);
}
+export function AttioIcon({ className, size = defaultSize }: Props) {
+ // Attio mark — a dark rounded square with a white "A".
+ return (
+
+ );
+}
+
export function PostHogIcon({ className, size = defaultSize }: Props) {
return (