WSB Analyst MCP Server

Instructions

Implementation Reference

• server.py:400-462 (handler)

  The primary handler function decorated with @mcp.tool() for automatic registration. It fetches top trending tickers from the ApeWisdom API based on upvotes and mentions, collects potential tickers, and validates them against NASDAQ listed symbols using the filter_valid_tickers helper. Returns a list of unique valid tickers.
  @mcp.tool() async def get_top_trending_tickers(num_stocks: int = 20, filter: str = "wallstreetbets") -> list[str]: """ Fetch top trending stock tickers from ApeWisdom, filtered by valid NASDAQ symbols. Args: num_stocks: Number of top stocks to consider based on upvotes and mentions. Defaults to 20. filter: ApeWisdom filter category (e.g., 'wallstreetbets', 'all'). Defaults to 'wallstreetbets'. Returns: A list of valid, trending tickers. """ api_url = f"https://apewisdom.io/api/v1.0/filter/{filter}" logger.info(f"Fetching trending tickers from {api_url}") async with httpx.AsyncClient() as client: try: response = await client.get(api_url) response.raise_for_status() # Raise exception for bad status codes data = response.json() logger.info(f"Successfully fetched data from ApeWisdom for filter '{filter}'") except httpx.RequestError as e: logger.error(f"Error fetching from ApeWisdom: {e}") return [] except httpx.HTTPStatusError as e: logger.error(f"ApeWisdom API returned error {e.response.status_code}: {e}") return [] except Exception as e: # Catch potential JSON decoding errors etc. logger.error(f"An unexpected error occurred fetching/parsing ApeWisdom data: {e}") return [] if "results" not in data or not isinstance(data["results"], list): logger.error("Invalid data structure received from ApeWisdom API.") return [] potential_tickers = set() metrics = ["upvotes", "mentions"] for metric in metrics: # Sort by the current metric and take top N # Add try-except for robustness against missing keys or non-numeric values try: sorted_stocks = sorted( [stock for stock in data["results"] if isinstance(stock.get(metric), (int, float))], key=lambda item: item.get(metric, 0), reverse=True ) for stock in sorted_stocks[:num_stocks]: if "ticker" in stock: potential_tickers.add(stock["ticker"]) except Exception as e: logger.warning(f"Could not process metric '{metric}' due to error: {e}") if not potential_tickers: logger.info("No potential tickers found after processing ApeWisdom results.") return [] logger.info(f"Found {len(potential_tickers)} potential tickers, filtering...") # Filter valid tickers valid_tickers = await filter_valid_tickers(list(potential_tickers)) logger.info(f"Filtered down to {len(valid_tickers)} valid trending tickers.") return valid_tickers
• server.py:81-115 (helper)

  Helper function used by get_top_trending_tickers to validate potential tickers against official NASDAQ listed symbols by fetching and parsing nasdaqlisted.txt and otherlisted.txt files, returning only valid uppercase tickers.
  async def filter_valid_tickers(tickers: list[str]) -> list[str]: """Filter out invalid tickers using NASDAQ's symbol list.""" async with httpx.AsyncClient() as client: try: nasdaq_response = await client.get("https://www.nasdaqtrader.com/dynamic/SymDir/nasdaqlisted.txt") other_response = await client.get("https://www.nasdaqtrader.com/dynamic/SymDir/otherlisted.txt") nasdaq_response.raise_for_status() other_response.raise_for_status() except httpx.RequestError as e: logger.error(f"Error fetching NASDAQ ticker lists: {e}") return list(set(ticker.upper() for ticker in tickers)) except httpx.HTTPStatusError as e: logger.error(f"Error fetching NASDAQ ticker lists (status {e.response.status_code}): {e}") return list(set(ticker.upper() for ticker in tickers)) valid_symbols = set() try: for line in nasdaq_response.text.splitlines()[1:-1]: # Skip header and footer parts = line.split('|') if len(parts) > 0: symbol = parts[0] valid_symbols.add(symbol) for line in other_response.text.splitlines()[1:-1]: # Skip header and footer parts = line.split('|') if len(parts) > 0: symbol = parts[0] valid_symbols.add(symbol) except Exception as e: logger.error(f"Error parsing NASDAQ ticker lists: {e}") return list(set(ticker.upper() for ticker in tickers)) unique_tickers = set(ticker.upper() for ticker in tickers) return [ticker for ticker in unique_tickers if ticker in valid_symbols]