Add Attio as a connected source (call recording transcripts)

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(

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

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())

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

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")

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")

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

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.