server.py•35.3 kB
#!/usr/bin/env python3
"""
SocialData MCP Server
A Model Context Protocol server that provides tools for interacting with the SocialData API.
This server allows you to fetch Twitter/X data including tweets, user profiles, followers,
and create monitoring alerts.
"""
import os
import json
import httpx
from datetime import datetime
from typing import Optional, Literal, List, Dict, Any, Union
from fastmcp import FastMCP
# Initialize MCP server
mcp = FastMCP("SocialData API Server")
# Base URL for SocialData API
BASE_URL = "https://api.socialdata.tools"
def get_api_key() -> str:
"""Get API key from environment variable"""
api_key = os.getenv("SOCIALDATA_API_KEY")
if not api_key:
raise ValueError("SOCIALDATA_API_KEY environment variable is required")
return api_key
def get_headers() -> Dict[str, str]:
"""Get headers for API requests"""
return {
"Authorization": f"Bearer {get_api_key()}",
"Content-Type": "application/json"
}
# =============================================================================
# DATA API - Search and User Data
# =============================================================================
@mcp.tool()
def search_twitter(
query: str,
cursor: Optional[str] = None,
search_type: Optional[Literal["Top", "Latest"]] = None
) -> Dict[str, Any]:
"""
Get Twitter search results for a given query.
Args:
query: Search query. Supports Twitter advanced search operators (e.g., "from:elonmusk doge -filter:replies").
cursor: Pagination cursor for additional results.
search_type: Type of search results ("Top" or "Latest").
Returns:
Dict containing search results with tweets and pagination info.
"""
params = {"query": query}
if cursor:
params["cursor"] = cursor
if search_type:
params["type"] = search_type
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/search",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to search Twitter: {e}")
@mcp.tool()
def get_user_by_username(username: str) -> Dict[str, Any]:
"""
Get detailed user profile information by Twitter username.
Args:
username: Twitter username (without @).
Returns:
Dict containing user profile information.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{username}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user by username: {e}")
@mcp.tool()
def get_user_by_id(user_id: str) -> Dict[str, Any]:
"""
Get detailed user profile information by Twitter user ID.
Args:
user_id: Twitter user numerical ID.
Returns:
Dict containing user profile information.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user by ID: {e}")
@mcp.tool()
def get_multiple_users_by_ids(user_ids: List[str]) -> Dict[str, Any]:
"""
Get user information for up to 100 Twitter users by their IDs.
Args:
user_ids: List of Twitter user IDs (max 100).
Returns:
Dict containing user profiles.
"""
if len(user_ids) > 100:
raise ValueError("Maximum 100 user IDs allowed")
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/twitter/users-by-ids",
headers=get_headers(),
json={"user_ids": user_ids}
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get multiple users by IDs: {e}")
@mcp.tool()
def get_multiple_users_by_usernames(usernames: List[str]) -> Dict[str, Any]:
"""
Get user information for up to 100 Twitter users by their usernames.
Args:
usernames: List of Twitter usernames (max 100).
Returns:
Dict containing user profiles.
"""
if len(usernames) > 100:
raise ValueError("Maximum 100 usernames allowed")
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/twitter/users-by-usernames",
headers=get_headers(),
json={"usernames": usernames}
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get multiple users by usernames: {e}")
# =============================================================================
# DATA API - User Connections (Followers/Following)
# =============================================================================
@mcp.tool()
def get_user_followers(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get followers of a Twitter user.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing follower profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/followers",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user followers: {e}")
@mcp.tool()
def get_user_verified_followers(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get verified followers of a Twitter user.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing verified follower profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/verified-followers",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get verified followers: {e}")
@mcp.tool()
def get_user_following(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get users that a Twitter user is following.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing following profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/following",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user following: {e}")
# =============================================================================
# DATA API - User Content (Tweets, Mentions, etc.)
# =============================================================================
@mcp.tool()
def get_user_tweets(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get tweets from a user's timeline.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing user tweets and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/tweets",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user tweets: {e}")
@mcp.tool()
def get_user_tweets_and_replies(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get tweets and replies from a user's timeline.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing user tweets and replies with pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/tweets-and-replies",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user tweets and replies: {e}")
@mcp.tool()
def get_user_mentions(username: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get tweets that mention a specific user.
Args:
username: Twitter username (without @).
cursor: Pagination cursor for additional results.
Returns:
Dict containing mentions and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{username}/mentions",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user mentions: {e}")
@mcp.tool()
def get_user_highlights(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get highlighted tweets from a user's profile.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing highlighted tweets and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/highlights",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user highlights: {e}")
@mcp.tool()
def get_user_affiliates(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get affiliated accounts for verified organizations.
Args:
user_id: Twitter user ID (must be verified organization).
cursor: Pagination cursor for additional results.
Returns:
Dict containing affiliated user profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/affiliates",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user affiliates: {e}")
@mcp.tool()
def get_user_lists(user_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get lists created by or subscribed to by a user.
Args:
user_id: Twitter user ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing user lists and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{user_id}/lists",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user lists: {e}")
@mcp.tool()
def get_user_extended_bio(username: str) -> Dict[str, Any]:
"""
Get extended bio information for a user.
Args:
username: Twitter username (without @).
Returns:
Dict containing extended bio information or empty object if none exists.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{username}/extended-bio",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get user extended bio: {e}")
# =============================================================================
# DATA API - Tweet Operations
# =============================================================================
@mcp.tool()
def get_tweet(tweet_id: str) -> Dict[str, Any]:
"""
Get complete details of a tweet by its ID.
Args:
tweet_id: Twitter tweet ID.
Returns:
Dict containing tweet details.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get tweet: {e}")
@mcp.tool()
def get_tweet_comments(tweet_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get comments/replies to a tweet.
Args:
tweet_id: Twitter tweet ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing tweet comments and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}/comments",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get tweet comments: {e}")
@mcp.tool()
def get_tweet_retweeters(tweet_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get users who retweeted a specific tweet.
Args:
tweet_id: Twitter tweet ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing retweeter profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}/retweeted_by",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get tweet retweeters: {e}")
@mcp.tool()
def get_tweet_quotes(tweet_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get quotes of a specific tweet.
Args:
tweet_id: Twitter tweet ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing quote tweets and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}/quotes",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get tweet quotes: {e}")
@mcp.tool()
def get_twitter_thread(thread_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get all tweets in a Twitter thread.
Args:
thread_id: Twitter thread ID (usually the first tweet's ID).
cursor: Pagination cursor for additional results.
Returns:
Dict containing thread tweets and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/thread/{thread_id}",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get Twitter thread: {e}")
# =============================================================================
# DATA API - Spaces, Lists, and Communities
# =============================================================================
@mcp.tool()
def get_twitter_space_details(space_id: str) -> Dict[str, Any]:
"""
Get detailed information about a Twitter Space.
Args:
space_id: Twitter Space ID.
Returns:
Dict containing Space details.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/space/{space_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get Twitter Space details: {e}")
@mcp.tool()
def get_list_details(list_id: str) -> Dict[str, Any]:
"""
Get detailed information about a Twitter List.
Args:
list_id: Twitter List ID.
Returns:
Dict containing List details.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/list/{list_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get list details: {e}")
@mcp.tool()
def get_list_members(list_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get members of a Twitter List.
Args:
list_id: Twitter List ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing list member profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/list/{list_id}/members",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get list members: {e}")
@mcp.tool()
def get_list_tweets(list_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get tweets from a specific Twitter List.
Args:
list_id: Twitter List ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing tweets from the list and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/list/{list_id}/tweets",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get list tweets: {e}")
@mcp.tool()
def get_community_tweets(community_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get tweets from a Twitter Community.
Args:
community_id: Twitter Community ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing community tweets and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/community/{community_id}/tweets",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get community tweets: {e}")
@mcp.tool()
def get_community_members(community_id: str, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get members of a Twitter Community.
Args:
community_id: Twitter Community ID.
cursor: Pagination cursor for additional results.
Returns:
Dict containing community member profiles and pagination info.
"""
params = {}
if cursor:
params["cursor"] = cursor
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/community/{community_id}/members",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get community members: {e}")
# =============================================================================
# SOCIAL ACTIONS API - Verification
# =============================================================================
@mcp.tool()
def verify_user_following(source_user_id: str, target_user_id: str) -> Dict[str, Any]:
"""
Verify if one user is following another user.
Args:
source_user_id: ID of the user to check.
target_user_id: ID of the user being followed.
Returns:
Dict containing following relationship status.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/user/{source_user_id}/following/{target_user_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to verify user following: {e}")
@mcp.tool()
def verify_user_retweeted(tweet_id: str, user_id: str) -> Dict[str, Any]:
"""
Verify if a user retweeted a specific tweet.
Args:
tweet_id: Twitter tweet ID.
user_id: Twitter user ID to check.
Returns:
Dict containing retweet verification status.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}/retweeted_by/{user_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to verify user retweeted: {e}")
@mcp.tool()
def verify_user_commented(tweet_id: str, user_id: str) -> Dict[str, Any]:
"""
Verify if a user commented on a specific tweet.
Args:
tweet_id: Twitter tweet ID.
user_id: Twitter user ID to check.
Returns:
Dict containing comment verification status and comment IDs.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/twitter/tweets/{tweet_id}/commented_by/{user_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to verify user commented: {e}")
# =============================================================================
# ACCOUNT MANAGEMENT
# =============================================================================
@mcp.tool()
def get_account_balance() -> Dict[str, Any]:
"""
Get remaining SocialData account balance in USD.
Returns:
Dict containing account balance information.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/user/balance",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get account balance: {e}")
@mcp.tool()
def set_global_webhook(webhook_url: str) -> Dict[str, Any]:
"""
Set global webhook URL for all monitors.
Args:
webhook_url: URL to receive webhook notifications.
Returns:
Dict containing webhook setup confirmation.
"""
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/user/webhook",
headers=get_headers(),
json={"webhook_url": webhook_url}
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to set global webhook: {e}")
# =============================================================================
# MONITORING API
# =============================================================================
@mcp.tool()
def create_user_tweets_monitor(
user_id: str,
webhook_url: Optional[str] = None
) -> Dict[str, Any]:
"""
Create a monitor for new tweets from a specific user.
Args:
user_id: Twitter user ID to monitor.
webhook_url: Optional monitor-specific webhook URL.
Returns:
Dict containing monitor creation confirmation.
"""
payload = {"user_id": user_id}
if webhook_url:
payload["webhook_url"] = webhook_url
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/monitors/user-tweets",
headers=get_headers(),
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to create user tweets monitor: {e}")
@mcp.tool()
def create_user_following_monitor(
user_id: str,
webhook_url: Optional[str] = None
) -> Dict[str, Any]:
"""
Create a monitor for when a user follows someone new.
Args:
user_id: Twitter user ID to monitor.
webhook_url: Optional monitor-specific webhook URL.
Returns:
Dict containing monitor creation confirmation.
"""
payload = {"user_id": user_id}
if webhook_url:
payload["webhook_url"] = webhook_url
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/monitors/user-following",
headers=get_headers(),
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to create user following monitor: {e}")
@mcp.tool()
def create_user_profile_monitor(
user_id: str,
webhook_url: Optional[str] = None
) -> Dict[str, Any]:
"""
Create a monitor for user profile changes.
Args:
user_id: Twitter user ID to monitor.
webhook_url: Optional monitor-specific webhook URL.
Returns:
Dict containing monitor creation confirmation.
"""
payload = {"user_id": user_id}
if webhook_url:
payload["webhook_url"] = webhook_url
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/monitors/user-profile",
headers=get_headers(),
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to create user profile monitor: {e}")
@mcp.tool()
def create_pump_fun_monitor(
webhook_url: Optional[str] = None
) -> Dict[str, Any]:
"""
Create a monitor for pump.fun coin mentions on Twitter.
Args:
webhook_url: Optional monitor-specific webhook URL.
Returns:
Dict containing monitor creation confirmation.
"""
payload = {}
if webhook_url:
payload["webhook_url"] = webhook_url
try:
with httpx.Client() as client:
response = client.post(
f"{BASE_URL}/monitors/search-pump-fun",
headers=get_headers(),
json=payload if payload else None
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to create pump.fun monitor: {e}")
@mcp.tool()
def list_monitors(page: Optional[int] = None) -> Dict[str, Any]:
"""
List all active monitors.
Args:
page: Page number for pagination.
Returns:
Dict containing list of active monitors.
"""
params = {}
if page:
params["page"] = page
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/monitors",
headers=get_headers(),
params=params
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to list monitors: {e}")
@mcp.tool()
def get_monitor_details(monitor_id: str) -> Dict[str, Any]:
"""
Get details about a specific monitor.
Args:
monitor_id: Monitor ID.
Returns:
Dict containing monitor details.
"""
try:
with httpx.Client() as client:
response = client.get(
f"{BASE_URL}/monitors/{monitor_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to get monitor details: {e}")
@mcp.tool()
def edit_monitor_webhook(monitor_id: str, webhook_url: Optional[str] = None) -> Dict[str, Any]:
"""
Edit or remove webhook URL for a specific monitor.
Args:
monitor_id: Monitor ID.
webhook_url: New webhook URL (None to remove).
Returns:
Dict containing webhook update confirmation.
"""
payload = {"webhook_url": webhook_url}
try:
with httpx.Client() as client:
response = client.patch(
f"{BASE_URL}/monitors/{monitor_id}",
headers=get_headers(),
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to edit monitor webhook: {e}")
@mcp.tool()
def delete_monitor(monitor_id: str) -> Dict[str, Any]:
"""
Delete an active monitor.
Args:
monitor_id: Monitor ID to delete.
Returns:
Dict containing deletion confirmation.
"""
try:
with httpx.Client() as client:
response = client.delete(
f"{BASE_URL}/monitors/{monitor_id}",
headers=get_headers()
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Failed to delete monitor: {e}")
# =============================================================================
# CONVENIENCE FUNCTIONS
# =============================================================================
@mcp.tool()
def get_user_complete_profile(username: str) -> Dict[str, Any]:
"""
Get complete user profile including basic info, extended bio, and recent tweets.
Args:
username: Twitter username (without @).
Returns:
Dict containing comprehensive user profile data.
"""
try:
# Get basic profile
profile = get_user_by_username(username)
# Get extended bio if available
try:
extended_bio = get_user_extended_bio(username)
profile["extended_bio"] = extended_bio
except:
profile["extended_bio"] = {}
# Get recent tweets
user_id = profile.get("id_str", "")
if user_id:
try:
recent_tweets = get_user_tweets(user_id)
profile["recent_tweets"] = recent_tweets.get("tweets", [])[:5] # Last 5 tweets
except:
profile["recent_tweets"] = []
return profile
except Exception as e:
raise Exception(f"Failed to get complete user profile: {e}")
@mcp.tool()
def analyze_tweet_engagement(tweet_id: str) -> Dict[str, Any]:
"""
Get comprehensive engagement data for a tweet including comments, retweets, and quotes.
Args:
tweet_id: Twitter tweet ID.
Returns:
Dict containing tweet and all engagement data.
"""
try:
# Get main tweet
tweet = get_tweet(tweet_id)
# Get engagement data
engagement = {
"tweet": tweet,
"comments": [],
"retweeters": [],
"quotes": []
}
# Get comments
try:
comments = get_tweet_comments(tweet_id)
engagement["comments"] = comments.get("tweets", [])
except:
pass
# Get retweeters
try:
retweeters = get_tweet_retweeters(tweet_id)
engagement["retweeters"] = retweeters.get("users", [])
except:
pass
# Get quotes
try:
quotes = get_tweet_quotes(tweet_id)
engagement["quotes"] = quotes.get("tweets", [])
except:
pass
return engagement
except Exception as e:
raise Exception(f"Failed to analyze tweet engagement: {e}")
if __name__ == "__main__":
mcp.run()