WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-11 12:17:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[shawn] refactor: vendor Instagram MCP into backend/apps/instagram_mcp, drop external setup

Browse Source
This commit is contained in:
TheAchiever6823
2026-05-17 18:43:43 -07:00
parent 789fb5e088
commit a9f6338227
8 changed files with 1145 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
GETTING_STARTED.md
+9 -21
View File
@@ -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/<username>_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/<username>_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/<username>_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.
---
backend/apps/instagram_mcp/__init__.py
View File
backend/apps/instagram_mcp/__main__.py
+12
View File
@@ -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/<username>_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()
backend/apps/instagram_mcp/rate_limiter.py
+172
View File
@@ -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
backend/apps/instagram_mcp/server.py
+932
View File
@@ -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/<username>_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()
backend/apps/tools_lib/tools_lib.py
+15 -15
View File
@@ -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"
frontend/src/app/pages/Tools/Tools.tsx
+5 -5
View File
@@ -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' },
scripts/setup-instagram-mcp.sh
-47
View File
@@ -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."