feat: add Intercom integration

backend/app/main.py

@@ -23,6 +23,7 @@
 from app.routers import hubspot_router
 from app.routers import salesforce_router
 from app.routers import ga4_router
+from app.routers import intercom_router
 from app.models import User
 from app.auth import hash_password, GUEST_USER_ID, ADMIN_USER_ID
 
@@ -249,6 +250,7 @@ async def httpx_connect_error_handler(_request: Request, exc: httpx.ConnectError
 app.include_router(hubspot_router.router, prefix=prefix)
 app.include_router(salesforce_router.router, prefix=prefix)
 app.include_router(ga4_router.router, prefix=prefix)
+app.include_router(intercom_router.router, prefix=prefix)
 app.include_router(pipedrive_router.router, prefix=prefix)
 
 

backend/app/routers/ask.py

@@ -37,6 +37,7 @@
 from app.scripts.ask_stripe import ask_stripe
 from app.scripts.ask_pipedrive import ask_pipedrive
 from app.scripts.ask_ga4 import ask_ga4
+from app.scripts.ask_intercom import ask_intercom
 from app.scripts.ask_salesforce import ask_salesforce
 from app.scripts.ask_sql import ask_sql
 from app.scripts.ask_sql_multi import ask_sql_multi_source
@@ -544,6 +545,23 @@ async def dispatch_question(
             history=history,
             channel=channel,
         )
+    elif source.type == "intercom":
+        meta = source.metadata_ or {}
+        ic_token = meta.get("accessToken", "")
+        if not ic_token:
+            raise HTTPException(400, "Intercom source missing accessToken in metadata")
+        result = await ask_intercom(
+            access_token=ic_token,
+            selected_resources=meta.get("selectedResources"),
+            question=question,
+            agent_description=agent.description or "",
+            source_name=source.name,
+            schema_text=meta.get("schemaText"),
+            preview=meta.get("preview"),
+            llm_overrides=llm_overrides,
+            history=history,
+            channel=channel,
+        )
     else:
         raise HTTPException(400, f"Unsupported source type: {source.type}")
 

backend/app/routers/crud.py

@@ -135,7 +135,7 @@ async def create_source(
     user: User = Depends(require_user),
 ):
     """Create a non-file source (BigQuery, Google Sheets, SQL). Credentials stored locally in metadata."""
-    valid_types = ("bigquery", "google_sheets", "sql_database", "firebase", "mongodb", "snowflake", "notion", "excel_online", "s3", "rest_api", "jira", "hubspot", "stripe", "pipedrive", "salesforce", "ga4")
+    valid_types = ("bigquery", "google_sheets", "sql_database", "firebase", "mongodb", "snowflake", "notion", "excel_online", "s3", "rest_api", "jira", "hubspot", "stripe", "pipedrive", "salesforce", "ga4", "intercom")
     if body.type not in valid_types:
         raise HTTPException(400, f"type must be one of: {', '.join(valid_types)}")
     source_id = str(uuid.uuid4())

backend/app/routers/intercom_router.py

@@ -0,0 +1,141 @@
+"""
+Intercom integration: test connection, discover resources, refresh metadata.
+- Test connection (validate access token via GET /me)
+- Discover resources (list available Intercom tables and counts)
+- Refresh source metadata (schema + preview)
+"""
+import asyncio
+from fastapi import APIRouter, Depends, HTTPException
+from app.auth import require_user
+from app.models import User, Source
+from app.database import get_db
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.routers.crud import _sanitize_for_json
+
+router = APIRouter(prefix="/intercom", tags=["intercom"])
+
+
+@router.post("/test-connection")
+async def test_connection(
+    body: dict,
+    user: User = Depends(require_user),
+):
+    """Test an Intercom access token. Body: { "accessToken": "..." }."""
+    token = body.get("accessToken")
+    if not token:
+        raise HTTPException(400, "accessToken is required")
+
+    from app.scripts.ask_intercom import _test_connection_sync
+
+    loop = asyncio.get_event_loop()
+    try:
+        me = await loop.run_in_executor(None, lambda: _test_connection_sync(token))
+        return {"ok": True, "app": me.get("app", {}).get("name", ""), "admin": me.get("name", "")}
+    except Exception as e:
+        raise HTTPException(400, f"Connection failed: {e}")
+
+
+@router.post("/discover")
+async def discover_resources(
+    body: dict,
+    user: User = Depends(require_user),
+):
+    """Discover available Intercom resources. Body: { "accessToken": "..." }."""
+    token = body.get("accessToken")
+    if not token:
+        raise HTTPException(400, "accessToken is required")
+
+    from app.scripts.ask_intercom import _discover_resources_sync
+
+    loop = asyncio.get_event_loop()
+    try:
+        resources = await loop.run_in_executor(
+            None, lambda: _discover_resources_sync(token)
+        )
+    except Exception as e:
+        raise HTTPException(400, str(e))
+
+    return {"resources": resources}
+
+
+@router.post("/sources/{source_id}/refresh-metadata")
+async def refresh_source_metadata(
+    source_id: str,
+    db: AsyncSession = Depends(get_db),
+    user: User = Depends(require_user),
+):
+    """Fetch Intercom resource schemas and preview data, update source metadata."""
+    r = await db.execute(
+        select(Source).where(Source.id == source_id, Source.user_id == user.id)
+    )
+    source = r.scalar_one_or_none()
+    if not source:
+        raise HTTPException(404, "Source not found")
+    if source.type != "intercom":
+        raise HTTPException(400, "Source is not Intercom")
+
+    meta = dict(source.metadata_ or {})
+    token = meta.get("accessToken")
+
+    if not token:
+        raise HTTPException(400, "Source missing accessToken")
+
+    from app.scripts.ask_intercom import (
+        _fetch_contacts_sync,
+        _fetch_companies_sync,
+        _fetch_conversations_sync,
+        _fetch_tags_sync,
+        _fetch_teams_sync,
+        _fetch_admins_sync,
+        _fetch_articles_sync,
+        _flatten,
+        _records_to_columns,
+        INTERCOM_TABLES,
+    )
+
+    selected = meta.get("selectedResources") or INTERCOM_TABLES
+    loop = asyncio.get_event_loop()
+
+    fetchers = {
+        "contacts": lambda: _fetch_contacts_sync(token),
+        "companies": lambda: _fetch_companies_sync(token),
+        "conversations": lambda: _fetch_conversations_sync(token),
+        "tags": lambda: _fetch_tags_sync(token),
+        "teams": lambda: _fetch_teams_sync(token),
+        "admins": lambda: _fetch_admins_sync(token),
+        "articles": lambda: _fetch_articles_sync(token),
+    }
+
+    table_infos: list[dict] = []
+    preview_data: dict[str, list[dict]] = {}
+
+    for resource in selected:
+        if resource == "conversation_parts":
+            continue
+        if resource not in fetchers:
+            continue
+        try:
+            data = await loop.run_in_executor(None, fetchers[resource])
+            columns = _records_to_columns(data[:100]) if data else []
+            table_infos.append({
+                "table": resource,
+                "columns": columns,
+                "row_count": len(data),
+            })
+            preview_data[resource] = [_flatten(r) for r in data[:5]]
+        except Exception:
+            table_infos.append({
+                "table": resource,
+                "columns": [],
+                "row_count": 0,
+            })
+
+    meta["table_infos"] = table_infos
+    meta["preview"] = preview_data
+    meta["schema"] = {"tables": [ti["table"] for ti in table_infos]}
+    source.metadata_ = _sanitize_for_json(meta)
+    await db.commit()
+    await db.refresh(source)
+    return {"metaJSON": source.metadata_}