search_reddit_all

Instructions

Implementation Reference

• reddit_mcp/server.py:110-178 (handler)

  The FastMCP tool handler for 'search_reddit_all'. Decorated with @mcp.tool() for automatic registration and schema inference from signature/docstring. Handles client check, calls RedditClient.search_all_reddit, formats and returns results as string.

  @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)}"

• reddit_mcp/reddit_client.py:153-194 (helper)

  Supporting method in RedditClient that uses asyncpraw to search subreddit('all') with the given parameters and returns formatted post data dictionaries.

  async def search_all_reddit(
      self, 
      query: str, 
      limit: int = 10,
      sort: str = "relevance",
      time_filter: str = "all"
  ) -> List[Dict[str, Any]]:
      """Search for posts across all of Reddit (site-wide search)."""
      try:
          # Search all of reddit using the 'all' subreddit
          all_subreddit = await self.reddit.subreddit("all")
          
          posts = []
          search_results = all_subreddit.search(
              query, 
              limit=limit, 
              sort=sort, 
              time_filter=time_filter
          )
          
          async for submission in search_results:
              post_data = {
                  "id": submission.id,
                  "title": submission.title,
                  "author": str(submission.author) if submission.author else "[deleted]",
                  "score": submission.score,
                  "upvote_ratio": submission.upvote_ratio,
                  "url": submission.url,
                  "permalink": f"https://reddit.com{submission.permalink}",
                  "created_utc": submission.created_utc,
                  "num_comments": submission.num_comments,
                  "selftext": submission.selftext[:500] + "..." if len(submission.selftext) > 500 else submission.selftext,
                  "is_self": submission.is_self,
                  "domain": submission.domain,
                  "subreddit": str(submission.subreddit),
              }
              posts.append(post_data)
          
          return posts
          
      except Exception as e:
          raise Exception(f"Error searching all Reddit for query '{query}': {str(e)}")