Reddit MCP Server

Instructions

Args:
    subreddit: Name of the subreddit to post in (with or without 'r/' prefix)
    title: Title of the post (max 300 characters)
    content: Content of the post (text for self posts, URL for link posts)
    flair: Flair to add to the post. Must be an available flair in the subreddit
    is_self: Whether this is a self (text) post (True) or link post (False)

Returns:
    Dictionary containing information about the created post

Raises:
    ValueError: If input validation fails or flair is invalid
    RuntimeError: For other errors during post creation

Implementation Reference

• server.py:1150-1277 (handler)

  The core handler function for the 'create_post' MCP tool. It validates inputs, checks subreddit and flair availability, creates self or link posts using PRAW's subreddit.submit(), formats the result using _format_post, and handles various errors like rate limits and subreddit access issues.

  @mcp.tool()
  @require_write_access
  def create_post(
      subreddit: str,
      title: str,
      content: str,
      flair: Optional[str] = None,
      is_self: bool = True,
  ) -> Dict[str, Any]:
      """Create a new post in a subreddit.
  
      Args:
          subreddit: Name of the subreddit to post in (with or without 'r/' prefix)
          title: Title of the post (max 300 characters)
          content: Content of the post (text for self posts, URL for link posts)
          flair: Flair to add to the post. Must be an available flair in the subreddit
          is_self: Whether this is a self (text) post (True) or link post (False)
  
      Returns:
          Dictionary containing information about the created post
  
      Raises:
          ValueError: If input validation fails or flair is invalid
          RuntimeError: For other errors during post creation
      """
      manager = RedditClientManager()
      if not manager.client:
          raise RuntimeError("Reddit client not initialized")
  
      # Input validation
      if not subreddit or not isinstance(subreddit, str):
          raise ValueError("Subreddit name is required")
      if not title or not isinstance(title, str):
          raise ValueError("Post title is required")
      if len(title) > 300:
          raise ValueError("Title must be 300 characters or less")
      if not content or not isinstance(content, str):
          raise ValueError("Post content/URL is required")
  
      # Clean up subreddit name (remove r/ prefix if present)
      clean_subreddit = subreddit[2:] if subreddit.startswith("r/") else subreddit
  
      try:
          logger.info(f"Creating post in r/{clean_subreddit}")
          subreddit_obj = manager.client.subreddit(clean_subreddit)
  
          # Verify subreddit exists and is postable
          _ = subreddit_obj.display_name
  
          # Check if flair is valid if provided
          if flair:
              try:
                  available_flairs = [
                      f["text"] for f in subreddit_obj.flair.link_templates
                  ]
                  if flair not in available_flairs:
                      raise ValueError(
                          f"Invalid flair. Available flairs: {', '.join(available_flairs)}"
                      )
              except Exception as flair_error:
                  logger.warning(f"Error checking flairs: {flair_error}")
                  raise ValueError(
                      "Failed to verify flair. The subreddit may not allow link flairs."
                  ) from flair_error
  
          # Create the post
          try:
              if is_self:
                  submission = subreddit_obj.submit(
                      title=title[:300],  # Ensure title is within limit
                      selftext=content,
                      flair_id=flair,
                      send_replies=True,
                  )
              else:
                  # Validate URL for link posts
                  if not content.startswith(("http://", "https://")):
                      content = f"https://{content}"
                  submission = subreddit_obj.submit(
                      title=title[:300],  # Ensure title is within limit
                      url=content,
                      flair_id=flair,
                      send_replies=True,
                  )
  
              logger.info(f"Post created successfully: {submission.permalink}")
  
              return {
                  "post": _format_post(submission),
                  "metadata": {
                      "created_at": _format_timestamp(time.time()),
                      "subreddit": clean_subreddit,
                      "is_self_post": is_self,
                      "permalink": f"https://reddit.com{submission.permalink}",
                      "id": submission.id,
                  },
              }
  
          except Exception as post_error:
              logger.error(f"Failed to create post in r/{clean_subreddit}: {post_error}")
              if "RATELIMIT" in str(post_error).upper():
                  raise RuntimeError(
                      "You're doing that too much. Please wait before posting again."
                  ) from post_error
              if "TOO_OLD" in str(post_error):
                  raise RuntimeError(
                      "This subreddit only allows posts from accounts with a minimum age or karma."
                  ) from post_error
              if "SUBREDDIT_NOEXIST" in str(post_error):
                  raise ValueError(f"r/{clean_subreddit} does not exist") from post_error
              raise RuntimeError(f"Failed to create post: {post_error}") from post_error
  
      except Exception as e:
          logger.error(f"Error in create_post for r/{clean_subreddit}: {e}")
          if "private" in str(e).lower():
              raise ValueError(
                  f"r/{clean_subreddit} is private or cannot be accessed"
              ) from e
          if "banned" in str(e).lower():
              raise ValueError(
                  f"r/{clean_subreddit} has been banned or doesn't exist"
              ) from e
          if "not found" in str(e).lower():
              raise ValueError(f"r/{clean_subreddit} not found") from e
          if isinstance(e, (ValueError, RuntimeError)):
              raise
          raise RuntimeError(f"Failed to create post: {e}") from e

• server.py:1150-1150 (registration)

  The @mcp.tool() decorator registers the create_post function as an MCP tool.

  @mcp.tool()

• server.py:117-134 (helper)

  Decorator applied to create_post that enforces write permissions by checking RedditClientManager for non-read-only mode and user authentication.

  def require_write_access(func: F) -> F:
      """Decorator to ensure write access is available."""
  
      @functools.wraps(func)
      def wrapper(*args: Any, **kwargs: Any) -> Any:
          reddit_manager = RedditClientManager()
          if reddit_manager.is_read_only:
              raise ValueError(
                  "Write operation not allowed in read-only mode. Please provide valid credentials."
              )
          if not reddit_manager.check_user_auth():
              raise Exception(
                  "Authentication required for write operations. "
                  "Please provide valid REDDIT_USERNAME and REDDIT_PASSWORD environment variables."
              )
          return func(*args, **kwargs)
  
      return cast(F, wrapper)

• server.py:180-224 (helper)

  Helper function used by create_post to format the response post information with stats, metadata, links, and engagement analysis.

  def _format_post(post: praw.models.Submission) -> str:
      """Format post information with AI-driven insights."""
      content_type = "Text Post" if post.is_self else "Link Post"
      content = post.selftext if post.is_self else post.url
  
      flags = []
      if post.over_18:
          flags.append("NSFW")
      if hasattr(post, "spoiler") and post.spoiler:
          flags.append("Spoiler")
      if post.edited:
          flags.append("Edited")
  
      # Add image URL section for non-self posts
      image_url_section = (
          f"""
          • Image URL: {post.url}"""
          if not post.is_self
          else ""
      )
  
      return f"""
          • Title: {post.title}
          • Type: {content_type}
          • Content: {content}
          • Author: u/{str(post.author)}
          • Subreddit: r/{str(post.subreddit)}{image_url_section}
          • Stats:
          - Score: {post.score:,}
          - Upvote Ratio: {post.upvote_ratio * 100:.1f}%
          - Comments: {post.num_comments:,}
          • Metadata:
          - Posted: {_format_timestamp(post.created_utc)}
          - Flags: {", ".join(flags) if flags else "None"}
          - Flair: {post.link_flair_text or "None"}
          • Links:
          - Full Post: https://reddit.com{post.permalink}
          - Short Link: https://redd.it/{post.id}
  
          📈 Engagement Analysis:
          - {_analyze_post_engagement(post.score, post.upvote_ratio, post.num_comments)}
  
          🎯 Best Time to Engage:
          - {_get_best_engagement_time(post.created_utc, post.score)}
          """