server.py•6.92 kB
#!/usr/bin/env python3
"""
Hacker News MCP Server
A FastMCP server that provides access to Hacker News data.
"""
import asyncio
import json
import os
from typing import Any, Dict, List, Optional
import httpx
from fastmcp import FastMCP
# Create the FastMCP server
mcp = FastMCP("Hacker News MCP Server")
# Hacker News API base URL
HN_API_BASE = "https://hacker-news.firebaseio.com/v0"
@mcp.tool()
def get_top_stories(limit: int = 10) -> str:
"""Get top stories from Hacker News.
Args:
limit: Number of stories to fetch (default: 10, max: 30)
Returns:
JSON string with top stories data
"""
try:
limit = min(max(limit, 1), 30) # Clamp between 1 and 30
# Fetch top story IDs
response = httpx.get(f"{HN_API_BASE}/topstories.json", timeout=10.0)
response.raise_for_status()
story_ids = response.json()[:limit]
# Fetch story details
stories = []
for story_id in story_ids:
story_response = httpx.get(f"{HN_API_BASE}/item/{story_id}.json", timeout=5.0)
story_response.raise_for_status()
story = story_response.json()
if story and story.get('type') == 'story':
stories.append({
'id': story.get('id'),
'title': story.get('title', ''),
'url': story.get('url', f"https://news.ycombinator.com/item?id={story_id}"),
'score': story.get('score', 0),
'author': story.get('by', ''),
'comments': story.get('descendants', 0),
'time': story.get('time', 0),
'time_iso': story.get('time') and str(story.get('time')) or None
})
return json.dumps(stories, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def get_new_stories(limit: int = 10) -> str:
"""Get newest stories from Hacker News.
Args:
limit: Number of stories to fetch (default: 10, max: 30)
Returns:
JSON string with newest stories data
"""
try:
limit = min(max(limit, 1), 30) # Clamp between 1 and 30
# Fetch new story IDs
response = httpx.get(f"{HN_API_BASE}/newstories.json", timeout=10.0)
response.raise_for_status()
story_ids = response.json()[:limit]
# Fetch story details
stories = []
for story_id in story_ids:
story_response = httpx.get(f"{HN_API_BASE}/item/{story_id}.json", timeout=5.0)
story_response.raise_for_status()
story = story_response.json()
if story and story.get('type') == 'story':
stories.append({
'id': story.get('id'),
'title': story.get('title', ''),
'url': story.get('url', f"https://news.ycombinator.com/item?id={story_id}"),
'score': story.get('score', 0),
'author': story.get('by', ''),
'time': story.get('time', 0),
'time_iso': story.get('time') and str(story.get('time')) or None
})
return json.dumps(stories, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def get_story(story_id: int) -> str:
"""Get details of a specific Hacker News story by ID.
Args:
story_id: The ID of the story to fetch
Returns:
JSON string with story details
"""
try:
response = httpx.get(f"{HN_API_BASE}/item/{story_id}.json", timeout=10.0)
response.raise_for_status()
story = response.json()
if not story:
return json.dumps({"error": f"Story {story_id} not found"}, indent=2)
if story.get('type') != 'story':
return json.dumps({"error": f"Item {story_id} is not a story"}, indent=2)
formatted_story = {
'id': story.get('id'),
'title': story.get('title', ''),
'url': story.get('url', f"https://news.ycombinator.com/item?id={story_id}"),
'score': story.get('score', 0),
'author': story.get('by', ''),
'comments': story.get('descendants', 0),
'time': story.get('time', 0),
'text': story.get('text', ''),
'kids': story.get('kids', []) # Comment IDs
}
return json.dumps(formatted_story, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def search_stories(query: str, limit: int = 10) -> str:
"""Search for stories on Hacker News (using Algolia API).
Args:
query: Search query string
limit: Number of results to return (default: 10, max: 20)
Returns:
JSON string with search results
"""
try:
limit = min(max(limit, 1), 20) # Clamp between 1 and 20
# Use Algolia search API
search_url = "https://hn.algolia.com/api/v1/search"
params = {
"query": query,
"tags": "story",
"hitsPerPage": limit
}
response = httpx.get(search_url, params=params, timeout=10.0)
response.raise_for_status()
data = response.json()
stories = []
for hit in data.get('hits', []):
stories.append({
'id': hit.get('objectID'),
'title': hit.get('title', ''),
'url': hit.get('url', f"https://news.ycombinator.com/item?id={hit.get('objectID')}"),
'score': hit.get('points', 0),
'author': hit.get('author', ''),
'comments': hit.get('num_comments', 0),
'time': hit.get('created_at_i', 0),
'time_iso': hit.get('created_at', '')
})
return json.dumps(stories, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
if __name__ == "__main__":
port = int(os.environ.get("PORT", 8000))
mcp.run(
transport="http",
host="0.0.0.0",
port=port,
stateless_http=True
)