Add Notion and Google Drive connector routes

src/api/app.py

@@ -32,6 +32,7 @@
 from src.api.routes.auth import router as auth_router
 from src.api.routes.billing import router as billing_router
 from src.api.routes.code import router as code_router
+from src.api.routes.connectors import router as connectors_router
 from src.api.routes.enterprise import router as enterprise_router
 from src.api.routes.health import router as health_router
 from src.api.routes.memory import router as memory_router
@@ -226,6 +227,7 @@ async def lifespan(app: FastAPI):
     app.include_router(scanner_router)
     app.include_router(auth_router)
     app.include_router(api_keys_router)
+    app.include_router(connectors_router)
     app.include_router(billing_router)
     app.include_router(enterprise_router)
     app.include_router(telemetry_router)

src/api/routes/connectors.py

@@ -0,0 +1,254 @@
+"""Connector OAuth routes for external knowledge sources."""
+
+from __future__ import annotations
+
+import secrets
+from datetime import datetime, timedelta, timezone
+from typing import Dict, List, Literal, Optional
+from urllib.parse import urlencode
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
+
+from src.api.dependencies import require_user
+from src.config import settings
+
+
+router = APIRouter(prefix="/api/connectors", tags=["Connectors"])
+
+ConnectorId = Literal["notion", "google-drive"]
+ConnectorState = Literal["connected", "not_connected", "pending"]
+
+STATE_TTL_MINUTES = 10
+
+
+class ConnectorDefinition(BaseModel):
+    id: ConnectorId
+    name: str
+    description: str
+    auth_url: str
+    token_url: str
+    scopes: List[str]
+    docs_url: str
+
+
+class ConnectorStatusResponse(BaseModel):
+    id: ConnectorId
+    name: str
+    state: ConnectorState
+    connected_at: Optional[datetime] = None
+    scopes: List[str] = Field(default_factory=list)
+    detail: str
+
+
+class ConnectorListResponse(BaseModel):
+    connectors: List[ConnectorStatusResponse]
+
+
+class ConnectorStartResponse(BaseModel):
+    connector_id: ConnectorId
+    authorization_url: str
+    state: str
+    expires_at: datetime
+
+
+class ConnectorDisconnectResponse(BaseModel):
+    connector_id: ConnectorId
+    disconnected: bool
+
+
+class PendingOAuthState(BaseModel):
+    connector_id: ConnectorId
+    user_id: str
+    expires_at: datetime
+
+
+class StoredConnection(BaseModel):
+    connector_id: ConnectorId
+    user_id: str
+    connected_at: datetime
+    scopes: List[str]
+
+
+CONNECTORS: Dict[ConnectorId, ConnectorDefinition] = {
+    "notion": ConnectorDefinition(
+        id="notion",
+        name="Notion",
+        description="Sync selected Notion pages and workspace notes into XMem memory.",
+        auth_url="https://api.notion.com/v1/oauth/authorize",
+        token_url="https://api.notion.com/v1/oauth/token",
+        scopes=[],
+        docs_url="https://developers.notion.com/docs/authorization",
+    ),
+    "google-drive": ConnectorDefinition(
+        id="google-drive",
+        name="Google Drive",
+        description="Bring Google Drive docs and files into XMem as searchable memory.",
+        auth_url="https://accounts.google.com/o/oauth2/v2/auth",
+        token_url="https://oauth2.googleapis.com/token",
+        scopes=[
+            "https://www.googleapis.com/auth/drive.readonly",
+            "https://www.googleapis.com/auth/documents.readonly",
+        ],
+        docs_url="https://developers.google.com/identity/protocols/oauth2",
+    ),
+}
+
+_pending_states: Dict[str, PendingOAuthState] = {}
+_connections: Dict[str, StoredConnection] = {}
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _connection_key(user_id: str, connector_id: ConnectorId) -> str:
+    return f"{user_id}:{connector_id}"
+
+
+def _client_id(connector_id: ConnectorId) -> Optional[str]:
+    if connector_id == "notion":
+        return settings.notion_client_id
+    return settings.google_drive_client_id
+
+
+def _redirect_uri(connector_id: ConnectorId) -> str:
+    if connector_id == "notion":
+        return settings.notion_redirect_uri
+    return settings.google_drive_redirect_uri
+
+
+def _get_connector(connector_id: str) -> ConnectorDefinition:
+    connector = CONNECTORS.get(connector_id)  # type: ignore[arg-type]
+    if not connector:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Unknown connector")
+    return connector
+
+
+def _status_for(user_id: str, connector: ConnectorDefinition) -> ConnectorStatusResponse:
+    connection = _connections.get(_connection_key(user_id, connector.id))
+    if connection:
+        return ConnectorStatusResponse(
+            id=connector.id,
+            name=connector.name,
+            state="connected",
+            connected_at=connection.connected_at,
+            scopes=connection.scopes,
+            detail="Connected",
+        )
+
+    return ConnectorStatusResponse(
+        id=connector.id,
+        name=connector.name,
+        state="not_connected",
+        scopes=connector.scopes,
+        detail="Not connected",
+    )
+
+
+def _build_authorization_url(connector: ConnectorDefinition, state: str) -> str:
+    client_id = _client_id(connector.id)
+    if not client_id:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"{connector.name} OAuth client ID is not configured",
+        )
+
+    params = {
+        "client_id": client_id,
+        "redirect_uri": _redirect_uri(connector.id),
+        "response_type": "code",
+        "state": state,
+    }
+    if connector.id == "google-drive":
+        params.update(
+            {
+                "access_type": "offline",
+                "include_granted_scopes": "true",
+                "prompt": "consent",
+                "scope": " ".join(connector.scopes),
+            }
+        )
+    if connector.id == "notion":
+        params["owner"] = "user"
+
+    return f"{connector.auth_url}?{urlencode(params)}"
+
+
+@router.get("", response_model=ConnectorListResponse)
+async def list_connectors(current_user: dict = Depends(require_user)) -> ConnectorListResponse:
+    user_id = str(current_user.get("id"))
+    return ConnectorListResponse(
+        connectors=[_status_for(user_id, connector) for connector in CONNECTORS.values()]
+    )
+
+
+@router.get("/{connector_id}/status", response_model=ConnectorStatusResponse)
+async def connector_status(
+    connector_id: str,
+    current_user: dict = Depends(require_user),
+) -> ConnectorStatusResponse:
+    connector = _get_connector(connector_id)
+    return _status_for(str(current_user.get("id")), connector)
+
+
+@router.post("/{connector_id}/oauth/start", response_model=ConnectorStartResponse)
+async def start_connector_oauth(
+    connector_id: str,
+    current_user: dict = Depends(require_user),
+) -> ConnectorStartResponse:
+    connector = _get_connector(connector_id)
+    state = secrets.token_urlsafe(32)
+    expires_at = _now() + timedelta(minutes=STATE_TTL_MINUTES)
+    _pending_states[state] = PendingOAuthState(
+        connector_id=connector.id,
+        user_id=str(current_user.get("id")),
+        expires_at=expires_at,
+    )
+
+    return ConnectorStartResponse(
+        connector_id=connector.id,
+        authorization_url=_build_authorization_url(connector, state),
+        state=state,
+        expires_at=expires_at,
+    )
+
+
+@router.get("/{connector_id}/oauth/callback")
+async def connector_oauth_callback(
+    connector_id: str,
+    code: str = Query(..., min_length=1),
+    state: str = Query(..., min_length=1),
+) -> dict:
+    connector = _get_connector(connector_id)
+    pending = _pending_states.pop(state, None)
+    if not pending or pending.connector_id != connector.id or pending.expires_at <= _now():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Invalid or expired connector authorization state",
+        )
+
+    # Token exchange and source ingestion are intentionally separate follow-up steps.
+    # This callback validates the flow and records a pending connection marker only.
+    _connections[_connection_key(pending.user_id, connector.id)] = StoredConnection(
+        connector_id=connector.id,
+        user_id=pending.user_id,
+        connected_at=_now(),
+        scopes=connector.scopes,
+    )
+    return {
+        "status": "connected",
+        "connector_id": connector.id,
+        "detail": f"{connector.name} authorization received",
+    }
+
+
+@router.post("/{connector_id}/disconnect", response_model=ConnectorDisconnectResponse)
+async def disconnect_connector(
+    connector_id: str,
+    current_user: dict = Depends(require_user),
+) -> ConnectorDisconnectResponse:
+    connector = _get_connector(connector_id)
+    key = _connection_key(str(current_user.get("id")), connector.id)
+    disconnected = _connections.pop(key, None) is not None
+    return ConnectorDisconnectResponse(connector_id=connector.id, disconnected=disconnected)

src/config/settings.py

@@ -401,6 +401,30 @@ class Settings(BaseSettings):
         default="http://localhost:8000/auth/callback",
         description="Google OAuth redirect URI"
     )
+    notion_client_id: Optional[str] = Field(
+        default=None,
+        description="Notion OAuth client ID for the Notion connector"
+    )
+    notion_client_secret: Optional[str] = Field(
+        default=None,
+        description="Notion OAuth client secret for the Notion connector"
+    )
+    notion_redirect_uri: str = Field(
+        default="http://localhost:8000/api/connectors/notion/oauth/callback",
+        description="Notion OAuth redirect URI"
+    )
+    google_drive_client_id: Optional[str] = Field(
+        default=None,
+        description="Google OAuth client ID for the Google Drive connector"
+    )
+    google_drive_client_secret: Optional[str] = Field(
+        default=None,
+        description="Google OAuth client secret for the Google Drive connector"
+    )
+    google_drive_redirect_uri: str = Field(
+        default="http://localhost:8000/api/connectors/google-drive/oauth/callback",
+        description="Google Drive OAuth redirect URI"
+    )
     jwt_secret_key: str = Field(
         default="your-secret-key-change-in-production",
         description="Secret key for JWT token signing (change in production!)"

tests/api/test_connectors.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.dependencies import require_user
+from src.api.routes import connectors
+
+
+def _user() -> dict:
+    return {"id": "user-1", "email": "user@example.com", "username": "user"}
+
+
+def _client() -> TestClient:
+    app = FastAPI()
+    app.dependency_overrides[require_user] = _user
+    app.include_router(connectors.router)
+    return TestClient(app)
+
+
+def test_lists_supported_connectors() -> None:
+    response = _client().get("/api/connectors")
+
+    assert response.status_code == 200
+    body = response.json()
+    ids = {item["id"] for item in body["connectors"]}
+    assert ids == {"notion", "google-drive"}
+    assert {item["state"] for item in body["connectors"]} == {"not_connected"}
+
+
+def test_oauth_start_requires_configured_client_id(monkeypatch) -> None:
+    monkeypatch.setattr(connectors.settings, "notion_client_id", None, raising=False)
+
+    response = _client().post("/api/connectors/notion/oauth/start")
+
+    assert response.status_code == 503
+    assert "client ID is not configured" in response.json()["detail"]
+
+
+def test_oauth_start_builds_authorization_url_without_secret(monkeypatch) -> None:
+    monkeypatch.setattr(connectors.settings, "google_drive_client_id", "drive-client", raising=False)
+    monkeypatch.setattr(connectors.settings, "google_drive_client_secret", "do-not-leak", raising=False)
+    monkeypatch.setattr(
+        connectors.settings,
+        "google_drive_redirect_uri",
+        "http://localhost:8000/api/connectors/google-drive/oauth/callback",
+        raising=False,
+    )
+
+    response = _client().post("/api/connectors/google-drive/oauth/start")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["connector_id"] == "google-drive"
+    assert "accounts.google.com" in body["authorization_url"]
+    assert "client_id=drive-client" in body["authorization_url"]
+    assert "do-not-leak" not in body["authorization_url"]
+    assert body["state"]
+
+
+def test_callback_marks_connection_then_disconnects(monkeypatch) -> None:
+    monkeypatch.setattr(connectors.settings, "notion_client_id", "notion-client", raising=False)
+    client = _client()
+
+    started = client.post("/api/connectors/notion/oauth/start")
+    state = started.json()["state"]
+
+    callback = client.get(f"/api/connectors/notion/oauth/callback?code=abc&state={state}")
+    assert callback.status_code == 200
+    assert callback.json()["status"] == "connected"
+
+    connected = client.get("/api/connectors/notion/status")
+    assert connected.status_code == 200
+    assert connected.json()["state"] == "connected"
+
+    disconnected = client.post("/api/connectors/notion/disconnect")
+    assert disconnected.status_code == 200
+    assert disconnected.json() == {"connector_id": "notion", "disconnected": True}