feat(connectors): restore multi-service sync infrastructure

Author: haasonsaas

agent_pm/connectors/__init__.py

@@ -0,0 +1,19 @@
+"""Connector package exports."""
+
+from .base import Connector
+from .calendar import CalendarConnector
+from .email import EmailConnector
+from .github import GitHubConnector
+from .google_drive import GoogleDriveConnector
+from .notion import NotionConnector
+from .slack import SlackConnector
+
+__all__ = [
+    "Connector",
+    "CalendarConnector",
+    "EmailConnector",
+    "GitHubConnector",
+    "GoogleDriveConnector",
+    "NotionConnector",
+    "SlackConnector",
+]

agent_pm/connectors/base.py

@@ -0,0 +1,30 @@
+"""Base connector protocol."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from datetime import datetime
+from typing import Any
+
+
+class Connector(ABC):
+    """Abstract connector interface used by the sync manager."""
+
+    name: str
+
+    def __init__(self, *, name: str) -> None:
+        self.name = name
+
+    @property
+    @abstractmethod
+    def enabled(self) -> bool:
+        """Return True if the connector has the credentials it needs."""
+
+    @abstractmethod
+    async def sync(self, *, since: datetime | None = None) -> list[dict[str, Any]]:
+        """Fetch data updated since the provided timestamp."""
+
+    def format_metadata(self, **kwargs: Any) -> dict[str, Any]:
+        data = {"connector": self.name}
+        data.update(kwargs)
+        return data

agent_pm/connectors/calendar.py

@@ -0,0 +1,104 @@
+"""Google Calendar connector with dry-run support."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from agent_pm.connectors.base import Connector
+from agent_pm.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+class CalendarConnector(Connector):
+    def __init__(self) -> None:
+        super().__init__(name="calendar")
+        self._scopes = settings.google_calendar_scopes
+        self._delegated_user = settings.google_calendar_delegated_user
+
+    @property
+    def enabled(self) -> bool:
+        json_creds = settings.google_service_account_json
+        file_creds = settings.google_service_account_file
+        return bool(settings.calendar_id and (json_creds or (file_creds and Path(file_creds).exists())))
+
+    async def sync(self, *, since: datetime | None = None) -> list[dict[str, Any]]:
+        calendar_id = settings.calendar_id
+        window_days = settings.calendar_sync_window_days
+        if settings.dry_run or not self.enabled:
+            return [
+                {
+                    "dry_run": True,
+                    "calendar_id": calendar_id,
+                    "since": since.isoformat() if since else None,
+                    "window_days": window_days,
+                }
+            ]
+
+        token = await self._get_token()
+        headers = {"Authorization": f"Bearer {token}"}
+        lower_bound = datetime.now(tz=UTC) - timedelta(days=window_days)
+        time_min = (since if since and since > lower_bound else lower_bound).isoformat()
+        time_max = (datetime.now(tz=UTC) + timedelta(days=window_days)).isoformat()
+        params = {"timeMin": time_min, "timeMax": time_max, "singleEvents": "true", "orderBy": "startTime"}
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}/events",
+                headers=headers,
+                params=params,
+                timeout=30,
+            )
+        response.raise_for_status()
+        return [response.json()]
+
+    async def _get_token(self) -> str:
+        creds = await asyncio.to_thread(self._load_credentials)
+        if not creds:
+            raise RuntimeError("Google Calendar credentials missing")
+        if not creds.valid or creds.expired:
+            await asyncio.to_thread(creds.refresh, self._request())
+        if not creds.token:
+            raise RuntimeError("Failed to obtain calendar token")
+        return creds.token
+
+    def _load_credentials(self):
+        info: dict[str, Any] | None = None
+        if settings.google_service_account_json:
+            try:
+                info = json.loads(settings.google_service_account_json)
+            except json.JSONDecodeError as exc:  # pragma: no cover - defensive
+                logger.error("Invalid GOOGLE_SERVICE_ACCOUNT_JSON: %s", exc)
+        elif settings.google_service_account_file:
+            try:
+                info = json.loads(Path(settings.google_service_account_file).read_text(encoding="utf-8"))
+            except FileNotFoundError:
+                logger.error("GOOGLE_SERVICE_ACCOUNT_FILE not found: %s", settings.google_service_account_file)
+            except json.JSONDecodeError as exc:  # pragma: no cover - defensive
+                logger.error("Invalid GOOGLE_SERVICE_ACCOUNT_FILE: %s", exc)
+        if not info:
+            return None
+
+        try:
+            from google.oauth2 import service_account
+        except ImportError as exc:  # pragma: no cover - optional dependency
+            raise RuntimeError("google-auth not installed") from exc
+
+        credentials = service_account.Credentials.from_service_account_info(info, scopes=self._scopes)
+        if self._delegated_user:
+            credentials = credentials.with_subject(self._delegated_user)
+        return credentials
+
+    @staticmethod
+    def _request():
+        try:
+            from google.auth.transport.requests import Request
+        except ImportError as exc:  # pragma: no cover - optional dependency
+            raise RuntimeError("google-auth not installed") from exc
+        return Request()

agent_pm/connectors/email.py

@@ -0,0 +1,140 @@
+"""Gmail connector returning structured message payloads."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from agent_pm.connectors.base import Connector
+from agent_pm.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+class EmailConnector(Connector):
+    def __init__(self) -> None:
+        super().__init__(name="gmail")
+        self._delegated_user = settings.gmail_delegated_user
+        self._labels = settings.gmail_label_filter
+        self._scopes = settings.gmail_scopes
+
+    @property
+    def enabled(self) -> bool:
+        json_creds = settings.gmail_service_account_json
+        file_creds = settings.gmail_service_account_file
+        return bool(json_creds or (file_creds and Path(file_creds).exists()))
+
+    def _build_query(self, since: datetime | None) -> str:
+        clauses: list[str] = []
+        if since:
+            clauses.append(f"after:{since.strftime('%Y/%m/%d')}")
+        if self._labels:
+            clauses.append(" OR ".join(f"label:{label}" for label in self._labels))
+        return " ".join(filter(None, clauses))
+
+    async def sync(self, *, since: datetime | None = None) -> list[dict[str, Any]]:
+        query = self._build_query(since)
+        if settings.dry_run or not self.enabled:
+            return [
+                {
+                    "dry_run": True,
+                    "query": query,
+                    "labels": self._labels,
+                    "delegated_user": self._delegated_user,
+                }
+            ]
+
+        token = await self._get_token()
+        headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+        params = {"q": query} if query else {}
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                "https://gmail.googleapis.com/gmail/v1/users/me/messages",
+                headers=headers,
+                params=params,
+                timeout=30,
+            )
+        response.raise_for_status()
+        message_list = response.json()
+
+        ids = [item.get("id") for item in message_list.get("messages", []) if item.get("id")]
+        if not ids:
+            return [message_list]
+
+        messages: list[dict[str, Any]] = []
+        async with httpx.AsyncClient() as client:
+            for message_id in ids:
+                detail = await client.get(
+                    f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{message_id}",
+                    headers=headers,
+                    timeout=30,
+                )
+                detail.raise_for_status()
+                data = detail.json()
+                payload = data.get("payload", {})
+                headers_list = payload.get("headers", [])
+                header_map = {item.get("name"): item.get("value") for item in headers_list}
+                messages.append(
+                    {
+                        "id": message_id,
+                        "thread_id": data.get("threadId"),
+                        "snippet": data.get("snippet"),
+                        "subject": header_map.get("Subject"),
+                        "from": header_map.get("From"),
+                        "to": header_map.get("To"),
+                        "date": header_map.get("Date"),
+                        "labels": data.get("labelIds", []),
+                    }
+                )
+        return [message_list, {"messages": messages}]
+
+    async def _get_token(self) -> str:
+        creds = await asyncio.to_thread(self._load_credentials)
+        if not creds:
+            raise RuntimeError("Gmail credentials missing")
+        if not creds.valid or creds.expired:
+            await asyncio.to_thread(creds.refresh, self._request())
+        if not creds.token:
+            raise RuntimeError("Failed to refresh Gmail access token")
+        return creds.token
+
+    def _load_credentials(self):
+        info: dict[str, Any] | None = None
+        if settings.gmail_service_account_json:
+            try:
+                info = json.loads(settings.gmail_service_account_json)
+            except json.JSONDecodeError as exc:  # pragma: no cover - defensive
+                logger.error("Invalid GMAIL_SERVICE_ACCOUNT_JSON: %s", exc)
+        elif settings.gmail_service_account_file:
+            try:
+                info = json.loads(Path(settings.gmail_service_account_file).read_text(encoding="utf-8"))
+            except FileNotFoundError:
+                logger.error("GMAIL_SERVICE_ACCOUNT_FILE not found: %s", settings.gmail_service_account_file)
+            except json.JSONDecodeError as exc:  # pragma: no cover - defensive
+                logger.error("Invalid GMAIL_SERVICE_ACCOUNT_FILE: %s", exc)
+        if not info:
+            return None
+
+        try:
+            from google.oauth2 import service_account
+        except ImportError as exc:  # pragma: no cover - optional dependency
+            raise RuntimeError("google-auth not installed") from exc
+
+        credentials = service_account.Credentials.from_service_account_info(info, scopes=self._scopes)
+        if self._delegated_user:
+            credentials = credentials.with_subject(self._delegated_user)
+        return credentials
+
+    @staticmethod
+    def _request():
+        try:
+            from google.auth.transport.requests import Request
+        except ImportError as exc:  # pragma: no cover - optional dependency
+            raise RuntimeError("google-auth not installed") from exc
+        return Request()

agent_pm/connectors/github.py

@@ -0,0 +1,59 @@
+"""GitHub connector for synchronizing repositories, issues, and pull requests."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+import httpx
+
+from agent_pm.connectors.base import Connector
+from agent_pm.settings import settings
+
+
+class GitHubConnector(Connector):
+    def __init__(self) -> None:
+        super().__init__(name="github")
+        self._token = settings.github_token
+        self._repositories = settings.github_repositories
+        self._base_url = "https://api.github.com"
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self._token and self._repositories)
+
+    async def _fetch(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        if settings.dry_run or not self.enabled:
+            return {"dry_run": True, "path": path, "params": params or {}, "repositories": self._repositories}
+
+        headers = {
+            "Authorization": f"Bearer {self._token}",
+            "Accept": "application/vnd.github+json",
+        }
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"{self._base_url}{path}", headers=headers, params=params, timeout=30)
+        response.raise_for_status()
+        return response.json()
+
+    async def _sync_repo(self, repo: str, since: datetime | None) -> dict[str, Any]:
+        params: dict[str, Any] = {}
+        if since:
+            params["since"] = since.astimezone(UTC).isoformat()
+
+        repo_data = await self._fetch(f"/repos/{repo}")
+        issues = await self._fetch(f"/repos/{repo}/issues", params)
+        pulls = await self._fetch(f"/repos/{repo}/pulls", params)
+        return {
+            "repository": repo,
+            "repository_data": repo_data,
+            "issues": issues,
+            "pull_requests": pulls,
+        }
+
+    async def sync(self, *, since: datetime | None = None) -> list[dict[str, Any]]:
+        if not self._repositories:
+            return []
+        payloads = []
+        for repo in self._repositories:
+            payloads.append(await self._sync_repo(repo, since))
+        return payloads