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()