Reddit MCP Server

Instructions

Implementation Reference

• server.py:1497-1559 (handler)

  The handler function implementing the 'join_subreddit' tool. It subscribes or unsubscribes the authenticated Reddit user to/from the specified subreddit using PRAW's subreddit.subscribe() and subreddit.unsubscribe() methods. Includes input validation, error handling, and logging. Registered via @mcp.tool() decorator and protected by @require_write_access.
  @mcp.tool() @require_write_access def join_subreddit(subreddit_name: str, unsubscribe: bool = False) -> Dict[str, Any]: """Join (subscribe to) or leave (unsubscribe from) a subreddit. Args: subreddit_name: Name of the subreddit to join/leave (with or without 'r/' prefix) unsubscribe: If True, leave the subreddit instead of joining Returns: Dictionary containing information about the action and subreddit Raises: ValueError: If subreddit name is invalid or subreddit not found RuntimeError: For other errors during the operation """ manager = RedditClientManager() if not manager.client: raise RuntimeError("Reddit client not initialized") if not subreddit_name or not isinstance(subreddit_name, str): raise ValueError("Subreddit name is required") # Clean up subreddit name clean_name = subreddit_name[2:] if subreddit_name.startswith("r/") else subreddit_name action = "leave" if unsubscribe else "join" try: logger.info(f"Attempting to {action} r/{clean_name}") sub = manager.client.subreddit(clean_name) # Verify subreddit exists try: display_name = sub.display_name except Exception as e: raise ValueError(f"Subreddit r/{clean_name} not found or inaccessible") from e if unsubscribe: sub.unsubscribe() message = f"Successfully unsubscribed from r/{display_name}" else: sub.subscribe() message = f"Successfully subscribed to r/{display_name}" logger.info(message) return { "success": True, "action": action, "subreddit": display_name, "message": message, "metadata": { "timestamp": time.time(), "subscribers": getattr(sub, "subscribers", None) } } except Exception as e: logger.error(f"Error {action}ing r/{clean_name}: {e}") if isinstance(e, (ValueError, RuntimeError)): raise raise RuntimeError(f"Failed to {action} r/{clean_name}: {e}") from e
• server.py:1497-1497 (registration)

  The @mcp.tool() decorator registers the join_subreddit function as an MCP tool.
  @mcp.tool()
• server.py:117-134 (helper)

  The @require_write_access decorator used on join_subreddit to ensure the Reddit client has write access and proper authentication.
  def require_write_access(func: F) -> F: """Decorator to ensure write access is available.""" @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> Any: reddit_manager = RedditClientManager() if reddit_manager.is_read_only: raise ValueError( "Write operation not allowed in read-only mode. Please provide valid credentials." ) if not reddit_manager.check_user_auth(): raise Exception( "Authentication required for write operations. " "Please provide valid REDDIT_USERNAME and REDDIT_PASSWORD environment variables." ) return func(*args, **kwargs) return cast(F, wrapper)