"""
MCP Client - Wrapper for calling MCP server functions
This module provides a client interface to call the MCP server's
5 core functions for college football data.
"""
import httpx
import logging
from typing import Optional, Dict, Any
logger = logging.getLogger(__name__)
class MCPClient:
"""Client for interacting with the MCP server"""
def __init__(self, base_url: str = "http://localhost:8000"):
"""
Initialize MCP client.
Args:
base_url: Base URL of the MCP server
"""
self.base_url = base_url.rstrip('/')
self.client = httpx.AsyncClient(timeout=30.0)
async def close(self):
"""Close the HTTP client"""
await self.client.aclose()
async def get_game_odds_and_score(
self,
team1: str,
team2: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""
Get game odds and score for a matchup.
Args:
team1: First team name (required)
team2: Second team name (optional)
Returns:
Game odds and score data or None if error
"""
try:
url = f"{self.base_url}/api/get_game_odds_and_score"
params = {"team1": team1}
if team2:
params["team2"] = team2
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error calling get_game_odds_and_score: {e}")
return None
async def get_recent_player_stats(
self,
player_name: str,
team: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""
Get recent player statistics (last 5 games).
Args:
player_name: Player name (required)
team: Team name (optional, for disambiguation)
Returns:
Player stats data or None if error
"""
try:
url = f"{self.base_url}/api/get_recent_player_stats"
params = {"player_name": player_name}
if team:
params["team"] = team
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error calling get_recent_player_stats: {e}")
return None
async def get_team_recent_results(
self,
team: str
) -> Optional[Dict[str, Any]]:
"""
Get team's recent game results (last 5 games).
Args:
team: Team name (required)
Returns:
Team results data or None if error
"""
try:
url = f"{self.base_url}/api/get_team_recent_results"
params = {"team": team}
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
error_detail = "Team not found"
try:
error_data = e.response.json()
error_detail = error_data.get("detail", error_detail)
except:
pass
logger.warning(f"Team not found: {team} - {error_detail}")
return {"error": f"Team '{team}' not found. Try using just the school name (e.g., 'Alabama' instead of 'Alabama Crimson Tide')."}
logger.error(f"HTTP error calling get_team_recent_results: {e.response.status_code} - {e.response.text}")
return {"error": f"API error: {e.response.status_code}"}
except Exception as e:
logger.error(f"Error calling get_team_recent_results: {e}")
return {"error": f"Request failed: {str(e)}"}
async def get_team_info(
self,
team: str
) -> Optional[Dict[str, Any]]:
"""
Get team's current season overview.
Args:
team: Team name (required)
Returns:
Team info data or None if error
"""
try:
url = f"{self.base_url}/api/get_team_info"
params = {"team": team}
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error calling get_team_info: {e}")
return None
async def get_next_game_odds(
self,
team: str
) -> Optional[Dict[str, Any]]:
"""
Get next game and odds for a team.
Args:
team: Team name (required)
Returns:
Next game odds data or None if error
"""
try:
url = f"{self.base_url}/api/get_next_game_odds"
params = {"team": team}
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Error calling get_next_game_odds: {e}")
return None