Reddit MCP Tool

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