feat(connector): implement social platform connectors (#47)

Author: dineshpinto

.env.example

@@ -162,6 +162,27 @@ SOCIAL__X_API_KEY="your_x_api_key"
 
 # API key for Telegram.
 # SOCIAL__TELEGRAM_API_TOKEN=
+# SOCIAL__TELEGRAM_BOT_TOKEN=
+# SOCIAL__TELEGRAM_CHAT_ID=
+
+# --- discord ---
+
+#SOCIAL__DISCORD_BOT_TOKEN=
+#SOCIAL__DISCORD_GUILD_ID=
+#SOCIAL__DISCORD_CHANNEL_ID=
+
+# --- slack ---
+#SOCIAL__SLACK_BOT_TOKEN=
+#SOCIAL__SLACK_CHANNEL_ID=
+
+# --- farcaster ---
+# SOCIAL__FARCASTER_API_KEY=
+# SOCIAL__FARCASTER_API_URL=
+# SOCIAL__FARCASTER_SIGNER_UUID=
+
+# --- GITHUB ---
+# SOCIAL__GITHUB_TOKEN =
+# SOCIAL__GITHUB_REPO =
 
 # ==============================================================================
 #                      TRUSTED EXECUTION ENVIRONMENT (TEE)

pyproject.toml

@@ -53,11 +53,15 @@ social = [
     "python-telegram-bot>=22.0",
     "tweepy>=4.15.0",
     "async_lru>=2.0.5",
+    "slack-sdk>=3.26.2",
+    "discord.py >=2.3.2",
+    "httpx >=0.27.0"
 ]
 tee = [
     "cryptography>=44.0.2",
     "pyjwt>=2.10.1",
     "pyopenssl>=25.0.0",
+    "python-dotenv >=1.0.1" 
 ]
 
 [build-system]

src/flare_ai_kit/social/connector/__init__.py

@@ -0,0 +1,27 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class SocialConnector(ABC):
+    """Abstract base class for all social platform connectors."""
+
+    @abstractmethod
+    def __init__(self) -> None:
+        """Initialize connector (optional for future shared setup)."""
+
+    @property
+    @abstractmethod
+    def platform(self) -> str:
+        """The name of the platform (e.g., 'x', 'github', 'discord')."""
+
+    @abstractmethod
+    async def fetch_mentions(self, query: str) -> list[dict[str, Any]]:
+        """
+        Fetch public mentions or messages related to a given query.
+
+        Returns a list of dictionaries with at least.
+            - platform
+            - content
+            - author_id
+            - timestamp
+        """

src/flare_ai_kit/social/connector/discord.py

@@ -0,0 +1,94 @@
+"""Discord Connector for Flare AI Kit."""
+
+import asyncio
+from typing import Any
+
+from discord import Client, Intents, Message, TextChannel
+
+from flare_ai_kit.config import AppSettings
+from flare_ai_kit.social.connector import SocialConnector
+
+
+class DiscordConnector(SocialConnector):
+    """Discord connector implementation."""
+
+    def __init__(self) -> None:
+        """Initialize the DiscordConnector with API token and channel ID."""
+        settings = AppSettings().social
+        self.token: str = (
+            settings.discord_bot_token.get_secret_value()
+            if settings.discord_bot_token
+            else ""
+        )
+        self.channel_id: int = int(
+            settings.discord_channel_id.get_secret_value()
+            if settings.discord_channel_id
+            else 0
+        )
+        self.client: Client = Client(intents=Intents.default())
+        self._ready_event: asyncio.Event = asyncio.Event()
+        self._messages: list[dict[str, Any]] = []
+
+        # Explicitly register event handlers
+        self.client.event(self._on_ready)
+        self.client.event(self._on_message)
+
+    async def _on_ready(self) -> None:
+        """Handle bot ready event."""
+        self._ready_event.set()
+
+    async def _on_message(self, message: Message) -> None:
+        """Handle new messages."""
+        if message.author == self.client.user:
+            return
+
+        self._messages.append(
+            {
+                "platform": "discord",
+                "content": message.content,
+                "author_id": str(message.author.id),
+                "timestamp": str(message.created_at),
+            }
+        )
+
+    @property
+    def platform(self) -> str:
+        """Return the platform name."""
+        return "discord"
+
+    async def _start_if_needed(self) -> None:
+        if not self.client.is_ready():
+            self._client_task = asyncio.create_task(self.client.start(self.token))
+            await self._ready_event.wait()
+
+    async def fetch_mentions(
+        self, query: str = "", limit: int = 10
+    ) -> list[dict[str, Any]]:
+        """Fetch messages that mention the query."""
+        await self._start_if_needed()
+        await asyncio.sleep(1)  # let messages collect
+
+        results: list[dict[str, Any]] = []
+        for msg in self._messages:
+            if query.lower() in msg["content"].lower():
+                results.append(msg)
+                if len(results) >= limit:
+                    break
+        return results
+
+    async def post_message(self, content: str) -> dict[str, Any]:
+        """Post a message to the Discord channel."""
+        await self._start_if_needed()
+        channel = self.client.get_channel(self.channel_id)
+        if isinstance(channel, TextChannel):
+            message = await channel.send(content)
+            return {
+                "platform": "discord",
+                "message_id": message.id,
+                "content": message.content,
+            }
+        return {
+            "platform": "discord",
+            "message_id": None,
+            "error": "Channel not found or not a text channel.",
+        }

src/flare_ai_kit/social/connector/farcaster.py

@@ -0,0 +1,89 @@
+"""Farcaster Connector for Flare AI Kit."""
+
+from typing import Any
+
+import httpx
+
+from flare_ai_kit.config import AppSettings
+from flare_ai_kit.social.connector import SocialConnector
+
+
+class FarcasterConnector(SocialConnector):
+    """Farcaster Connector for Flare AI Kit."""
+
+    def __init__(self, client: httpx.AsyncClient | None = None) -> None:
+        """Initialize the FarcasterConnector with API key."""
+        settings = AppSettings().social
+        self.api_key = (
+            settings.farcaster_api_key.get_secret_value()
+            if settings.farcaster_api_key
+            else ""
+        )
+        self.signer_uuid = (
+            settings.farcaster_signer_uuid.get_secret_value()
+            if settings.farcaster_signer_uuid
+            else ""
+        )
+        self.fid = (
+            settings.farcaster_fid.get_secret_value() if settings.farcaster_fid else ""
+        )
+        self.api_url = (
+            settings.farcaster_api_url.get_secret_value()
+            if settings.farcaster_api_url
+            else ""
+        )
+        self.endpoint = f"{self.api_url}/v2/farcaster/feed/search"
+        self.client = client or httpx.AsyncClient()
+
+    @property
+    def platform(self) -> str:
+        """Return the platform name."""
+        return "farcaster"
+
+    async def fetch_mentions(self, query: str, limit: int = 10) -> list[dict[str, Any]]:
+        """Fetch mentions from Farcaster based on a query."""
+        try:
+            response = await self.client.get(
+                self.endpoint,
+                params={"text": query, "limit": limit},
+                headers={"api_key": self.api_key},
+            )
+            response.raise_for_status()
+            json_data = response.json()  # Already a dict in httpx
+            casts = json_data.get("casts", [])
+
+            return [
+                {
+                    "platform": self.platform,
+                    "content": cast.get("text", ""),
+                    "author_id": cast.get("author", {}).get("fid", ""),
+                    "timestamp": cast.get("timestamp", ""),
+                }
+                for cast in casts
+            ]
+        except httpx.HTTPError:
+            return []
+
+    async def post_message(self, content: str) -> dict[str, Any]:
+        """Post a message to Farcaster."""
+        try:
+            response = await self.client.post(
+                f"{self.api_url}/v2/casts",
+                headers={
+                    "Content-Type": "application/json",
+                    "x-api-key": self.api_key,
+                },
+                json={
+                    "text": content,
+                    "signer_uuid": self.signer_uuid,
+                },
+            )
+            response.raise_for_status()
+
+        except httpx.HTTPError as e:
+            return {"error": str(e)}
+        else:
+            return {
+                "platform": "farcaster",
+                "content": content,
+            }

src/flare_ai_kit/social/connector/github.py

@@ -0,0 +1,53 @@
+"""GitHub Connector for Flare AI Kit."""
+
+from typing import Any
+
+import httpx
+
+from flare_ai_kit.config import AppSettings
+from flare_ai_kit.social.connector import SocialConnector
+
+
+class GitHubConnector(SocialConnector):
+    """GitHub Connector for Flare AI Kit."""
+
+    def __init__(self) -> None:
+        """Initialize the GitHubConnector with API token and repository."""
+        settings = AppSettings().social
+        self.token = (
+            settings.github_token.get_secret_value() if settings.github_token else ""
+        )
+        self.repo = (
+            settings.github_repo.get_secret_value() if settings.github_repo else ""
+        )
+        self.client = httpx.AsyncClient()
+        self.endpoint = f"https://api.github.com/repos/{self.repo}/issues/comments"
+
+    @property
+    def platform(self) -> str:
+        """Return the platform name."""
+        return "github"
+
+    async def fetch_mentions(self, query: str, limit: int = 10) -> list[dict[str, Any]]:
+        """Fetch issue comments containing the query."""
+        try:
+            response = await self.client.get(
+                self.endpoint,
+                headers={"Authorization": f"Bearer {self.token}"},
+                params={"per_page": limit},
+            )
+            response.raise_for_status()
+            comments = await response.json()
+
+            return [
+                {
+                    "platform": self.platform,
+                    "content": comment.get("body", ""),
+                    "author_id": comment.get("user", {}).get("login", ""),
+                    "timestamp": comment.get("created_at", ""),
+                }
+                for comment in comments
+                if query.lower() in comment.get("body", "").lower()
+            ]
+        except httpx.HTTPError:
+            return []

src/flare_ai_kit/social/connector/slack.py

@@ -0,0 +1,81 @@
+"""Slack Connector for Flare AI Kit."""
+
+import logging
+from typing import Any
+
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
+
+from flare_ai_kit.config import AppSettings
+from flare_ai_kit.social.connector import SocialConnector
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+class SlackConnector(SocialConnector):
+    """Slack Connector for Flare AI Kit."""
+
+    def __init__(self, client: WebClient | None) -> None:
+        """Initialize the SlackConnector with API token and channel ID."""
+        settings = AppSettings().social
+        self.token = (
+            settings.slack_bot_token.get_secret_value()
+            if settings.slack_bot_token
+            else ""
+        )
+        self.channel_id = (
+            settings.slack_channel_id.get_secret_value()
+            if settings.slack_channel_id
+            else ""
+        )
+        self.client: WebClient = client or WebClient(token=self.token)
+
+    @property
+    def platform(self) -> str:
+        """Return the platform name."""
+        return "slack"
+
+    async def fetch_mentions(
+        self, query: str = "", limit: int = 10
+    ) -> list[dict[str, Any]]:
+        """Fetch messages from Slack channel that match the query."""
+        if not self.token or not self.channel_id:
+            return []
+
+        try:
+            response = self.client.conversations_history(  # type: ignore[reportUnknownMemberType]
+                channel=self.channel_id,
+                limit=100,
+            )
+            messages = response.get("messages", [])
+
+            results = [
+                {
+                    "platform": self.platform,
+                    "content": msg.get("text", ""),
+                    "author_id": msg.get("user", ""),
+                    "timestamp": msg.get("ts", ""),
+                }
+                for msg in messages
+                if query.lower() in msg.get("text", "").lower()
+            ]
+
+            return results[-limit:]
+        except SlackApiError:
+            logger.exception("Slack connector error: %s")
+            return []
+
+    def post_message(self, content: str) -> dict[str, Any]:
+        """Post a message to the Slack channel."""
+        try:
+            result = self.client.chat_postMessage(  # type: ignore[reportUnknownMemberType]
+                channel=self.channel_id, text=content
+            )
+            return {
+                "platform": "slack",
+                "message_ts": result["ts"],
+                "content": content,
+            }
+        except SlackApiError as e:
+            return {"error": str(e)}