analyze_driver_performance

Instructions

Implementation Reference

• src/f1_mcp_server/f1_data.py:296-391 (handler)

  Core handler function that loads F1 session data via fastf1 library, retrieves driver laps, calculates fastest lap, average lap time, and compiles detailed lap information for analysis.

  def analyze_driver_performance(
      year: Any, event_identifier: str, session_name: str, driver_identifier: str
  ) -> dict[str, Any]:
      """
      Analyze a driver's performance in a Formula One session.
  
      Args:
          year (int or str): The year of the F1 season
          event_identifier (str): Event name or round number
          session_name (str): Session type (Race, Qualifying, Sprint, etc.)
          driver_identifier (str): Driver number, code, or name
  
      Returns:
          dict: Status and performance analysis or error information
      """
      try:
          # Validate year
          year_int = validate_year(year)
  
          logger.debug(
              f"Analyzing driver performance for {year_int}, "
              f"event: {event_identifier}, session: {session_name}, "
              f"driver: {driver_identifier}"
          )
          session = fastf1.get_session(year_int, event_identifier, session_name)
          session.load()
  
          # Get laps for the specified driver
          driver_laps = session.laps.pick_driver(driver_identifier)
  
          if len(driver_laps) == 0:
              return {
                  "status": "error",
                  "message": f"No laps found for driver {driver_identifier}",
              }
  
          # Basic statistics
          fastest_lap = driver_laps.pick_fastest()
  
          # Calculate average lap time (excluding outliers)
          valid_lap_times = []
          for _, lap in driver_laps.iterrows():
              if lap["LapTime"] is not None and not pd.isna(lap["LapTime"]):
                  valid_lap_times.append(lap["LapTime"].total_seconds())
  
          avg_lap_time = (
              sum(valid_lap_times) / len(valid_lap_times) if valid_lap_times else None
          )
  
          # Format lap time as minutes:seconds.milliseconds
          formatted_fastest = (
              str(fastest_lap["LapTime"])
              if fastest_lap is not None and not pd.isna(fastest_lap["LapTime"])
              else None
          )
  
          # Get all lap times - limit to avoid excessive data
          max_laps = min(len(driver_laps), 100)  # Safety limit
          lap_times = []
  
          for _, lap in driver_laps.iloc[:max_laps].iterrows():
              lap_dict = {
                  "LapNumber": int(lap["LapNumber"])
                  if not pd.isna(lap["LapNumber"])
                  else None,
                  "LapTime": str(lap["LapTime"]) if not pd.isna(lap["LapTime"]) else None,
                  "Compound": lap["Compound"] if not pd.isna(lap["Compound"]) else None,
                  "TyreLife": int(lap["TyreLife"])
                  if not pd.isna(lap["TyreLife"])
                  else None,
                  "Stint": int(lap["Stint"]) if not pd.isna(lap["Stint"]) else None,
                  "FreshTyre": bool(lap["FreshTyre"])
                  if not pd.isna(lap["FreshTyre"])
                  else None,
                  "LapStartTime": json_serial(lap["LapStartTime"])
                  if not pd.isna(lap["LapStartTime"])
                  else None,
              }
              lap_times.append(lap_dict)
  
          # Format results
          result = {
              "DriverCode": fastest_lap["Driver"]
              if fastest_lap is not None and not pd.isna(fastest_lap["Driver"])
              else None,
              "TotalLaps": len(driver_laps),
              "FastestLap": formatted_fastest,
              "AverageLapTime": avg_lap_time,
              "LapTimes": lap_times,
          }
  
          logger.info(f"Successfully analyzed performance for driver {driver_identifier}")
          return {"status": "success", "data": result}
      except Exception as e:
          return {"status": "error", "message": str(e)}

• src/f1_mcp_server/server.py:386-424 (registration)

  MCP tool registration in list_tools(), defining the tool name, description, and JSON schema for input parameters including year, event, session, and driver.

  types.Tool(
      name="analyze_driver_performance",
      description=("Analyze a driver's performance in a Formula One session"),
      inputSchema={
          "type": "object",
          "properties": {
              "year": {
                  "type": "number",
                  "description": "Season year (e.g., 2023)",
              },
              "event_identifier": {
                  "type": "string",
                  "description": (
                      "Event name or round number (e.g., 'Monaco' or '7')"
                  ),
              },
              "session_name": {
                  "type": "string",
                  "description": (
                      "Session name (e.g., 'Race', 'Qualifying', "
                      "'Sprint', 'FP1', 'FP2', 'FP3')"
                  ),
              },
              "driver_identifier": {
                  "type": "string",
                  "description": (
                      "Driver identifier (number, code, or name; "
                      "e.g., '44', 'HAM', 'Hamilton')"
                  ),
              },
          },
          "required": [
              "year",
              "event_identifier",
              "session_name",
              "driver_identifier",
          ],
      },
  ),

• src/f1_mcp_server/server.py:389-424 (schema)

  Input schema definition for the analyze_driver_performance tool, specifying required parameters and their types/descriptions.

      inputSchema={
          "type": "object",
          "properties": {
              "year": {
                  "type": "number",
                  "description": "Season year (e.g., 2023)",
              },
              "event_identifier": {
                  "type": "string",
                  "description": (
                      "Event name or round number (e.g., 'Monaco' or '7')"
                  ),
              },
              "session_name": {
                  "type": "string",
                  "description": (
                      "Session name (e.g., 'Race', 'Qualifying', "
                      "'Sprint', 'FP1', 'FP2', 'FP3')"
                  ),
              },
              "driver_identifier": {
                  "type": "string",
                  "description": (
                      "Driver identifier (number, code, or name; "
                      "e.g., '44', 'HAM', 'Hamilton')"
                  ),
              },
          },
          "required": [
              "year",
              "event_identifier",
              "session_name",
              "driver_identifier",
          ],
      },
  ),

• src/f1_mcp_server/server.py:210-216 (handler)

  Dispatch handler in the MCP server's call_tool function that invokes the analyze_driver_performance implementation after input sanitization.

  elif name == "analyze_driver_performance":
      result = analyze_driver_performance(
          sanitized_args["year"],
          str(arguments["event_identifier"]),
          str(arguments["session_name"]),
          str(arguments["driver_identifier"]),
      )

• f1-messenger-app/mcp-bridge.py:108-108 (registration)

  Tool mapping registration in the HTTP bridge for direct function invocation.

  'analyze_driver_performance': analyze_driver_performance,