Reddit MCP Server

Instructions

Implementation Reference

• server.py:1394-1495 (handler)

  The primary handler implementation for the 'reply_to_comment' MCP tool. This function handles replying to a specific Reddit comment by fetching the comment object via PRAW, performing validations (e.g., length, accessibility), checking parent post status, and executing the reply operation. It returns formatted information about the new reply and parent comment.
  @mcp.tool() @require_write_access def reply_to_comment( comment_id: str, content: str ) -> Dict[str, Any]: """Post a reply to an existing Reddit comment. Args: comment_id: The ID of the comment to reply to (can be full URL, permalink, or just ID) content: The content of the reply (1-10000 characters) Returns: Dictionary containing information about the created reply and parent comment Raises: ValueError: If input validation fails or comment is not found RuntimeError: For other errors during reply creation """ manager = RedditClientManager() if not manager.client: raise RuntimeError("Reddit client not initialized") # Input validation if not comment_id or not isinstance(comment_id, str): raise ValueError("Comment ID is required") if not content or not isinstance(content, str): raise ValueError("Reply content is required") if len(content) < 1 or len(content) > 10000: raise ValueError("Reply must be between 1 and 10000 characters") try: # Clean up the comment_id if it's a full URL or permalink clean_comment_id = _extract_reddit_id(comment_id) logger.info(f"Creating reply to comment ID: {clean_comment_id}") # Get the comment object comment = manager.client.comment(id=clean_comment_id) # Verify the comment exists by accessing its attributes try: # Force fetch to verify existence _ = comment.body comment_author = getattr(comment, "author", None) comment_subreddit = comment.subreddit logger.info( f"Replying to comment: " f"Author: {comment_author}, " f"Subreddit: r/{comment_subreddit.display_name}" ) except Exception as e: logger.exception(f"Failed to access comment {clean_comment_id}: {e}") raise ValueError(f"Comment {clean_comment_id} not found or inaccessible") from e # Check if the parent submission is archived or locked try: submission = comment.submission if getattr(submission, "archived", False): raise ValueError("Cannot reply to a comment in an archived thread") if getattr(submission, "locked", False): raise ValueError("Cannot reply to a comment in a locked thread") except ValueError: raise except Exception as check_error: logger.debug(f"Could not check submission status: {check_error}") # Create the reply logger.info(f"Posting reply with content length: {len(content)} characters") try: reply = comment.reply(body=content) logger.info(f"Reply created successfully: {reply.id}") return { "reply": _format_comment(reply), "parent_comment": _format_comment(comment), "metadata": { "created_at": _format_timestamp(time.time()), "reply_id": reply.id, "parent_id": clean_comment_id, "subreddit": comment_subreddit.display_name, }, } except Exception as reply_error: logger.exception(f"Failed to create reply: {reply_error}") if "RATELIMIT" in str(reply_error).upper(): raise RuntimeError( "You're doing that too much. Please wait before replying again." ) from reply_error if "TOO_OLD" in str(reply_error): raise RuntimeError( "This thread is archived and cannot be replied to" ) from reply_error raise RuntimeError(f"Failed to post reply: {reply_error}") from reply_error except Exception as e: logger.exception(f"Error in reply_to_comment for ID {comment_id}: {e}") if isinstance(e, (ValueError, RuntimeError)): raise raise RuntimeError(f"Failed to create comment reply: {e}") from e
• server.py:1394-1395 (registration)

  The @mcp.tool() decorator registers the reply_to_comment function as an MCP tool in the FastMCP server. The @require_write_access decorator ensures write permissions are available.
  @mcp.tool() @require_write_access
• server.py:240-261 (helper)

  Helper function used by reply_to_comment to extract the clean comment ID from potentially full URLs or permalinks.
  def _extract_reddit_id(reddit_id: str) -> str: """Extract the base ID from a Reddit URL or ID. Args: reddit_id: Either a Reddit ID or a URL containing the ID Returns: The extracted Reddit ID """ if not reddit_id: raise ValueError("Empty ID provided") # If it's a URL, extract the ID part if "/" in reddit_id: # Handle both standard URLs and permalinks parts = [p for p in reddit_id.split("/") if p] # The ID is typically the last non-empty part reddit_id = parts[-1] logger.debug(f"Extracted ID {reddit_id} from URL") return reddit_id
• server.py:263-288 (helper)

  Helper function used to format comment details for the response, including analysis, called for both parent and new reply.
  def _format_comment(comment: praw.models.Comment) -> str: """Format comment information with AI-driven insights.""" flags = [] if comment.edited: flags.append("Edited") if hasattr(comment, "is_submitter") and comment.is_submitter: flags.append("OP") return f""" • Author: u/{str(comment.author)} • Content: {comment.body} • Stats: - Score: {comment.score:,} - Controversiality: {comment.controversiality if hasattr(comment, "controversiality") else "Unknown"} • Context: - Subreddit: r/{str(comment.subreddit)} - Thread: {comment.submission.title} • Metadata: - Posted: {_format_timestamp(comment.created_utc)} - Flags: {", ".join(flags) if flags else "None"} • Link: https://reddit.com{comment.permalink} 💬 Comment Analysis: - {_analyze_comment_impact(comment.score, bool(comment.edited), hasattr(comment, "is_submitter"))} """