diff --git a/GETTING_STARTED.md b/GETTING_STARTED.md index dc7332a3..d57972c2 100644 --- a/GETTING_STARTED.md +++ b/GETTING_STARTED.md @@ -163,23 +163,9 @@ Your agents can now use Google Calendar, Gmail, Drive, etc. through MCP tools. --- -## Instagram (`instagram_dm_mcp` via local install) (optional) +## Instagram (built in) -OpenSwarm uses [ShawnMadadha/instagram_dm_mcp](https://github.com/ShawnMadadha/instagram_dm_mcp), a rate-limited fork of trypeggy/instagram_dm_mcp. 25 tools for DMs, user/follower lookup, post engagement, and story reads, powered by `instagrapi` (pure HTTP, no browser). The server enforces per-category rate limits on sends, likes, searches, lookups, and modifications to protect the connected account from being flagged for automation. - -### Prerequisites - -`git` and `python3` on `PATH`. - -### One-time install - -From the repo root: - -```bash -bash scripts/setup-instagram-mcp.sh -``` - -This clones the server into `~/.openswarm/instagram-mcp/`, creates a venv, and pip-installs `instagrapi` + the rest of the dependencies. Re-running upgrades to the latest fork commit. +The Instagram MCP server is vendored at `backend/apps/instagram_mcp/` and ships with OpenSwarm. 25 tools for DMs, user/follower lookup, post engagement, and story reads, powered by `instagrapi` (HTTP only, no browser). No setup script, no clone, no external dependencies beyond the backend's existing Python env. ### Connect from the UI @@ -188,9 +174,11 @@ This clones the server into `~/.openswarm/instagram-mcp/`, creates a venv, and p 3. Enter the username and password of the Instagram account the agent should use. 4. Tile flips to **Connected**. -Session state is cached at `~/.instagram_dm_mcp/sessions/_session.json` (per OS user, isolated from any project checkout) so future restarts skip the password prompt. +The backend validates the credentials via instagrapi and saves the session to `~/.instagram_dm_mcp/sessions/_session.json`. Future tool calls spawn `python -m backend.apps.instagram_mcp` which reloads that session, so subsequent restarts skip the password prompt. -### Rate limits (built into the server) +### Built-in rate limiting + +The server enforces per-category caps to protect the connected account from anti-abuse bans. All caps are well below Instagram's documented thresholds and apply across server restarts. | Category | Tools | per_min | per_hour | per_day | |---|---|---:|---:|---:| @@ -200,15 +188,15 @@ Session state is cached at `~/.instagram_dm_mcp/sessions/_session.json | `lookup` | 16 read tools | 30 | 300 | 2000 | | `modify` | `mark_message_seen`, `mute_conversation`, `delete_message` | 10 | 100 | 500 | -Plus randomized jitter (1.5–4s before DMs, 0.5–2s before likes, smaller elsewhere) so action timing isn't bot-perfect. +Plus randomized jitter (1.5 to 4s before DMs, 0.5 to 2s before likes, smaller elsewhere) so action timing isn't bot-perfect. -Overridable per env var, e.g.: +Override any cap via env var, for example: ```bash export IG_RATE_LIMIT_DM_SEND_PER_DAY=40 ``` -Rate-limit state persists at `~/.instagram-mcp-rate-limits.json` so a server restart doesn't reset the daily budget. When a cap is hit, the tool returns a structured `{ok: false, rate_limited: true, retry_after_seconds: ...}` response the agent can surface as *"hit DM cap, retry in 4h"* instead of failing opaquely. +State persists at `~/.instagram-mcp-rate-limits.json`. When a cap is hit, the tool returns a structured `{ok: false, rate_limited: true, retry_after_seconds: ...}` response the agent can surface as *"hit DM cap, retry in 4h"* instead of failing opaquely. --- diff --git a/backend/apps/instagram_mcp/__init__.py b/backend/apps/instagram_mcp/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/instagram_mcp/__main__.py b/backend/apps/instagram_mcp/__main__.py new file mode 100644 index 00000000..9852fc15 --- /dev/null +++ b/backend/apps/instagram_mcp/__main__.py @@ -0,0 +1,12 @@ +"""Module entrypoint so `python -m backend.apps.instagram_mcp` works. + +Spawned by OpenSwarm when an agent invokes an Instagram tool. Authentication +happens before this server starts — the backend's /credentials/instagram/* +endpoints validate the user's credentials, dump an instagrapi session to +~/.instagram_dm_mcp/sessions/_session.json, and set +INSTAGRAM_USERNAME in the env that this server inherits. +""" +from backend.apps.instagram_mcp.server import main + +if __name__ == "__main__": + main() diff --git a/backend/apps/instagram_mcp/rate_limiter.py b/backend/apps/instagram_mcp/rate_limiter.py new file mode 100644 index 00000000..e2fa567c --- /dev/null +++ b/backend/apps/instagram_mcp/rate_limiter.py @@ -0,0 +1,172 @@ +""" +Rate limiter for Instagram MCP tools. + +Why this exists: Instagram's anti-abuse system flags accounts on action +velocity (DMs, likes, follows). Even legitimate agent use can trigger +"action blocks" or permanent bans. This module enforces hard caps well +below Instagram's documented thresholds plus jitter to mimic human pacing. + +State persists across server restarts to ~/.instagram-mcp-rate-limits.json +so a relaunch does not reset the daily budget. + +Per-category defaults (conservative; Instagram's real limits are higher): + + category per_minute per_hour per_day jitter + dm_send 2 20 80 1.5-4.0s + like 6 30 200 0.5-2.0s + search 30 200 1000 0.0-0.5s + lookup 30 300 2000 0.0-0.5s + modify 10 100 500 0.0-0.5s + +Override any cap via env var, e.g.: + IG_RATE_LIMIT_DM_SEND_PER_DAY=40 + IG_RATE_LIMIT_LIKE_PER_HOUR=15 +""" +from __future__ import annotations + +import functools +import json +import logging +import os +import random +import time +from pathlib import Path +from typing import Any, Callable, Dict, List, Tuple + +logger = logging.getLogger(__name__) + +_STATE_PATH = Path.home() / ".instagram-mcp-rate-limits.json" + +DEFAULTS: Dict[str, Dict[str, Any]] = { + "dm_send": {"per_minute": 2, "per_hour": 20, "per_day": 80, "jitter": (1.5, 4.0)}, + "like": {"per_minute": 6, "per_hour": 30, "per_day": 200, "jitter": (0.5, 2.0)}, + "search": {"per_minute": 30, "per_hour": 200, "per_day": 1000, "jitter": (0.0, 0.5)}, + "lookup": {"per_minute": 30, "per_hour": 300, "per_day": 2000, "jitter": (0.0, 0.5)}, + "modify": {"per_minute": 10, "per_hour": 100, "per_day": 500, "jitter": (0.0, 0.5)}, +} + + +def _env_override(category: str, key: str, default: int) -> int: + var = f"IG_RATE_LIMIT_{category.upper()}_{key.upper()}" + raw = os.environ.get(var) + if raw is None: + return default + try: + value = int(raw) + if value <= 0: + return default + return value + except ValueError: + return default + + +def _get_limits(category: str) -> Dict[str, Any]: + d = DEFAULTS[category] + return { + "per_minute": _env_override(category, "per_minute", d["per_minute"]), + "per_hour": _env_override(category, "per_hour", d["per_hour"]), + "per_day": _env_override(category, "per_day", d["per_day"]), + "jitter": d["jitter"], + } + + +def _load_state() -> Dict[str, List[float]]: + if not _STATE_PATH.exists(): + return {} + try: + data = json.loads(_STATE_PATH.read_text()) + return {k: [float(t) for t in v] for k, v in data.items() if isinstance(v, list)} + except Exception as exc: + logger.warning("Could not load rate-limit state, starting fresh: %s", exc) + return {} + + +def _save_state(state: Dict[str, List[float]]) -> None: + try: + _STATE_PATH.write_text(json.dumps(state)) + except Exception as exc: + logger.warning("Failed to persist rate-limit state: %s", exc) + + +def _prune(timestamps: List[float], now: float, window_s: int) -> List[float]: + cutoff = now - window_s + return [t for t in timestamps if t >= cutoff] + + +def _fmt_duration(seconds: int) -> str: + if seconds < 60: + return f"{seconds}s" + if seconds < 3600: + return f"{seconds // 60}m {seconds % 60}s" + return f"{seconds // 3600}h {(seconds % 3600) // 60}m" + + +def _check_budget( + category: str, + limits: Dict[str, Any], + state: Dict[str, List[float]], +) -> Tuple[bool, str, int, Dict[str, int]]: + """Returns (ok, reason_if_blocked, retry_after_seconds, current_counts).""" + now = time.time() + pruned_day = _prune(state.get(category, []), now, 24 * 3600) + state[category] = pruned_day + + counts: Dict[str, int] = {} + for window_name, window_s in (("per_minute", 60), ("per_hour", 3600), ("per_day", 86400)): + in_window = _prune(pruned_day, now, window_s) + counts[window_name] = len(in_window) + + for window_name, window_s in (("per_minute", 60), ("per_hour", 3600), ("per_day", 86400)): + in_window = _prune(pruned_day, now, window_s) + limit = limits[window_name] + if len(in_window) >= limit: + oldest = min(in_window) + retry_after = int((oldest + window_s) - now) + 1 + label = window_name.replace("per_", "") + return ( + False, + f"{category} hit {limit}/{label} cap (currently {len(in_window)}). Retry in {_fmt_duration(retry_after)}.", + retry_after, + counts, + ) + return (True, "", 0, counts) + + +def rate_limited(category: str) -> Callable[[Callable[..., Dict[str, Any]]], Callable[..., Dict[str, Any]]]: + """Decorator: enforce per-category limits and apply jitter before the call. + + Returns a structured error dict to the MCP client when blocked instead of + raising, so the agent can surface "try again in 4h 12m" to the user + instead of failing opaquely. + """ + if category not in DEFAULTS: + raise ValueError(f"Unknown rate-limit category: {category}") + + def decorator(func: Callable[..., Dict[str, Any]]) -> Callable[..., Dict[str, Any]]: + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Dict[str, Any]: + limits = _get_limits(category) + state = _load_state() + ok, reason, retry_after, current = _check_budget(category, limits, state) + if not ok: + logger.warning("Rate limit blocked %s: %s", func.__name__, reason) + return { + "success": False, + "rate_limited": True, + "category": category, + "message": ( + "Rate limit hit to protect this Instagram account from being " + f"flagged for automation: {reason}" + ), + "retry_after_seconds": retry_after, + "limits": {k: limits[k] for k in ("per_minute", "per_hour", "per_day")}, + "current": current, + } + state.setdefault(category, []).append(time.time()) + _save_state(state) + lo, hi = limits["jitter"] + if hi > 0: + time.sleep(random.uniform(lo, hi)) + return func(*args, **kwargs) + return wrapper + return decorator diff --git a/backend/apps/instagram_mcp/server.py b/backend/apps/instagram_mcp/server.py new file mode 100644 index 00000000..e6cf928b --- /dev/null +++ b/backend/apps/instagram_mcp/server.py @@ -0,0 +1,932 @@ +import logging +import os +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional + +from instagrapi import Client +from mcp.server.fastmcp import FastMCP + +from .rate_limiter import rate_limited + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + +INSTRUCTIONS = """ +Instagram via instagrapi. 25 tools: DMs, user/follower lookup, post +engagement, story reads, media downloads. Per-category rate limiting is +enforced server-side to protect the connected account from anti-abuse bans. +""" + +client = Client() + +mcp = FastMCP( + name="Instagram DMs", + instructions=INSTRUCTIONS, +) + +SESSION_DIR = Path.home() / ".instagram_dm_mcp" / "sessions" + + +@mcp.tool() +@rate_limited("dm_send") +def send_message(username: str, message: str) -> Dict[str, Any]: + """Send an Instagram direct message to a user by username. + + Args: + username: Instagram username of the recipient. + message: The message text to send. + Returns: + A dictionary with success status and a status message. + """ + if not username or not message: + return {"success": False, "message": "Username and message must be provided."} + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + dm = client.direct_send(message, [user_id]) + if dm: + return {"success": True, "message": "Message sent to user.", "direct_message_id": getattr(dm, 'id', None)} + else: + return {"success": False, "message": "Failed to send message."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("dm_send") +def send_photo_message(username: str, photo_path: str) -> Dict[str, Any]: + """Send a photo via Instagram direct message to a user by username. + + Args: + username: Instagram username of the recipient. + photo_path: Path to the photo file to send. + message: Optional message text to accompany the photo. + Returns: + A dictionary with success status and a status message. + """ + if not username or not photo_path: + return {"success": False, "message": "Username and photo_path must be provided."} + + if not os.path.exists(photo_path): + return {"success": False, "message": f"Photo file not found: {photo_path}"} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + result = client.direct_send_photo(Path(photo_path), [user_id]) + if result: + return {"success": True, "message": "Photo sent successfully.", "direct_message_id": getattr(result, 'id', None)} + else: + return {"success": False, "message": "Failed to send photo."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("dm_send") +def send_video_message(username: str, video_path: str) -> Dict[str, Any]: + """Send a video via Instagram direct message to a user by username. + + Args: + username: Instagram username of the recipient. + video_path: Path to the video file to send. + Returns: + A dictionary with success status and a status message. + """ + if not username or not video_path: + return {"success": False, "message": "Username and video_path must be provided."} + + if not os.path.exists(video_path): + return {"success": False, "message": f"Video file not found: {video_path}"} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + result = client.direct_send_video(Path(video_path), [user_id]) + if result: + return {"success": True, "message": "Video sent successfully.", "direct_message_id": getattr(result, 'id', None)} + else: + return {"success": False, "message": "Failed to send video."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def list_chats( + amount: int = 20, + selected_filter: str = "", + thread_message_limit: Optional[int] = None, + full: bool = False, + fields: Optional[List[str]] = None, +) -> Dict[str, Any]: + """Get Instagram Direct Message threads (chats) from the user's account, with optional filters and limits. + + Args: + amount: Number of threads to fetch (default 20). + selected_filter: Filter for threads ("", "flagged", or "unread"). + thread_message_limit: Limit for messages per thread. + full: If True, return the full thread object for each chat (default False). + fields: If provided, return only these fields for each thread. + Returns: + A dictionary with success status and the list of threads or error message. + """ + def thread_summary(thread): + t = thread if isinstance(thread, dict) else thread.dict() + users = t.get("users", []) + user_summaries = [ + { + "username": u.get("username"), + "full_name": u.get("full_name"), + "pk": u.get("pk") + } + for u in users + ] + return { + "thread_id": t.get("id"), + "thread_title": t.get("thread_title"), + "users": user_summaries, + "last_activity_at": t.get("last_activity_at"), + "last_message": t.get("messages", [{}])[-1] if t.get("messages") else None + } + + def filter_fields(thread, fields): + t = thread if isinstance(thread, dict) else thread.dict() + return {field: t.get(field) for field in fields} + + try: + threads = client.direct_threads(amount, selected_filter, thread_message_limit) + if full: + return {"success": True, "threads": [t.dict() if hasattr(t, 'dict') else str(t) for t in threads]} + elif fields: + return {"success": True, "threads": [filter_fields(t, fields) for t in threads]} + else: + return {"success": True, "threads": [thread_summary(t) for t in threads]} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def list_messages(thread_id: str, amount: int = 20) -> Dict[str, Any]: + """Get messages from a specific Instagram Direct Message thread by thread ID, with an optional limit. + + Args: + thread_id: The thread ID to fetch messages from. + amount: Number of messages to fetch (default 20). + Returns: + A dictionary with success status and the list of messages or error message. + """ + if not thread_id: + return {"success": False, "message": "Thread ID must be provided."} + try: + messages = client.direct_messages(thread_id, amount) + result_msgs = [] + for m in messages: + msg = m.dict() if hasattr(m, 'dict') else (m if isinstance(m, dict) else {}) + # Expose item_type and shared post/reel info if present + item_type = getattr(m, 'item_type', None) or msg.get('item_type') + shared_info = None + shared_url = None + shared_code = None + if item_type in ["clip", "media_share", "reel_share", "xma_media_share", "post_share"]: + # Try to extract code/url from known attributes + clip = getattr(m, 'clip', None) or msg.get('clip') + media_share = getattr(m, 'media_share', None) or msg.get('media_share') + xma = getattr(m, 'xma_media_share', None) or msg.get('xma_media_share') + post_share = getattr(m, 'post_share', None) or msg.get('post_share') + # Try to get code/url from any of these + for obj in [clip, media_share, xma, post_share]: + if obj: + shared_code = obj.get('code') or obj.get('pk') + shared_url = obj.get('url') or (f"https://www.instagram.com/reel/{shared_code}/" if shared_code else None) + shared_info = obj + break + msg['item_type'] = item_type + msg['shared_post_info'] = shared_info + msg['shared_post_url'] = shared_url + msg['shared_post_code'] = shared_code + result_msgs.append(msg) + return {"success": True, "messages": result_msgs} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("modify") +def mark_message_seen(thread_id: str, message_id: str) -> Dict[str, Any]: + """Mark a message as seen in a direct message thread. + + Args: + thread_id: The thread ID containing the message. + message_id: The ID of the message to mark as seen. + Returns: + A dictionary with success status and a status message. + """ + if not thread_id or not message_id: + return {"success": False, "message": "Both thread_id and message_id must be provided."} + + try: + result = client.direct_message_seen(int(thread_id), int(message_id)) + if result: + return {"success": True, "message": "Message marked as seen."} + else: + return {"success": False, "message": "Failed to mark message as seen."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def list_pending_chats(amount: int = 20) -> Dict[str, Any]: + """Get Instagram Direct Message threads (chats) from the user's pending inbox. + + Args: + amount: Number of pending threads to fetch (default 20). + Returns: + A dictionary with success status and the list of pending threads or error message. + """ + try: + threads = client.direct_pending_inbox(amount) + return {"success": True, "threads": [t.dict() if hasattr(t, 'dict') else str(t) for t in threads]} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("search") +def search_threads(query: str) -> Dict[str, Any]: + """Search Instagram Direct Message threads by username or keyword. + + Args: + query: The search term (username or keyword). + Returns: + A dictionary with success status and the search results or error message. + """ + if not query: + return {"success": False, "message": "Query must be provided."} + try: + results = client.direct_search(query) + return {"success": True, "results": [r.dict() if hasattr(r, 'dict') else str(r) for r in results]} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_thread_by_participants(user_ids: List[int]) -> Dict[str, Any]: + """Get an Instagram Direct Message thread by participant user IDs. + + Args: + user_ids: List of user IDs (ints). + Returns: + A dictionary with success status and the thread or error message. + """ + if not user_ids or not isinstance(user_ids, list): + return {"success": False, "message": "user_ids must be a non-empty list of user IDs."} + try: + thread = client.direct_thread_by_participants(user_ids) + return {"success": True, "thread": thread.dict() if hasattr(thread, 'dict') else str(thread)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_thread_details(thread_id: str, amount: int = 20) -> Dict[str, Any]: + """Get details and messages for a specific Instagram Direct Message thread by thread ID, with an optional message limit. + + Args: + thread_id: The thread ID to fetch details for. + amount: Number of messages to fetch (default 20). + Returns: + A dictionary with success status and the thread details or error message. + """ + if not thread_id: + return {"success": False, "message": "Thread ID must be provided."} + try: + thread = client.direct_thread(thread_id, amount) + return {"success": True, "thread": thread.dict() if hasattr(thread, 'dict') else str(thread)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_id_from_username(username: str) -> Dict[str, Any]: + """Get the Instagram user ID for a given username. + + Args: + username: Instagram username. + Returns: + A dictionary with success status and the user ID or error message. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + try: + user_id = client.user_id_from_username(username) + if user_id: + return {"success": True, "user_id": user_id} + else: + return {"success": False, "message": f"User '{username}' not found."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_username_from_user_id(user_id: str) -> Dict[str, Any]: + """Get the Instagram username for a given user ID. + + Args: + user_id: Instagram user ID. + Returns: + A dictionary with success status and the username or error message. + """ + if not user_id: + return {"success": False, "message": "User ID must be provided."} + try: + username = client.username_from_user_id(user_id) + if username: + return {"success": True, "username": username} + else: + return {"success": False, "message": f"User ID '{user_id}' not found."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_info(username: str) -> Dict[str, Any]: + """Get detailed information about an Instagram user. + + Args: + username: Instagram username to get information about. + Returns: + A dictionary with success status and user information. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + + try: + user = client.user_info_by_username(username) + if user: + user_data = { + "user_id": str(user.pk), + "username": user.username, + "full_name": user.full_name, + "biography": user.biography, + "follower_count": user.follower_count, + "following_count": user.following_count, + "media_count": user.media_count, + "is_private": user.is_private, + "is_verified": user.is_verified, + "profile_pic_url": str(user.profile_pic_url) if user.profile_pic_url else None, + "external_url": str(user.external_url) if user.external_url else None, + "category": user.category, + } + return {"success": True, "user_info": user_data} + else: + return {"success": False, "message": f"User '{username}' not found."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def check_user_online_status(usernames: List[str]) -> Dict[str, Any]: + """Check the online status of Instagram users. + + Args: + usernames: List of Instagram usernames to check status for. + Returns: + A dictionary with success status and users' presence information. + """ + if not usernames or not isinstance(usernames, list): + return {"success": False, "message": "A list of usernames must be provided."} + + try: + user_ids = [] + username_to_id = {} + + # Get user IDs for the usernames + for username in usernames: + try: + user_id = client.user_id_from_username(username) + if user_id: + user_ids.append(int(user_id)) + username_to_id[user_id] = username + except: + continue + + if not user_ids: + return {"success": False, "message": "No valid users found."} + + presence_data = client.direct_users_presence(user_ids) + + # Convert back to usernames + result = {} + for user_id_str, presence in presence_data.items(): + username = username_to_id.get(user_id_str, f"user_{user_id_str}") + result[username] = presence + + return {"success": True, "presence_data": result} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("search") +def search_users(query: str) -> Dict[str, Any]: + """Search for Instagram users by name or username. + + Args: + query: Search term (name or username). + count: Maximum number of users to return (default 10, max 50). + Returns: + A dictionary with success status and search results. + """ + if not query: + return {"success": False, "message": "Search query must be provided."} + + try: + users = client.search_users(query) + + user_results = [] + for user in users: + user_data = { + "user_id": str(user.pk), + "username": user.username, + "full_name": user.full_name, + "is_private": user.is_private, + "profile_pic_url": str(user.profile_pic_url) if user.profile_pic_url else None, + "follower_count": getattr(user, 'follower_count', None), + } + user_results.append(user_data) + + return {"success": True, "users": user_results, "count": len(user_results)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_stories(username: str) -> Dict[str, Any]: + """Get Instagram stories from a user. + + Args: + username: Instagram username to get stories from. + Returns: + A dictionary with success status and stories information. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + stories = client.user_stories(user_id) + + story_results = [] + for story in stories: + story_data = { + "story_id": str(story.pk), + "media_type": story.media_type, # 1=photo, 2=video + "taken_at": str(story.taken_at), + "user": { + "username": story.user.username, + "full_name": story.user.full_name, + "user_id": str(story.user.pk) + }, + "media_url": str(story.thumbnail_url) if story.thumbnail_url else None, + } + + if story.media_type == 2 and story.video_url: + story_data["video_url"] = str(story.video_url) + story_data["video_duration"] = story.video_duration + + story_results.append(story_data) + + return {"success": True, "stories": story_results, "count": len(story_results)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("like") +def like_media(media_url: str, like: bool = True) -> Dict[str, Any]: + """Like or unlike an Instagram post. + + Args: + media_url: URL of the Instagram post. + like: True to like, False to unlike the post. + Returns: + A dictionary with success status and a status message. + """ + if not media_url: + return {"success": False, "message": "Media URL must be provided."} + + try: + media_pk = client.media_pk_from_url(media_url) + if not media_pk: + return {"success": False, "message": "Invalid media URL or post not found."} + + if like: + result = client.media_like(media_pk) + action = "liked" + else: + result = client.media_unlike(media_pk) + action = "unliked" + + if result: + return {"success": True, "message": f"Post {action} successfully."} + else: + return {"success": False, "message": f"Failed to {action.rstrip('d')} post."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_followers(username: str, count: int = 20) -> Dict[str, Any]: + """Get followers of an Instagram user. + + Args: + username: Instagram username to get followers for. + count: Maximum number of followers to return (default 20). + Returns: + A dictionary with success status and followers list. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + followers = client.user_followers(user_id, amount=count) + + follower_results = [] + for follower_id, follower in followers.items(): + follower_data = { + "user_id": str(follower.pk), + "username": follower.username, + "full_name": follower.full_name, + "is_private": follower.is_private, + "profile_pic_url": str(follower.profile_pic_url) if follower.profile_pic_url else None, + } + follower_results.append(follower_data) + + return {"success": True, "followers": follower_results, "count": len(follower_results)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_following(username: str, count: int = 20) -> Dict[str, Any]: + """Get users that an Instagram user is following. + + Args: + username: Instagram username to get following list for. + count: Maximum number of following to return (default 20). + Returns: + A dictionary with success status and following list. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + following = client.user_following(user_id, amount=count) + + following_results = [] + for following_id, followed_user in following.items(): + following_data = { + "user_id": str(followed_user.pk), + "username": followed_user.username, + "full_name": followed_user.full_name, + "is_private": followed_user.is_private, + "profile_pic_url": str(followed_user.profile_pic_url) if followed_user.profile_pic_url else None, + } + following_results.append(following_data) + + return {"success": True, "following": following_results, "count": len(following_results)} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("lookup") +def get_user_posts(username: str, count: int = 12) -> Dict[str, Any]: + """Get recent posts from an Instagram user. + + Args: + username: Instagram username to get posts from. + count: Maximum number of posts to return (default 12). + Returns: + A dictionary with success status and posts list. + """ + if not username: + return {"success": False, "message": "Username must be provided."} + + try: + user_id = client.user_id_from_username(username) + if not user_id: + return {"success": False, "message": f"User '{username}' not found."} + + medias = client.user_medias(user_id, amount=count) + + media_results = [] + for media in medias: + media_data = { + "media_id": str(media.pk), + "media_type": media.media_type, # 1=photo, 2=video, 8=album + "caption": media.caption_text if media.caption_text else "", + "like_count": media.like_count, + "comment_count": media.comment_count, + "taken_at": str(media.taken_at), + "media_url": str(media.thumbnail_url) if media.thumbnail_url else None, + } + + if media.media_type == 2 and media.video_url: + media_data["video_url"] = str(media.video_url) + media_data["video_duration"] = media.video_duration + + media_results.append(media_data) + + return {"success": True, "posts": media_results, "count": len(media_results)} + except Exception as e: + return {"success": False, "message": str(e)} + + +def _ensure_download_directory(download_path: str) -> None: + """Ensure download directory exists.""" + Path(download_path).mkdir(parents=True, exist_ok=True) + + +def _download_single_media(media, download_path: str) -> str: + """Download a single media item and return the file path.""" + media_type = media.media_type + if media_type == 1: # Photo + return str(client.photo_download(media.pk, download_path)) + elif media_type == 2: # Video + return str(client.video_download(media.pk, download_path)) + else: + raise ValueError(f"Unsupported media type: {media_type}") + + +def _find_message_in_thread(thread_id: str, message_id: str): + """Find a specific message in a thread.""" + messages = client.direct_messages(thread_id, 100) + return next((m for m in messages if str(m.id) == message_id), None) + + +@mcp.tool() +@rate_limited("lookup") +def list_media_messages(thread_id: str, limit: int = 100) -> Dict[str, Any]: + """List all messages containing media in an Instagram direct message thread. + Args: + thread_id: The ID of the thread to check for media messages + limit: Maximum number of messages to check (default 100, max 200) + Returns: + A dictionary containing success status and list of all media messages found + """ + try: + limit = min(limit, 200) + messages = client.direct_messages(thread_id, limit) + media_messages = [] + for message in messages: + if message.media: + media_messages.append({ + "message_id": str(message.id), + "media_type": "photo" if message.media.media_type == 1 else "video", + "timestamp": str(message.timestamp) if hasattr(message, 'timestamp') else None, + "sender_user_id": message.user_id if hasattr(message, 'user_id') else None + }) + return { + "success": True, + "message": f"Found {len(media_messages)} messages with media", + "total_messages_checked": len(messages), + "media_messages": media_messages + } + except Exception as e: + return { + "success": False, + "message": f"Failed to list media messages: {str(e)}" + } + +@mcp.tool() +@rate_limited("lookup") +def download_media_from_message(message_id: str, thread_id: str, download_path: str = "./downloads") -> Dict[str, Any]: + """Download media from a specific Instagram direct message and get the local file path. + Args: + message_id: The ID of the message containing the media + thread_id: The ID of the thread containing the message + download_path: Directory to save the downloaded file (default: ./downloads) + Returns: + A dictionary containing success status, a status message, and the file path if successful + """ + try: + _ensure_download_directory(download_path) + target_message = _find_message_in_thread(thread_id, message_id) + if not target_message: + return { + "success": False, + "message": f"Message {message_id} not found in thread {thread_id}" + } + if not target_message.media: + return { + "success": False, + "message": "This message does not contain media" + } + file_path = _download_single_media(target_message.media, download_path) + return { + "success": True, + "message": "Media downloaded successfully", + "file_path": file_path, + "media_type": "photo" if target_message.media.media_type == 1 else "video", + "message_id": message_id, + "thread_id": thread_id + } + except Exception as e: + return { + "success": False, + "message": f"Failed to download media: {str(e)}" + } + + +@mcp.tool() +@rate_limited("lookup") +def download_shared_post_from_message(message_id: str, thread_id: str, download_path: str = "./downloads") -> Dict[str, Any]: + """Download media from a shared post/reel/clip in a DM message and get the local file path. + Args: + message_id: The ID of the message containing the shared post/reel/clip + thread_id: The ID of the thread containing the message + download_path: Directory to save the downloaded file (default: ./downloads) + Returns: + A dictionary containing success status, a status message, and the file path if successful + """ + try: + _ensure_download_directory(download_path) + target_message = _find_message_in_thread(thread_id, message_id) + if not target_message: + return {"success": False, "message": f"Message {message_id} not found in thread {thread_id}"} + item_type = getattr(target_message, 'item_type', None) + # Extract shared post/reel/clip URL + shared_url = None + shared_code = None + shared_obj = None + if item_type in ["clip", "media_share", "reel_share", "xma_media_share", "post_share"]: + for attr in ['clip', 'media_share', 'xma_media_share', 'post_share']: + obj = getattr(target_message, attr, None) + if obj: + shared_code = obj.get('code') or obj.get('pk') + shared_url = obj.get('url') or (f"https://www.instagram.com/reel/{shared_code}/" if shared_code else None) + shared_obj = obj + break + if not shared_url: + return {"success": False, "message": "This message does not contain a supported shared post/reel/clip"} + # Download using Instagrapi + try: + media_pk = client.media_pk_from_url(shared_url) + media = client.media_info(media_pk) + if media.media_type == 1: + file_path = str(client.photo_download(media_pk, download_path)) + media_type = "photo" + elif media.media_type == 2: + file_path = str(client.video_download(media_pk, download_path)) + media_type = "video" + elif media.media_type == 8: # album + # Download all items in album + album_paths = client.album_download(media_pk, download_path) + file_path = str(album_paths) + media_type = "album" + else: + return {"success": False, "message": f"Unsupported media type: {media.media_type}"} + return { + "success": True, + "message": "Shared post/reel/clip downloaded successfully", + "file_path": file_path, + "media_type": media_type, + "shared_post_url": shared_url, + "message_id": message_id, + "thread_id": thread_id + } + except Exception as e: + return {"success": False, "message": f"Failed to download shared post/reel/clip: {str(e)}"} + except Exception as e: + return {"success": False, "message": f"Failed to process message: {str(e)}"} + + +@mcp.tool() +@rate_limited("modify") +def delete_message(thread_id: str, message_id: str) -> Dict[str, Any]: + """Delete a message from a direct message thread. + + Args: + thread_id: The thread ID containing the message. + message_id: The ID of the message to delete. + Returns: + A dictionary with success status and a status message. + """ + if not thread_id or not message_id: + return {"success": False, "message": "Both thread_id and message_id must be provided."} + + try: + result = client.direct_message_delete(int(thread_id), int(message_id)) + if result: + return {"success": True, "message": "Message deleted successfully."} + else: + return {"success": False, "message": "Failed to delete message."} + except Exception as e: + return {"success": False, "message": str(e)} + + +@mcp.tool() +@rate_limited("modify") +def mute_conversation(thread_id: str, mute: bool = True) -> Dict[str, Any]: + """Mute or unmute a direct message conversation. + + Args: + thread_id: The thread ID to mute/unmute. + mute: True to mute, False to unmute the conversation. + Returns: + A dictionary with success status and a status message. + """ + if not thread_id: + return {"success": False, "message": "Thread ID must be provided."} + + try: + if mute: + result = client.direct_thread_mute(int(thread_id)) + action = "muted" + else: + result = client.direct_thread_unmute(int(thread_id)) + action = "unmuted" + + if result: + return {"success": True, "message": f"Conversation {action} successfully."} + else: + return {"success": False, "message": f"Failed to {action.rstrip('d')} conversation."} + except Exception as e: + return {"success": False, "message": str(e)} + + +def _resolve_username() -> Optional[str]: + """INSTAGRAM_USERNAME env var (set by OpenSwarm at spawn) > current_user.txt fallback.""" + env_user = os.getenv("INSTAGRAM_USERNAME") + if env_user: + return env_user.strip() or None + marker = SESSION_DIR / "current_user.txt" + if marker.exists(): + return marker.read_text().strip() or None + return None + + +def main() -> None: + """Spawn entrypoint. Loads the prebuilt instagrapi session and runs stdio. + + Authentication happens before this process starts: the OpenSwarm backend's + /credentials/instagram/{validate,from_browser} endpoints dump an instagrapi + session to SESSION_DIR/_session.json. This server just loads it. + """ + SESSION_DIR.mkdir(parents=True, exist_ok=True) + username = _resolve_username() + if not username: + logger.error("No INSTAGRAM_USERNAME and no current_user.txt — Connect Instagram via the OpenSwarm Tools page first.") + sys.exit(1) + + session_file = SESSION_DIR / f"{username}_session.json" + if not session_file.exists(): + logger.error(f"No session at {session_file}. Connect Instagram via the OpenSwarm Tools page first.") + sys.exit(1) + + try: + client.load_settings(session_file) + # We deliberately skip an account_info() probe here. Instagram's mobile + # API (i.instagram.com) returns 467 on browser-derived sessions even + # when those cookies still authorize other calls; we trust the loaded + # session and let real tool calls surface any auth issues at use time. + logger.info(f"Loaded Instagram session for @{username}") + mcp.run(transport="stdio") + except Exception as e: + logger.error(f"Failed to authenticate to Instagram: {e}") + print(f"Error: Failed to authenticate to Instagram - {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/backend/apps/tools_lib/tools_lib.py b/backend/apps/tools_lib/tools_lib.py index aa6fb392..0a7b6bb0 100644 --- a/backend/apps/tools_lib/tools_lib.py +++ b/backend/apps/tools_lib/tools_lib.py @@ -1330,17 +1330,17 @@ async def validate_instagram_credentials(payload: dict) -> dict: msg = str(e) return {"ok": False, "error": msg[:300] if msg else type(e).__name__} - # Cache session where trypeggy expects it so the MCP server's startup - # account_info() probe has a fresh session and skips client.login() on - # first spawn. Trypeggy reads SESSION_FILE = REPO_ROOT/f"{username}_session.json". + # Cache the instagrapi session where the vendored MCP server expects it + # (backend/apps/instagram_mcp/server.py loads from this path). Per OS user, + # isolated from any project checkout — works for every OpenSwarm install. try: from pathlib import Path - trypeggy_root = Path("/Users/shawnmadadha/dev/instagram_dm_mcp") - if trypeggy_root.exists(): - cl.dump_settings(trypeggy_root / f"{username}_session.json") - (trypeggy_root / "current_user.txt").write_text(username + "\n") + session_dir = Path.home() / ".instagram_dm_mcp" / "sessions" + session_dir.mkdir(parents=True, exist_ok=True) + cl.dump_settings(session_dir / f"{username}_session.json") + (session_dir / "current_user.txt").write_text(username + "\n") except Exception as cache_err: # noqa: BLE001 - logger.warning(f"validate_instagram: could not cache session for trypeggy: {cache_err}") + logger.warning(f"validate_instagram: could not cache session: {cache_err}") tool.credentials = {"INSTAGRAM_USERNAME": username, "INSTAGRAM_PASSWORD": password} tool.auth_type = "env_vars" @@ -1415,16 +1415,16 @@ async def instagram_from_browser(payload: dict) -> dict: user_id_str = result["user_id"] settings = result["settings"] - trypeggy_root = Path("/Users/shawnmadadha/dev/instagram_dm_mcp") + session_dir = Path.home() / ".instagram_dm_mcp" / "sessions" try: - if trypeggy_root.exists(): - (trypeggy_root / f"{username}_session.json").write_text(json.dumps(settings)) - (trypeggy_root / "current_user.txt").write_text(username + "\n") + session_dir.mkdir(parents=True, exist_ok=True) + (session_dir / f"{username}_session.json").write_text(json.dumps(settings)) + (session_dir / "current_user.txt").write_text(username + "\n") except Exception as cache_err: # noqa: BLE001 - logger.warning(f"instagram_from_browser: could not write trypeggy session file: {cache_err}") + logger.warning(f"instagram_from_browser: could not write session file: {cache_err}") - # No password from the browser flow. trypeggy will load the session and - # tolerate a 467 on its startup account_info() probe (patched locally). + # No password from the browser flow. The vendored server skips the + # account_info() probe and trusts the loaded session. tool.credentials = {"INSTAGRAM_USERNAME": username} tool.auth_type = "env_vars" tool.auth_status = "connected" diff --git a/frontend/src/app/pages/Tools/Tools.tsx b/frontend/src/app/pages/Tools/Tools.tsx index e7488fa3..bcc25d4b6 100644 --- a/frontend/src/app/pages/Tools/Tools.tsx +++ b/frontend/src/app/pages/Tools/Tools.tsx @@ -148,16 +148,16 @@ const INTEGRATIONS: Integration[] = [ id: 'instagram', name: 'Instagram', description: - 'Instagram DM outreach plus user/follower lookup, post engagement, and story reads. 25 tools from ShawnMadadha/instagram_dm_mcp (a rate-limited fork of trypeggy/instagram_dm_mcp). Sign in with your Instagram username and password. First-time setup: run scripts/setup-instagram-mcp.sh. NOTE: mass-DMing from personal accounts triggers Instagram anti-abuse detection; per-tool rate limits are enforced by the server to protect your account.', + 'Instagram DM outreach plus user/follower lookup, post engagement, and story reads. 25 tools powered by instagrapi (HTTP, no browser). Per-category rate limits are enforced server-side to protect the connected account from anti-abuse bans.', mcp_config: { type: 'stdio', - command: 'bash', - args: ['-c', 'exec "$HOME/.openswarm/instagram-mcp/.venv/bin/python" "$HOME/.openswarm/instagram-mcp/src/mcp_server.py"'], + command: 'python', + args: ['-m', 'backend.apps.instagram_mcp'], }, color: '#E4405F', - website: 'https://github.com/ShawnMadadha/instagram_dm_mcp', + website: 'https://github.com/openswarm-ai/openswarm', connectLabel: 'Connect Instagram', - connectInstructions: 'First-time on this machine: run `bash scripts/setup-instagram-mcp.sh` in a terminal. It installs the MCP server into ~/.openswarm/instagram-mcp/. Then sign in below with the Instagram username and password the agent should use. Credentials are stored locally in OpenSwarm; session files live in ~/.instagram_dm_mcp/sessions/ and are reused on every restart.', + connectInstructions: 'Sign in with the Instagram username and password the agent should use. Credentials validate once, then a session file at ~/.instagram_dm_mcp/sessions/ is reused on every spawn. NOTE: mass-DMing from personal accounts triggers anti-abuse detection; use established accounts and respect the built-in rate limits.', credentialFields: [ { key: 'INSTAGRAM_USERNAME', label: 'Instagram Username', placeholder: 'your_handle (no @)', type: 'text' }, { key: 'INSTAGRAM_PASSWORD', label: 'Instagram Password', placeholder: '••••••••', type: 'password' }, diff --git a/scripts/setup-instagram-mcp.sh b/scripts/setup-instagram-mcp.sh deleted file mode 100755 index bf909d4b..00000000 --- a/scripts/setup-instagram-mcp.sh +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env bash -# One-time setup for the Instagram MCP server. Each OpenSwarm user runs this -# once on their machine; it installs the ShawnMadadha/instagram_dm_mcp fork -# (carries the per-tool rate limiter that protects accounts from anti-abuse -# bans) into ~/.openswarm/instagram-mcp/ and pip-installs its dependencies -# into a local venv. Re-running upgrades to the latest version. -# -# Why a fork: trypeggy/instagram_dm_mcp upstream does not have the rate -# limiter yet. PR is open at trypeggy/instagram_dm_mcp#12. - -set -euo pipefail - -DEST="$HOME/.openswarm/instagram-mcp" -REPO_URL="https://github.com/ShawnMadadha/instagram_dm_mcp.git" - -if ! command -v git >/dev/null 2>&1; then - echo "error: git not found on PATH." >&2 - exit 1 -fi -if ! command -v python3 >/dev/null 2>&1; then - echo "error: python3 not found on PATH." >&2 - exit 1 -fi - -if [ -d "$DEST/.git" ]; then - echo "Updating existing checkout at $DEST..." - git -C "$DEST" fetch --quiet - git -C "$DEST" reset --hard origin/main --quiet -else - echo "Cloning $REPO_URL into $DEST..." - mkdir -p "$(dirname "$DEST")" - git clone --depth=1 --quiet "$REPO_URL" "$DEST" -fi - -VENV="$DEST/.venv" -if [ ! -d "$VENV" ]; then - echo "Creating venv at $VENV..." - python3 -m venv "$VENV" -fi - -echo "Installing requirements into venv..." -"$VENV/bin/python" -m pip install --quiet --upgrade pip -"$VENV/bin/python" -m pip install --quiet -r "$DEST/requirements.txt" - -echo "Installed: $DEST" -"$VENV/bin/python" -c "from pathlib import Path; print(' python:', Path('$VENV/bin/python').resolve()); print(' server:', '$DEST/src/mcp_server.py')" -echo "Done. Connect Instagram from the OpenSwarm Tools page next."