MCP Metaculus Server

from typing import Annotated from fastmcp import FastMCP from datetime import datetime, timezone import httpx import os import re import json from dotenv import load_dotenv mcp = FastMCP( name="metaculus", instructions="Provides historical prediction data from Metaculus forecasting platform. Use this to retrieve question details, community predictions, and user comments for specific Metaculus questions." ) @mcp.tool(exclude_args=["cutoff_date"]) async def get_metaculus_question_info( question_url: Annotated[str, "The full Metaculus question URL (e.g., 'https://www.metaculus.com/questions/39771/')"], cutoff_date: Annotated[str, "ISO format date (YYYY-MM-DD) - only return predictions made before this date"] = datetime.now().strftime("%Y-%m-%d") ) -> str: """Get Metaculus question information with historical predictions filtered by cutoff date. Args: question_url: Full Metaculus question URL cutoff_date: Only return predictions before this date Returns: Formatted string with question details, predictions, and comments """ try: load_dotenv() # Extract post ID from URL post_id_match = re.search(r'/questions/(\d+)', question_url) if not post_id_match: return f"Error: Could not extract post ID from URL: {question_url}. Expected format: https://www.metaculus.com/questions/[ID]" post_id = post_id_match.group(1) # Fetch question from Metaculus API api_url = f"https://www.metaculus.com/api/posts/{post_id}/" async with httpx.AsyncClient() as client: response = await client.get(api_url, timeout=30.0) if not response.is_success: return f"Error fetching question from Metaculus API: {response.status_code} {response.text[:200]}" data = response.json() question = data.get("question") if not question: return f"Error: No question found in post {post_id}" # Extract metadata title = question.get("title", "N/A") question_id = question.get("id", "N/A") question_type = question.get("type", "N/A") background_info = question.get("background_info", "") resolution_criteria = question.get("resolution_criteria", "") fine_print = question.get("fine_print", "") # Extract and filter prediction history aggregations = question.get("aggregations", {}) recency_weighted = aggregations.get("recency_weighted", {}) history = recency_weighted.get("history", []) if not history: return f"Error: No prediction history found for question {question_id}" # Filter by cutoff date cutoff_dt = datetime.fromisoformat(cutoff_date).replace(tzinfo=timezone.utc) cutoff_timestamp = cutoff_dt.timestamp() filtered_history = [ entry for entry in history if entry.get("end_time") is not None and entry.get("end_time") <= cutoff_timestamp ] if not filtered_history: return f"No prediction history available before cutoff date {cutoff_date} for question {question_id}" # Fetch comments using Firecrawl (optional) comments_text = "N/A" if firecrawl_key := os.environ.get("FIRECRAWL_API_KEY"): try: firecrawl_url = "https://api.firecrawl.dev/v2/scrape" firecrawl_payload = { "url": question_url if question_url.endswith('/') else f"{question_url}/", "onlyMainContent": False, "maxAge": 172800000, "parsers": ["pdf"], "formats": [{ "type": "json", "schema": { "type": "object", "required": ["comments"], "properties": { "comments": { "type": "array", "items": { "type": "object", "required": ["content"], "properties": { "content": {"type": "string"}, "time_posted": {"type": "string"}, "upvotes": {"type": "number"}, "downvotes": {"type": "number"}, "changed_my_mind_votes": {"type": "number"}, "author": {"type": "string"} } } } } } }] } async with httpx.AsyncClient() as client: firecrawl_response = await client.post( firecrawl_url, json=firecrawl_payload, headers={ "Authorization": f"Bearer {firecrawl_key}", "Content-Type": "application/json" }, timeout=60.0 ) if firecrawl_response.is_success: firecrawl_data = firecrawl_response.json() if "data" in firecrawl_data and "json" in firecrawl_data["data"]: raw_comments = firecrawl_data["data"]["json"].get("comments", "") # Filter comments by cutoff_date if time_posted is available if isinstance(raw_comments, list): filtered_comments = [] for comment in raw_comments: if "time_posted" in comment: try: comment_time = datetime.fromisoformat(comment["time_posted"].replace('Z', '+00:00')) if comment_time <= cutoff_dt: filtered_comments.append(comment) except: # If we can't parse date, include the comment to be safe filtered_comments.append(comment) else: # No timestamp, include it filtered_comments.append(comment) comments_text = json.dumps(filtered_comments, indent=2) if filtered_comments else "No comments before cutoff date" else: comments_text = str(raw_comments) except Exception as e: comments_text = f"Error fetching comments: {str(e)}" # Format output parts = [ "--- Metaculus Question ---", f"Title: {title}", f"Question ID: {question_id}", f"Post ID: {post_id}", f"Type: {question_type}", f"URL: {question_url}", "", "Background:", background_info, "", "Resolution Criteria:", resolution_criteria, ] if fine_print and fine_print != "N/A": parts.extend(["", "Fine Print:", fine_print]) # Add comments if comments_text and comments_text != "N/A": parts.extend(["", "User Comments:", comments_text]) parts.extend([ "", f"Community Prediction History:", "Recent predictions:" ]) # Show last 15 entries recent_entries = filtered_history[-15:] for entry in recent_entries: start_dt = datetime.fromtimestamp(entry["start_time"], tz=timezone.utc) end_dt = datetime.fromtimestamp(entry["end_time"], tz=timezone.utc) forecaster_count = entry.get("forecaster_count", 0) # Format prediction by question type if question_type == "binary": centers = entry.get("centers", []) prediction = f"{centers[0]*100:.1f}%" if centers and centers[0] is not None else "N/A" elif question_type in ["numeric", "date"]: lower = entry.get("interval_lower_bounds") upper = entry.get("interval_upper_bounds") centers = entry.get("centers") if centers and centers[0] is not None: prediction = f"median: {centers[0]:.2f}" if lower and upper and lower[0] is not None and upper[0] is not None: prediction += f" (range: {lower[0]:.2f} - {upper[0]:.2f})" else: prediction = "N/A" elif question_type == "multiple_choice": centers = entry.get("centers", []) prediction = f"options: {[f'{c*100:.1f}%' if c is not None else 'N/A' for c in centers]}" if centers else "N/A" else: prediction = str(entry.get("centers", "N/A")) time_range = f"{start_dt.strftime('%Y-%m-%d %H:%M')} to {end_dt.strftime('%Y-%m-%d %H:%M')}" parts.append(f" {time_range}: {prediction} ({forecaster_count} forecasters)") # Mark most recent if recent_entries: parts[-1] += " [MOST RECENT]" parts.append(f"\nTotal historical entries: {len(filtered_history)}") return "\n".join(parts) except ValueError as e: return f"Error parsing date '{cutoff_date}'. Please use ISO format (YYYY-MM-DD): {str(e)}" except Exception as e: return f"Error retrieving Metaculus question history: {str(e)}"