Sleeper API MCP

#!/usr/bin/env python3 import json import logging import os import sys import uuid from typing import Dict, List, Any, Optional, Union import requests # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stderr)] ) logger = logging.getLogger("sleeper-mcp") # Base URL for Sleeper API SLEEPER_API_BASE_URL = "https://api.sleeper.app/v1" class SleeperMCP: def __init__(self): self.tools = { "getUserInfo": self.get_user_info, "getUserLeagues": self.get_user_leagues, "getLeagueInfo": self.get_league_info, "getLeagueRosters": self.get_league_rosters, "getLeagueUsers": self.get_league_users, "getLeagueMatchups": self.get_league_matchups, "getLeagueWinnersBracket": self.get_league_winners_bracket, "getLeagueLosersBracket": self.get_league_losers_bracket, "getLeagueTransactions": self.get_league_transactions, "getLeagueTradedPicks": self.get_league_traded_picks, "getNFLState": self.get_nfl_state, "getUserDrafts": self.get_user_drafts, "getLeagueDrafts": self.get_league_drafts, "getDraftInfo": self.get_draft_info, "getDraftPicks": self.get_draft_picks, "getDraftTradedPicks": self.get_draft_traded_picks, "getAllPlayers": self.get_all_players, "getTrendingPlayers": self.get_trending_players, } def _make_request(self, endpoint: str) -> Any: """ Make a request to the Sleeper API. Args: endpoint: The API endpoint to request Returns: The JSON response from the API Raises: Exception: If the request fails """ url = f"{SLEEPER_API_BASE_URL}{endpoint}" logger.debug(f"Making request to {url}") try: response = requests.get(url) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logger.error(f"Error making request to {url}: {str(e)}") raise Exception(f"Failed to fetch data from Sleeper API: {str(e)}") def get_user_info(self, params: Dict[str, Any]) -> Dict[str, Any]: """ Fetch user information by username or user_id. Args: params: Dictionary with username_or_user_id Returns: User information """ username_or_user_id = params.get("username_or_user_id") return self._make_request(f"/user/{username_or_user_id}") def get_user_leagues(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all leagues for a user for a specified sport and season. Args: params: Dictionary with user_id, sport, and season Returns: List of leagues """ user_id = params.get("user_id") sport = params.get("sport", "nfl") season = params.get("season") return self._make_request(f"/user/{user_id}/leagues/{sport}/{season}") def get_league_info(self, params: Dict[str, Any]) -> Dict[str, Any]: """ Fetch information about a specific league. Args: params: Dictionary with league_id Returns: League information """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}") def get_league_rosters(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all rosters in a league. Args: params: Dictionary with league_id Returns: List of rosters """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/rosters") def get_league_users(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all users in a league. Args: params: Dictionary with league_id Returns: List of users """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/users") def get_league_matchups(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch matchups in a league for a specific week. Args: params: Dictionary with league_id and week Returns: List of matchups """ league_id = params.get("league_id") week = params.get("week") return self._make_request(f"/league/{league_id}/matchups/{week}") def get_league_winners_bracket(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch the playoff winners bracket for a league. Args: params: Dictionary with league_id Returns: List of bracket entries """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/winners_bracket") def get_league_losers_bracket(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch the playoff losers bracket for a league. Args: params: Dictionary with league_id Returns: List of bracket entries """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/losers_bracket") def get_league_transactions(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch transactions in a league for a specific week. Args: params: Dictionary with league_id and week Returns: List of transactions """ league_id = params.get("league_id") week = params.get("week") return self._make_request(f"/league/{league_id}/transactions/{week}") def get_league_traded_picks(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all traded picks in a league. Args: params: Dictionary with league_id Returns: List of traded picks """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/traded_picks") def get_nfl_state(self, params: Dict[str, Any]) -> Dict[str, Any]: """ Fetch the current NFL state. Args: params: Empty dictionary Returns: NFL state information """ return self._make_request(f"/state/nfl") def get_user_drafts(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all drafts for a user for a specific sport and season. Args: params: Dictionary with user_id, sport, and season Returns: List of drafts """ user_id = params.get("user_id") sport = params.get("sport", "nfl") season = params.get("season") return self._make_request(f"/user/{user_id}/drafts/{sport}/{season}") def get_league_drafts(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all drafts for a league. Args: params: Dictionary with league_id Returns: List of drafts """ league_id = params.get("league_id") return self._make_request(f"/league/{league_id}/drafts") def get_draft_info(self, params: Dict[str, Any]) -> Dict[str, Any]: """ Fetch information about a specific draft. Args: params: Dictionary with draft_id Returns: Draft information """ draft_id = params.get("draft_id") return self._make_request(f"/draft/{draft_id}") def get_draft_picks(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all picks in a draft. Args: params: Dictionary with draft_id Returns: List of draft picks """ draft_id = params.get("draft_id") return self._make_request(f"/draft/{draft_id}/picks") def get_draft_traded_picks(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch all traded picks in a draft. Args: params: Dictionary with draft_id Returns: List of traded picks """ draft_id = params.get("draft_id") return self._make_request(f"/draft/{draft_id}/traded_picks") def get_all_players(self, params: Dict[str, Any]) -> Dict[str, Any]: """ Fetch information about all players for a specific sport. Args: params: Dictionary with sport Returns: Dictionary of players """ sport = params.get("sport", "nfl") return self._make_request(f"/players/{sport}") def get_trending_players(self, params: Dict[str, Any]) -> List[Dict[str, Any]]: """ Fetch trending players based on add/drop activity. Args: params: Dictionary with sport, type, lookback_hours, and limit Returns: List of trending players """ sport = params.get("sport", "nfl") trend_type = params.get("type") lookback_hours = params.get("lookback_hours", 24) limit = params.get("limit", 25) url = f"/players/{sport}/trending/{trend_type}" if lookback_hours or limit: url += "?" if lookback_hours: url += f"lookback_hours={lookback_hours}" if limit: url += "&" if limit: url += f"limit={limit}" return self._make_request(url) def handle_message(self, message: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ Handle incoming JSON-RPC message. Args: message: The JSON-RPC message Returns: JSON-RPC response if applicable """ if "jsonrpc" not in message or message["jsonrpc"] != "2.0": return { "jsonrpc": "2.0", "id": message.get("id"), "error": { "code": -32600, "message": "Invalid Request", "data": "Message does not conform to JSON-RPC 2.0 spec" } } method = message.get("method") if not method: return { "jsonrpc": "2.0", "id": message.get("id"), "error": { "code": -32600, "message": "Invalid Request", "data": "Method is required" } } # Check if this is a notification (no id) is_notification = "id" not in message # Special handling for listTools method if method == "listTools": tool_descriptions = [] for tool_name in self.tools: tool_descriptions.append({ "name": tool_name, "description": getattr(self.tools[tool_name], "__doc__", "").strip() }) if is_notification: return None return { "jsonrpc": "2.0", "id": message.get("id"), "result": tool_descriptions } # Handle tool calls if method in self.tools: params = message.get("params", {}) try: result = self.tools[method](params) if is_notification: return None return { "jsonrpc": "2.0", "id": message.get("id"), "result": result } except Exception as e: logger.error(f"Error executing {method}: {str(e)}") if is_notification: return None return { "jsonrpc": "2.0", "id": message.get("id"), "error": { "code": -32000, "message": "Server error", "data": str(e) } } else: if is_notification: return None return { "jsonrpc": "2.0", "id": message.get("id"), "error": { "code": -32601, "message": "Method not found", "data": f"The method {method} is not supported" } } def run_stdio_server(): """Run the server using stdio as the transport.""" sleeper_mcp = SleeperMCP() logger.info("Starting Sleeper MCP stdio server...") while True: try: line = sys.stdin.readline() if not line: break message = json.loads(line) logger.debug(f"Received message: {message}") response = sleeper_mcp.handle_message(message) if response: logger.debug(f"Sending response: {response}") print(json.dumps(response), flush=True) except json.JSONDecodeError as e: logger.error(f"Error decoding JSON: {str(e)}") error_response = { "jsonrpc": "2.0", "id": None, "error": { "code": -32700, "message": "Parse error", "data": str(e) } } print(json.dumps(error_response), flush=True) except Exception as e: logger.error(f"Unexpected error: {str(e)}") error_response = { "jsonrpc": "2.0", "id": None, "error": { "code": -32000, "message": "Server error", "data": str(e) } } print(json.dumps(error_response), flush=True) logger.info("Shutting down Sleeper MCP stdio server...") if __name__ == "__main__": run_stdio_server()