jarianc/reddit-cli
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Compare commits

base: jarianc:4eb9d08d29bcef80a7feba7e51df878c5d0f80bc
Branches Tags
jarianc:main
..
compare: jarianc:b3db83cbdde499e204c3a950749a5e53d3484a81
Branches Tags
jarianc:main

No commits in common. "4eb9d08d29bcef80a7feba7e51df878c5d0f80bc" and "b3db83cbdde499e204c3a950749a5e53d3484a81" have entirely different histories.
4eb9d08d29 ... b3db83cbdd

13 changed files with 7 additions and 1379 deletions
Show Stats Expand all files Collapse all files

32
.dockerignore
View File

@ -1,32 +0,0 @@
# Docker ignore file
# Virtual environment
venv/
env/
.venv/
# Python cache
__pycache__/
*.pyc
*.pyo
*.pyd
.Python
.pytest_cache/
.coverage
# Logs
logs/
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Test files
test_*.py
*_test.py

143
.gitea/workflows/ci.yml
View File

@ -1,143 +0,0 @@
name: CI
on:
push:
branches: [main, master]
pull_request:
branches: [main, master]
env:
GITEA_URL: https://git.home.ms
jobs:
lint:
runs-on: ubuntu-latest
container:
image: gitea-job-image
steps:
- name: Clone repo
run: |
rm -rf $GITHUB_WORKSPACE/*
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
- name: Run ruff (Python lint)
if: always()
run: |
if [[ -f pyproject.toml ]]; then
pip3 install ruff
ruff check .
else
echo "No Python project detected, skipping ruff"
fi
- name: Run npm lint (JS/TS)
if: always()
run: |
if [[ -f package.json ]]; then
npm ci
npm run lint --if-present || true
else
echo "No Node.js project detected, skipping npm lint"
fi
test:
runs-on: ubuntu-latest
container:
image: gitea-job-image
steps:
- name: Clone repo
run: |
rm -rf $GITHUB_WORKSPACE/*
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
- name: Run pytest (Python)
if: always()
run: |
if [[ -f pyproject.toml ]]; then
python3 -m pip install --upgrade pip
pip3 install -e ".[dev]" 2>/dev/null || pip3 install -e . 2>/dev/null || true
pip3 install pytest
pytest tests/ -v --tb=short 2>/dev/null || true
else
echo "No Python project detected, skipping pytest"
fi
- name: Run npm test (JS/TS)
if: always()
run: |
if [[ -f package.json ]]; then
npm ci
npm run test --if-present || true
else
echo "No Node.js project detected, skipping npm test"
fi
- name: Run Go tests
if: always()
run: |
if [[ -f go.mod ]]; then
go test ./...
else
echo "No Go project detected, skipping go test"
fi
docker-build:
runs-on: ubuntu-latest
steps:
- name: Clone repo
run: |
rm -rf $GITHUB_WORKSPACE/*
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
- name: Build Docker image
if: always()
run: |
if [[ -f Dockerfile ]]; then
docker build -t $GITHUB_REPOSITORY:test .
else
echo "No Dockerfile found, skipping docker build"
fi
security:
runs-on: ubuntu-latest
container:
image: gitea-job-image
steps:
- name: Clone repo
run: |
rm -rf $GITHUB_WORKSPACE/*
git clone --depth 1 $GITEA_URL/$GITHUB_REPOSITORY $GITHUB_WORKSPACE
git -C $GITHUB_WORKSPACE checkout $GITHUB_SHA 2>/dev/null || true
- name: Run bandit (Python SAST)
if: always()
run: |
if [[ -f pyproject.toml ]]; then
pip3 install bandit
bandit -r . --severity-level high --confidence-level high --exclude tests/,test_*
else
echo "No Python project detected, skipping bandit"
fi
- name: Run npm audit (JS/TS)
if: always()
run: |
if [[ -f package.json ]]; then
npm ci
npm audit --audit-level=high 2>/dev/null || echo "npm audit: vulnerabilities found (non-blocking)"
else
echo "No Node.js project detected, skipping npm audit"
fi
build-result:
needs: [lint, test, docker-build, security]
runs-on: ubuntu-latest
container:
image: gitea-job-image
if: always()
steps:
- name: Summary
run: echo "All CI checks completed"

2
.gitignore vendored
View File

@ -5,5 +5,3 @@ lib/
target/ target/
pyvenv.cfg pyvenv.cfg
.aider*

28
Dockerfile
View File

@ -1,28 +0,0 @@
# Dockerfile for Reddit MCP Server
FROM python:3.9-slim
# Set working directory
WORKDIR /app
# Copy requirements first (for better caching)
COPY requirements.txt .
# Install dependencies
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY src/ ./src/
COPY run_mcp_server.sh .
# Make the run script executable
RUN chmod +x run_mcp_server.sh
# Expose port
EXPOSE 5000
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:5000/health || exit 1
# Default command
CMD ["./run_mcp_server.sh"]

21
LICENSE
View File

@ -1,21 +0,0 @@
MIT License
Copyright (c) 2025 Jarian Cottingham
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

57
README.md
View File

@ -14,7 +14,6 @@ A command-line interface for browsing Reddit posts with search capabilities, int
- **Comments Viewer**: Cycle through comments using 'c' key - **Comments Viewer**: Cycle through comments using 'c' key
- **AI Integration**: Get AI-generated summaries using Ollama API - **AI Integration**: Get AI-generated summaries using Ollama API
- **Pagination**: Navigate between pages of results ('n' key) - **Pagination**: Navigate between pages of results ('n' key)
- **MCP Server**: REST API endpoints for Reddit querying with resilience and retry logic
## Project Structure ## Project Structure
@ -23,7 +22,7 @@ reddit-cli/
├── src/ ├── src/
│ ├── __init__.py │ ├── __init__.py
│ ├── main.py # Main application entrypoint │ ├── main.py # Main application entrypoint
│ ├── mcp_server.py # MCP server implementation with REST endpoints │ ├── reddit_client.py # Reddit API client implementation
│ └── ai_client.py # Ollama AI integration │ └── ai_client.py # Ollama AI integration
├── tests/ ├── tests/
│ └── test_cli.py # Unit tests │ └── test_cli.py # Unit tests
@ -54,7 +53,6 @@ pip install -r requirements.txt
## Usage ## Usage
### Running the CLI Interface
Run the CLI interface with a search query: Run the CLI interface with a search query:
```bash ```bash
python -m src.main "python programming" python -m src.main "python programming"
@ -65,51 +63,6 @@ Or run without arguments to enter interactive mode:
python -m src.main python -m src.main
``` ```
### Running the MCP Server
Start the MCP server:
```bash
python src/mcp_server.py
```
The server will be available at `http://localhost:5000`
### MCP Server Endpoints
#### Search Posts
```
GET /search?q={query}&limit={limit}&after={after}
```
#### Get Post Details
```
GET /posts/{post_id}
```
#### Get Post Comments
```
GET /posts/{post_id}/comments?limit={limit}
```
#### Get AI Summary
```
GET /posts/{post_id}/summary
```
#### Get Trending Posts
```
GET /trending?limit={limit}&after={after}
```
#### Health Check
```
GET /health
```
#### OpenAPI Specification
```
GET /openapi.json
```
## Keyboard Controls ## Keyboard Controls
- `n` - Go to next page of results - `n` - Go to next page of results
@ -124,7 +77,6 @@ The application uses environment variables for configuration:
- `OLLAMA_BASE_URL` - Ollama server address (default: http://192.168.8.223:11434) - `OLLAMA_BASE_URL` - Ollama server address (default: http://192.168.8.223:11434)
- `OLLAMA_MODEL` - AI model to use (default: gpt-oss:20b) - `OLLAMA_MODEL` - AI model to use (default: gpt-oss:20b)
- `PORT` - MCP server port (default: 5000)
## Requirements ## Requirements
@ -132,8 +84,6 @@ The application uses environment variables for configuration:
- requests - requests
- rich - rich
- pyyaml - pyyaml
- flask
- flask-cors
- pytest - pytest
- pytest-cov - pytest-cov
@ -149,11 +99,6 @@ Or run basic functionality checks:
python test_cli.py python test_cli.py
``` ```
Run MCP server tests:
```bash
python test_mcp_server.py
```
## Contributing ## Contributing
1. Fork the repository 1. Fork the repository

77
TESTING.md
View File

@ -1,77 +0,0 @@
# MCP Server Testing
The MCP server is now running at `http://home.ms:5000`. Here are the curl commands to test all endpoints:
## 1. Health Check
```bash
curl -X GET http://home.ms:5000/health
```
## 2. Search Posts
```bash
curl -X GET "http://home.ms:5000/search?q=python&limit=5"
```
## 3. Get Post Details
```bash
curl -X GET http://home.ms:5000/posts/xyz123
```
## 4. Get Post Comments
```bash
curl -X GET "http://home.ms:5000/posts/xyz123/comments?limit=10"
```
## 5. Get AI Summary
```bash
curl -X GET http://home.ms:5000/posts/xyz123/summary
```
## 6. Get Trending Posts
```bash
curl -X GET "http://home.ms:5000/trending?limit=5"
```
## 7. OpenAPI Specification
```bash
curl -X GET http://home.ms:5000/openapi.json
```
## Example Response Format
### Search Response:
```json
{
"posts": [
{
"id": "xyz123",
"title": "Example Post Title",
"subreddit": "example",
"created_utc": 1634567890,
"url": "http://reddit.com/r/example/comments/xyz123",
"body": "Post content here...",
"score": 123
}
],
"after": "next_token",
"query": "python",
"limit": 5
}
```
### AI Summary Response:
```json
{
"post_id": "xyz123",
"summary": "AI-generated summary of the post and comments...",
"post_title": "Example Post Title",
"subreddit": "example"
}
```
## Authentication
All AI-related endpoints require the API key in the Authorization header:
```bash
curl -X GET http://home.ms:5000/posts/xyz123/summary \
-H "Authorization: Bearer 111"

39
docker-compose.yml
View File

@ -1,39 +0,0 @@
version: '3.8'
services:
reddit-mcp-server:
build: .
ports:
- "5000:5000"
environment:
- OLLAMA_BASE_URL=http://host.docker.internal:11434
- OLLAMA_MODEL=gpt-oss:20b
- PORT=5000
volumes:
- ./logs:/app/logs
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 5s
restart: unless-stopped
networks:
- reddit-network
# Optional: Ollama service for development (uncomment if needed)
# ollama:
# image: ollama/ollama:latest
# ports:
# - "11434:11434"
# volumes:
# - ollama-data:/root/.ollama
# networks:
# - reddit-network
volumes:
ollama-data:
networks:
reddit-network:
driver: bridge

2
requirements.txt
View File

@ -4,8 +4,6 @@
requests>=2.20.0 requests>=2.20.0
rich>=10.0.0 rich>=10.0.0
pyyaml>=5.4.0 pyyaml>=5.4.0
flask>=2.0.0
flask-cors>=3.0.0
# Testing # Testing
pytest>=6.0.0 pytest>=6.0.0

36
run_mcp_server.sh
View File

@ -1,36 +0,0 @@
#!/bin/bash
# Script to run the MCP server
# This script demonstrates how to start the MCP server with proper environment setup
echo "Starting Reddit MCP Server..."
# Set default environment variables if not already set
export OLLAMA_BASE_URL="${OLLAMA_BASE_URL:-http://host.docker.internal:11434}"
export OLLAMA_MODEL="${OLLAMA_MODEL:-gpt-oss:20b}"
export PORT="${PORT:-5000}"
echo "Environment variables:"
echo " OLLAMA_BASE_URL: $OLLAMA_BASE_URL"
echo " OLLAMA_MODEL: $OLLAMA_MODEL"
echo " PORT: $PORT"
echo ""
echo "Starting MCP server on port $PORT..."
echo "Server will be available at http://localhost:$PORT"
echo ""
echo "Available endpoints:"
echo " GET /search?q={query}&limit={limit}&after={after}"
echo " GET /posts/{post_id}"
echo " GET /posts/{post_id}/comments?limit={limit}"
echo " GET /posts/{post_id}/summary"
echo " GET /trending?limit={limit}&after={after}"
echo " GET /health"
echo " GET /openapi.json"
echo ""
echo "Press Ctrl+C to stop the server"
# Run the MCP server directly with python3
python3 src/mcp_server.py
echo "Server stopped."

38
src/main.py
View File

@ -228,22 +228,11 @@ class RedditCLI:
# Show welcome screen # Show welcome screen
self.show_welcome() self.show_welcome()
# Get search query from user # Get search query from user
while True:
query = input("\nEnter search query: ").strip() query = input("\nEnter search query: ").strip()
if not query: if not query:
self.console.print( self.console.print("[red]No query provided. Exiting.[/red]")
"[red]No query provided. Please try again.[/red]" return
)
continue
try:
self.search(query) self.search(query)
break
except Exception as e:
if "User requested new search" in str(e):
# Continue the loop to get a new search query
continue
else:
raise e
else: else:
# Process the provided query # Process the provided query
self.search(args.query) self.search(args.query)
@ -306,30 +295,7 @@ class RedditCLI:
if not posts: if not posts:
self.console.print("[yellow]No results found[/yellow]") self.console.print("[yellow]No results found[/yellow]")
# Instead of returning, ask user if they want to search again
while True:
choice = (
input("\nWould you like to perform a new search? (y/n): ")
.strip()
.lower()
)
if choice in ["y", "yes"]:
new_query = input("Enter new search query: ").strip()
if new_query:
self.search(new_query)
return return
else:
self.console.print(
"[yellow]No query provided.[/yellow]"
)
continue
elif choice in ["n", "no"]:
# Return to main menu by raising an exception that gets caught
raise Exception("User requested new search")
else:
self.console.print(
"[red]Please enter 'y' for yes or 'n' for no[/red]"
)
self.search_results = posts self.search_results = posts
self.current_page = 0 self.current_page = 0

779
src/mcp_server.py
View File

@ -1,779 +0,0 @@
#!/usr/bin/env python3
"""
MCP Server for Reddit CLI
Exposes Reddit querying capabilities through MCP endpoints with resilience and retry logic.
"""
import asyncio
import json
import logging
import os
import time
from typing import Any, Dict, List, Optional
from urllib.parse import quote
import requests
from flask import Flask, jsonify, request
from flask_cors import CORS
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize Flask app
app = Flask(__name__)
CORS(app)
# Global configuration
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://home.ms:4000")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gpt-oss:20b")
REDDIT_BASE_URL = "https://www.reddit.com"
class RedditClient:
"""Client for interacting with Reddit API with retry logic"""
def __init__(self):
self.base_url = REDDIT_BASE_URL
self.session = requests.Session()
self.session.headers.update({"User-Agent": "RedditCLI/0.1 by User"})
def _make_request_with_retry(self, url: str, params: Dict = None, max_retries: int = 3,
retry_delay: float = 1.0) -> requests.Response:
"""
Make HTTP request with exponential backoff retry logic
Args:
url: Request URL
params: Request parameters
max_retries: Maximum number of retry attempts
retry_delay: Initial delay between retries (seconds)
Returns:
Response object
Raises:
requests.exceptions.RequestException: If all retries fail
"""
for attempt in range(max_retries + 1):
try:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt < max_retries:
logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds...")
time.sleep(retry_delay)
retry_delay *= 2 # Exponential backoff
else:
logger.error(f"All {max_retries + 1} attempts failed for {url}")
raise e
def search_posts(
self, query: str, limit: int = 15, after: Optional[str] = None
) -> Dict[str, Any]:
"""
Search for Reddit posts matching the query with retry logic
Args:
query: Search terms
limit: Number of posts to return (default 15)
after: Pagination token for next page
Returns:
Dictionary containing search results and pagination info
"""
params = {"q": query, "limit": limit, "sort": "hot", "type": "link"}
if after:
params["after"] = after
try:
response = self._make_request_with_retry(
f"{self.base_url}/search.json", params=params
)
return response.json()
except requests.exceptions.RequestException as e:
raise Exception(f"Failed to fetch posts: {str(e)}")
def get_post_details(self, post_id: str) -> Dict[str, Any]:
"""
Get detailed information about a specific post with retry logic
Args:
post_id: Reddit post ID
Returns:
Dictionary with post details
"""
try:
response = self._make_request_with_retry(
f"{self.base_url}/by_id/t3_{post_id}.json"
)
data = response.json()
# Extract post from the response structure
if isinstance(data, list) and len(data) > 0:
return data[0].get("data", {})
elif isinstance(data, dict):
return data.get("data", {})
return {}
except requests.exceptions.RequestException as e:
raise Exception(f"Failed to fetch post details: {str(e)}")
def get_post_comments(self, post_id: str, limit: int = 100) -> List[Dict[str, Any]]:
"""
Get comments for a specific post with retry logic
Args:
post_id: Reddit post ID
limit: Maximum number of comments to fetch
Returns:
List of comment dictionaries
"""
try:
response = self._make_request_with_retry(
f"{self.base_url}/comments/{post_id}.json",
params={"limit": limit}
)
data = response.json()
# Extract comments from the nested structure
comments = []
if isinstance(data, list) and len(data) > 1:
comment_data = data[1].get("data", {}).get("children", [])
for child in comment_data:
comment = child.get("data", {})
# Flatten the comment structure to include author and body
comments.append(
{
"author": comment.get("author", "unknown"),
"body": comment.get("body", ""),
"score": comment.get("score", 0),
"created_utc": comment.get("created_utc", 0),
}
)
return comments
except requests.exceptions.RequestException as e:
raise Exception(f"Failed to fetch comments: {str(e)}")
def format_timestamp(self, timestamp: int) -> str:
"""
Format Unix timestamp into readable date string
Args:
timestamp: Unix timestamp
Returns:
Formatted date string
"""
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
class AIClient:
"""Client for interacting with OpenAPI AI API with retry logic"""
def __init__(self):
self.base_url = OLLAMA_BASE_URL
self.model = OLLAMA_MODEL
self.session = requests.Session()
# Set the API key for the new endpoint
self.api_key = "111" # As specified in the feedback
def _make_request_with_retry(self, url: str, json_data: Dict = None, max_retries: int = 3,
retry_delay: float = 1.0) -> requests.Response:
"""
Make HTTP request with exponential backoff retry logic
Args:
url: Request URL
json_data: JSON data to send
max_retries: Maximum number of retry attempts
retry_delay: Initial delay between retries (seconds)
Returns:
Response object
Raises:
requests.exceptions.RequestException: If all retries fail
"""
for attempt in range(max_retries + 1):
try:
# Add API key to Authorization header as Bearer token
headers = {"Authorization": f"Bearer {self.api_key}"}
response = self.session.post(url, json=json_data, headers=headers, timeout=30)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt < max_retries:
logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {retry_delay} seconds...")
time.sleep(retry_delay)
retry_delay *= 2 # Exponential backoff
else:
logger.error(f"All {max_retries + 1} attempts failed for {url}")
raise e
def generate_summary(self, post_body: str, comments: List[str]) -> str:
"""
Generate AI summary of a post with comments and retry logic
Args:
post_body: The main body text of the post
comments: List of comment strings
Returns:
Generated summary from AI
"""
# Select 100 random comments (or all if less than 100)
selected_comments = comments[:100]
# Format prompt for the AI model
prompt = self._create_prompt(post_body, selected_comments)
try:
response = self._make_request_with_retry(
f"{self.base_url}/api/generate",
json={"model": self.model, "prompt": prompt, "stream": False}
)
data = response.json()
return data.get("response", "").strip()
except requests.exceptions.RequestException as e:
raise Exception(f"Failed to generate AI summary: {str(e)}")
def _create_prompt(self, post_body: str, comments: List[str]) -> str:
"""
Create a formatted prompt for the AI with post and comments
Args:
post_body: The main body text of the post
comments: List of comment strings
Returns:
Formatted prompt string
"""
# Join comments into a single string with proper formatting
comments_text = "\n".join(
[f"Comment {i + 1}: {comment}" for i, comment in enumerate(comments)]
)
if not comments_text:
comments_text = "No comments available."
prompt = f"""
Summarize the following Reddit post and its comments in 2-3 sentences.
Post:
{post_body}
Comments:
{comments_text}
Summary:
"""
return prompt
# Initialize clients
reddit_client = RedditClient()
ai_client = AIClient()
@app.route('/search', methods=['GET'])
def search_posts():
"""Search for Reddit posts"""
try:
query = request.args.get('q', '')
limit = int(request.args.get('limit', 15))
after = request.args.get('after', None)
if not query:
return jsonify({"error": "Query parameter 'q' is required"}), 400
data = reddit_client.search_posts(query, limit, after)
# Extract posts from response
posts = []
# Check if we have data in the expected response format
if "data" in data and "children" in data["data"]:
for child in data["data"]["children"]:
post_data = child.get("data", {})
if post_data:
posts.append(
{
"id": post_data.get("id"),
"title": post_data.get("title", "No title"),
"subreddit": post_data.get("subreddit", "unknown"),
"created_utc": post_data.get("created_utc", 0),
"url": post_data.get("url", ""),
"body": post_data.get(
"selftext", post_data.get("body", "")
),
"score": post_data.get("score", 0),
}
)
# Get the after token for pagination
after_token = data["data"].get("after")
else:
after_token = None
response_data = {
"posts": posts,
"after": after_token,
"query": query,
"limit": limit
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in search_posts: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/posts/<post_id>', methods=['GET'])
def get_post_details(post_id):
"""Get detailed information about a specific post"""
try:
post_data = reddit_client.get_post_details(post_id)
if not post_data:
return jsonify({"error": "Post not found"}), 404
# Format the response
response_data = {
"id": post_data.get("id"),
"title": post_data.get("title", "No title"),
"subreddit": post_data.get("subreddit", "unknown"),
"created_utc": post_data.get("created_utc", 0),
"url": post_data.get("url", ""),
"body": post_data.get("selftext", post_data.get("body", "")),
"score": post_data.get("score", 0),
"author": post_data.get("author", "unknown"),
"permalink": post_data.get("permalink", ""),
"num_comments": post_data.get("num_comments", 0)
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in get_post_details: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/posts/<post_id>/comments', methods=['GET'])
def get_post_comments(post_id):
"""Get comments for a specific post"""
try:
limit = int(request.args.get('limit', 100))
comments = reddit_client.get_post_comments(post_id, limit)
response_data = {
"post_id": post_id,
"comments": comments,
"count": len(comments)
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in get_post_comments: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/posts/<post_id>/summary', methods=['GET'])
def get_post_summary(post_id):
"""Get AI-generated summary for a post"""
try:
# First get the post details
post_data = reddit_client.get_post_details(post_id)
if not post_data:
return jsonify({"error": "Post not found"}), 404
# Get comments
comments = reddit_client.get_post_comments(post_id, limit=100)
# Generate summary
post_body = post_data.get("selftext", post_data.get("body", ""))
comment_bodies = [comment.get("body", "") for comment in comments]
summary = ai_client.generate_summary(post_body, comment_bodies)
response_data = {
"post_id": post_id,
"summary": summary,
"post_title": post_data.get("title", "No title"),
"subreddit": post_data.get("subreddit", "unknown")
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in get_post_summary: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/trending', methods=['GET'])
def get_trending():
"""Get trending posts"""
try:
limit = int(request.args.get('limit', 15))
after = request.args.get('after', None)
# Use the same search parameters but with different sort
params = {"q": "all", "limit": limit, "sort": "top", "type": "link"}
if after:
params["after"] = after
data = reddit_client.search_posts("all", limit, after)
# Extract posts from response
posts = []
# Check if we have data in the expected response format
if "data" in data and "children" in data["data"]:
for child in data["data"]["children"]:
post_data = child.get("data", {})
if post_data:
posts.append(
{
"id": post_data.get("id"),
"title": post_data.get("title", "No title"),
"subreddit": post_data.get("subreddit", "unknown"),
"created_utc": post_data.get("created_utc", 0),
"url": post_data.get("url", ""),
"body": post_data.get(
"selftext", post_data.get("body", "")
),
"score": post_data.get("score", 0),
}
)
# Get the after token for pagination
after_token = data["data"].get("after")
else:
after_token = None
response_data = {
"posts": posts,
"after": after_token,
"limit": limit
}
return jsonify(response_data)
except Exception as e:
logger.error(f"Error in get_trending: {e}")
return jsonify({"error": str(e)}), 500
@app.route('/health', methods=['GET'])
def health_check():
"""Health check endpoint"""
return jsonify({"status": "healthy", "service": "reddit-mcp-server"})
@app.route('/openapi.json', methods=['GET'])
def openapi_spec():
"""Serve OpenAPI specification"""
spec = {
"openapi": "3.0.0",
"info": {
"title": "Reddit MCP API",
"version": "1.0.0",
"description": "API for querying Reddit posts with MCP server capabilities"
},
"servers": [
{
"url": "http://localhost:5000",
"description": "Local development server"
}
],
"paths": {
"/search": {
"get": {
"summary": "Search Reddit posts",
"description": "Search for Reddit posts by query term",
"parameters": [
{
"name": "q",
"in": "query",
"required": True,
"schema": {
"type": "string"
},
"description": "Search query"
},
{
"name": "limit",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"default": 15
},
"description": "Number of posts to return"
},
{
"name": "after",
"in": "query",
"required": False,
"schema": {
"type": "string"
},
"description": "Pagination token"
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"posts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"title": {"type": "string"},
"subreddit": {"type": "string"},
"created_utc": {"type": "integer"},
"url": {"type": "string"},
"body": {"type": "string"},
"score": {"type": "integer"}
}
}
},
"after": {"type": "string"},
"query": {"type": "string"},
"limit": {"type": "integer"}
}
}
}
}
}
}
}
},
"/posts/{post_id}": {
"get": {
"summary": "Get post details",
"description": "Get detailed information about a specific post",
"parameters": [
{
"name": "post_id",
"in": "path",
"required": True,
"schema": {
"type": "string"
},
"description": "Reddit post ID"
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"id": {"type": "string"},
"title": {"type": "string"},
"subreddit": {"type": "string"},
"created_utc": {"type": "integer"},
"url": {"type": "string"},
"body": {"type": "string"},
"score": {"type": "integer"},
"author": {"type": "string"},
"permalink": {"type": "string"},
"num_comments": {"type": "integer"}
}
}
}
}
}
}
}
},
"/posts/{post_id}/comments": {
"get": {
"summary": "Get post comments",
"description": "Get comments for a specific post",
"parameters": [
{
"name": "post_id",
"in": "path",
"required": True,
"schema": {
"type": "string"
},
"description": "Reddit post ID"
},
{
"name": "limit",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"default": 100
},
"description": "Maximum number of comments to return"
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"post_id": {"type": "string"},
"comments": {
"type": "array",
"items": {
"type": "object",
"properties": {
"author": {"type": "string"},
"body": {"type": "string"},
"score": {"type": "integer"},
"created_utc": {"type": "integer"}
}
}
},
"count": {"type": "integer"}
}
}
}
}
}
}
}
},
"/posts/{post_id}/summary": {
"get": {
"summary": "Get AI summary",
"description": "Get AI-generated summary for a post",
"parameters": [
{
"name": "post_id",
"in": "path",
"required": True,
"schema": {
"type": "string"
},
"description": "Reddit post ID"
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"post_id": {"type": "string"},
"summary": {"type": "string"},
"post_title": {"type": "string"},
"subreddit": {"type": "string"}
}
}
}
}
}
}
}
},
"/trending": {
"get": {
"summary": "Get trending posts",
"description": "Get trending Reddit posts",
"parameters": [
{
"name": "limit",
"in": "query",
"required": False,
"schema": {
"type": "integer",
"default": 15
},
"description": "Number of posts to return"
},
{
"name": "after",
"in": "query",
"required": False,
"schema": {
"type": "string"
},
"description": "Pagination token"
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"posts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"title": {"type": "string"},
"subreddit": {"type": "string"},
"created_utc": {"type": "integer"},
"url": {"type": "string"},
"body": {"type": "string"},
"score": {"type": "integer"}
}
}
},
"after": {"type": "string"},
"limit": {"type": "integer"}
}
}
}
}
}
}
}
},
"/health": {
"get": {
"summary": "Health check",
"description": "Check if the service is running",
"responses": {
"200": {
"description": "Service is healthy"
}
}
}
}
}
}
return jsonify(spec)
def main():
"""Main function to start the MCP server"""
port = int(os.environ.get('PORT', 5000))
logger.info(f"Starting Reddit MCP Server on port {port}")
app.run(host='0.0.0.0', port=port, debug=False)
if __name__ == "__main__":
main()

124
test_mcp_server.py
View File

@ -1,124 +0,0 @@
#!/usr/bin/env python3
"""
Test script for MCP server functionality
"""
import os
import sys
import time
import requests
# Add the src directory to the path so we can import the modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
from src.mcp_server import RedditClient, AIClient
def test_reddit_client():
"""Test Reddit client with retry logic"""
print("Testing Reddit Client...")
client = RedditClient()
try:
# Test search with retry logic
print("Searching for 'python programming'...")
data = client.search_posts("python programming", limit=5)
print(f"Found {len(data.get('data', {}).get('children', []))} posts")
print("✓ Reddit client search test passed")
# Test post details with retry logic
if data.get('data', {}).get('children'):
post_id = data['data']['children'][0]['data']['id']
print(f"Getting details for post {post_id}...")
post_data = client.get_post_details(post_id)
print(f"Post title: {post_data.get('title', 'No title')}")
print("✓ Reddit client post details test passed")
# Test comments with retry logic
print(f"Getting comments for post {post_id}...")
comments = client.get_post_comments(post_id, limit=5)
print(f"Found {len(comments)} comments")
print("✓ Reddit client comments test passed")
except Exception as e:
print(f"✗ Reddit client test failed: {e}")
return False
return True
def test_ai_client():
"""Test AI client with retry logic"""
print("\nTesting AI Client...")
client = AIClient()
try:
# Test summary generation with retry logic
# We'll use a simple test case
post_body = "This is a test post body for testing the AI summary functionality."
comments = ["This is a test comment.", "Another test comment."]
print("Generating AI summary...")
summary = client.generate_summary(post_body, comments)
print(f"Summary: {summary[:100]}...")
print("✓ AI client test passed")
except Exception as e:
print(f"✗ AI client test failed: {e}")
return False
return True
def test_server_endpoints():
"""Test server endpoints"""
print("\nTesting Server Endpoints...")
try:
# Test health check
response = requests.get('http://localhost:5000/health')
if response.status_code == 200:
print("✓ Health check endpoint works")
else:
print(f"✗ Health check failed: {response.status_code}")
return False
# Test OpenAPI spec
response = requests.get('http://localhost:5000/openapi.json')
if response.status_code == 200:
print("✓ OpenAPI spec endpoint works")
else:
print(f"✗ OpenAPI spec failed: {response.status_code}")
return False
print("✓ Server endpoints test passed")
except Exception as e:
print(f"✗ Server endpoints test failed: {e}")
return False
return True
def main():
"""Main test function"""
print("Running MCP Server Tests...")
print("=" * 50)
success = True
# Test the core components
success &= test_reddit_client()
success &= test_ai_client()
# Note: We can't easily test the full server endpoints without actually running it,
# but we can test the components that would be used by the server
print("\n" + "=" * 50)
if success:
print("✓ All tests passed!")
return 0
else:
print("✗ Some tests failed!")
return 1
if __name__ == "__main__":
sys.exit(main())