Reddit MCP Tool

Instructions

Implementation Reference

• reddit_mcp/server.py:36-107 (handler)

  The primary handler function for the 'search_reddit_posts' MCP tool. Decorated with @mcp.tool() for automatic registration and schema inference from signature/docstring. Performs client check, calls helper method, formats output, handles errors.

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

• reddit_mcp/reddit_client.py:29-71 (helper)

  Core helper method in RedditClient that executes the Reddit API search using asyncpraw.Reddit.subreddit.search(), extracts relevant post fields into structured dicts, and returns a list of post data.

  async def search_posts(
      self, 
      subreddit_name: str, 
      query: str, 
      limit: int = 10,
      sort: str = "relevance",
      time_filter: str = "all"
  ) -> List[Dict[str, Any]]:
      """Search for posts in a subreddit."""
      try:
          subreddit = await self.reddit.subreddit(subreddit_name)
          
          # Search posts
          posts = []
          search_results = 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 posts in r/{subreddit_name}: {str(e)}")

• reddit_mcp/server.py:37-56 (schema)

  Input schema defined by function parameters with type hints and comprehensive docstring describing args, defaults, and return type. Used by FastMCP for tool schema generation.

  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
      """

• reddit_mcp/server.py:36-36 (registration)

  FastMCP decorator that registers the function as an MCP tool, inferring name from function name, schema from signature/docstring.

  @mcp.tool()