Pitstop

Instructions

Implementation Reference

• tools/weather/session_weather.py:10-51 (handler)

  The core handler function that retrieves historical weather data for a specific F1 session using FastF1 client, processes the pandas DataFrame into individual WeatherDataPoint models, and constructs the SessionWeatherDataResponse.

  def get_session_weather(year: int, gp: Union[str, int], session: str) -> SessionWeatherDataResponse:
      """
      Get time-series weather data - temp, humidity, pressure, wind, rainfall.
  
      Args:
          year: Season year (2018+)
          gp: Grand Prix name or round
          session: 'FP1', 'FP2', 'FP3', 'Q', 'S', 'R'
  
      Returns:
          SessionWeatherDataResponse with weather points
  
      Example:
          get_session_weather(2024, "Spa", "R") → Weather throughout race
      """
      session_obj = fastf1_client.get_session(year, gp, session)
      session_obj.load(laps=False, telemetry=False, weather=True, messages=False)
  
      event = session_obj.event
      weather_df = session_obj.weather_data
  
      # Convert to Pydantic models
      weather_points = []
      for idx, row in weather_df.iterrows():
          point = WeatherDataPoint(
              time=str(row['Time']) if pd.notna(row.get('Time')) else None,
              air_temp=float(row['AirTemp']) if pd.notna(row.get('AirTemp')) else None,
              track_temp=float(row['TrackTemp']) if pd.notna(row.get('TrackTemp')) else None,
              humidity=float(row['Humidity']) if pd.notna(row.get('Humidity')) else None,
              pressure=float(row['Pressure']) if pd.notna(row.get('Pressure')) else None,
              wind_speed=float(row['WindSpeed']) if pd.notna(row.get('WindSpeed')) else None,
              wind_direction=float(row['WindDirection']) if pd.notna(row.get('WindDirection')) else None,
              rainfall=bool(row['Rainfall']) if pd.notna(row.get('Rainfall')) else None,
          )
          weather_points.append(point)
  
      return SessionWeatherDataResponse(
          session_name=session_obj.name,
          event_name=event['EventName'],
          weather_data=weather_points,
          total_points=len(weather_points)
      )

• models/weather/weather.py:5-25 (schema)

  Pydantic models defining the input/output structure: WeatherDataPoint for individual readings and SessionWeatherDataResponse for the full session weather response.

  class WeatherDataPoint(BaseModel):
      """Single weather data point."""
  
      time: Optional[str] = Field(None, description="Timestamp")
      air_temp: Optional[float] = Field(None, description="Air temperature (°C)")
      track_temp: Optional[float] = Field(None, description="Track surface temperature (°C)")
      humidity: Optional[float] = Field(None, description="Relative humidity (%)")
      pressure: Optional[float] = Field(None, description="Atmospheric pressure (mbar)")
      wind_speed: Optional[float] = Field(None, description="Wind speed (m/s)")
      wind_direction: Optional[float] = Field(None, description="Wind direction (degrees)")
      rainfall: Optional[bool] = Field(None, description="Whether it's raining")
  
  
  class SessionWeatherDataResponse(BaseModel):
      """Session weather data response."""
  
      session_name: str = Field(description="Session name")
      event_name: str = Field(description="Grand Prix name")
      weather_data: list[WeatherDataPoint] = Field(description="Weather data points throughout session")
      total_points: int = Field(description="Total number of weather data points")

• server.py:161-161 (registration)

  Registers the get_session_weather function as an MCP tool in the main server.

  mcp.tool()(get_session_weather)

• tools/__init__.py:19-19 (registration)

  Package-level import of the tool for exposure through tools namespace.

  from .weather import get_session_weather

• tools/weather/__init__.py:1-3 (registration)

  Subpackage export of the tool function.

  from .session_weather import get_session_weather
  
  __all__ = ["get_session_weather"]