Reddit MCP Server

Instructions

Implementation Reference

• server.py:707-895 (handler)

  Primary handler function implementing the 'get_top_posts' MCP tool. Fetches top posts from specified subreddit using PRAW Reddit API, formats into structured dict with optional full comment trees, includes validation and error handling.
  @mcp.tool() def get_top_posts( subreddit: str, time_filter: str = "week", limit: int = 10, include_comments: bool = False, comment_replace_more_limit: int = 0, ) -> Dict[str, Any]: """Get top posts from a subreddit. Args: subreddit: Name of the subreddit (with or without 'r/' prefix) time_filter: Time period to filter posts (e.g. "day", "week", "month", "year", "all") limit: Number of posts to fetch (1-100) include_comments: If True, load and return the full comment forest for each post comment_replace_more_limit: Limit for replacing "MoreComments" objects (0 for none, None for all) Returns: Dictionary containing structured post information with the following structure: { 'subreddit': str, # Subreddit name 'time_filter': str, # The time period used for filtering 'posts': [ # List of posts, each with the following structure: { 'id': str, # Post ID 'title': str, # Post title 'author': str, # Author's username 'score': int, # Post score (upvotes - downvotes) 'upvote_ratio': float, # Ratio of upvotes to total votes 'num_comments': int, # Number of comments 'created_utc': float, # Post creation timestamp 'url': str, # URL to the post 'permalink': str, # Relative URL to the post 'is_self': bool, # Whether it's a self (text) post 'selftext': str, # Content of self post (if any) 'link_url': str, # URL for link posts (if any) 'over_18': bool, # Whether marked as NSFW 'spoiler': bool, # Whether marked as spoiler 'stickied': bool, # Whether stickied in the subreddit 'locked': bool, # Whether comments are locked 'distinguished': Optional[str], # Distinguishing type (e.g., 'moderator') 'flair': Optional[Dict], # Post flair information if any 'comments': Optional[List[Dict]], # present if include_comments is True }, ... ], 'metadata': { 'fetched_at': float, # Timestamp when data was fetched 'post_count': int, # Number of posts returned } } Raises: ValueError: If subreddit is invalid or time_filter is not valid RuntimeError: For other errors during the operation """ manager = RedditClientManager() if not manager.client: raise RuntimeError("Reddit client not initialized") if not subreddit or not isinstance(subreddit, str): raise ValueError("Subreddit name is required") valid_time_filters = ["hour", "day", "week", "month", "year", "all"] if time_filter not in valid_time_filters: raise ValueError( f"Invalid time filter. Must be one of: {', '.join(valid_time_filters)}" ) limit = max(1, min(100, limit)) # Ensure limit is between 1 and 100 # Clean up subreddit name (remove r/ prefix if present) clean_subreddit = subreddit[2:] if subreddit.startswith("r/") else subreddit try: logger.info( f"Getting top {limit} posts from r/{clean_subreddit} " f"(time_filter={time_filter}, include_comments={include_comments})" ) # Get the subreddit sub = manager.client.subreddit(clean_subreddit) # Verify subreddit exists and is accessible _ = sub.display_name # Fetch posts posts = list(sub.top(time_filter=time_filter, limit=limit)) if not posts: return { "subreddit": clean_subreddit, "time_filter": time_filter, "posts": [], "metadata": {"fetched_at": time.time(), "post_count": 0}, } # Format posts into structured data formatted_posts = [] for post in posts: try: # Get post data with error handling for each field post_data: Dict[str, Any] = { "id": post.id, "title": post.title, "author": str(post.author) if getattr(post, "author", None) else "[deleted]", "score": getattr(post, "score", 0), "upvote_ratio": getattr(post, "upvote_ratio", 0.0), "num_comments": getattr(post, "num_comments", 0), "created_utc": post.created_utc, "url": f"https://www.reddit.com{post.permalink}" if hasattr(post, "permalink") else "", "permalink": getattr(post, "permalink", ""), "is_self": getattr(post, "is_self", False), "selftext": getattr(post, "selftext", ""), "link_url": getattr(post, "url", ""), "over_18": getattr(post, "over_18", False), "spoiler": getattr(post, "spoiler", False), "stickied": getattr(post, "stickied", False), "locked": getattr(post, "locked", False), "distinguished": getattr(post, "distinguished", None), } # Add flair information if available if hasattr(post, "link_flair_text") and post.link_flair_text: post_data["flair"] = { "text": post.link_flair_text, "css_class": getattr(post, "link_flair_css_class", ""), "template_id": getattr(post, "link_flair_template_id", None), "text_color": getattr(post, "link_flair_text_color", None), "background_color": getattr( post, "link_flair_background_color", None ), } else: post_data["flair"] = None # Add comments if requested if include_comments: try: # Resolve all MoreComments to get the complete tree # limit=0 removes no MoreComments, limit=None removes all (slow!) post.comments.replace_more(limit=comment_replace_more_limit) top_level_comments = [ c for c in post.comments if isinstance(c, praw.models.Comment) ] post_data["comments"] = [ _serialize_comment_tree(c) for c in top_level_comments ] except Exception as comments_error: logger.exception( f"Error loading comments for post {getattr(post, 'id', 'unknown')}" ) post_data["comments"] = [] formatted_posts.append(post_data) except Exception as post_error: logger.error( f"Error processing post {getattr(post, 'id', 'unknown')}: {post_error}" ) continue return { "subreddit": clean_subreddit, "time_filter": time_filter, "posts": formatted_posts, "metadata": { "fetched_at": time.time(), "post_count": len(formatted_posts), }, } except Exception as e: logger.error(f"Error getting top posts from r/{clean_subreddit}: {e}") if "private" in str(e).lower(): raise ValueError( f"r/{clean_subreddit} is private or cannot be accessed" ) from e if "banned" in str(e).lower(): raise ValueError( f"r/{clean_subreddit} has been banned or doesn't exist" ) from e if "not found" in str(e).lower(): raise ValueError(f"r/{clean_subreddit} not found") from e raise RuntimeError(f"Failed to get top posts: {e}") from e
• server.py:307-333 (helper)

  Recursive helper to serialize PRAW comment trees into flat JSON-compatible dicts, used by get_top_posts when include_comments=True to include full comment forests.
  def _serialize_comment_tree(comment: praw.models.Comment) -> Dict[str, Any]: """Serialize a PRAW comment into a JSON-serializable tree structure.""" try: replies = [] if getattr(comment, "replies", None): replies = [ _serialize_comment_tree(reply) for reply in comment.replies if isinstance(reply, praw.models.Comment) ] except Exception as e: logger.error(f"Error while serializing replies for comment {getattr(comment, 'id', 'unknown')}: {e}") replies = [] return { "id": comment.id, "author": str(comment.author) if comment.author else "[deleted]", "body": getattr(comment, "body", ""), "score": getattr(comment, "score", 0), "created_utc": getattr(comment, "created_utc", 0.0), "permalink": getattr(comment, "permalink", ""), "is_submitter": getattr(comment, "is_submitter", False), "distinguished": getattr(comment, "distinguished", None), "stickied": getattr(comment, "stickied", False), "locked": getattr(comment, "locked", False), "replies": replies, }
• server.py:19-115 (helper)

  Singleton manager for PRAW Reddit client instance, used by all tools including get_top_posts for API access with fallback to read-only mode.
  class RedditClientManager: """Manages the Reddit client and its state.""" _instance = None _client = None _is_read_only = True def __new__(cls) -> "RedditClientManager": if cls._instance is None: cls._instance = super(RedditClientManager, cls).__new__(cls) cls._instance._initialize_client() return cls._instance def _initialize_client(self) -> None: """Initialize the Reddit client with appropriate credentials.""" client_id = getenv("REDDIT_CLIENT_ID") client_secret = getenv("REDDIT_CLIENT_SECRET") user_agent = getenv("REDDIT_USER_AGENT", "RedditMCPServer v1.0") username = getenv("REDDIT_USERNAME") password = getenv("REDDIT_PASSWORD") self._is_read_only = True try: # Try authenticated access first if credentials are provided if all([username, password, client_id, client_secret]): logger.info( f"Attempting to initialize Reddit client with user authentication for u/{username}" ) try: self._client = praw.Reddit( client_id=client_id, client_secret=client_secret, user_agent=user_agent, username=username, password=password, check_for_updates=False, ) # Test authentication if self._client.user.me() is None: raise ValueError(f"Failed to authenticate as u/{username}") logger.info(f"Successfully authenticated as u/{username}") self._is_read_only = False return except Exception as auth_error: logger.warning(f"Authentication failed: {auth_error}") logger.info("Falling back to read-only access") # Fall back to read-only with client credentials if client_id and client_secret: logger.info("Initializing Reddit client with read-only access") self._client = praw.Reddit( client_id=client_id, client_secret=client_secret, user_agent=user_agent, check_for_updates=False, read_only=True, ) return # Last resort: read-only without credentials logger.info( "Initializing Reddit client in read-only mode without credentials" ) self._client = praw.Reddit( user_agent=user_agent, check_for_updates=False, read_only=True, ) # Test read-only access self._client.subreddit("popular").hot(limit=1) except Exception as e: logger.error(f"Error initializing Reddit client: {e}") self._client = None @property def client(self) -> Optional[praw.Reddit]: """Get the Reddit client instance.""" return self._client @property def is_read_only(self) -> bool: """Check if the client is in read-only mode.""" return self._is_read_only def check_user_auth(self) -> bool: """Check if user authentication is available for write operations.""" if not self._client: logger.error("Reddit client not initialized") return False if self._is_read_only: logger.error("Reddit client is in read-only mode") return False return True
• server.py:707-707 (registration)

  MCP tool registration decorator for the get_top_posts function, handled by FastMCP instance at line 137.
  @mcp.tool()