# Directory Structure
```
├── __init__.py
├── .gitignore
├── .python-version
├── CHANGELOG.md
├── CONTRIBUTING.md
├── docs
│ └── demo.mp4
├── examples
│ └── autogen
│ ├── __init__.py
│ ├── main.py
│ ├── pyproject.toml
│ └── README.md
├── LICENSE
├── pyproject.toml
├── pyrightconfig.json
├── README.md
├── scripts
│ └── dump_schemas.py
├── src
│ └── reddit_mcp
│ ├── __init__.py
│ ├── __main.py__
│ ├── server.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── get_comments.py
│ │ ├── get_submission.py
│ │ ├── get_subreddit.py
│ │ ├── search_posts.py
│ │ └── search_subreddits.py
│ └── util
│ ├── __init__.py
│ ├── date_utils.py
│ └── reddit_client.py
├── tests
│ ├── __init__.py
│ └── test_reddit_tools.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
.venv/
env/
venv/
ENV/
dist/
build/
*.egg-info/
.eggs/
*.egg
# IDE
.idea/
.vscode/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Project specific
*.db
.env
.cache/
```
--------------------------------------------------------------------------------
/examples/autogen/README.md:
--------------------------------------------------------------------------------
```markdown
# Autogen with MCP server
This example demonstrates how to use the MCP server with AutoGen agents.
> **NOTE**: There seems to be a bug in autogen-mcp that causes Pydantic types with refs to raise an error. To use this example, comment out the `search_subreddits` tool in `tools/__init__.py`.
## Running the example
Add your OpenAI API key to an `.env` file:
```bash
OPENAI_API_KEY=<your-openai-api-key>
```
Install the dependencies:
```bash
uv sync
```
Run the example:
```bash
uv run main.py
```
## Example output
> Here are some interesting Reddit posts about AI agents:
>
> 1. **[Moore's Law for AI Agents](https://i.redd.it/2zht3052qupe1.png)**
> _Score: 76 | Comments: 42_
> _(Posted on March 20, 2025)_
>
> 2. **[custom AI agents](https://www.reddit.com/r/BlackboxAI_/comments/1jfgzxu/custom_ai_agents/)**
> _Score: 5 | Comments: 2_
> _(Posted on March 20, 2025)_
>
> 3. **[Stop Calling AI Automation "AI Agents" – It’s Misleading!](https://i.redd.it/wg127obp2wpe1.gif)**
> _Score: 2 | Comments: 0_
> _(Posted on March 20, 2025)_
>
> 4. **[Building an AI Agent with Memory and Adaptability](https://www.reddit.com/r/PromptEngineering/comments/1jfs3mt/building_an_ai_agent_with_memory_and_adaptability/)**
> _Score: 85 | Comments: 6_
> _(Posted on March 20, 2025)_
>
> 5. **[AI Mailing Agent MVP](https://www.reddit.com/r/SaaS/comments/1jfqrhk/ai_mailing_agent_mvp/)**
> _Score: 3 | Comments: 0_
> _(Posted on March 20, 2025)_
>
> 6. **[Moore's Law for AI Agents: if the length of tasks AIs can do continues doubling every 7 months, then the singularity is near](https://i.redd.it/hua9inf9jupe1.png)**
> _Score: 14 | Comments: 7_
> _(Posted on March 20, 2025)_
>
> 7. **[Optimizing AI Agents with Open-source High-Performance RAG framework](https://www.reddit.com/r/AI_Agents/comments/1jexngk/optimizing_ai_agents_with_opensouce/)**
> _Score: 17 | Comments: 5_
> _(Posted on March 19, 2025)_
>
> 8. **[Built an AI Agent to find and apply to jobs automatically](https://www.reddit.com/r/RemoteJobs/comments/1jfxfj9/built_an_ai_agent_to_find_and_apply_to_jobs/)**
> _Score: 73 | Comments: 31_
> _(Posted on March 20, 2025)_
>
> Please let me know if you need further information or details on any specific post! TERMINATE
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Reddit MCP
[](https://opensource.org/licenses/MIT)
A plug-and-play [MCP](https://modelcontextprotocol.io) server to browse, search, and read Reddit.
## Demo
Here's a short video showing how to use this in Claude Desktop:
https://github.com/user-attachments/assets/a2e9f2dd-a9ac-453f-acd9-1791380ebdad
## Features
- Detailed parameter validation with [pydantic](https://docs.pydantic.dev)
- Uses the reliable [PRAW](https://praw.readthedocs.io/) library under the hood
- Built-in rate limiting protection thanks to PRAW
## Caveats
- Only supports read features for now. If you want to use write features, upvote the [issue](https://github.com/GridfireAI/reddit-mcp/issues/1) or [send a PR](CONTRIBUTING.md)! 🙌
- Tools use tokens. To use this with Claude, you may need to be a Pro user to use many tool calls. Free tier users should be fine with lighter tool usage. Your token usage is your responsibility.
## Installation
### Prerequisite: Reddit API credentials
Create a [developer app](https://www.reddit.com/prefs/apps) in your Reddit account if you don't already have one. This will give you a `client_id` and `client_secret` to use in the following steps. If you already have these, you can skip this step.
### Claude Desktop
To install into Claude Desktop:
- Follow the instructions [here](https://modelcontextprotocol.io/quickstart/user) until the section "Open up the configuration file in any text editor."
- Add the following to the file depending on your preferred installation method:
### Using [uvx](https://docs.astral.sh/uv/guides/tools/) (recommended)
```json
"mcpServers": {
"reddit": {
"command": "uvx",
"args": ["reddit-mcp"],
"env": {
"REDDIT_CLIENT_ID": "<client_id>",
"REDDIT_CLIENT_SECRET": "<client_secret>"
}
}
}
```
### Using PIP
First install the package:
```bash
pip install reddit-mcp
```
Then add the following to the configuration file:
```json
"mcpServers": {
"reddit": {
"command": "python",
"args": ["-m", "reddit_mcp"],
"env": {
"REDDIT_CLIENT_ID": "<client_id>",
"REDDIT_CLIENT_SECRET": "<client_secret>"
}
}
}
```
### Others
You can use this server with any [MCP client](https://modelcontextprotocol.io/docs/clients), including agent frameworks (LangChain, LlamaIndex, AutoGen, etc). For an example AutoGen integration, check out the [example](examples/autogen).
## Tools
The tools the server will expose are:
| Name | Description |
| ---------------------------- | ---------------------------------------- |
| `get_comment` | Access a comment |
| `get_comments_by_submission` | Access comments of a submission |
| `get_submission` | Access a submission |
| `get_subreddit` | Access a subreddit by name |
| `search_posts` | Search posts in a subreddit |
| `search_subreddits` | Search subreddits by name or description |
## Contributing
Contributions are welcome! See [CONTRIBUTING.md](CONTRIBUTING.md) for more information.
## Acknowledgments
- [PRAW](https://praw.readthedocs.io/) for an amazingly reliable library 💙
```
--------------------------------------------------------------------------------
/CONTRIBUTING.md:
--------------------------------------------------------------------------------
```markdown
# Contributing to Reddit MCP
## Getting Started
1. Fork the repository
2. Clone your fork locally
3. Create a new branch for your feature (`git checkout -b feature/amazing-feature`)
4. Make your changes
5. Commit your changes (`git commit -m 'Add some amazing feature'`)
6. Push to your branch (`git push origin feature/amazing-feature`)
7. Open a Pull Request
## Development Setup
1. Set up your Python environment (we recommend using `uv`)
```bash
uv sync
```
2. Create a `.env` file in the project root with your Reddit API credentials:
```sh
REDDIT_CLIENT_ID=your_client_id
REDDIT_CLIENT_SECRET=your_client_secret
```
## Running your dev server
### Claude Desktop
Add the following to your Claude Desktop configuration file:
```json
{
"mcpServers": {
"reddit": {
"command": "<absolute path to your uv executable>",
"args": [
"run",
"--with",
"<absolute path to the project root>",
"reddit-mcp"
]
}
}
}
```
### Running directly
```bash
reddit-mcp
```
### MCP Inspector
```bash
npx @modelcontextprotocol/inspector $(which uv) --directory <absolute path to the project root> run main.py
```
## Testing
Run the test suite:
```bash
uv run pytest
```
## Development Tools
Check tool schemas:
```bash
uv run dump-schemas
```
```
--------------------------------------------------------------------------------
/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/reddit_mcp/util/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/examples/autogen/__init__.py:
--------------------------------------------------------------------------------
```python
"""Reddit agent example using autogen."""
```
--------------------------------------------------------------------------------
/src/reddit_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
from .server import serve
def main():
"""MCP Reddit Server - Reddit functionality for MCP"""
import asyncio
asyncio.run(serve())
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/scripts/dump_schemas.py:
--------------------------------------------------------------------------------
```python
from reddit_mcp.tools import tools
from pydantic import TypeAdapter
import json
def main():
schemas = {tool.__name__: TypeAdapter(tool).json_schema() for tool in tools}
print(json.dumps(schemas, indent=2))
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/reddit_mcp/server.py:
--------------------------------------------------------------------------------
```python
from mcp.server.fastmcp import FastMCP
from .tools import tools
import logging
def serve():
logger = logging.getLogger("mcp")
mcp = FastMCP("Reddit")
for tool in tools:
logger.info(f"Registering tool: {tool.__name__}")
mcp.tool()(tool)
logger.info("Starting MCP server...")
mcp.run(transport="stdio")
```
--------------------------------------------------------------------------------
/src/reddit_mcp/util/date_utils.py:
--------------------------------------------------------------------------------
```python
from datetime import datetime
def format_utc_timestamp(timestamp: float, format: str = "%Y-%m-%d") -> str:
"""
Convert a UTC timestamp to a formatted date string.
Args:
timestamp (float): UTC timestamp to convert
format (str, optional): Date format string. Defaults to "%Y-%m-%d".
Returns:
str: Formatted date string
"""
return datetime.utcfromtimestamp(timestamp).strftime(format)
```
--------------------------------------------------------------------------------
/examples/autogen/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "reddit-mcp-autogen-example"
version = "0.1.0"
description = "Reddit agent example using autogen"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"reddit-mcp", # Parent package
"autogen-ext[mcp,openai]>=0.4.9.2",
"mcp[cli]>=1.4.1",
"python-dotenv>=1.0.1",
"autogen-agentchat>=0.4.9.2",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["."]
[tool.uv.sources]
reddit-mcp = { workspace = true }
```
--------------------------------------------------------------------------------
/src/reddit_mcp/util/reddit_client.py:
--------------------------------------------------------------------------------
```python
import os
import praw
from typing import Optional
from dotenv import load_dotenv
load_dotenv()
class RedditClient:
_instance: Optional["RedditClient"] = None
def __init__(self):
self.reddit = praw.Reddit(
client_id=os.getenv("REDDIT_CLIENT_ID"),
client_secret=os.getenv("REDDIT_CLIENT_SECRET"),
user_agent="Interesting Posts Reader",
)
@classmethod
def get_instance(cls) -> "RedditClient":
if cls._instance is None:
cls._instance = cls()
return cls._instance
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/__init__.py:
--------------------------------------------------------------------------------
```python
from .get_submission import get_submission
from .get_subreddit import get_subreddit
from .get_comments import get_comments_by_submission, get_comment_by_id
from .search_posts import search_posts
from .search_subreddits import search_subreddits
__all__ = [
"get_submission",
"get_subreddit",
"get_comments_by_submission",
"get_comment_by_id",
"search_posts",
"search_subreddits",
]
# Registry of all available tools
tools = [
get_submission,
get_subreddit,
get_comments_by_submission,
get_comment_by_id,
search_posts,
search_subreddits,
]
```
--------------------------------------------------------------------------------
/CHANGELOG.md:
--------------------------------------------------------------------------------
```markdown
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [0.1.0] - 2024-03-20
### Added
- Initial release
- Basic Reddit API integration using PRAW
- Read-only tools for accessing Reddit content:
- `get_comment`
- `get_comments_by_submission`
- `get_submission`
- `get_subreddit`
- `search_posts`
- `search_subreddits`
- MCP server implementation
- Documentation and examples
```
--------------------------------------------------------------------------------
/pyrightconfig.json:
--------------------------------------------------------------------------------
```json
{
"include": ["tools", "util", "tests"],
"exclude": ["**/node_modules", "**/__pycache__", ".venv"],
"strict": ["tools", "util"],
"typeCheckingMode": "strict",
"useLibraryCodeForTypes": true,
"reportMissingTypeStubs": "warning",
"reportUnknownMemberType": "warning",
"reportUnknownVariableType": "warning",
"reportUnknownArgumentType": "warning",
"reportMissingParameterType": "error",
"reportMissingTypeArgument": "error",
"reportInvalidTypeVarUse": "error",
"reportUntypedFunctionDecorator": "warning",
"reportUnknownParameterType": "warning",
"reportPrivateUsage": "warning",
"pythonVersion": "3.12",
"venvPath": ".",
"venv": ".venv"
}
```
--------------------------------------------------------------------------------
/examples/autogen/main.py:
--------------------------------------------------------------------------------
```python
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.agents import AssistantAgent
from autogen_core import CancellationToken
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StdioServerParams, mcp_server_tools
from dotenv import load_dotenv
import asyncio
load_dotenv()
async def main() -> None:
server_params = StdioServerParams(
command="/Users/josh/.local/bin/uv",
args=[
"run",
"--directory",
"../../",
"--with",
"mcp[cli]",
"mcp",
"run",
"main.py",
],
)
# Get all available tools from the server
tools = await mcp_server_tools(server_params)
print("Tools discovered:", [f"{t.name}: {t.description}" for t in tools])
# Create an agent that can use all the tools
agent = AssistantAgent(
name="researcher",
model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
tools=tools, # type: ignore
reflect_on_tool_use=True,
)
# The agent can now use any of the Reddit tools
result = await agent.run(
task="Find some interesting posts on Reddit about AI agents",
cancellation_token=CancellationToken(),
)
print(result.messages[-1].content)
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/get_subreddit.py:
--------------------------------------------------------------------------------
```python
from pydantic import BaseModel, Field, validate_call
from ..util.reddit_client import RedditClient
class SubredditResult(BaseModel):
"""Subreddit details"""
display_name: str = Field(description="Display name of the subreddit")
title: str = Field(description="Title of the subreddit")
description: str = Field(description="Full subreddit description")
public_description: str = Field(description="Short public description")
subscribers: int = Field(description="Number of subscribers")
created_utc: float = Field(description="UTC timestamp when subreddit was created")
over18: bool = Field(description="Whether the subreddit is NSFW")
url: str = Field(description="URL of the subreddit")
@validate_call(validate_return=True)
def get_subreddit(subreddit_name: str) -> SubredditResult:
"""
Retrieve a subreddit by name.
Args:
subreddit_name: Name of the subreddit to retrieve
Returns:
Detailed information about the subreddit
"""
client = RedditClient.get_instance()
subreddit = client.reddit.subreddit(subreddit_name)
return SubredditResult(
display_name=subreddit.display_name,
title=subreddit.title,
description=subreddit.description,
public_description=subreddit.public_description,
subscribers=subreddit.subscribers,
created_utc=subreddit.created_utc,
over18=subreddit.over18,
url=subreddit.url,
)
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/get_submission.py:
--------------------------------------------------------------------------------
```python
from pydantic import BaseModel, Field, validate_call
from ..util.reddit_client import RedditClient
from ..util.date_utils import format_utc_timestamp
class SubmissionResult(BaseModel):
"""Reddit submission details"""
title: str = Field(description="Title of the submission")
url: str = Field(description="URL of the submission")
author: str | None = Field(description="Username of the author, or None if deleted")
subreddit: str = Field(description="Name of the subreddit")
score: int = Field(description="Number of upvotes minus downvotes")
num_comments: int = Field(description="Number of comments on the submission")
selftext: str = Field(description="Text content of the submission")
created_utc: str = Field(description="UTC timestamp when submission was created")
@validate_call(validate_return=True)
def get_submission(submission_id: str) -> SubmissionResult:
"""
Retrieve a specific submission by ID.
Args:
submission_id: ID of the submission to retrieve
Returns:
Detailed information about the submission
"""
client = RedditClient.get_instance()
submission = client.reddit.submission(submission_id)
return SubmissionResult(
title=submission.title,
url=submission.url,
author=None if submission.author is None else submission.author.name,
subreddit=submission.subreddit.display_name,
score=submission.score,
num_comments=submission.num_comments,
selftext=submission.selftext,
created_utc=format_utc_timestamp(submission.created_utc),
)
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "reddit-mcp"
version = "0.1.2"
description = "Reddit API tools and examples"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"certifi>=2025.1.31",
"charset-normalizer>=3.4.1",
"idna>=3.10",
"praw>=7.8.1",
"prawcore>=2.4.0",
"python-dotenv>=1.0.1",
"requests>=2.32.3",
"update-checker>=0.18.0",
"urllib3>=2.3.0",
"websocket-client>=1.8.0",
"pydantic>=2.10.0",
"mcp[cli]>=1.4.1",
]
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src", "."]
include = ["reddit_mcp*", "scripts*"]
[tool.uv.workspace]
members = ["examples/autogen", "examples/openai-function-calling"]
[tool.uv.sources]
reddit-mcp = { workspace = true }
[dependency-groups]
dev = [
"pytest>=7.0.0",
"pytest-asyncio>=0.23.0",
]
[project.scripts]
dump-schemas = "scripts.dump_schemas:main"
reddit-mcp = "reddit_mcp:main"
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
addopts = "-v"
[tool.pyright]
include = ["src", "tests"]
exclude = ["**/node_modules", "**/__pycache__", ".venv"]
strict = ["src"]
typeCheckingMode = "strict"
useLibraryCodeForTypes = true
reportMissingTypeStubs = "warning"
reportUnknownMemberType = "warning"
reportUnknownVariableType = "warning"
reportUnknownArgumentType = "warning"
reportMissingParameterType = "error"
reportMissingTypeArgument = "error"
reportInvalidTypeVarUse = "error"
reportUntypedFunctionDecorator = "warning"
reportUnknownParameterType = "warning"
reportPrivateUsage = "warning"
[tool.ruff]
line-length = 88
target-version = "py312"
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"N", # pep8-naming
"UP", # pyupgrade
"ANN", # flake8-annotations
]
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/search_posts.py:
--------------------------------------------------------------------------------
```python
from typing import List, Literal
from pydantic import BaseModel, Field, validate_call
from ..util.reddit_client import RedditClient
from ..util.date_utils import format_utc_timestamp
class PostResult(BaseModel):
"""Reddit post search result"""
id: str = Field(description="Unique identifier of the post")
title: str = Field(description="Title of the post")
url: str = Field(description="URL of the post")
score: int = Field(description="Number of upvotes minus downvotes")
num_comments: int = Field(description="Number of comments on the post")
created_utc: str = Field(description="UTC timestamp when post was created")
class SearchPostsParams(BaseModel):
"""Parameters for searching posts within a subreddit"""
subreddit_name: str = Field(description="Name of the subreddit to search in")
query: str = Field(description="Search query string")
sort: Literal["relevance", "hot", "top", "new", "comments"] = Field(
default="relevance", description="How to sort the results"
)
syntax: Literal["cloudsearch", "lucene", "plain"] = Field(
default="lucene", description="Query syntax to use"
)
time_filter: Literal["all", "year", "month", "week", "day", "hour"] = Field(
default="all", description="Time period to limit results to"
)
@validate_call(validate_return=True)
def search_posts(params: SearchPostsParams) -> List[PostResult]:
"""
Search for posts within a subreddit.
Args:
params: Search parameters including subreddit name, query, and filters
Returns:
List of matching posts with their details
"""
client = RedditClient.get_instance()
subreddit = client.reddit.subreddit(params.subreddit_name)
posts = subreddit.search(
query=params.query,
sort=params.sort,
syntax=params.syntax,
time_filter=params.time_filter,
)
return [
PostResult(
id=post.id,
title=post.title,
url=post.url,
score=post.score,
num_comments=post.num_comments,
created_utc=format_utc_timestamp(post.created_utc),
)
for post in posts
]
```
--------------------------------------------------------------------------------
/tests/test_reddit_tools.py:
--------------------------------------------------------------------------------
```python
import pytest
from reddit_mcp.tools.search_subreddits import (
search_subreddits,
SearchByName,
SearchByDescription,
)
from reddit_mcp.tools.get_subreddit import get_subreddit, SubredditResult
from reddit_mcp.tools.search_posts import search_posts, SearchPostsParams
from reddit_mcp.tools.get_submission import get_submission, SubmissionResult
from reddit_mcp.tools.get_comments import (
get_comments_by_submission,
get_comment_by_id,
CommentResult,
)
def test_search_subreddits():
# Test name-based search
results = search_subreddits(SearchByName(type="name", query="computer"))
assert isinstance(results, list)
if results:
assert isinstance(results[0].name, str)
assert isinstance(results[0].public_description, str)
# Test description-based search
results = search_subreddits(
SearchByDescription(type="description", query="computers")
)
assert isinstance(results, list)
if results:
assert isinstance(results[0].name, str)
assert isinstance(results[0].public_description, str)
def test_get_subreddit():
result = get_subreddit("ChatGPT")
assert isinstance(result, SubredditResult)
assert isinstance(result.display_name, str)
assert isinstance(result.subscribers, int)
def test_search_posts():
# Test with Pydantic model params
results = search_posts(
SearchPostsParams(subreddit_name="ChatGPT", query="artificial intelligence")
)
assert isinstance(results, list)
if results:
assert isinstance(results[0].id, str)
assert isinstance(results[0].title, str)
def test_get_submission():
result = get_submission("1j66jbs")
assert isinstance(result, SubmissionResult)
assert isinstance(result.title, str)
assert isinstance(result.url, str)
def test_get_comments():
# Test getting comments by submission
comments = get_comments_by_submission("1j66jbs", replace_more=False)
assert isinstance(comments, list)
if comments:
assert isinstance(comments[0], CommentResult)
assert isinstance(comments[0].id, str)
assert isinstance(comments[0].body, str)
# Test getting single comment
comment = get_comment_by_id("mgmk7d2")
assert isinstance(comment, CommentResult)
assert isinstance(comment.id, str)
assert isinstance(comment.body, str)
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/get_comments.py:
--------------------------------------------------------------------------------
```python
from typing import List
from pydantic import BaseModel, Field, validate_call
from ..util.reddit_client import RedditClient
from ..util.date_utils import format_utc_timestamp
from praw.models import MoreComments
class CommentResult(BaseModel):
"""Reddit comment details"""
id: str = Field(description="Unique identifier of the comment")
body: str = Field(description="Text content of the comment")
author: str | None = Field(description="Username of the author, or None if deleted")
created_utc: str = Field(description="UTC timestamp when comment was created")
is_submitter: bool = Field(
description="Whether the comment author is the submission author"
)
score: int = Field(description="Number of upvotes minus downvotes")
replies: List["CommentResult"] = Field(
description="List of reply comments", default_factory=list
)
CommentResult.model_rebuild() # Required for self-referential models
def comment_to_model(comment) -> CommentResult:
"""Convert PRAW comment object to CommentResult model."""
# Skip MoreComments objects
if isinstance(comment, MoreComments):
return None
return CommentResult(
id=comment.id,
body=comment.body,
author=None if comment.author is None else comment.author.name,
created_utc=format_utc_timestamp(comment.created_utc),
is_submitter=comment.is_submitter,
score=comment.score,
replies=[
result
for reply in comment.replies
if (result := comment_to_model(reply)) is not None
],
)
@validate_call(validate_return=True)
def get_comments_by_submission(
submission_id: str, replace_more: bool = True
) -> List[CommentResult]:
"""
Retrieve comments from a specific submission.
Args:
submission_id: ID of the submission to get comments from
replace_more: Whether to replace MoreComments objects with actual comments
Returns:
List of comments with their replies
"""
client = RedditClient.get_instance()
submission = client.reddit.submission(submission_id)
if replace_more:
submission.comments.replace_more()
return [
result
for comment in submission.comments.list()
if (result := comment_to_model(comment)) is not None
]
@validate_call(validate_return=True)
def get_comment_by_id(comment_id: str) -> CommentResult:
"""
Retrieve a specific comment by ID.
Args:
comment_id: ID of the comment to retrieve
Returns:
Comment details with any replies
"""
client = RedditClient.get_instance()
return comment_to_model(client.reddit.comment(comment_id))
```
--------------------------------------------------------------------------------
/src/reddit_mcp/tools/search_subreddits.py:
--------------------------------------------------------------------------------
```python
from typing import List, Literal, Union
from ..util.reddit_client import RedditClient
from ..util.date_utils import format_utc_timestamp
from pydantic import BaseModel, Field, validate_call
class SubredditResult(BaseModel):
"""Subreddit search result"""
name: str = Field(description="Display name of the subreddit")
public_description: str = Field(description="Short description shown publicly")
url: str = Field(description="URL of the subreddit")
subscribers: int | None = Field(default=None, description="Number of subscribers")
created_utc: str = Field(description="UTC date when subreddit was created")
description: str | None = Field(
default=None,
description="Full subreddit description with markdown formatting",
)
class SearchByName(BaseModel):
"""Parameters for searching subreddits by name"""
type: Literal["name"]
query: str
include_nsfw: bool = Field(
default=False,
description="Whether to include NSFW subreddits in search results",
)
exact_match: bool = Field(
default=False, description="If True, only return exact name matches"
)
class SearchByDescription(BaseModel):
"""Parameters for searching subreddits by description"""
type: Literal["description"]
query: str
include_full_description: bool = Field(
default=False,
description="Whether to include the full subreddit description (aka sidebar description) in results -- can be very long and contain markdown formatting",
)
SearchParams = Union[SearchByName, SearchByDescription]
@validate_call(validate_return=True)
def search_subreddits(by: SearchParams) -> List[SubredditResult]:
"""
Search for subreddits using either name-based or description-based search.
Args:
by: Search parameters, either SearchByName or SearchByDescription
Returns:
List of matching subreddits with their details
"""
client = RedditClient.get_instance()
if by.type == "name":
subreddits = client.reddit.subreddits.search_by_name(
by.query, exact=by.exact_match, include_nsfw=by.include_nsfw
)
else: # by.type == "description"
subreddits = client.reddit.subreddits.search(by.query)
return [
SubredditResult(
name=subreddit.display_name,
public_description=subreddit.public_description,
description=(
subreddit.description
if (by.type == "description" and by.include_full_description)
else None
),
url=subreddit.url,
subscribers=subreddit.subscribers,
created_utc=format_utc_timestamp(subreddit.created_utc),
)
for subreddit in subreddits
]
```