College Football MCP

""" MCP Server for College Football Data This module implements the FastAPI server that exposes MCP endpoints for college football game information, odds, and statistics. """ from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse from pydantic import BaseModel from typing import Optional import os from dotenv import load_dotenv import logging from src.odds_api import ( get_ncaaf_odds, find_game_by_teams, find_next_game_for_team, format_game_odds_response, format_next_game_odds ) from src.cfbd_api import ( search_players, get_player_game_stats, get_player_game_logs, get_team_games, get_team_records, get_team_rankings, format_player_recent_stats, format_team_recent_results, format_team_info ) # Load environment variables load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load API keys from environment ODDS_API_KEY = os.getenv("ODDS_API_KEY") CFB_API_KEY = os.getenv("CFB_API_KEY") # Validate API keys on startup def validate_api_keys(): """Validate that required API keys are present""" missing_keys = [] if not ODDS_API_KEY: missing_keys.append("ODDS_API_KEY") if not CFB_API_KEY: missing_keys.append("CFB_API_KEY") if missing_keys: logger.warning(f"Missing API keys: {', '.join(missing_keys)}. Some functions may not work.") else: logger.info("All API keys loaded successfully") # Validate on module load validate_api_keys() app = FastAPI( title="College Football MCP Server", description="MCP server for college football game data and betting odds", version="0.1.0" ) @app.get("/") async def root(): """Health check endpoint""" return { "status": "running", "service": "cfb-mcp", "version": "0.1.0" } @app.get("/health") async def health(): """Health check endpoint""" return { "status": "healthy", "api_keys_configured": { "odds_api": bool(ODDS_API_KEY), "cfb_api": bool(CFB_API_KEY) } } # Request models for MCP endpoints class GameOddsRequest(BaseModel): """Request model for get_game_odds_and_score""" team1: str team2: Optional[str] = None class PlayerStatsRequest(BaseModel): """Request model for get_recent_player_stats""" player_name: str team: Optional[str] = None class TeamResultsRequest(BaseModel): """Request model for get_team_recent_results""" team: str class TeamInfoRequest(BaseModel): """Request model for get_team_info""" team: str class NextGameOddsRequest(BaseModel): """Request model for get_next_game_odds""" team: str # MCP Function Endpoints @app.post("/mcp/get_game_odds_and_score") async def get_game_odds_and_score(request: GameOddsRequest): """ Get live game scores and betting odds for a specific matchup. This endpoint retrieves real-time odds and scores from The Odds API for NCAA college football games. Args: request: GameOddsRequest with team1 (required) and team2 (optional) Returns: JSON response with game details, scores, and odds """ if not ODDS_API_KEY: raise HTTPException( status_code=503, detail="ODDS_API_KEY is not configured" ) try: # Fetch odds from The Odds API games = get_ncaaf_odds(ODDS_API_KEY) if not games: raise HTTPException( status_code=404, detail="No games found or API request failed" ) # Find the matching game game = find_game_by_teams(games, request.team1, request.team2) if not game: raise HTTPException( status_code=404, detail=f"Game not found for team: {request.team1}" ) # Format the response formatted_response = format_game_odds_response(game) return JSONResponse(content=formatted_response) except HTTPException: raise except Exception as e: logger.error(f"Error in get_game_odds_and_score: {e}") raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" ) @app.get("/api/get_game_odds_and_score") async def get_game_odds_and_score_get( team1: str = Query(..., description="First team name (required)"), team2: Optional[str] = Query(None, description="Second team name (optional, for exact match)") ): """ GET endpoint for get_game_odds_and_score (for easier testing). Same functionality as POST endpoint but accessible via GET with query parameters. """ request = GameOddsRequest(team1=team1, team2=team2) return await get_game_odds_and_score(request) @app.post("/mcp/get_recent_player_stats") async def get_recent_player_stats(request: PlayerStatsRequest): """ Get player statistics from the last 5 games. This endpoint retrieves detailed game-by-game statistics for a player from the CollegeFootballData API. Args: request: PlayerStatsRequest with player_name (required) and team (optional) Returns: JSON response with player stats from last 5 games """ if not CFB_API_KEY: raise HTTPException( status_code=503, detail="CFB_API_KEY is not configured" ) try: # Search for the player players = search_players(CFB_API_KEY, request.player_name, request.team) if not players or len(players) == 0: raise HTTPException( status_code=404, detail=f"Player not found: {request.player_name}" ) # Use the first matching player (could be enhanced to handle multiple matches) player = players[0] player_id = player.get("id") player_name = player.get("name", request.player_name) player_team = player.get("team", request.team) if not player_id: raise HTTPException( status_code=404, detail=f"Player ID not found for: {request.player_name}" ) # Get current year for stats from datetime import datetime current_year = datetime.now().year # Try to get game-by-game stats game_stats = get_player_game_logs(CFB_API_KEY, player_id, current_year) # If game logs not available, try season stats if not game_stats: game_stats = get_player_game_stats(CFB_API_KEY, player_id, current_year) # If still no stats, try previous year if not game_stats: game_stats = get_player_game_logs(CFB_API_KEY, player_id, current_year - 1) if not game_stats: game_stats = get_player_game_stats(CFB_API_KEY, player_id, current_year - 1) if not game_stats: raise HTTPException( status_code=404, detail=f"No game statistics found for player: {request.player_name}" ) # Sort games by date (most recent first) if date information is available # For now, we'll take the first/last entries depending on API response order # Format the response formatted_response = format_player_recent_stats( player_name, player_team, game_stats ) return JSONResponse(content=formatted_response) except HTTPException: raise except Exception as e: logger.error(f"Error in get_recent_player_stats: {e}") import traceback logger.error(traceback.format_exc()) raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" ) @app.get("/api/get_recent_player_stats") async def get_recent_player_stats_get( player_name: str = Query(..., description="Player name (required)"), team: Optional[str] = Query(None, description="Team name (optional, for disambiguation)") ): """ GET endpoint for get_recent_player_stats (for easier testing). Same functionality as POST endpoint but accessible via GET with query parameters. """ request = PlayerStatsRequest(player_name=player_name, team=team) return await get_recent_player_stats(request) @app.post("/mcp/get_team_recent_results") async def get_team_recent_results(request: TeamResultsRequest): """ Get a team's last 5 game results. This endpoint retrieves recent game results for a team from the CollegeFootballData API, including opponents, dates, scores, and outcomes. Args: request: TeamResultsRequest with team name (required) Returns: JSON response with team's last 5 games including scores and win/loss """ if not CFB_API_KEY: raise HTTPException( status_code=503, detail="CFB_API_KEY is not configured" ) try: # Get current year for stats from datetime import datetime from src.team_normalizer import normalize_team_name current_year = datetime.now().year team_name = request.team original_team = request.team # Start with previous year (most recent completed season) as college football season # typically ends in January, so December 2025 means 2024 season is most recent year_to_try = current_year - 1 normalized_name = normalize_team_name(original_team) # Try combinations: [original, normalized] x [previous_year, current_year] name_variations = [original_team] if normalized_name and normalized_name != original_team.lower(): name_variations.append(normalized_name) games = None # Try current year first (most recent), then previous year # This ensures 2025 games are prioritized when they exist for year in [current_year, current_year - 1]: for name_var in name_variations: logger.info(f"Trying team '{name_var}' for year {year}") games = get_team_games(CFB_API_KEY, name_var, year, "both") if games and isinstance(games, list) and len(games) > 0: team_name = name_var logger.info(f"Found {len(games)} games for '{name_var}' in {year}") break if games and isinstance(games, list) and len(games) > 0: break if not games or (isinstance(games, list) and len(games) == 0): raise HTTPException( status_code=404, detail=f"No games found for team: {request.team} (tried variations: {team_name})" ) # Format the response (use the team name that worked) formatted_response = format_team_recent_results(team_name, games) return JSONResponse(content=formatted_response) except HTTPException: raise except Exception as e: logger.error(f"Error in get_team_recent_results: {e}") import traceback logger.error(traceback.format_exc()) raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" ) @app.get("/api/get_team_recent_results") async def get_team_recent_results_get( team: str = Query(..., description="Team name (required)") ): """ GET endpoint for get_team_recent_results (for easier testing). Same functionality as POST endpoint but accessible via GET with query parameters. """ request = TeamResultsRequest(team=team) return await get_team_recent_results(request) @app.post("/mcp/get_team_info") async def get_team_info(request: TeamInfoRequest): """ Get a team's current season overview. This endpoint retrieves team season information from the CollegeFootballData API, including overall record, conference record, and rankings. Args: request: TeamInfoRequest with team name (required) Returns: JSON response with team season overview including record and rankings """ if not CFB_API_KEY: raise HTTPException( status_code=503, detail="CFB_API_KEY is not configured" ) try: # Get current year from datetime import datetime from src.team_normalizer import normalize_team_name current_year = datetime.now().year team_name = request.team # Try original team name first record = get_team_records(CFB_API_KEY, team_name, current_year) # If no record, try normalized team name if not record: normalized_name = normalize_team_name(team_name) if normalized_name and normalized_name != team_name.lower(): logger.info(f"No record found for '{team_name}', trying normalized name '{normalized_name}'") record = get_team_records(CFB_API_KEY, normalized_name, current_year) if record: team_name = normalized_name # If no record for current year, try previous year if not record: record = get_team_records(CFB_API_KEY, team_name, current_year - 1) if not record: normalized_name = normalize_team_name(request.team) if normalized_name and normalized_name != team_name.lower(): record = get_team_records(CFB_API_KEY, normalized_name, current_year - 1) if record: team_name = normalized_name # Get rankings (AP Poll) - try both names rankings = get_team_rankings( CFB_API_KEY, team=team_name, year=current_year, poll="ap" ) if not rankings: normalized_name = normalize_team_name(request.team) if normalized_name and normalized_name != team_name.lower(): rankings = get_team_rankings( CFB_API_KEY, team=normalized_name, year=current_year, poll="ap" ) # Format the response formatted_response = format_team_info(team_name, record, rankings) return JSONResponse(content=formatted_response) except HTTPException: raise except Exception as e: logger.error(f"Error in get_team_info: {e}") import traceback logger.error(traceback.format_exc()) raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" ) @app.get("/api/get_team_info") async def get_team_info_get( team: str = Query(..., description="Team name (required)") ): """ GET endpoint for get_team_info (for easier testing). Same functionality as POST endpoint but accessible via GET with query parameters. """ request = TeamInfoRequest(team=team) return await get_team_info(request) @app.post("/mcp/get_next_game_odds") async def get_next_game_odds(request: NextGameOddsRequest): """ Get next scheduled game and betting odds for a team. This endpoint finds a team's next upcoming game from The Odds API and returns the betting odds for that matchup. Args: request: NextGameOddsRequest with team name (required) Returns: JSON response with next game details and betting odds """ if not ODDS_API_KEY: raise HTTPException( status_code=503, detail="ODDS_API_KEY is not configured" ) try: # Fetch upcoming games from The Odds API games = get_ncaaf_odds(ODDS_API_KEY) if not games: raise HTTPException( status_code=404, detail="No upcoming games found or API request failed" ) # Find the next game for the team next_game = find_next_game_for_team(games, request.team) if not next_game: raise HTTPException( status_code=404, detail=f"No upcoming game found for team: {request.team}" ) # Format the response formatted_response = format_next_game_odds(request.team, next_game) return JSONResponse(content=formatted_response) except HTTPException: raise except Exception as e: logger.error(f"Error in get_next_game_odds: {e}") import traceback logger.error(traceback.format_exc()) raise HTTPException( status_code=500, detail=f"Internal server error: {str(e)}" ) @app.get("/api/get_next_game_odds") async def get_next_game_odds_get( team: str = Query(..., description="Team name (required)") ): """ GET endpoint for get_next_game_odds (for easier testing). Same functionality as POST endpoint but accessible via GET with query parameters. """ request = NextGameOddsRequest(team=team) return await get_next_game_odds(request) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)