Reddit MCP Tool

"""Reddit MCP Server implementation using FastMCP.""" import logging from typing import Optional from fastmcp import FastMCP from .config import RedditConfig from .reddit_client import RedditClient # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialize FastMCP mcp = FastMCP("reddit-mcp-tool") # Global client instance reddit_client: Optional[RedditClient] = None def initialize_reddit_client(): """Initialize the Reddit client with configuration.""" global reddit_client try: config = RedditConfig.from_env() reddit_client = RedditClient(config) logger.info("Reddit client initialized successfully in read-only mode") except Exception as e: logger.error(f"Failed to initialize Reddit client: {str(e)}") logger.error("Please ensure you have created a .env file with your Reddit API credentials.") logger.error("Copy env.example to .env and fill in your Reddit API details.") reddit_client = None @mcp.tool() async def search_reddit_posts( subreddit: str, query: str, limit: int = 10, sort: str = "relevance", time_filter: str = "all" ) -> str: """ Search for posts in a specific subreddit Args: subreddit: The name of the subreddit to search in (without r/) query: The search query limit: Number of posts to return (default: 10, max: 100) sort: Sort method - "relevance", "hot", "top", "new", "comments" (default: "relevance") time_filter: Time filter - "all", "day", "week", "month", "year" (default: "all") Returns: Human readable string containing search results """ if reddit_client is None: return """Error: Reddit client not initialized. To fix this: 1. Copy env.example to .env: cp env.example .env 2. Edit .env with your Reddit API credentials: - Get credentials from https://old.reddit.com/prefs/apps/ - Create a 'script' type app - Fill in REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT 3. Restart the MCP server Example .env content: REDDIT_CLIENT_ID=your_14_char_client_id REDDIT_CLIENT_SECRET=your_27_char_client_secret REDDIT_USER_AGENT=reddit-mcp-tool:v0.2.0 (by /u/yourusername)""" try: posts = await reddit_client.search_posts( subreddit_name=subreddit, query=query, limit=min(limit, 100), sort=sort, time_filter=time_filter ) if not posts: return f"No posts found in r/{subreddit} for query: '{query}'" result = f"Found {len(posts)} posts in r/{subreddit} for query: '{query}'\n\n" for i, post in enumerate(posts, 1): result += ( f"{i}. **{post['title']}**\n" f" Author: {post['author']}\n" f" Score: {post['score']} (upvote ratio: {post['upvote_ratio']:.0%})\n" f" Comments: {post['num_comments']}\n" f" Link: {post['permalink']}\n" f" Subreddit: r/{post['subreddit']}\n" ) if post['selftext'] and len(post['selftext']) > 0: preview = post['selftext'][:200] + "..." if len(post['selftext']) > 200 else post['selftext'] result += f" Content: {preview}\n" result += "\n" return result except Exception as e: logger.error(f"Error searching posts in r/{subreddit}: {str(e)}") return f"Error searching posts in r/{subreddit}: {str(e)}" @mcp.tool() async def search_reddit_all( query: str, limit: int = 10, sort: str = "relevance", time_filter: str = "all" ) -> str: """ Search for posts across all of Reddit (site-wide search) Args: query: The search query to search across all Reddit limit: Number of posts to return (default: 10, max: 100) sort: Sort method - "relevance", "hot", "top", "new", "comments" (default: "relevance") time_filter: Time filter - "all", "day", "week", "month", "year" (default: "all") Returns: Human readable string containing search results from across Reddit """ if reddit_client is None: return """Error: Reddit client not initialized. To fix this: 1. Copy env.example to .env: cp env.example .env 2. Edit .env with your Reddit API credentials: - Get credentials from https://old.reddit.com/prefs/apps/ - Create a 'script' type app - Fill in REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT 3. Restart the MCP server Example .env content: REDDIT_CLIENT_ID=your_14_char_client_id REDDIT_CLIENT_SECRET=your_27_char_client_secret REDDIT_USER_AGENT=reddit-mcp-tool:v0.2.0 (by /u/yourusername)""" try: posts = await reddit_client.search_all_reddit( query=query, limit=min(limit, 100), sort=sort, time_filter=time_filter ) if not posts: return f"No posts found across Reddit for query: '{query}'" result = f"Found {len(posts)} posts across all Reddit for query: '{query}'\n\n" for i, post in enumerate(posts, 1): result += ( f"{i}. **{post['title']}**\n" f" Author: {post['author']}\n" f" Score: {post['score']} (upvote ratio: {post['upvote_ratio']:.0%})\n" f" Comments: {post['num_comments']}\n" f" Link: {post['permalink']}\n" f" Subreddit: r/{post['subreddit']}\n" ) if post['selftext'] and len(post['selftext']) > 0: preview = post['selftext'][:200] + "..." if len(post['selftext']) > 200 else post['selftext'] result += f" Content: {preview}\n" result += "\n" return result except Exception as e: logger.error(f"Error searching all Reddit for query '{query}': {str(e)}") return f"Error searching all Reddit for query '{query}': {str(e)}" @mcp.tool() async def get_reddit_post_details(post_id: str) -> str: """ Get detailed information about a specific Reddit post Args: post_id: The Reddit post ID Returns: Human readable string containing detailed post information """ if reddit_client is None: return """Error: Reddit client not initialized. To fix this: 1. Copy env.example to .env: cp env.example .env 2. Edit .env with your Reddit API credentials: - Get credentials from https://old.reddit.com/prefs/apps/ - Create a 'script' type app - Fill in REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT 3. Restart the MCP server Example .env content: REDDIT_CLIENT_ID=your_14_char_client_id REDDIT_CLIENT_SECRET=your_27_char_client_secret REDDIT_USER_AGENT=reddit-mcp-tool:v0.2.0 (by /u/yourusername)""" try: post_details = await reddit_client.get_post_details(post_id) result = ( f"**{post_details['title']}**\n\n" f"Author: {post_details['author']}\n" f"Score: {post_details['score']} (upvote ratio: {post_details['upvote_ratio']:.0%})\n" f"Comments: {post_details['num_comments']}\n" f"Link: {post_details['permalink']}\n" f"Subreddit: r/{post_details['subreddit']}\n" f"Domain: {post_details['domain']}\n" f"Locked: {post_details['locked']}\n" f"Stickied: {post_details['stickied']}\n" ) if post_details.get('flair_text'): result += f"Flair: {post_details['flair_text']}\n" if post_details['selftext']: result += f"\nContent:\n{post_details['selftext']}\n" return result except Exception as e: logger.error(f"Error getting post details for {post_id}: {str(e)}") return f"Error getting post details for {post_id}: {str(e)}" @mcp.tool() async def get_subreddit_info(subreddit: str) -> str: """ Get information about a subreddit Args: subreddit: The name of the subreddit (without r/) Returns: Human readable string containing subreddit information """ if reddit_client is None: return """Error: Reddit client not initialized. To fix this: 1. Copy env.example to .env: cp env.example .env 2. Edit .env with your Reddit API credentials: - Get credentials from https://old.reddit.com/prefs/apps/ - Create a 'script' type app - Fill in REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT 3. Restart the MCP server Example .env content: REDDIT_CLIENT_ID=your_14_char_client_id REDDIT_CLIENT_SECRET=your_27_char_client_secret REDDIT_USER_AGENT=reddit-mcp-tool:v0.2.0 (by /u/yourusername)""" try: subreddit_info = await reddit_client.get_subreddit_info(subreddit) result = ( f"**r/{subreddit_info['name']}**\n\n" f"Title: {subreddit_info['title']}\n" f"Subscribers: {subreddit_info['subscribers']:,}\n" f"Active Users: {subreddit_info['active_user_count'] or 'N/A'}\n" f"NSFW: {subreddit_info['over18']}\n" f"URL: {subreddit_info['url']}\n\n" f"Description:\n{subreddit_info['public_description']}\n" ) if subreddit_info['description'] and subreddit_info['description'] != subreddit_info['public_description']: result += f"\nFull Description:\n{subreddit_info['description']}\n" return result except Exception as e: logger.error(f"Error getting subreddit info for r/{subreddit}: {str(e)}") return f"Error getting subreddit info for r/{subreddit}: {str(e)}" @mcp.tool() async def get_hot_reddit_posts(subreddit: str, limit: int = 10) -> str: """ Get hot posts from a subreddit Args: subreddit: The name of the subreddit (without r/) limit: Number of posts to return (default: 10, max: 100) Returns: Human readable string containing hot posts """ if reddit_client is None: return """Error: Reddit client not initialized. To fix this: 1. Copy env.example to .env: cp env.example .env 2. Edit .env with your Reddit API credentials: - Get credentials from https://old.reddit.com/prefs/apps/ - Create a 'script' type app - Fill in REDDIT_CLIENT_ID, REDDIT_CLIENT_SECRET, and REDDIT_USER_AGENT 3. Restart the MCP server Example .env content: REDDIT_CLIENT_ID=your_14_char_client_id REDDIT_CLIENT_SECRET=your_27_char_client_secret REDDIT_USER_AGENT=reddit-mcp-tool:v0.2.0 (by /u/yourusername)""" try: posts = await reddit_client.get_hot_posts(subreddit, min(limit, 100)) if not posts: return f"No hot posts found in r/{subreddit}" result = f"Hot posts from r/{subreddit}:\n\n" for i, post in enumerate(posts, 1): result += ( f"{i}. **{post['title']}**\n" f" Author: {post['author']}\n" f" Score: {post['score']} (upvote ratio: {post['upvote_ratio']:.0%})\n" f" Comments: {post['num_comments']}\n" f" Link: {post['permalink']}\n" ) if post['selftext'] and len(post['selftext']) > 0: preview = post['selftext'][:150] + "..." if len(post['selftext']) > 150 else post['selftext'] result += f" Content: {preview}\n" result += "\n" return result except Exception as e: logger.error(f"Error getting hot posts from r/{subreddit}: {str(e)}") return f"Error getting hot posts from r/{subreddit}: {str(e)}" def run_server(): """Entry point for the CLI command.""" # Initialize Reddit client initialize_reddit_client() # Run the FastMCP server mcp.run() if __name__ == "__main__": run_server()