|
| 1 | +from datetime import datetime, timedelta |
| 2 | +import json |
| 3 | +import time |
| 4 | +from uuid import UUID |
| 5 | +import dotenv |
| 6 | +from nba_api.live.nba.endpoints import scoreboard |
| 7 | +from nba_api.stats.endpoints import scheduleleaguev2, leaguestandingsv3 |
| 8 | +from strands import tool |
| 9 | +from strands.models.bedrock import BedrockModel |
| 10 | +from strands.session.file_session_manager import FileSessionManager |
| 11 | +import boto3 |
| 12 | +from strands import Agent |
| 13 | +import pandas as pd |
| 14 | + |
| 15 | +dotenv.load_dotenv() |
| 16 | + |
| 17 | +@tool |
| 18 | +def get_nba_live_scores(name:str): |
| 19 | + """Get today's NBA live scores for a given team. |
| 20 | + This tool doesn't provide past scores. |
| 21 | + Args: |
| 22 | + name (str): team name or city of the NBA team. |
| 23 | + """ |
| 24 | + # Today's Score Board |
| 25 | + games = scoreboard.ScoreBoard() |
| 26 | + games_json = json.loads(games.get_json()) |
| 27 | + name = name.lower() |
| 28 | + scores = [] |
| 29 | + for game in games_json['scoreboard']['games']: |
| 30 | + home_team = game['homeTeam']['teamName'].lower() |
| 31 | + away_team = game['awayTeam']['teamName'].lower() |
| 32 | + if name in [home_team, away_team]: |
| 33 | + scores.append({ |
| 34 | + 'home_team': game['homeTeam']['teamName'], |
| 35 | + 'home_score': game['homeTeam']['score'], |
| 36 | + 'away_team': game['awayTeam']['teamName'], |
| 37 | + 'away_score': game['awayTeam']['score'], |
| 38 | + 'quarter': game['period'], |
| 39 | + 'time_remaining': game['gameClock'] |
| 40 | + }) |
| 41 | + break |
| 42 | + if not scores: |
| 43 | + return f"No games found for team: {name}" |
| 44 | + return scores |
| 45 | + |
| 46 | +@tool |
| 47 | +def get_nba_past_scores(name:str, game_date:str): |
| 48 | + """Get past NBA scores for a given team for a specific date (MM/DD/YYYY). |
| 49 | + Use 'today' or 'yesterday' for convenience. |
| 50 | + Args: |
| 51 | + name (str): team name or city of the NBA team. |
| 52 | + """ |
| 53 | + if game_date.lower() == "today": |
| 54 | + date = datetime.now().strftime("%m/%d/%Y") |
| 55 | + elif game_date.lower() in ["yesterday", "yday", "prev day", "y'day"]: |
| 56 | + date = (datetime.now() - timedelta(days=1)).strftime("%m/%d/%Y") |
| 57 | + else: |
| 58 | + try: |
| 59 | + date_obj = datetime.strptime(game_date, "%m/%d/%Y") |
| 60 | + date = date_obj.strftime("%m/%d/%Y") |
| 61 | + except ValueError: |
| 62 | + return "Please provide the date in 'MM/DD/YYYY' format or use 'today'/'yesterday'." |
| 63 | + schedule = scheduleleaguev2.ScheduleLeagueV2() |
| 64 | + json_schedule = json.loads(schedule.get_json()) |
| 65 | + games = json_schedule['leagueSchedule']['gameDates'] |
| 66 | + past_scores = [] |
| 67 | + for day_games in games: |
| 68 | + if day_games['gameDate'].startswith(date): |
| 69 | + for game in day_games['games']: |
| 70 | + if name.lower() in [ |
| 71 | + game['homeTeam']['teamName'].lower(), |
| 72 | + game['homeTeam']['teamCity'].lower(), |
| 73 | + game['awayTeam']['teamName'].lower(), |
| 74 | + game['awayTeam']['teamCity'].lower() |
| 75 | + ]: |
| 76 | + past_scores.append({ |
| 77 | + 'home_team': game['homeTeam']['teamName'], |
| 78 | + 'home_score': game['homeTeam']['score'], |
| 79 | + 'away_team': game['awayTeam']['teamName'], |
| 80 | + 'away_score': game['awayTeam']['score'] |
| 81 | + }) |
| 82 | + break |
| 83 | + if not past_scores: |
| 84 | + if game_date.lower() == "today": |
| 85 | + return f"Also check live scores tool. No past games found for team: {name} today." |
| 86 | + return f"No past games found for team: {name} on date: {date}" |
| 87 | + return past_scores |
| 88 | + |
| 89 | +@tool |
| 90 | +def get_team_standings(name:str): |
| 91 | + """Get standings/ranking for a given team. |
| 92 | + |
| 93 | + Args: |
| 94 | + name (str): team name or city of the NBA team. |
| 95 | + """ |
| 96 | + standings = leaguestandingsv3.LeagueStandingsV3() |
| 97 | + json_standings = json.loads(standings.get_json()) |
| 98 | + df = pd.DataFrame(json_standings['resultSets'][0]['rowSet'], |
| 99 | + columns=json_standings['resultSets'][0]['headers']) |
| 100 | + team_status = None |
| 101 | + for index, row in df.iterrows(): |
| 102 | + if row['TeamName'].lower() == name.lower() or row['TeamCity'].lower() == name.lower(): |
| 103 | + team_status = { |
| 104 | + 'Rank': row['PlayoffRank'], |
| 105 | + 'Record': row['Record'], |
| 106 | + 'Conference': row['Conference'] |
| 107 | + } |
| 108 | + break |
| 109 | + if team_status is None: |
| 110 | + return f"No standings found for team: {name}" |
| 111 | + return team_status |
| 112 | + |
| 113 | +def setup_agents() -> Agent: |
| 114 | + boto_session = boto3.Session() |
| 115 | + claud_on_bedrock = BedrockModel(boto_session=boto_session, streaming=False) |
| 116 | + # Create a session manager with a unique session ID |
| 117 | + session_manager = FileSessionManager(session_id=str(UUID(int=time.time_ns()))) |
| 118 | + |
| 119 | + nba_score_agent = \ |
| 120 | + Agent(name="Nba Score Agent", model=claud_on_bedrock, |
| 121 | + system_prompt= """You are an agent who provides NBA scores |
| 122 | + and standings for the given team. Look at the tool output and |
| 123 | + anwer the user query only from that output. |
| 124 | + Be very concise and to the point. |
| 125 | + If anything other than NBA scores or standings is asked, |
| 126 | + respond with 'I can only provide NBA scores and standings.'. |
| 127 | + When providing live scores use following format: |
| 128 | + 'Team A is playing against Team B with current score X-Y.' |
| 129 | + If no games found, just say 'I didn't find any game for the team' |
| 130 | + When providing response to game status use following format: |
| 131 | + 'Team A won/lost against Team B with score X-Y.' |
| 132 | + If no games found, just say 'I didn't find any game for the team' |
| 133 | + When providing team status or standings use following format: |
| 134 | + 'Team A is currently ranked X in the CONFERENCE with a record of W-L.' |
| 135 | + """, |
| 136 | + tools = [get_nba_live_scores, get_nba_past_scores, get_team_standings], |
| 137 | + description="NBA scores agent", callback_handler=None, |
| 138 | + session_manager=session_manager |
| 139 | + ) |
| 140 | + return nba_score_agent |
| 141 | + |
| 142 | +def get_scores(message: str): |
| 143 | + travel_agent = setup_agents() |
| 144 | + response = travel_agent(message) |
| 145 | + return response.message['content'][0]['text'] |
| 146 | + |
| 147 | +if __name__ == "__main__": |
| 148 | + nba_agent = setup_agents() |
| 149 | + while True: |
| 150 | + try: |
| 151 | + user_request = input("\nHey, got a question on NBA scores or standings? ") |
| 152 | + except EOFError: |
| 153 | + user_request = "exit" |
| 154 | + if user_request.lower() in ["exit", "quit", ""]: |
| 155 | + print("Exiting the NBA scores agent. Goodbye!") |
| 156 | + break |
| 157 | + response = nba_agent(prompt=user_request) |
| 158 | + print(response.message['content'][0]['text']) |
0 commit comments